"""Shared HTTP infrastructure for notification provider clients.

Slack/Discord/ntfy/Matrix/Webhook all follow the same pattern: build a
JSON payload, POST/PUT it, decode 200-range as success, decode 4xx/5xx
into a stable error dict, and retry transient 429/503 responses with a
capped ``Retry-After``. ``HttpProviderClient`` centralizes that pattern
so every provider gets the same SSRF guard, timeouts, secret-redacted
errors, and bounded retry policy by construction — adding a new
provider doesn't get to forget any one of them.
"""

from __future__ import annotations

import asyncio
import logging
from typing import Any, Final, Mapping

import aiohttp

from .redact import redact, redact_exc
from .ssrf import UnsafeURLError, avalidate_outbound_url

_LOGGER = logging.getLogger(__name__)

_DEFAULT_TIMEOUT: Final = aiohttp.ClientTimeout(total=30, connect=10)
_MAX_RETRIES: Final = 3
_MAX_RETRY_AFTER_S: Final = 60.0
_RETRY_STATUSES: Final = frozenset({429, 503})

# Hop-by-hop / framing headers a caller must not be able to override via
# user-supplied target config. Letting them through enables request
# smuggling, host-header bypasses of WAFs, and cache poisoning.
_FORBIDDEN_HEADERS: Final = frozenset({
    "host",
    "content-length",
    "transfer-encoding",
    "connection",
    "keep-alive",
    "te",
    "upgrade",
    "expect",
    "proxy-authorization",
    "proxy-connection",
})


def safe_headers(headers: Mapping[str, str] | None) -> dict[str, str]:
    """Return a copy of ``headers`` with hop-by-hop/forbidden names dropped
    and CRLF-bearing values rejected.

    A target config that lets a user inject ``"X-Foo": "bar\\r\\nHost: evil"``
    can perform request smuggling depending on aiohttp's framing. We strip
    those values rather than letting them reach the wire.
    """
    if not headers:
        return {}
    safe: dict[str, str] = {}
    for raw_name, raw_value in headers.items():
        name = str(raw_name).strip()
        if not name or name.lower() in _FORBIDDEN_HEADERS:
            continue
        if any(c in name for c in "\r\n:"):
            continue
        value = str(raw_value)
        if "\r" in value or "\n" in value:
            continue
        safe[name] = value
    return safe


def make_error(message: str, *, status: int | None = None, body: str | None = None) -> dict[str, Any]:
    """Build a stable failure dict shape used by every provider client."""
    err: dict[str, Any] = {"success": False, "error": redact(message)}
    if status is not None:
        err["status_code"] = status
    if body:
        err["body"] = redact(body)[:200]
    return err


def make_success(**extra: Any) -> dict[str, Any]:
    """Build a stable success dict shape used by every provider client."""
    out: dict[str, Any] = {"success": True}
    out.update(extra)
    return out


def _retry_after_seconds(headers: Mapping[str, str], cap_s: float) -> float:
    raw = headers.get("Retry-After") or headers.get("retry-after") or "2"
    try:
        seconds = float(raw)
    except (TypeError, ValueError):
        seconds = 2.0
    return max(0.0, min(seconds, cap_s))


class HttpProviderClient:
    """Base for JSON-over-HTTP notification providers.

    Subclasses call :meth:`request` instead of using ``self._session``
    directly.  ``request`` runs the SSRF guard (skippable for known-safe
    upstreams via ``ssrf_validate=False``), enforces a per-request
    timeout, retries 429/503 with a capped ``Retry-After``, and turns
    transport/HTTP errors into the canonical ``{"success": False, ...}``
    shape with secrets redacted.
    """

    _max_retries: int = _MAX_RETRIES
    # Settable per-instance so tests / hostile-upstream tuning can
    # tighten the cap. Reads of this attribute fall through to the
    # class default when no instance value has been set.
    _MAX_RETRY_AFTER: float = _MAX_RETRY_AFTER_S

    def __init__(
        self,
        session: aiohttp.ClientSession,
        *,
        timeout: aiohttp.ClientTimeout | None = None,
        provider_name: str = "http",
    ) -> None:
        self._session = session
        self._timeout = timeout or _DEFAULT_TIMEOUT
        self._provider = provider_name

    async def request(
        self,
        method: str,
        url: str,
        *,
        json: Any = None,
        headers: Mapping[str, str] | None = None,
        ssrf_validate: bool = True,
        retry_statuses: frozenset[int] = _RETRY_STATUSES,
    ) -> dict[str, Any]:
        """Send a single request with retry + redaction. Always returns a dict.

        On 2xx returns ``{"success": True, "status_code": int, "json": ...
        OR "body": str}``. On non-2xx returns the canonical error dict.
        """
        if ssrf_validate:
            try:
                await avalidate_outbound_url(url)
            except UnsafeURLError as err:
                return make_error(f"Unsafe URL: {redact_exc(err)}")

        outbound_headers: dict[str, str] = {"Content-Type": "application/json"}
        outbound_headers.update(safe_headers(headers))

        for attempt in range(1, self._max_retries + 1):
            try:
                async with self._session.request(
                    method,
                    url,
                    json=json,
                    headers=outbound_headers,
                    timeout=self._timeout,
                    allow_redirects=False,
                ) as resp:
                    if resp.status in retry_statuses and attempt < self._max_retries:
                        delay = _retry_after_seconds(resp.headers, self._MAX_RETRY_AFTER)
                        _LOGGER.warning(
                            "%s %s %s: HTTP %d, retrying after %.2fs (attempt %d/%d)",
                            self._provider, method, redact(url), resp.status,
                            delay, attempt, self._max_retries,
                        )
                        await resp.read()  # drain body so connection can return to pool
                        await asyncio.sleep(delay)
                        continue
                    if 200 <= resp.status < 300:
                        try:
                            payload: Any = await resp.json(content_type=None)
                        except (aiohttp.ContentTypeError, ValueError):
                            payload = await resp.text()
                        return make_success(status_code=resp.status, json=payload)
                    body = await resp.text()
                    return make_error(
                        f"HTTP {resp.status}",
                        status=resp.status,
                        body=body,
                    )
            except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
                # asyncio.CancelledError inherits from BaseException on
                # 3.8+, so it is not caught here — good: cancellation
                # must propagate.
                if attempt < self._max_retries and isinstance(err, asyncio.TimeoutError):
                    _LOGGER.warning(
                        "%s %s %s: timeout, retrying (attempt %d/%d)",
                        self._provider, method, redact(url),
                        attempt, self._max_retries,
                    )
                    await asyncio.sleep(min(2 ** (attempt - 1), 5))
                    continue
                return make_error(redact_exc(err))

        # Retry budget exhausted on a retriable status.
        return make_error("Rate limited (retries exhausted)")
