"""Transports: the stdio runner and the streamable HTTP app with its runner.

Startup is fail-closed. A non-loopback bind without HTTP authentication, a
half-configured OAuth setup or a missing Gitea token end in exit code 2 with an
actionable message on stderr (never on stdout, which carries JSON-RPC on stdio).
"""

from __future__ import annotations

import asyncio
import errno
import ipaddress
import logging
import sys
import time
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlsplit

import anyio
import httpx
import uvicorn
from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions
from mcp.server.stdio import stdio_server
from mcp.server.transport_security import TransportSecuritySettings
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.routing import Route
from starlette.types import ASGIApp, Receive, Scope, Send
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

from aegis_gitea_mcp import __version__
from aegis_gitea_mcp.audit import get_audit_logger
from aegis_gitea_mcp.auth import (
    MCP_SCOPE,
    CompositeVerifier,
    IdentityResolver,
    StaticBearerVerifier,
)
from aegis_gitea_mcp.config import Settings, is_loopback_host
from aegis_gitea_mcp.gitea import GiteaExecutor
from aegis_gitea_mcp.oauth import OAuthSetupError
from aegis_gitea_mcp.rate_limit import RequestRateLimiter
from aegis_gitea_mcp.server import (
    SERVER_NAME,
    AegisServer,
    ServerConfigError,
    build_server,
    max_request_body_size,
)

logger = logging.getLogger(__name__)

HEALTH_CACHE_SECONDS = 10.0
_PROBE_TIMEOUT_SECONDS = 5.0
_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::", "[::]"})  # noqa: S104 (compared, never bound)
EXIT_USAGE = 2


def _fail(message: str) -> int:
    """Print an actionable startup error to stderr and return the usage exit code."""
    print(f"aegis-gitea-mcp: {message}", file=sys.stderr)
    return EXIT_USAGE


@dataclass(frozen=True)
class HttpAuthExtras:
    """Optional per-user OAuth pieces plugged into the HTTP app (admin-token mode needs none).

    Attributes:
        token_verifier: Verifier accepting OAuth-issued tokens (combined with the
            static bearer verifier when ``MCP_BEARER_TOKEN`` is set).
        auth_server_provider: SDK authorization server provider (mounts the OAuth routes).
        resolver: Identity resolver for OAuth tokens.
        extra_routes: Extra unauthenticated routes (for example ``/oauth/callback``).
        client_registration_options: Dynamic client registration settings.
        revocation_options: Token revocation settings.
        executor: Gitea executor shared by the server and the resolver.
        rate_limited_paths: Unauthenticated paths that get the per-IP rate limit.
    """

    token_verifier: TokenVerifier | None = None
    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None
    resolver: IdentityResolver | None = None
    extra_routes: tuple[Route, ...] = ()
    client_registration_options: ClientRegistrationOptions | None = None
    revocation_options: RevocationOptions | None = None
    executor: GiteaExecutor | None = None
    rate_limited_paths: tuple[str, ...] = ()


def load_oauth_components(settings: Settings) -> HttpAuthExtras:
    """Build the per-user OAuth components for ``OAUTH_ENABLED``.

    Raises:
        OAuthSetupError: Incomplete configuration or an unusable token store.
    """
    from aegis_gitea_mcp.oauth.components import build_oauth_components

    return build_oauth_components(settings)


# Health probe


class GiteaProbe:
    """Cached reachability check of the Gitea API (``GET /api/v1/version``).

    Never exposes URLs, tokens or exception text: callers only learn up or down.
    """

    def __init__(self, settings: Settings, client: httpx.AsyncClient | None = None) -> None:
        self._url = f"{settings.gitea_api_url}/version"
        token = settings.gitea_token
        self._headers = {"Authorization": f"token {token.get_secret_value()}"} if token else {}
        self._client = client or httpx.AsyncClient(
            timeout=httpx.Timeout(
                min(_PROBE_TIMEOUT_SECONDS, float(settings.request_timeout_seconds))
            ),
            follow_redirects=False,
        )
        self._owns_client = client is None
        self._lock = asyncio.Lock()
        self._checked_at = 0.0
        self._reachable = False

    async def reachable(self) -> bool:
        """Return whether Gitea answered successfully within the last 10 seconds."""
        async with self._lock:
            now = time.monotonic()
            if self._checked_at and now - self._checked_at < HEALTH_CACHE_SECONDS:
                return self._reachable
            try:
                response = await self._client.get(self._url, headers=self._headers)
                self._reachable = response.is_success
            except Exception:
                self._reachable = False
            self._checked_at = time.monotonic()
            return self._reachable

    async def aclose(self) -> None:
        """Close the HTTP client when this probe created it."""
        if self._owns_client:
            await self._client.aclose()


