src/api/routers/system.py

"""``/system/*`` endpoints — health + version + events.

Auth posture (current): only ``/system/ping`` is unauthenticated (it returns
just ``{"ok": true}`` as a liveness probe). ``/health``, ``/version``, and
``/events`` all require ``current_user`` per ``BACKEND_BUILD.md §7`` ("user"
tier) because they reveal topology/config, and ``/events`` scopes non-admin
callers to their own jobs. Config values surfaced by ``/health`` are redacted
(see :mod:`src.api.config`).

The **stale-pricing** field on ``/health`` is mandatory (CONTRACT.md §12 /
BACKEND_BUILD.md §11.6): it surfaces tools whose ``cost_model.last_updated``
is older than 90 days. The dashboard renders a warning chip when this is
non-empty so the user gets nudged to refresh ``tools_registry.yaml``
quarterly — without us having to nag in-app.
"""

from __future__ import annotations

import asyncio
import json
import logging
from datetime import datetime, timezone
from pathlib import Path

from fastapi import APIRouter, Depends, Request
from fastapi.responses import StreamingResponse

from src.api.config import current_git_sha, masked_config
from src.api.deps import forbid_readers
from src.api.dto.system import HealthCheck, HealthResponse, VersionResponse
from src.api.jobs import store as jobs_store
from src.lib.auth.models import User
from src.lib.system.health import (
    check_env_keys,
    check_inventory_db,
    check_qdrant,
)
from src.lib.system.models import CheckResult
from src.lib.tools.repo import load_registry

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/system", tags=["system"])

# Poll cadence for the active-jobs feed. 2.5s keeps the sidebar chip
# responsive without hammering Turso — one cheap SELECT per tick across
# all attached clients is fine even on the free tier.
_ACTIVE_JOBS_POLL_SEC = 2.5
# Heartbeat cadence per CONTRACT.md §4. Browsers / proxies will close
# an idle SSE connection without these.
_HEARTBEAT_SEC = 15.0


# ---------- Helpers -------------------------------------------------------
def _to_check_dto(result: CheckResult) -> HealthCheck:
    """Adapt the existing :class:`CheckResult` dataclass to the wire DTO.

    The repo's ``CheckResult`` carries a ``fix_command`` field we don't ship
    on the wire (running shell snippets from a JSON response is a footgun in
    a future remote-host deployment). The textual ``fix_hint`` is enough for
    the dashboard chip.
    """
    return HealthCheck(
        name=result.name,
        ok=result.ok,
        detail=result.detail,
        fix_hint=result.fix_hint,
    )


def _check_storage() -> HealthCheck:
    """Probe the active storage backend.

    Resolves :func:`src.lib.storage.get_storage` and confirms the backend
    can answer ``list_book_ids()``. Cheap on the local backend
    (``listdir`` over ``raw_dir``); the R2 backend issues one ``list_objects_v2``
    call which is fine to do at startup-frequency intervals. We don't open
    each PDF — that's what ``ensure_local`` is for.
    """
    try:
        from src.lib.storage import get_storage

        backend = get_storage()
        ids = backend.list_book_ids()
    except Exception as e:  # noqa: BLE001
        return HealthCheck(
            name="Storage",
            ok=False,
            detail=f"{type(e).__name__}: {e}",
            fix_hint="Check backends.storage in config.yaml.",
        )
    return HealthCheck(
        name="Storage",
        ok=True,
        detail=f"{backend.name} · {len(ids)} PDF(s)",
    )


def _stale_pricing(*, max_age_days: int = 90) -> list[str]:
    """Return tool names with stale ``cost_model.last_updated`` entries.

    Stale = older than ``max_age_days`` from today (UTC). Tools without
    ``last_updated`` are also returned so the user knows which entries
    never got dated. Free tools (no pricing block) are excluded.
    """
    today = datetime.now(timezone.utc).date()
    stale: list[str] = []
    for tool in load_registry():
        if tool.pricing is None:
            continue  # not a paid tool; skip
        raw = tool.pricing.last_updated
        if not raw:
            stale.append(tool.name)
            continue
        try:
            updated = datetime.strptime(str(raw), "%Y-%m-%d").date()
        except (TypeError, ValueError):
            stale.append(tool.name)
            continue
        if (today - updated).days > max_age_days:
            stale.append(tool.name)
    return stale


# ---------- Endpoints -----------------------------------------------------
@router.get("/ping")
def ping() -> dict:
    """Unauthenticated liveness probe — zero information disclosure.

    Exists so an external keep-warm pinger (and any load-balancer health
    check) can confirm the process is up without exposing the backend
    topology. The detailed :func:`health` below — which lists Turso/Qdrant
    hosts, model names, the active collection, etc. — is auth-gated.
    """
    return {"ok": True}


@router.get("/health", response_model=HealthResponse)
def health(_: User = Depends(forbid_readers)) -> HealthResponse:
    """Aggregate health: DB, Qdrant, storage, API keys, stale pricing.

    Each probe is independent — a failure in one doesn't short-circuit the
    others. ``ok`` at the top level is the AND of the four dependency
    probes (stale pricing is informational, not a fail condition).

    Auth-gated: the per-probe ``detail`` strings disclose backend topology
    (hostnames, the active collection, PDF counts), so this is for
    logged-in operators only. Unauthenticated callers use ``/system/ping``.
    """
    db = _to_check_dto(check_inventory_db())
    qdrant = _to_check_dto(check_qdrant())
    storage = _check_storage()
    api_keys = _to_check_dto(check_env_keys())
    stale = _stale_pricing()
    return HealthResponse(
        ok=db.ok and qdrant.ok and storage.ok and api_keys.ok,
        db=db,
        qdrant=qdrant,
        storage=storage,
        api_keys=api_keys,
        stale_pricing=stale,
    )


