"""Generic outbound-resilience guards: a registry of per-resource circuit breakers + adapters.

This module is service-agnostic mechanism and imports nothing service-specific, so it is the part
that lifts cleanly into a shared library. A consumer builds a `GuardRegistry` and registers each
downstream it calls -- `(name, policy, adapter, classifier)` -- then calls through it. The HTTP
adapters here are generic (httpx / requests / no-per-call-timeout shapes), each normalizing its
client's failures to `DownstreamServerError` (5xx) or `DownstreamTransportError` (timeout/connect)
so one classifier works regardless of client. A consumer supplies any service-specific adapters or
classifiers (see this service's catalog in `core/hardening/downstream.py`).

A `policy` is any object exposing `connect_timeout`, `read_timeout`, `fail_max`, `reset_timeout`,
`success_threshold`, and `retries` (e.g. `DownstreamPolicy`).
"""

import logging
import random
import time
from dataclasses import dataclass
from typing import Any, Callable, Hashable, Iterable, Protocol, TypeVar

from core.hardening.breaker import CircuitBreaker

_log = logging.getLogger(__name__)

T = TypeVar('T')


class Policy(Protocol):
    """Structural type of a downstream policy (e.g. DownstreamPolicy); read-only by design."""

    @property
    def connect_timeout(self) -> float:
        """TCP connect timeout in seconds."""

    @property
    def read_timeout(self) -> float:
        """Read/response timeout in seconds."""

    @property
    def fail_max(self) -> int:
        """Consecutive failures that trip the breaker."""

    @property
    def reset_timeout(self) -> float:
        """Seconds the breaker stays open before admitting a probe."""

    @property
    def success_threshold(self) -> int:
        """Consecutive half-open successes needed to re-close."""

    @property
    def retries(self) -> int:
        """Bonus attempts for idempotent calls (0 = no retry)."""


Adapter = Callable[[Hashable, Policy, Callable[..., Any]], Any]
Classifier = Callable[[BaseException], bool]


class DownstreamServerError(Exception):
    """A downstream returned a server-side (5xx) failure that should count against the breaker."""

    def __init__(self, resource: Hashable, detail: object = None) -> None:
        """Record the resource and an optional detail (e.g. the HTTP status)."""
        super().__init__(f'{resource} server error')
        self.resource = resource
        self.detail = detail


class DownstreamTransportError(Exception):
    """A downstream call failed at the transport layer (timeout / connection error)."""

    def __init__(self, resource: Hashable, detail: object = None) -> None:
        """Record the resource and an optional detail (e.g. the client exception name)."""
        super().__init__(f'{resource} transport error')
        self.resource = resource
        self.detail = detail


def _http_status(resp: object) -> int | None:
    code: int | None = getattr(resp, 'status_code', None)
    if code is None:
        code = getattr(resp, 'status', None)
    return code


def _raise_on_5xx(resource: Hashable, resp: T) -> T:
    status = _http_status(resp)
    if status is not None and status >= 500:
        raise DownstreamServerError(resource, status)
    return resp


# --- generic HTTP adapters: (resource, policy, call) -> result. Each injects the right timeout
# shape and normalizes its client's transport failures to DownstreamTransportError + 5xx to
# DownstreamServerError, so the breaker actually trips on slow/unreachable downstreams. ---


def requests_timeout_adapter(
    resource: Hashable, policy: Policy, call: Callable[..., Any]
) -> Any:
    """Drive a requests-style client: inject a (connect, read) timeout tuple; normalize failures."""
    import requests

    try:
        resp = call(timeout=(policy.connect_timeout, policy.read_timeout))
    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
        raise DownstreamTransportError(resource, type(e).__name__) from e
    return _raise_on_5xx(resource, resp)


def httpx_timeout_adapter(
    resource: Hashable, policy: Policy, call: Callable[..., Any]
) -> Any:
    """Drive an httpx-based client: inject an httpx.Timeout; normalize failures."""
    import httpx

    try:
        resp = call(
            timeout=httpx.Timeout(
                connect=policy.connect_timeout,
                read=policy.read_timeout,
                write=policy.read_timeout,
                pool=policy.read_timeout,
            )
        )
    except httpx.TransportError as e:
        raise DownstreamTransportError(resource, type(e).__name__) from e
    return _raise_on_5xx(resource, resp)