# Rate limiting


class RateLimitMiddleware:
    """ASGI middleware applying the per-IP and per-token limits to the MCP path.

    ``extra_paths`` (for example the unauthenticated OAuth endpoints) share the limits.
    """

    def __init__(
        self,
        app: ASGIApp,
        mcp_path: str,
        limiter: RequestRateLimiter,
        extra_paths: Sequence[str] = (),
    ) -> None:
        self.app = app
        self._paths = frozenset({mcp_path, *(path.rstrip("/") for path in extra_paths)})
        self._limiter = limiter

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http" or scope["path"].rstrip("/") not in self._paths:
            await self.app(scope, receive, send)
            return
        client = scope.get("client")
        client_ip = client[0] if client else "unknown"
        authorization: str | None = None
        for name, value in scope.get("headers", []):
            if name == b"authorization":
                authorization = value.decode("latin-1")
                break
        decision = self._limiter.check(client_ip, authorization)
        if decision.allowed:
            await self.app(scope, receive, send)
            return
        response = JSONResponse(
            {"error": {"kind": "rate_limited", "message": "Too many requests"}},
            status_code=429,
            headers={"Retry-After": "60"},
        )
        await response(scope, receive, send)


# Forwarded headers


class ForwardedHeadersMiddleware:
    """Applies ``X-Forwarded-For`` and ``-Proto`` from trusted peers and warns about the rest.

    Replaces uvicorn's own proxy-header handling (the runner passes
    ``proxy_headers=False``) so the original peer address is still known here:
    a request that carries ``X-Forwarded-For`` from a peer outside
    ``FORWARDED_ALLOW_IPS`` is not honoured (the peer's own address is used) and
    logs one warning, because that usually means the reverse proxy is not listed
    (for example a Docker network address) and every client then shares the
    proxy's rate-limit bucket.

    Args:
        app: The ASGI application.
        trusted: ``FORWARDED_ALLOW_IPS`` (``*`` or a comma list of IPs and networks).
    """

    def __init__(self, app: ASGIApp, trusted: str) -> None:
        self.trusted = trusted
        self._all = trusted.strip() == "*"
        self._networks = [
            ipaddress.ip_network(item.strip(), strict=False)
            for item in trusted.split(",")
            if item.strip() and item.strip() != "*"
        ]
        self._app = ProxyHeadersMiddleware(app, trusted_hosts=trusted)  # type: ignore[arg-type]
        self._warned = False

    def _peer_trusted(self, host: str) -> bool:
        if self._all:
            return True
        try:
            address = ipaddress.ip_address(host.split("%")[0])
        except ValueError:
            return False
        return any(
            address in network for network in self._networks if network.version == address.version
        )

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "http" and not self._warned:
            client = scope.get("client")
            has_forwarded = any(name == b"x-forwarded-for" for name, _ in scope.get("headers", []))
            if has_forwarded and client and not self._peer_trusted(client[0]):
                self._warned = True
                logger.warning(
                    "a request carried X-Forwarded-For from a peer that is not in "
                    "FORWARDED_ALLOW_IPS, so the header is ignored and the peer address is used "
                    "as the client address; if this peer is your reverse proxy, add its address "
                    "to FORWARDED_ALLOW_IPS (all clients otherwise share one rate-limit bucket)"
                )
        await self._app(scope, receive, send)  # type: ignore[arg-type]


# HTTP app


def _base_url(settings: Settings) -> str:
    return settings.public_base_url or f"http://127.0.0.1:{settings.mcp_port}"


