"""FastAPI auth dependency for the External API surface.

``require_api_client(*scopes)`` is the single gate every external-API
route sits behind. It mirrors the defence stack proven by the MCP PAT
middleware and the directory-pull dependency, on the External API's own
``helix_api_`` credential store:

* uniform 401 for every credential failure (no oracle),
* per-token + per-IP failed-auth windows (anti-brute-force, pre-DB),
* per-client fixed-window rate limit on the success path,
* auth failures + scope denials audited in an independent session,
* last-used + daily-usage bookkeeping in its own short session so a
  read-only request still leaves an investigation trail.

All Redis windows fail OPEN (degrade to a process-local cap), never
taking the surface down on a Redis blip.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, datetime
from uuid import UUID, uuid4

from fastapi import Depends, Request
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.api_clients.models import ApiClientToken, ApiClientUsageDaily
from app.api_clients.tokens import API_TOKEN_PREFIX, ApiClientTokenService
from app.api_clients.visibility import cloud_visibility_is_malformed
from app.core import database as _db
from app.core.database import get_db
from app.core.exceptions import AppException, ForbiddenError, UnauthorizedError
from app.core.service_token import (
    fixed_window_incr,
    hash_token,
    peek_local_window,
    window_ttl,
)

logger = logging.getLogger(__name__)

_AUTH_FAILURE_ACTION = "external_api.auth_failure"
_SCOPE_DENIED_ACTION = "external_api.scope_denied"

#: Default per-client request cap per 60s (overridable per client in the
#: DB). Anti-brute-force caps on failed auth, per IP + per token.
_DEFAULT_RATE_LIMIT_PER_MINUTE = 300
_AUTH_FAILURE_CAP = 20
_AUTH_FAILURE_WINDOW_SECONDS = 60


@dataclass(slots=True)
class ApiClientContext:
    """Request-scoped snapshot of the authenticated API client."""

    client_id: UUID
    slug: str
    name: str
    scopes: frozenset[str]
    token_id: UUID | None = None
    token_prefix: str = ""
    #: Raw ``api_clients.cloud_visibility`` (see
    #: :mod:`app.api_clients.visibility`). Carried on the context so a
    #: cloud route enforces the whitelist without a second client read;
    #: ``None`` = unrestricted. Read together with
    #: ``cloud_visibility_malformed`` — ``None`` here is ambiguous on its
    #: own, since an unreadable stored value also coerces to ``None``.
    cloud_visibility: dict | None = None
    #: True when the stored ``cloud_visibility`` was present but NOT a
    #: JSON object. Such a value must fail **closed** — read as "no
    #: subscriptions" — never degrade to the ``None`` meaning of
    #: "unrestricted", which would hand a deliberately narrowed client the
    #: entire cloud inventory (and, with ``cloud.billing:read``, the whole
    #: spend picture). Same posture as the resolver's rule that an
    #: unmatched whitelist resolves to ``false()``, never to "no
    #: constraint". Routes get this for free through
    #: ``resolve_client_cloud_visibility``.
    cloud_visibility_malformed: bool = False
    #: Raw ``api_clients.person_visibility`` (see
    #: :mod:`app.api_clients.person_visibility`). The population ceiling
    #: for people routes, carried here so a route enforces it without a
    #: second client read; ``None`` = unrestricted.
    person_visibility: dict | None = None
    #: True when the stored ``person_visibility`` was present but NOT a
    #: JSON object (a hand-edited row, a restore from an older shape, a
    #: future writer that stored it serialized). Such a value must fail
    #: **closed** — read as "nobody" — never degrade to the ``None``
    #: meaning of "unrestricted", which would hand a narrowed client the
    #: whole roster. Same posture as ``resolve_in_scope_person_ids``,
    #: which returns the empty set on a filter it cannot parse.
    person_visibility_malformed: bool = False

    def has_scope(self, scope: str) -> bool:
        return scope in self.scopes


def _client_ip(request: Request) -> str:
    try:
        from app.core.dependencies import get_real_ip

        return get_real_ip(request)
    except Exception:  # noqa: BLE001
        return request.client.host if request.client else "unknown"


def _token_failure_key(token: str) -> str:
    return f"extapi_authfail:tok:{hash_token(token)[:32]}"


async def _peek_token_failures(token: str) -> tuple[bool, int]:
    """Read-only check of the per-token failed-auth window (pre-DB)."""
    from app.core.redis_client import get_redis

    key = _token_failure_key(token)
    try:
        raw = await get_redis().get(key)
        if raw is not None and int(raw) > _AUTH_FAILURE_CAP:
            return False, await window_ttl(key, _AUTH_FAILURE_WINDOW_SECONDS)
        return True, 0
    except Exception:  # noqa: BLE001 — consult process-local window
        local = await peek_local_window(key)
        if local is not None and local > _AUTH_FAILURE_CAP:
            return False, _AUTH_FAILURE_WINDOW_SECONDS
        return True, 0


async def _record_auth_failure(
    *, reason: str, request: Request, token: str | None,
) -> None:
    """Bump failure windows + audit. Never raises."""
    if token:
        await fixed_window_incr(_token_failure_key(token), _AUTH_FAILURE_WINDOW_SECONDS)
    client_ip = _client_ip(request)
    ip_count = await fixed_window_incr(
        f"extapi_authfail:{client_ip}", _AUTH_FAILURE_WINDOW_SECONDS,
    )
    if ip_count is not None and ip_count > _AUTH_FAILURE_CAP:
        # Over the per-IP cap: skip the audit write so a scanner can't
        # flood the table; the 401 still goes out.
        return
    try:
        from app.audit.service import log_denied_action

        await log_denied_action(
            trace_id=getattr(request.state, "trace_id", None) or str(uuid4()),
            actor_type="anonymous",
            action=_AUTH_FAILURE_ACTION,
            resource_type="api_client",
            resource_id="auth",
            detail={"reason": reason[:200], "path": request.url.path[:200]},
            ip_address=client_ip,
            user_agent=(request.headers.get("user-agent") or "")[:512] or None,
        )
    except Exception:  # noqa: BLE001 — audit must never block the 401
        logger.exception("external-api auth-failure audit write failed")


async def _record_usage(
    *, token_id: UUID, client_id: UUID, client_ip: str,
) -> None:
    """last_used_* on the token + daily counter upsert, own session.
    Best-effort: a DB blip here must not 500 a valid request."""
    now = datetime.now(UTC)
    try:
        async with _db.async_session_factory() as session:
            token = await session.get(ApiClientToken, token_id)
            if token is not None:
                token.last_used_at = now
                token.last_used_ip = client_ip[:64]
            stmt = mysql_insert(ApiClientUsageDaily).values(
                id=str(uuid4()),
                client_id=str(client_id),
                usage_date=now.date(),
                request_count=1,
                last_request_at=now,
            ).on_duplicate_key_update(
                request_count=ApiClientUsageDaily.request_count + 1,
                last_request_at=now,
            )
            await session.execute(stmt)
            await session.commit()
    except Exception as exc:  # noqa: BLE001
        logger.warning("external-api usage bookkeeping failed: %s", exc)


async def authorize_api_client_token(
    *,
    request: Request,
    db: AsyncSession,
    token: str,
    required_scopes: tuple[str, ...],
) -> ApiClientContext:
    """Authenticate a raw ``helix_api_`` token string and enforce every
    scope in ``required_scopes``, running the full defence stack
    (brute-force windows, per-client rate limit, scope-denial audit,
    usage bookkeeping). Raises ``UnauthorizedError`` /
    ``ForbiddenError`` / ``RATE_LIMITED`` on any failure.

    Shared by the header dependency (:func:`require_api_client`) and the
    body-token path (:func:`authorize_api_client_body_token`, used by the
    Feishu external-options webhook, which carries the credential in the
    request body rather than the ``Authorization`` header)."""
    if not token or not token.startswith(API_TOKEN_PREFIX):
        await _record_auth_failure(
            reason="non_api_client_token", request=request, token=None,
        )
        raise UnauthorizedError(
            "Only API-client tokens (helix_api_…) are accepted on this endpoint.",
        )

    allowed, retry_after = await _peek_token_failures(token)
    if not allowed:
        raise AppException(
            code="RATE_LIMITED",
            message=f"Too many failed attempts. Try again in {retry_after}s.",
            status_code=429,
            headers={"Retry-After": str(retry_after)},
        )

    resolved = await ApiClientTokenService(db).authenticate(token)
    if resolved is None:
        await _record_auth_failure(
            reason="invalid_or_expired_api_token", request=request, token=token,
        )
        raise UnauthorizedError("Invalid or expired token.")
    client, token_row = resolved

    # Rate-limit the authenticated client FIRST — before the scope
    # check — so a valid-but-underprivileged key cannot amplify
    # ``log_denied_action`` audit-table writes by hammering a scope it
    # lacks. The scope-denial path below writes an audit row but does
    # NOT feed the per-IP failed-auth counter, so without this ordering
    # an authenticated key could issue unbounded 403s, each an audit
    # INSERT. Bumping the per-client window here caps that.
    cap = client.rate_limit_per_minute or _DEFAULT_RATE_LIMIT_PER_MINUTE
    rl_key = f"extapi_ratelimit:{client.id}"
    count = await fixed_window_incr(rl_key, 60)
    if count is not None and count > cap:
        retry_after = await window_ttl(rl_key, 60)
        raise AppException(
            code="RATE_LIMITED",
            message=f"Rate limit exceeded. Try again in {retry_after}s.",
            status_code=429,
            headers={"Retry-After": str(retry_after)},
        )

    scopes = frozenset(
        s for s in (client.scopes or []) if isinstance(s, str)
    )
    missing = [s for s in required_scopes if s not in scopes]
    if missing:
        try:
            from app.audit.service import log_denied_action

            await log_denied_action(
                trace_id=getattr(request.state, "trace_id", None) or str(uuid4()),
                actor_type="service",
                action=_SCOPE_DENIED_ACTION,
                resource_type="api_client",
                resource_id=str(client.id),
                detail={
                    "client_slug": client.slug,
                    "missing_scopes": missing,
                    "path": request.url.path[:200],
                },
                ip_address=_client_ip(request),
            )
        except Exception:  # noqa: BLE001
            logger.exception("external-api scope-denied audit failed")
        raise ForbiddenError(
            f"API client lacks required scope(s): {', '.join(missing)}",
        )

    client_ip = _client_ip(request)
    await _record_usage(
        token_id=token_row.id, client_id=client.id, client_ip=client_ip,
    )
    logger.info(
        "external-api request client=%s path=%s ip=%s",
        client.slug, request.url.path, client_ip,
    )

    return ApiClientContext(
        client_id=client.id,
        slug=client.slug,
        name=client.name,
        scopes=scopes,
        token_id=token_row.id,
        token_prefix=token_row.token_prefix,
        cloud_visibility=(
            client.cloud_visibility
            if isinstance(client.cloud_visibility, dict)
            else None
        ),
        cloud_visibility_malformed=cloud_visibility_is_malformed(
            client.cloud_visibility,
        ),
        person_visibility=(
            client.person_visibility
            if isinstance(client.person_visibility, dict)
            else None
        ),
        person_visibility_malformed=(
            client.person_visibility is not None
            and not isinstance(client.person_visibility, dict)
        ),
    )


def require_api_client(*required_scopes: str):
    """Dependency factory: authenticate a ``helix_api_`` bearer and
    enforce that the client holds every scope in ``required_scopes``."""

    async def _dependency(
        request: Request,
        db: AsyncSession = Depends(get_db),
    ) -> ApiClientContext:
        auth_header = request.headers.get("authorization") or ""
        if not auth_header.lower().startswith("bearer "):
            await _record_auth_failure(
                reason="missing_or_malformed_authorization",
                request=request, token=None,
            )
            raise UnauthorizedError("Missing or malformed Authorization header.")

        token = auth_header.split(" ", 1)[1].strip()
        return await authorize_api_client_token(
            request=request, db=db, token=token,
            required_scopes=required_scopes,
        )

    return _dependency


async def authorize_api_client_body_token(
    *,
    request: Request,
    db: AsyncSession,
    token: str,
    required_scopes: tuple[str, ...],
) -> ApiClientContext:
    """Body-token variant of :func:`require_api_client` for webhook
    callers (Feishu external-options) that carry the ``helix_api_``
    credential inside the JSON body rather than the ``Authorization``
    header. Same credential plane and defence stack; raises identically."""
    return await authorize_api_client_token(
        request=request, db=db, token=(token or "").strip(),
        required_scopes=required_scopes,
    )


__all__ = [
    "ApiClientContext",
    "authorize_api_client_body_token",
    "authorize_api_client_token",
    "require_api_client",
]