from __future__ import annotations

import re
import threading
import uuid
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from typing import Any
from urllib.parse import parse_qsl, unquote, urlsplit

from backend.config import get_embed_link_settings
from backend.models.events import EventKind
from backend.utils.embed_url_utils import (
    normalize_embed_url,
    normalize_origin,
    normalize_path_prefix,
)
from backend.utils.logging_utils import get_logger

logger = get_logger(__name__)

URL_PATTERN = r"https?://[^\s<>\"]+"
TRAILING_PUNCT = ".,;:"
TRAILING_DELIMS = {
    ")": "(",
    "]": "[",
    "}": "{",
}


def _trim_trailing_punct(cand: str) -> str:
    trimmed = cand.rstrip(TRAILING_PUNCT)
    while trimmed and trimmed[-1] in TRAILING_DELIMS:
        trailing_delim = trimmed[-1]
        opening_delim = TRAILING_DELIMS[trailing_delim]
        decoded_trimmed = unquote(trimmed)
        if decoded_trimmed.count(trailing_delim) > decoded_trimmed.count(opening_delim):
            trimmed = trimmed[:-1]
            continue
        break
    return trimmed


def _match_path_pat(path: str, pat: object) -> tuple[int, int] | None:
    if not isinstance(pat, str):
        return None
    norm_prefix = normalize_path_prefix(pat)
    if not norm_prefix:
        return None
    if norm_prefix == "/":
        return (1, 0)

    is_subtree = pat.endswith("/*")
    if path == norm_prefix:
        return (len(norm_prefix), 0 if is_subtree else 1)
    if is_subtree and path.startswith(f"{norm_prefix}/"):
        return (len(norm_prefix), 0)
    return None


def _norm_preview_label(label: str | None) -> str | None:
    if not isinstance(label, str):
        return None
    trimmed = label.strip()
    return trimmed or None


def _rule_for_path(path: str, rules: list[dict]) -> dict | None:
    best: tuple[tuple[int, int], dict] | None = None
    for rule in rules:
        if not isinstance(rule, dict):
            continue
        match = _match_path_pat(path, rule.get("pathPrefix"))
        if match is None:
            continue
        if best is None or match > best[0]:
            best = (match, rule)
    return best[1] if best is not None else None


def _is_query_allowed(url: str, rule: dict) -> bool:
    query = urlsplit(url).query
    policy = rule.get("queryPolicy", "none")
    if policy == "any":
        return True
    if not query:
        return True
    if policy != "allow-list":
        return False
    if ";" in query:
        return False

    allowed_params = {param for param in (rule.get("allowedQueryParameters") or []) if isinstance(param, str)}
    parameters = parse_qsl(query, keep_blank_values=True)
    return bool(parameters) and all(name in allowed_params for name, _ in parameters)


def _blocked_res(
    *,
    url: str,
    reason: str,
    origin_url: str | None = None,
    path: str | None = None,
) -> dict[str, Any]:
    result: dict[str, Any] = {
        "originalUrl": url,
        "resolvedUrl": None,
        "embedAllowed": False,
        "reason": reason,
    }
    if origin_url is not None:
        result["origin"] = origin_url
    if path is not None:
        result["path"] = path
    return result


def resolve_streamed_link_candidate(
    url: str,
    *,
    allowed_origins_enabled: bool,
    allowed_origins: list[dict],
) -> dict[str, Any]:
    normalized = normalize_embed_url(url)
    if normalized is None:
        return _blocked_res(url=url, reason="invalid_url")

    normalized_url, origin, path = normalized
    origin_url = origin.url

    def blocked(reason: str) -> dict[str, Any]:
        return _blocked_res(url=url, reason=reason, origin_url=origin_url, path=path)

    if not allowed_origins_enabled:
        return blocked("allowed_origins_disabled")

    entry = next(
        (
            entry
            for entry in allowed_origins or []
            if isinstance(entry, dict) and normalize_origin(entry.get("origin")) == origin
        ),
        None,
    )
    if entry is None:
        return blocked("origin_not_allowed")

    excluded_pfx = entry.get("excludedPathPrefixes") or []
    if any(_match_path_pat(path, prefix) is not None for prefix in excluded_pfx):
        return blocked("excluded_path")

    rules = entry.get("rules")
    rules = rules if isinstance(rules, list) else []
    match_rule = _rule_for_path(path, rules)
    if match_rule is None:
        return blocked("path_not_allowed")

    if not _is_query_allowed(normalized_url, match_rule):
        return blocked("query_not_allowed")

    return {
        "originalUrl": url,
        "resolvedUrl": normalized_url,
        "embedAllowed": True,
        "previewLabel": _norm_preview_label(match_rule.get("previewLabel")),
        "showPreviewUrl": bool(match_rule.get("showPreviewUrl", True)),
        "origin": origin_url,
        "path": path,
        "reason": "allowed_origin",
    }


def resolve_links_in_text(
    text: str,
    *,
    allowed_origins_enabled: bool,
    allowed_origins: list[dict],
) -> list[dict[str, Any]]:
    if not text:
        return []

    links: list[dict[str, Any]] = []
    for index, match in enumerate(re.finditer(URL_PATTERN, text)):
        cand = _trim_trailing_punct(match.group(0))
        if not cand:
            continue
        links.append(
            {
                "matchId": f"resolved-link-{index}",
                **resolve_streamed_link_candidate(
                    cand,
                    allowed_origins_enabled=allowed_origins_enabled,
                    allowed_origins=allowed_origins,
                ),
            }
        )
    return links


def serialize_messages_with_enriched_links(messages: list) -> list[dict]:
    settings = get_embed_link_settings()
    serialized: list[dict] = []

    for msg in messages or []:
        msg_data = msg if isinstance(msg, dict) else msg.model_dump(mode="json", by_alias=True)
        if msg_data.get("role") == "assistant" and msg_data.get("content"):
            msg_data = {
                **msg_data,
                "resolvedLinks": resolve_links_in_text(
                    msg_data.get("content", ""),
                    allowed_origins_enabled=settings["allowed_origins_enabled"],
                    allowed_origins=settings["allowed_origins"],
                ),
            }
        serialized.append(msg_data)

    return serialized


class StreamLinkEnrichmentService:
    def __init__(
        self,
        *,
        emit_event: Callable[[dict[str, Any], bool], None],
        cancel_event,
        allowed_origins_enabled: bool,
        allowed_origins: list[dict],
        dispatch_resolution: Callable[[str, str], None] | None = None,
    ) -> None:
        self._emit_event = emit_event
        self._cancel_event = cancel_event
        self._allowed_origins_enabled = allowed_origins_enabled
        self._allowed_origins = allowed_origins or []
        self._dispatch_resolution = dispatch_resolution or self._dispatch_resolution_async
        self._buffer = ""
        self._seen_occurrences: set[tuple[int, int]] = set()
        self._lock = threading.Lock()
        self._executor: ThreadPoolExecutor | None = None
        if dispatch_resolution is None:
            self._executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="stream-link-enrichment")

    def consume_text(self, text: str) -> None:
        if not text:
            return
        cands = []
        with self._lock:
            self._buffer += text
            cands = self._collect_new_cands(final=False)
        for match_id, url in cands:
            self._dispatch_resolution(match_id, url)

    def flush(self) -> None:
        cands = []
        with self._lock:
            cands = self._collect_new_cands(final=True)
        for match_id, url in cands:
            self._dispatch_resolution(match_id, url)

    def close(self) -> None:
        if self._executor is not None:
            self._executor.shutdown(wait=True, cancel_futures=False)

    def _collect_new_cands(self, *, final: bool) -> list[tuple[str, str]]:
        cands: list[tuple[str, str]] = []
        for match in re.finditer(URL_PATTERN, self._buffer):
            if match.end() == len(self._buffer) and not final:
                continue

            occ = (match.start(), match.end())
            cand = _trim_trailing_punct(match.group(0))
            if not cand or occ in self._seen_occurrences:
                continue

            self._seen_occurrences.add(occ)
            cands.append((str(uuid.uuid4()), cand))
        return cands

    def _dispatch_resolution_async(self, match_id: str, url: str) -> None:
        if self._executor is None:
            self._resolve_and_emit(match_id, url)
            return
        self._executor.submit(self._resolve_and_emit, match_id, url)

    def _resolve_and_emit(self, match_id: str, url: str) -> None:
        if self._cancel_event.is_set():
            return

        try:
            resolved = resolve_streamed_link_candidate(
                url,
                allowed_origins_enabled=self._allowed_origins_enabled,
                allowed_origins=self._allowed_origins,
            )
            resolved_with_match = {
                "matchId": match_id,
                **resolved,
            }
            event_payload = {
                "eventKind": EventKind.LINK_RESOLVED,
                "eventData": resolved_with_match,
            }
            self._emit_event(event_payload, False)
        except Exception:
            logger.exception("Failed to resolve streamed link candidate: %s", url)