def transport_security(settings: Settings) -> TransportSecuritySettings:
    """DNS-rebinding protection: allow loopback, the public host and ``ALLOWED_HOSTS``."""
    port = settings.mcp_port
    hosts = [f"127.0.0.1:{port}", f"localhost:{port}", f"[::1]:{port}"]
    origins = [f"http://127.0.0.1:{port}", f"http://localhost:{port}", f"http://[::1]:{port}"]
    if settings.mcp_host.strip("[]") not in _WILDCARD_HOSTS and not settings.is_loopback_bind():
        hosts.append(f"{settings.mcp_host}:{port}")
    if settings.public_base_url:
        parts = urlsplit(settings.public_base_url)
        hosts.extend([parts.netloc, parts.hostname or parts.netloc])
        origins.append(f"{parts.scheme}://{parts.netloc}")
    for host in settings.allowed_hosts:
        hosts.append(host)
        origins.extend([f"https://{host}", f"http://{host}"])
    return TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=list(dict.fromkeys(hosts)),
        allowed_origins=list(dict.fromkeys(origins)),
    )


def _resource_url(settings: Settings) -> str:
    return f"{_base_url(settings)}{settings.mcp_path}"


def _auth_settings(
    settings: Settings,
    client_registration_options: ClientRegistrationOptions | None,
    revocation_options: RevocationOptions | None,
    *,
    authorization_server: bool,
) -> AuthSettings:
    issuer = _base_url(settings)
    return AuthSettings(
        issuer_url=issuer,  # type: ignore[arg-type]
        resource_server_url=_resource_url(settings),  # type: ignore[arg-type]
        # With our own authorization server every token (OAuth-issued and the
        # static bearer) names this endpoint as its RFC 8707 resource and carries
        # the one scope we grant, so the SDK enforces both. Without it, extra
        # verifiers are trusted to check the audience themselves.
        required_scopes=[MCP_SCOPE] if authorization_server else None,
        validate_token_resource=authorization_server,
        client_registration_options=client_registration_options,
        revocation_options=revocation_options,
    )


def build_http_app(
    settings: Settings,
    *,
    server: AegisServer | None = None,
    token_verifier: TokenVerifier | None = None,
    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
    extra_routes: Sequence[Route] = (),
    client_registration_options: ClientRegistrationOptions | None = None,
    revocation_options: RevocationOptions | None = None,
    probe: GiteaProbe | None = None,
    rate_limited_paths: Sequence[str] = (),
) -> Starlette:
    """Build the streamable HTTP ASGI app (MCP endpoint, ``/health``, ``/``).

    Authentication: ``MCP_BEARER_TOKEN`` (when set) and ``token_verifier`` (when
    passed) are combined; any accepted token reaches the MCP endpoint. With
    neither, the endpoint is unauthenticated, which is only acceptable on a
    loopback bind (a warning is logged).

    Args:
        settings: Application settings.
        server: Server to serve (defaults to ``build_server(settings)``).
        token_verifier: Extra verifier, for example the per-user OAuth verifier.
        auth_server_provider: SDK authorization server provider (needs a verifier).
        extra_routes: Additional routes served without MCP authentication.
        client_registration_options: Dynamic client registration settings.
        revocation_options: Token revocation settings.
        probe: Gitea reachability probe for ``/health`` (defaults to a new one).
        rate_limited_paths: Further paths (besides the MCP endpoint) that get the
            rate limit, for example the OAuth endpoints.

    Raises:
        ValueError: ``auth_server_provider`` given without any token verifier.
    """
    mcp_server = server or build_server(settings)
    gitea_probe = probe or GiteaProbe(settings)

    verifiers: list[TokenVerifier] = []
    if settings.mcp_bearer_token is not None:
        verifiers.append(
            StaticBearerVerifier(
                settings.mcp_bearer_token.get_secret_value(), resource=_resource_url(settings)
            )
        )
    if token_verifier is not None:
        verifiers.append(token_verifier)
    if auth_server_provider is not None and not verifiers:
        raise ValueError("auth_server_provider needs a token verifier to protect the MCP endpoint")
    verifier: TokenVerifier | None = None
    if verifiers:
        verifier = verifiers[0] if len(verifiers) == 1 else CompositeVerifier(*verifiers)
    else:
        logger.warning(
            "HTTP endpoint runs WITHOUT authentication (loopback development only); "
            "set MCP_BEARER_TOKEN before exposing it"
        )

    async def health(request: Request) -> Response:
        if await gitea_probe.reachable():
            return JSONResponse(
                {
                    "status": "ok",
                    "version": __version__,
                    "tools": mcp_server.tool_count,
                    "gitea": "reachable",
                }
            )
        return JSONResponse({"status": "degraded", "gitea": "unreachable"}, status_code=503)

    async def index(request: Request) -> Response:
        return JSONResponse(
            {"name": SERVER_NAME, "version": __version__, "mcp_endpoint": settings.mcp_path}
        )

    routes = [
        Route("/health", health, methods=["GET"]),
        Route("/", index, methods=["GET"]),
        *extra_routes,
    ]
    app = mcp_server.streamable_http_app(
        streamable_http_path=settings.mcp_path,
        stateless_http=True,
        json_response=False,
        max_request_body_size=max_request_body_size(settings),
        transport_security=transport_security(settings),
        host=settings.mcp_host,
        auth=(
            _auth_settings(
                settings,
                client_registration_options,
                revocation_options,
                authorization_server=auth_server_provider is not None,
            )
            if verifier is not None
            else None
        ),
        token_verifier=verifier,
        auth_server_provider=auth_server_provider,
        custom_starlette_routes=routes,
    )

    sdk_lifespan = app.router.lifespan_context

    @asynccontextmanager
    async def lifespan(application: Starlette) -> AsyncIterator[None]:
        try:
            async with sdk_lifespan(application):
                yield
        finally:
            await gitea_probe.aclose()

    app.router.lifespan_context = lifespan
    app.add_middleware(
        RateLimitMiddleware,
        mcp_path=settings.mcp_path,
        limiter=RequestRateLimiter(settings),
        extra_paths=tuple(rate_limited_paths),
    )
    return app


# Runners


def http_startup_errors(settings: Settings, extras: HttpAuthExtras | None) -> list[str]:
    """List reasons the HTTP server must not start (empty when startup may proceed).

    An OAuth-only deployment (OAuth on, no ``MCP_BEARER_TOKEN``, no
    ``GITEA_TOKEN``) may start: every user brings their own Gitea token and
    admin-token mode is simply unavailable. ``GITEA_TOKEN`` stays required
    whenever admin-token mode is reachable (a static bearer, or no HTTP auth).
    """
    errors: list[str] = []
    bearer = settings.mcp_bearer_token is not None
    if settings.oauth_enabled:
        errors.extend(settings.oauth_config_errors())
        if not settings.oauth_config_errors() and extras is None:
            errors.append("OAUTH_ENABLED=true but per-user OAuth could not be set up")
        public = urlsplit(settings.public_base_url or "")
        if public.scheme == "http" and not is_loopback_host(public.hostname or ""):
            errors.append("PUBLIC_BASE_URL must use https unless the host is loopback")
    authenticated = bearer or extras is not None
    if not settings.is_loopback_bind() and not authenticated:
        errors.append(
            f"refusing to bind to {settings.mcp_host} without HTTP authentication: set "
            "MCP_BEARER_TOKEN (or enable OAuth), or bind to 127.0.0.1"
        )
    if (bearer or extras is None) and settings.gitea_token is None:
        message = "GITEA_TOKEN is required for admin-token mode (static bearer or no auth)"
        if settings.oauth_enabled and bearer:
            message += "; for an OAuth-only deployment unset MCP_BEARER_TOKEN"
        errors.append(message)
    return errors


def http_startup_warnings(settings: Settings, extras: HttpAuthExtras | None) -> list[str]:
    """List non-fatal startup problems of the HTTP server (printed to stderr)."""
    warnings: list[str] = []
    if (
        not settings.is_loopback_bind()
        and not settings.public_base_url
        and not settings.allowed_hosts
    ):
        warnings.append(
            f"MCP_HOST={settings.mcp_host} is not loopback and neither PUBLIC_BASE_URL nor "
            "ALLOWED_HOSTS is set: requests whose Host header is not a loopback name or the "
            "bind address are rejected with 421. Set PUBLIC_BASE_URL (the public https URL) "
            "or ALLOWED_HOSTS (comma separated host names clients use)"
        )
    if settings.forwarded_allow_ips == "*":
        warnings.append(
            "FORWARDED_ALLOW_IPS=*: X-Forwarded-For is trusted from any peer, so clients can "
            "spoof their IP and dodge the per-IP rate limit. List the proxy addresses instead"
        )
    return warnings


def _audit_startup_error(settings: Settings) -> str | None:
    """Open the audit log now; return an error message when it is not writable.

    A server that cannot record its calls must not start: the dispatcher refuses
    calls without an audit log, so every call would fail anyway.
    """
    try:
        get_audit_logger()
    except OSError as exc:
        reason = exc.strerror or str(exc) or type(exc).__name__
        if exc.errno == errno.ELOOP:
            reason = "the path is a symbolic link, which is refused"
        return (
            f"cannot open the audit log {settings.resolved_audit_log_path} for append "
            f"({reason}). Fix the permissions or point AUDIT_LOG_PATH (or DATA_DIR) at a "
            "writable location"
        )
    return None


class AccessLogQueryFilter(logging.Filter):
    """Drops query strings from uvicorn access log lines.

    OAuth redirects carry one-time codes and states in the query string
    (``/oauth/callback?code=...&state=...``); they must not reach the logs.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        """Rewrite the path argument of an access log record; never drops the record."""
        args = record.args
        if isinstance(args, tuple) and len(args) >= 3 and isinstance(args[2], str):
            path, separator, _ = args[2].partition("?")
            if separator:
                record.args = (*args[:2], f"{path}?[redacted]", *args[3:])
        return True


def run_http(settings: Settings) -> int:
    """Run the streamable HTTP server; return a process exit code.

    Returns 2 without starting when a startup check fails (including an audit
    log that cannot be opened for append).
    """
    extras: HttpAuthExtras | None = None
    if settings.oauth_enabled and not settings.oauth_config_errors():
        try:
            extras = load_oauth_components(settings)
        except OAuthSetupError as exc:
            return _fail(f"per-user OAuth cannot start: {exc}")
    errors = http_startup_errors(settings, extras)
    if errors:
        for message in errors:
            _fail(message)
        return EXIT_USAGE
    audit_error = _audit_startup_error(settings)
    if audit_error is not None:
        return _fail(audit_error)
    for message in http_startup_warnings(settings, extras):
        print(f"aegis-gitea-mcp: warning: {message}", file=sys.stderr)
    try:
        server = build_server(
            settings,
            resolver=extras.resolver if extras else None,
            executor=extras.executor if extras else None,
        )
    except ServerConfigError as exc:
        return _fail(str(exc))
    app = build_http_app(
        settings,
        server=server,
        token_verifier=extras.token_verifier if extras else None,
        auth_server_provider=extras.auth_server_provider if extras else None,
        extra_routes=extras.extra_routes if extras else (),
        client_registration_options=extras.client_registration_options if extras else None,
        revocation_options=extras.revocation_options if extras else None,
        rate_limited_paths=extras.rate_limited_paths if extras else (),
    )
    access_logger = logging.getLogger("uvicorn.access")
    if not any(isinstance(item, AccessLogQueryFilter) for item in access_logger.filters):
        access_logger.addFilter(AccessLogQueryFilter())
    uvicorn.run(
        ForwardedHeadersMiddleware(app, settings.forwarded_allow_ips),
        host=settings.mcp_host,
        port=settings.mcp_port,
        log_level=settings.log_level.lower(),
        proxy_headers=False,  # ForwardedHeadersMiddleware applies them (and warns)
        server_header=False,
    )
    return 0


async def _serve_stdio(server: AegisServer) -> None:
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


def run_stdio(settings: Settings) -> int:
    """Run the local stdio server (single identity: the owner of ``GITEA_TOKEN``).

    Returns 2 without starting when ``GITEA_TOKEN`` is missing, the audit log is
    not writable or the tool filters are invalid. All logging goes to stderr; stdout carries JSON-RPC only.
    """
    if settings.gitea_token is None:
        return _fail("stdio needs GITEA_TOKEN; create a Gitea access token and set it in .env")
    audit_error = _audit_startup_error(settings)
    if audit_error is not None:
        return _fail(audit_error)
    try:
        server = build_server(settings)
    except ServerConfigError as exc:
        return _fail(str(exc))
    try:
        anyio.run(_serve_stdio, server)
    except KeyboardInterrupt:  # pragma: no cover - interactive shutdown
        return 0
    return 0