@router.get("/version", response_model=VersionResponse)
def version(_: User = Depends(forbid_readers)) -> VersionResponse:
    """Current git short-SHA + masked config snapshot.

    Used by the ops dashboard to confirm what's running. The config is
    redacted (api keys, tokens, URLs-with-userinfo) via :func:`masked_config`
    — we want operators to be able to send this payload to a maintainer
    without leaking secrets. Auth-gated: even masked, the config still
    reveals topology/model choices, so it's for logged-in operators only.
    """
    return VersionResponse(
        git_sha=current_git_sha(),
        config=masked_config(),
    )


# ---------- /system/events SSE -------------------------------------------
def _snapshot_active_jobs(username: str | None = None) -> dict:
    """Read the current active-jobs roster from the jobs table.

    Returns the wire shape ``{"activeCount": int, "runningJobIds": [str]}``
    consumed by the FE's :file:`SidebarJobStatus.tsx`. Statuses considered
    active: ``queued`` and ``running``. The Turso retry happens inside
    :func:`src.api.jobs.store.list_jobs` (and one layer deeper at the
    connection facade), so this helper is exception-safe for the SSE
    generator.

    ``username`` scopes the roster to one owner — the endpoint passes the
    caller's username for non-admins so the global feed doesn't broadcast
    other users' job ids (admins pass ``None`` to see everything).
    """
    ids: list[str] = []
    for status in ("queued", "running"):
        try:
            rows = jobs_store.list_jobs(status=status, username=username, limit=50)
        except Exception:  # noqa: BLE001
            logger.exception("/system/events: list_jobs(%s) failed", status)
            continue
        for r in rows:
            ids.append(r.id)
    return {"activeCount": len(ids), "runningJobIds": ids}


def _format_event(event_type: str, data: dict) -> str:
    """Encode one SSE frame per CONTRACT.md §4 wire format.

    Format: ``event: TYPE\\ndata: JSON\\n\\n``. No ``id:`` field because
    /system/events is stateless — there is no ``Last-Event-ID`` replay
    window for "what jobs are running right now". Clients reconnect and
    receive the current snapshot.
    """
    body = json.dumps(data, ensure_ascii=False, default=str)
    return f"event: {event_type}\ndata: {body}\n\n"


async def _system_event_stream(request: Request, username: str | None = None):
    """Yield ``active_jobs`` snapshots every poll tick + ``ping`` heartbeats.

    Architecture: a single polling loop drives both event types — every
    tick we check whether the active-jobs snapshot changed (and emit if
    so or on the first iteration), and whether enough time has passed
    since the last heartbeat (and emit a ping if so). This avoids the
    overhead of separate tasks while still bounding both cadences.

    A client-disconnect check runs on every tick so a closed tab stops
    generating within one poll interval.
    """
    loop = asyncio.get_event_loop()
    last_heartbeat = loop.time()
    last_snapshot: dict | None = None

    # _snapshot_active_jobs() is synchronous DB I/O (sqlite / libsql) and is
    # called every poll tick. It MUST run on the default thread-pool
    # executor — calling it inline would block the asyncio event loop on
    # every tick, and after a PUT /config backend flip the first call also
    # triggers a multi-second schema re-init that froze every other
    # coroutine in the process (2026-05-18 incident).
    async def _snapshot_async() -> dict:
        return await loop.run_in_executor(None, _snapshot_active_jobs, username)

    # Emit initial snapshot immediately so the FE chip flips off
    # "Connecting…" on first byte rather than waiting one poll tick.
    snapshot = await _snapshot_async()
    last_snapshot = snapshot
    yield _format_event("active_jobs", snapshot)

    while True:
        if await request.is_disconnected():
            return

        await asyncio.sleep(_ACTIVE_JOBS_POLL_SEC)
        if await request.is_disconnected():
            return

        # Refresh snapshot; only re-emit when it actually changed (cheap
        # JSON-equality check; avoids spamming clients with identical
        # frames during idle periods).
        snapshot = await _snapshot_async()
        if snapshot != last_snapshot:
            last_snapshot = snapshot
            yield _format_event("active_jobs", snapshot)

        # Heartbeat ticker. CONTRACT.md §4 spec'd 15s ping cadence.
        now = loop.time()
        if now - last_heartbeat >= _HEARTBEAT_SEC:
            last_heartbeat = now
            yield "event: ping\ndata: {}\n\n"


@router.get("/events")
async def system_events(
    request: Request,
    user: User = Depends(forbid_readers),
) -> StreamingResponse:
    """Global SSE feed: active job count + running job ids + heartbeats.

    Spec: BACKEND_BUILD.md §7 (``GET /system/events`` — user tier). Feeds
    the FE's sidebar chip (``SidebarJobStatus.tsx``) and is intended as
    the canonical "what's running app-wide" subscription point.

    Implementation note: the SSE pubsub in ``src.api.jobs.sse`` is
    per-job (each job_id has its own subscriber set). For the global
    "any job changed" feed a poll-driven snapshot is simpler and the
    Turso latency on a single status filter is in the low milliseconds,
    so polling every 2.5s costs less than wiring a separate global bus.
    """
    return StreamingResponse(
        _system_event_stream(request, None if user.is_admin else user.username),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )