"""
Shared async HTTP core: one pooled client, a TTL cache, per-host circuit
breakers and a deadline helper. Everything network-bound in DeepSearch
goes through here, which is what keeps the fan-out cheap.
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
import random
import time
from collections import OrderedDict, defaultdict
from typing import Any, Awaitable, Dict, Iterable, List, Optional, Tuple, TypeVar

import httpx

from .config import settings

log = logging.getLogger("deepsearch.core")
T = TypeVar("T")


# --------------------------------------------------------------------------
# TTL + LRU cache
# --------------------------------------------------------------------------
class TTLCache:
    __slots__ = ("_d", "_max", "hits", "misses", "_lock")

    def __init__(self, max_items: int = 4096) -> None:
        self._d: "OrderedDict[str, Tuple[float, Any]]" = OrderedDict()
        self._max = max_items
        self.hits = 0
        self.misses = 0
        self._lock = asyncio.Lock()

    @staticmethod
    def key(*parts: Any) -> str:
        raw = "||".join(str(p) for p in parts)
        return hashlib.sha1(raw.encode("utf-8", "ignore")).hexdigest()

    def get(self, key: str) -> Optional[Any]:
        item = self._d.get(key)
        if not item:
            self.misses += 1
            return None
        expiry, value = item
        if expiry < time.time():
            self._d.pop(key, None)
            self.misses += 1
            return None
        self._d.move_to_end(key)
        self.hits += 1
        return value

    def set(self, key: str, value: Any, ttl: int) -> None:
        if not settings.cache_enabled or ttl <= 0:
            return
        self._d[key] = (time.time() + ttl, value)
        self._d.move_to_end(key)
        while len(self._d) > self._max:
            self._d.popitem(last=False)

    def stats(self) -> Dict[str, Any]:
        total = self.hits + self.misses
        return {
            "items": len(self._d),
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 3) if total else 0.0,
        }

    def clear(self) -> None:
        self._d.clear()


serp_cache = TTLCache(settings.cache_max_items)
page_cache = TTLCache(settings.cache_max_items)
answer_cache = TTLCache(512)


# --------------------------------------------------------------------------
# Circuit breaker - stop hammering providers that are rate-limiting us
# --------------------------------------------------------------------------
class CircuitBreaker:
    def __init__(self, threshold: int = 3, cooldown: float = 120.0) -> None:
        self.fails: Dict[str, int] = defaultdict(int)
        self.open_until: Dict[str, float] = {}
        self.threshold = threshold
        self.cooldown = cooldown

    def is_open(self, name: str) -> bool:
        until = self.open_until.get(name, 0.0)
        if until and until > time.time():
            return True
        if until:
            self.open_until.pop(name, None)
            self.fails[name] = 0
        return False

    def record_success(self, name: str) -> None:
        self.fails[name] = 0

    def record_failure(self, name: str) -> None:
        self.fails[name] += 1
        if self.fails[name] >= self.threshold:
            # exponential-ish backoff, capped
            mult = min(4, self.fails[name] - self.threshold + 1)
            self.open_until[name] = time.time() + self.cooldown * mult
            log.warning("circuit open for %s (%.0fs)", name, self.cooldown * mult)

    def snapshot(self) -> Dict[str, Any]:
        now = time.time()
        return {
            k: round(v - now, 1)
            for k, v in self.open_until.items()
            if v > now
        }


breaker = CircuitBreaker()


# --------------------------------------------------------------------------
# Per-host pacing
# --------------------------------------------------------------------------
class HostThrottle:
    """
    Serialise + space out requests to the same host.

    Fanning 3 sub-queries x 15 providers out at once means several
    simultaneous hits per host, which is exactly what makes free engines
    return 403/429. A small enforced gap per host keeps them friendly and
    dramatically raises the number of providers that stay alive.
    """

    def __init__(self, min_interval: float = 0.35, max_parallel: int = 2) -> None:
        self.min_interval = min_interval
        self.max_parallel = max_parallel
        self._last: Dict[str, float] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
        self._sems: Dict[str, asyncio.Semaphore] = {}

    def _lock(self, host: str) -> asyncio.Lock:
        if host not in self._locks:
            self._locks[host] = asyncio.Lock()
        return self._locks[host]

    def _sem(self, host: str) -> asyncio.Semaphore:
        if host not in self._sems:
            self._sems[host] = asyncio.Semaphore(self.max_parallel)
        return self._sems[host]

    async def acquire(self, host: str) -> None:
        await self._sem(host).acquire()
        async with self._lock(host):
            last = self._last.get(host, 0.0)
            wait = self.min_interval - (time.monotonic() - last)
            if wait > 0:
                await asyncio.sleep(wait + random.uniform(0, 0.12))
            self._last[host] = time.monotonic()

    def release(self, host: str) -> None:
        try:
            self._sem(host).release()
        except ValueError:
            pass


throttle = HostThrottle()


# --------------------------------------------------------------------------
# HTTP client singleton
# --------------------------------------------------------------------------
_client: Optional[httpx.AsyncClient] = None
_client_lock = asyncio.Lock()

_ACCEPT_LANG = ["en-US,en;q=0.9", "en-GB,en;q=0.9", "en;q=0.8,en-US;q=0.7"]


def _supported_encodings() -> str:
    """
    Only advertise codecs we can actually decode. Claiming `br` without the
    brotli package installed makes upstreams return bodies we cannot read -
    a silent, total failure mode for every HTML provider.
    """
    encs = ["gzip", "deflate"]
    try:
        import brotli  # noqa: F401
        encs.append("br")
    except ImportError:
        try:
            import brotlicffi  # noqa: F401
            encs.append("br")
        except ImportError:
            pass
    try:
        import zstandard  # noqa: F401
        encs.append("zstd")
    except ImportError:
        pass
    return ", ".join(encs)


ACCEPT_ENCODING = _supported_encodings()


def base_headers(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
    h = {
        "User-Agent": settings.user_agent,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": random.choice(_ACCEPT_LANG),
        "Accept-Encoding": ACCEPT_ENCODING,
        "Cache-Control": "no-cache",
        "Upgrade-Insecure-Requests": "1",
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "none",
    }
    if extra:
        h.update(extra)
    return h


async def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None or _client.is_closed:
        async with _client_lock:
            if _client is None or _client.is_closed:
                _client = httpx.AsyncClient(
                    timeout=httpx.Timeout(
                        settings.http_timeout,
                        connect=5.0,
                        read=settings.http_timeout,
                        pool=5.0,
                    ),
                    limits=httpx.Limits(
                        max_connections=settings.max_connections,
                        max_keepalive_connections=settings.max_keepalive,
                        keepalive_expiry=30.0,
                    ),
                    follow_redirects=True,
                    headers=base_headers(),
                    http2=False,
                    verify=True,
                )
    return _client


async def close_client() -> None:
    global _client
    if _client and not _client.is_closed:
        await _client.aclose()
    _client = None


async def fetch_text(
    url: str,
    *,
    method: str = "GET",
    params: Optional[Dict[str, Any]] = None,
    data: Optional[Dict[str, Any]] = None,
    json_body: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: Optional[float] = None,
    max_bytes: Optional[int] = None,
    provider: str = "",
) -> Optional[str]:
    """Fetch a URL and return decoded text, or None on any failure."""
    if provider and breaker.is_open(provider):
        return None
    client = await get_client()
    limit = max_bytes or settings.max_page_bytes

    # Pace requests per host so free providers don't rate-limit us.
    host = ""
    if provider:
        try:
            from urllib.parse import urlparse as _up
            host = _up(url).netloc.lower()
        except Exception:
            host = ""
    if host:
        await throttle.acquire(host)
    try:
        req = client.build_request(
            method,
            url,
            params=params,
            data=data,
            json=json_body,
            headers=base_headers(headers),
            timeout=timeout or settings.http_timeout,
        )
        resp = await client.send(req, stream=True)
        try:
            if resp.status_code >= 400:
                if provider and resp.status_code in (403, 429, 503):
                    breaker.record_failure(provider)
                return None
            ctype = resp.headers.get("content-type", "").lower()
            if ctype and not any(
                t in ctype for t in ("text", "json", "xml", "html", "javascript")
            ):
                return None
            total = 0
            parts: List[bytes] = []
            async for chunk in resp.aiter_bytes(65536):
                parts.append(chunk)
                total += len(chunk)
                if total >= limit:
                    break
            raw = b"".join(parts)
        finally:
            await resp.aclose()

        if provider:
            breaker.record_success(provider)
        enc = resp.encoding or "utf-8"
        try:
            return raw.decode(enc, errors="replace")
        except (LookupError, TypeError):
            return raw.decode("utf-8", errors="replace")
    except (httpx.TimeoutException, httpx.ConnectError, httpx.ReadError):
        if provider:
            breaker.record_failure(provider)
        return None
    except Exception as e:  # noqa: BLE001
        log.debug("fetch failed %s: %s", url[:80], e)
        if provider:
            breaker.record_failure(provider)
        return None
    finally:
        if host:
            throttle.release(host)


async def fetch_json(url: str, **kw: Any) -> Optional[Any]:
    import orjson

    txt = await fetch_text(url, **kw)
    if not txt:
        return None
    try:
        return orjson.loads(txt)
    except Exception:
        try:
            import json

            return json.loads(txt)
        except Exception:
            return None


# --------------------------------------------------------------------------
# Concurrency helpers
# --------------------------------------------------------------------------
async def gather_capped(
    coros: Iterable[Awaitable[T]],
    limit: int,
    deadline: Optional[float] = None,
) -> List[T]:
    """
    Run awaitables with a concurrency cap and an optional absolute deadline
    (perf_counter seconds). Failures and timeouts yield no result rather than
    blowing up the whole batch - partial results always beat no results.
    """
    sem = asyncio.Semaphore(max(1, limit))
    out: List[T] = []

    async def run(c: Awaitable[T]) -> None:
        async with sem:
            if deadline and time.perf_counter() >= deadline:
                # Close the coroutine so Python doesn't warn about it.
                if hasattr(c, "close"):
                    c.close()  # type: ignore[attr-defined]
                return
            try:
                remaining = (deadline - time.perf_counter()) if deadline else None
                if remaining is not None:
                    res = await asyncio.wait_for(c, timeout=max(0.05, remaining))
                else:
                    res = await c
                out.append(res)
            except (asyncio.TimeoutError, asyncio.CancelledError):
                return
            except Exception as e:  # noqa: BLE001
                log.debug("task failed: %s", e)
                return

    tasks = [asyncio.create_task(run(c)) for c in coros]
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)
    return out


def deadline_in(seconds: float) -> float:
    return time.perf_counter() + seconds