def no_timeout_adapter(
    resource: Hashable, policy: Policy, call: Callable[..., Any]
) -> Any:
    """Drive a client with no per-call timeout (relies on the breaker; e.g. owsrequest behind harakiri); normalize 5xx."""
    # KNOWN GAP: with no per-call timeout, the breaker cannot trip on a merely-SLOW collaborator --
    # only on a 5xx or a harakiri-kill. A slow-but-200 downstream won't count. Revisit if the
    # collaborator client gains per-call timeouts (it would then look like the requests adapter).
    resp = call()
    return _raise_on_5xx(resource, resp)


def is_downstream_failure(e: BaseException) -> bool:
    """Default classifier: count server (5xx) + transport (timeout/connect) errors; ignore the rest."""
    return isinstance(e, (DownstreamServerError, DownstreamTransportError))


def _retry(
    fn: Callable[[], T],
    attempts: int,
    classifier: Classifier,
    sleep: Callable[[float], object],
    base_delay: float,
    rng: Callable[[], float],
) -> T:
    last: BaseException | None = None
    for i in range(attempts):
        try:
            return fn()
        except Exception as e:
            try:
                retryable = classifier(e)
            except Exception:
                # A buggy classifier must neither mask the real error nor trigger a retry.
                _log.exception('retry classifier raised; treating as non-retryable')
                retryable = False
            if not retryable:
                raise  # propagate the ORIGINAL error (e.g. a 4xx-ish bug or bad SQL)
            last = e
            if i + 1 < attempts:
                # Full jitter (AWS): random in [0, base_delay * 2**i) avoids lockstep re-hits.
                sleep(rng() * base_delay * (2**i))
    if (
        last is None
    ):  # attempts < 1; never happens via call() but keeps the type/contract honest
        raise RuntimeError('retry called with attempts < 1')
    raise last


@dataclass(frozen=True)
class ResourceGuard:
    """The policy, breaker, adapter and classifier that together guard one downstream."""

    name: Hashable
    policy: Policy
    breaker: CircuitBreaker
    adapter: Adapter
    classifier: Classifier


class GuardRegistry:
    """A per-consumer registry of resource guards: register downstreams, then call through them.

    The library owns this mechanism; each service constructs a registry and registers its own
    downstreams, so nothing here is tied to a fixed resource catalog.
    """

    def __init__(
        self,
        on_state_change: Callable[[str, str, str], None] | None = None,
        sleep: Callable[[float], object] = time.sleep,
        base_delay: float = 0.1,
        rng: Callable[[], float] = random.random,
    ) -> None:
        """Configure the registry; on_state_change is forwarded to every breaker (fired off-lock)."""
        self._guards: dict[Hashable, ResourceGuard] = {}
        self._on_state_change: Callable[[str, str, str], None] = on_state_change or (
            lambda *a: None
        )
        self._sleep = sleep
        self._base_delay = base_delay
        self._rng = rng

    def register(
        self, name: Hashable, policy: Policy, adapter: Adapter, classifier: Classifier
    ) -> 'GuardRegistry':
        """Register a downstream and build its CircuitBreaker (re-registering a name replaces it)."""
        breaker = CircuitBreaker(
            name=str(name),
            fail_max=policy.fail_max,
            reset_timeout=policy.reset_timeout,
            success_threshold=policy.success_threshold,
            count_failure=classifier,
            on_state_change=self._on_state_change,
        )
        self._guards[name] = ResourceGuard(name, policy, breaker, adapter, classifier)
        return self

    def guard(self, name: Hashable) -> ResourceGuard:
        """Return the ResourceGuard for name, or raise KeyError if it was never registered."""
        if name not in self._guards:
            raise KeyError(f'no downstream registered for {name!r}')
        return self._guards[name]

    def guards(self) -> Iterable[ResourceGuard]:
        """Return all registered guards (used by tests to reset breaker state between runs)."""
        return self._guards.values()

    def call(
        self, name: Hashable, call: Callable[..., Any], *, idempotent: bool = False
    ) -> Any:
        """Run an outbound `call` to `name` under its breaker + (idempotent-only) bounded retry."""
        g = self.guard(name)
        attempts = (g.policy.retries + 1) if idempotent else 1
        return g.breaker.call(
            lambda: _retry(
                lambda: g.adapter(g.name, g.policy, call),
                attempts,
                g.classifier,
                self._sleep,
                self._base_delay,
                self._rng,
            )
        )