thermograph/backend/web/app.py

"""Thermograph API — grade recent local weather against ~45 years of climatology."""
import contextlib
import datetime
import hashlib
import hmac
import ipaddress
import json
import os
import queue
import threading
import time

import httpx
from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response
from fastapi.concurrency import run_in_threadpool
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse, RedirectResponse

from accounts import api_accounts
from api import content_routes
from api import internal_routes
from core import audit
from data import climate
from notifications import digest
from notifications import discord_interactions as discord_interactions_mod
from notifications import discord_link
from accounts import db, oauth
from data import grid
from core import metrics
from notifications import notify
import paths
from data import places
from core import singleton
from data import store
from accounts import users
from api import payloads
from accounts.schemas import UserCreate, UserRead, UserUpdate

# Everything (pages, assets, API) is served under this base path so the app can
# live at https://<host>/thermograph/ behind a shared domain. Override with the
# THERMOGRAPH_BASE env var. The frontend uses base-relative URLs, so it follows
# this automatically without hardcoding the prefix.
#
# THERMOGRAPH_BASE=/ (or empty) means the app owns the whole domain: BASE becomes
# "" so every route sits at the domain root ("/", "/calendar", "/api/v2/…") with
# no prefix and no leading double slash. A non-empty value keeps the sub-path form
# ("/thermograph").
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""

# The SSR content service (repo-split Stage 4): backend no longer renders the
# climate hub/city/month/records/glossary/home pages itself (frontend_ssr does,
# over the content API above), and this process now needs a way back TO it for
# every environment with no reverse proxy in front to do the split (LAN dev,
# bare-metal run.sh) -- see _proxy_to_frontend below. Required, fails loud: every
# environment is dual-service from this stage on, no single-service fallback.
FRONTEND_BASE = os.environ["THERMOGRAPH_FRONTEND_BASE_INTERNAL"]
_frontend_client = httpx.AsyncClient(base_url=FRONTEND_BASE, timeout=10.0)

# Which duties this process performs: "web" (serve requests, no notifier),
# "worker" (own the notifier, still serves requests today — see _lifespan), or
# "all" (both — today's single-process default, so an unset ROLE is unchanged
# behavior). Read once at import: toggling requires a process restart, same as
# every other env-driven constant here (BASE, WORKERS, ...).
ROLE = os.environ.get("THERMOGRAPH_ROLE", "all").strip().lower()


def _weather_fetch_error(e) -> HTTPException:
    """A clean, retryable 503 when weather data is temporarily unavailable; the
    raw error as a 502 otherwise. The fetch layer classifies its own failures
    (climate.WeatherUnavailable carries the user-facing text) so no message
    parsing happens here; the is_rate_limit fallback covers a raw upstream 429
    from a fetcher that didn't classify it."""
    if isinstance(e, climate.WeatherUnavailable):
        return HTTPException(status_code=503, detail=str(e))
    if climate.is_rate_limit(e):
        return HTTPException(status_code=503, detail=climate.limit_message(False))
    return HTTPException(status_code=502, detail=f"weather data fetch failed: {e}")


def _parse_target_date(date: str | None, default: datetime.date) -> datetime.date:
    """Parse the ``date`` query param (YYYY-MM-DD) into a date, falling back to
    ``default`` when it's absent. A malformed or non-calendar date (e.g. 2026-13-40)
    is a client error, not a crash: surface it as a 422 rather than let
    fromisoformat's ValueError become a 500."""
    if not date:
        return default
    try:
        return datetime.date.fromisoformat(date[:10])
    except ValueError:
        raise HTTPException(status_code=422,
                            detail=f"invalid date: {date!r} (expected YYYY-MM-DD)")


# --- background neighbor warming ---------------------------------------------
# /cell?neighbors=1 asks the server to warm the 8 grid cells around the request
# (the frontend used to fire 8 staggered prefetch requests, mirroring grid.py's
# snapping math in JS). One worker drains the queue so warms never stack; the
# hard guarantee matches prefetch=1 — a cell with no cached archive is skipped,
# so no weather-API quota is ever spent, and reverse_geocode itself paces the
# at-most-one Nominatim call per never-labeled cell (~1/s policy).

_WARM_TTL = 3600.0                 # don't re-enqueue a cell within the hour
_WARM_SEEN: dict[str, float] = {}  # cell_id -> last enqueue time
_warm_queue: "queue.Queue[dict]" = queue.Queue()


def _warm_cell(cell: dict) -> None:
    """Materialize the history-derived slices for one cell — the same rows the
    prefetch=1 bundle serves. Never fetches weather upstream."""
    history = climate.load_cached_history(cell)
    if history is None or history.is_empty():
        return
    place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"])
    token = payloads.history_token(history)
    start_ts, end_ts = payloads.cal_span(history, None, None, 24)
    cal_key = payloads.calendar_key(start_ts, end_ts, 24)
    if store.get_payload("calendar", cell["id"], cal_key, token) is None:
        store.put_payload("calendar", cell["id"], cal_key, token,
                          payloads.build_calendar(cell, history, start_ts, end_ts, 24, place))
    last = history["date"].max()
    day_key = payloads.day_key(last)
    if store.get_payload("day", cell["id"], day_key, token) is None:
        store.put_payload("day", cell["id"], day_key, token,
                          payloads.build_day(cell, history, last, place))


def _enqueue_neighbor_warming(cell: dict) -> None:
    now = time.time()
    for n in grid.neighbors(cell):
        if now - _WARM_SEEN.get(n["id"], 0.0) < _WARM_TTL:
            continue
        _WARM_SEEN[n["id"]] = now
        _warm_queue.put(n)


def _neighbor_warmer() -> None:
    while True:
        cell = _warm_queue.get()
        try:
            _warm_cell(cell)
        except Exception:  # noqa: BLE001 - warming is best-effort, never fatal
            pass
        finally:
            _warm_queue.task_done()


def _should_run_notifier() -> bool:
    """Whether this process should start the subscription-notifier thread: ROLE must
    permit it, and it must win the leader election. Split out from _lifespan so this
    decision is unit-testable without booting the full app (DB init, places index,
    neighbor warmer) — see tests/web/test_role.py."""
    return ROLE in ("worker", "all") and singleton.claim_leader()


_HEARTBEAT_STOP = threading.Event()
_HEARTBEAT_THREAD: threading.Thread | None = None
HEARTBEAT_INTERVAL = float(os.environ.get("THERMOGRAPH_HEARTBEAT_INTERVAL", "300"))  # 5 min


def _heartbeat_loop() -> None:
    metrics.record_heartbeat(ROLE, HEARTBEAT_INTERVAL)
    audit.log_heartbeat(ROLE, HEARTBEAT_INTERVAL)
    while not _HEARTBEAT_STOP.wait(HEARTBEAT_INTERVAL):
        metrics.record_heartbeat(ROLE, HEARTBEAT_INTERVAL)
        audit.log_heartbeat(ROLE, HEARTBEAT_INTERVAL)


def _start_heartbeat() -> None:
    """Process-level liveness beat, tagged daemon=ROLE, independent of whether this
    process also runs the notifier. web's and worker's own stdout is near-silent by
    design (real logging goes to the app-json file source; worker's Docker/stdout
    stream is dropped entirely by Alloy), so container-log-silence alerting can't
    tell a live process from a dead one for either role — this heartbeat is the
    signal observability actually alerts on instead. Deliberately unguarded across
    every uvicorn worker process and every autoscaled replica: a beat is a single
    local file append (no upstream quota, no DB, no lock), so duplicate beats are
    harmless, and only the process actually being dead stops them — see
    _should_run_notifier's leader election for the contrasting case where
    duplication would be a real cost."""
    global _HEARTBEAT_THREAD
    if os.environ.get("THERMOGRAPH_ENABLE_HEARTBEAT", "1") == "0" or _HEARTBEAT_THREAD is not None:
        return
    _HEARTBEAT_STOP.clear()
    _HEARTBEAT_THREAD = threading.Thread(target=_heartbeat_loop, name=f"heartbeat-{ROLE}", daemon=True)
    _HEARTBEAT_THREAD.start()


@contextlib.asynccontextmanager
async def _lifespan(app):
    # Warm the local place-name index (/suggest's typo tolerance) in the
    # background; the app boots and serves fine without it — suggestions just
    # fall back to the upstream geocoder until it's ready. A server-startup
    # hook (not import time) so offline importers (tests, the migrate script's
    # dependencies) don't kick off a GeoNames download. The neighbor warmer is
    # also a server concern: enqueues before startup just wait in the queue.
    # Create the account DB tables (users/sessions/subscriptions/notifications)
    # before serving — a fast no-op once they exist.
    await db.create_db_and_tables()
    places.start_loading()
    threading.Thread(target=_neighbor_warmer, name="neighbor-warmer", daemon=True).start()
    _start_heartbeat()
    # Background subscription evaluator. Its periodic sweep fetches upstream on a timer
    # (not per request), so with multiple workers — or multiple *hosts* under Swarm —
    # it must run in exactly one: elect a leader (see singleton.claim_leader, which
    # picks a host-local flock or a cluster-wide Postgres advisory lock from env).
    # Unset (single worker / dev / tests) => always the leader. The neighbor warmer
    # above stays per-worker: each drains its own request-fed queue. ROLE further
    # restricts this to processes that are allowed to own it at all — "web" replicas
    # never start it even if they'd otherwise win the leader election, so a stateless
    # N-replica web tier can scale without also scaling notifier instances.
    # The Discord gateway bot and the recurring worker jobs (city warming,
    # IndexNow) used to ride this same leader gate in-process; both now live in
    # the thermograph-daemon Go binary, which owns the connection/timers and
    # calls back over api/internal_routes.py for anything needing data. The
    # notifier is the one singleton still running here.
    if _should_run_notifier():
        notify.start()
    # Deploy/restart marker: one line correlates a metric shift with a boot.
    audit.log_activity("app.start", {"version": app.version, "role": ROLE, "base": BASE,
                                     "notifier": _should_run_notifier(), "pid": os.getpid()})
    yield
    audit.log_activity("app.stop", {"role": ROLE, "pid": os.getpid()})
    notify.stop()
    _HEARTBEAT_STOP.set()
    await _frontend_client.aclose()


app = FastAPI(title="Thermograph", version="0.2.0", lifespan=_lifespan)

# Compress every sizeable response (the 2-year calendar JSON shrinks ~6-8×).
# Applies to API JSON and static assets alike; tiny responses are left alone.
# compresslevel: GZipMiddleware compresses synchronously ON THE EVENT LOOP, so
# this is CPU work competing with every other async route in the worker while it
# runs — zlib's default level 9 spends a lot of that time chasing the last couple
# percent of ratio for a payload this size. 6 is the standard speed/ratio sweet
# spot (roughly zlib's own recommended default): most of level 9's compression
# but a fraction of the CPU, so a big calendar/day/cell-bundle response doesn't
# stall other requests on this worker for as long. Not worth a custom threaded
# compression middleware for payloads this size — the level drop captures most
# of the win for near-zero complexity.
app.add_middleware(GZipMiddleware, minimum_size=1024, compresslevel=6)

# CORS (repo-split Stage 5): unset (default) means no CORSMiddleware at all --
# today's exact behavior, browsers already allow same-origin fetch with no
# help from us. Comma-separated THERMOGRAPH_CORS_ORIGINS opts in an explicit
# allowlist for a genuinely cross-origin frontend (never "*" -- FastAPI/
# Starlette itself refuses that combined with allow_credentials=True, since a
# credentialed wildcard is exactly the misconfiguration CORS exists to
# prevent). expose_headers=["ETag"] matters as much as allow_origins: without
# it, cache.js's res.headers.get("ETag") silently returns null cross-origin
# (the browser hides every response header except a fixed "simple" allowlist
# by default) and 304 revalidation quietly stops working with no console
# error at all.
_cors_origins = [o.strip() for o in os.environ.get("THERMOGRAPH_CORS_ORIGINS", "").split(",") if o.strip()]
if _cors_origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=_cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
        expose_headers=["ETag"],
    )


# Log a structured line for every unhandled 500. Without an exception handler the
# route exception propagates out past the request middleware (call_next re-raises),
# so a 500 is otherwise invisible in both the access log and metrics — this makes
# it visible and countable (tag="http.error" in the errors stream).
SLOW_REQUEST_MS = 2000.0   # http.slow: log any request slower than this


@app.exception_handler(Exception)
async def _log_unhandled_exception(request: Request, exc: Exception) -> Response:
    try:
        audit.log_event("http.error", {
            "phase": "http", "path": request.url.path, "method": request.method,
            "cat": metrics.classify_inbound(request.url.path, BASE),
            "exc_type": type(exc).__name__, "error": repr(exc)[:500]})
    except Exception:  # noqa: BLE001 - logging must never mask the original error
        pass
    return JSONResponse({"detail": "Internal server error."}, status_code=500)


@app.get("/healthz")
def healthz():
    """Liveness probe: the process is up and serving. Deliberately does no I/O (no
    DB query, no upstream call) so it stays cheap and reliable enough for a tight
    Swarm/Docker healthcheck interval — readiness (DB reachable, etc.) is a
    separate concern the ingress health-gate covers by hitting a real route. Not
    under BASE: a fixed, unauthenticated path a healthcheck can hit without
    knowing THERMOGRAPH_BASE, on both the web and worker roles."""
    return {"status": "ok", "role": ROLE}


API_CONTRACT_VERSION = "2"   # bump only on a breaking change to any /api/v{N} route or payload shape
MIN_SUPPORTED_FRONTEND = "1" # oldest frontend contract version this backend still serves correctly


@app.get(f"{BASE}/api/version")
def api_version():
    """Capability/version-negotiation endpoint: lets a split-out frontend detect at
    boot whether the backend it's talking to still supports its contract.
    Deliberately does no I/O, same posture as /healthz."""
    return {
        "backend_version": API_CONTRACT_VERSION,
        "min_frontend": MIN_SUPPORTED_FRONTEND,
        "payload_ver": payloads.PAYLOAD_VER,
    }


def _client_ip(request) -> "str | None":
    """The real client IP: the left-most X-Forwarded-For hop when a proxy fronts us
    (Caddy on prod sets it), otherwise the direct peer address (LAN dev). Full
    precision — the rate limiter (metrics._rate_ok) needs the exact address, since
    a truncated key would let one abuser exhaust a whole NAT'd office's quota.
    Truncate at the point of persistence instead (see _loggable_ip)."""
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else None


def _loggable_ip(ip: "str | None") -> "str | None":
    """Coarsen a client IP before it's written to the access log: /24 for IPv4,
    /48 for IPv6. Keeps rough geo/abuse signal without keeping a full, joinable
    address sitting in structured logs for the log's 30-day retention -- the
    public privacy page promises IPs aren't logged beyond normal request handling,
    and a raw address shipped to Loki was a live gap against that. Never the
    input to the rate limiter (_client_ip's callers pass the untruncated value
    there) -- this is only what gets persisted."""
    if not ip:
        return ip
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return ip  # not a parseable address (e.g. a test/placeholder value) -- pass through
    prefix = 24 if addr.version == 4 else 48
    return str(ipaddress.ip_network(f"{addr}/{prefix}", strict=False).network_address)


@app.middleware("http")
async def revalidate_static(request, call_next):
    """Serve the frontend with no-cache (NOT no-store): browsers may keep a copy
    but must revalidate it on every use. Starlette's FileResponse/StaticFiles
    already emit ETag + Last-Modified, so an unchanged asset costs one conditional
    request answered with an empty 304 instead of a full re-download — page-to-page
    navigation stops re-transferring the JS/CSS/HTML while still picking up every
    deploy immediately (no "my change isn't showing" bugs)."""
    t0 = time.perf_counter()
    response = await call_next(request)
    dur_ms = (time.perf_counter() - t0) * 1000.0
    path = request.url.path
    try:
        cat = metrics.classify_inbound(path, BASE)
        # Both are blocking (sync DB write, sync file write) — run off the event
        # loop so one request's instrumentation can't stall every other request
        # this worker is serving concurrently.
        await run_in_threadpool(metrics.record_inbound, cat, response.status_code)
        # Retain per-request client IPs for later analysis; skip static assets, the
        # dashboard's own metrics polling, the liveness probe (health -- by far the
        # highest-volume single path, and never real traffic), and the internal
        # SSR->API hop (internal -- frontend_ssr's own server-to-server calls, not
        # a page view) to keep the log to real, meaningful traffic. The IP itself is
        # truncated (see _loggable_ip) -- coarse geo/abuse signal, not a full address.
        if cat not in ("static", "metrics", "event", "health", "internal"):
            await run_in_threadpool(audit.log_access, {
                "ip": _loggable_ip(_client_ip(request)), "method": request.method,
                "path": path, "status": response.status_code, "cat": cat})
        # Threshold-gated slow-request line: RunAudit only times the 7 graded
        # endpoints, so this is the only latency signal for auth/account/content
        # routes. Kept rare (>2s) so it never becomes per-request hot-path noise.
        # Also a blocking file write, so keep it off the event loop.
        if dur_ms > SLOW_REQUEST_MS and cat != "static":
            await run_in_threadpool(audit.log_activity, "http.slow", {
                "path": path, "method": request.method, "cat": cat,
                "status": response.status_code, "duration_ms": round(dur_ms, 1)})
    except Exception:  # noqa: BLE001 - never let instrumentation break a response
        pass
    pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/score",
             f"{BASE}/compare", f"{BASE}/legend", f"{BASE}/alerts", f"{BASE}/privacy")
    if path.endswith((".js", ".css", ".html")) or path in pages:
        response.headers["Cache-Control"] = "no-cache"
    return response


# --- derived-store plumbing --------------------------------------------------
# Every graded payload is cached in SQLite under (kind, cell, key) and validated
# by a token that encodes what it was computed from (see store.py). The freshness
# drivers stay where they were — climate.get_history tops up the archive tail
# hourly and get_recent_forecast refetches hourly — and the tokens are derived
# from what those return, so a cached payload expires exactly when its inputs
# change and never before. The same tokens double as ETags: a client sending
# If-None-Match gets an empty 304 without the payload even being loaded.
# The payloads themselves are assembled in api/payloads.py, shared with the
# offline migrate script.

def _etag_for(kind: str, cell_id: str, key: str, token: str) -> str:
    """Deterministic weak ETag from a payload's identity + validity token. Weak
    because the same content may be served under different encodings (gzip)."""
    h = hashlib.sha1(f"{kind}:{cell_id}:{key}:{token}".encode()).hexdigest()[:20]
    return f'W/"{h}"'


def _not_modified(request: Request, etag: str) -> bool:
    """Does the client already hold this exact payload? Answerable from the token
    alone — no payload load needed."""
    inm = request.headers.get("if-none-match")
    if not inm:
        return False
    tags = {t.strip() for t in inm.split(",")}
    return "*" in tags or etag in tags or etag.removeprefix("W/") in tags


def _encode(payload: dict) -> bytes:
    """Compact JSON bytes, matching store.put_payload's encoding (allow_nan=False
    mirrors Starlette — a NaN from grading should fail loudly, not reach a client)."""
    return json.dumps(payload, default=str, allow_nan=False, separators=(",", ":")).encode()


def _json_response(body: bytes, etag: str | None = None) -> Response:
    headers = {"ETag": etag} if etag else {}
    return Response(content=body, media_type="application/json", headers=headers)


def _fetch_history(run, cell, recent_too=False, recent_phase="recent"):
    """The shared fetch preamble for every data route: the archive record (and
    optionally the hourly recent/forecast bundle) with upstream failures mapped
    to clean HTTP errors and an empty record to a 404."""
    try:
        with run.phase("history"):
            history, cache_meta = climate.get_history(cell)
        recent = None
        if recent_too:
            with run.phase(recent_phase):
                recent = climate.get_recent_forecast(cell)
    except Exception as e:  # noqa: BLE001
        raise _weather_fetch_error(e)
    if history.is_empty():
        raise HTTPException(status_code=404, detail="No historical data for this cell.")
    return history, cache_meta, recent


def _cached_response(request, run, kind, cell, key, token, build, cache_placeless=True):
    """The shared derived-store flow behind every per-view endpoint:
    If-None-Match -> empty 304; a token-valid store row -> replay its exact
    bytes; otherwise resolve the place label, build the payload, persist, serve.

    ``build(place) -> payload`` does any endpoint-specific audit tagging itself.
    ``cache_placeless=False`` skips persisting a payload whose place label
    failed to resolve (a transient reverse-geocode miss) — otherwise the
    bare-coordinates fallback would stick for the life of the token."""
    cid = cell["id"]
    etag = _etag_for(kind, cid, key, token)
    if _not_modified(request, etag):
        run.set(run_type="cache", not_modified=True)
        return Response(status_code=304, headers={"ETag": etag})
    body = store.get_payload(kind, cid, key, token)
    if body is not None:
        run.set(run_type="cache", history_source="store")
        return _json_response(body, etag)
    with run.phase("reverse_geocode"):
        place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"])
    payload = build(place)
    return _json_response(
        store.put_payload(kind, cid, key, token, payload,
                          cache=cache_placeless or place is not None), etag)



# --- endpoints ----------------------------------------------------------------

_GEOCODE_LIMIT = 5


def api_geocode(q: str = Query(..., min_length=1)):
    """Forward place search. The local GeoNames index answers first (no network);
    only a genuine miss — or a query the index can't hold (neighbourhoods,
    postcodes, tiny villages, native-language names) — falls back to Nominatim.
    `search` returns None while the index is still loading, which also falls
    through to the network so search keeps working during warmup."""
    local = places.search(q, _GEOCODE_LIMIT)
    if local:
        # Strip the internal prefix/fuzzy tag so the wire shape matches the
        # Nominatim branch (and the former Open-Meteo contract) exactly.
        return {"results": [{k: v for k, v in r.items() if k != "match"} for r in local]}
    try:
        return {"results": climate.geocode_nominatim(q, _GEOCODE_LIMIT)}
    except Exception as e:  # noqa: BLE001 - surface upstream failures to the client
        raise HTTPException(status_code=502, detail=f"geocoding failed: {e}")


_SUGGEST_LIMIT = 5


def _no_upstream(q: str) -> tuple:
    """Autocomplete is served purely from the local GeoNames index — no per-keystroke
    call to any external geocoder. Typo-correction still runs against the local
    index, so respellings ("pest seattle" → "west seattle") keep working offline."""
    return ()


def api_suggest(q: str = Query(..., min_length=1)):
    """Type-ahead location suggestions: the top 5 places for a (possibly
    typo'd) query prefix; `corrected` reports the respelling that produced the
    results ("pest seattle" → "west seattle"), or null. The ranking/typo policy
    lives in places.suggest, now driven entirely by the local index."""
    try:
        results, corrected = places.suggest(q, _no_upstream, _SUGGEST_LIMIT)
    except Exception as e:  # noqa: BLE001 - nothing at all to serve
        raise HTTPException(status_code=502, detail=f"geocoding failed: {e}")
    return {"results": results, "corrected": corrected}


def api_place(
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
):
    """Best-effort neighbourhood/city label for a point — the same string the view
    endpoints expose as ``place`` (snapped to the grid cell, reverse-geocoded and
    cached). Resolved on its own so the compare page can show a location's name as
    soon as it's added, before its full series loads."""
    cell = grid.snap(lat, lon)
    with audit.RunAudit(endpoint="place", lat=round(lat, 4), lon=round(lon, 4),
                        cell_id=cell["id"]) as run:
        with run.phase("reverse_geocode"):
            place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"])
    return {"place": place,
            "cell": {"center_lat": cell["center_lat"], "center_lon": cell["center_lon"]}}


def api_grade(
    request: Request,
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
    date: str | None = Query(None, description="target date YYYY-MM-DD (default today)"),
    days: int = Query(14, ge=1, le=60, description="days of history to grade before the target"),
    after: int = Query(14, ge=0, le=14,
                       description="days to grade after the target (observed, or forecast when future; "
                                   "the forecast reaches ~7 days out so future days cap there)"),
):
    target = _parse_target_date(date, datetime.date.today())
    cell = grid.snap(lat, lon)

    with audit.RunAudit(
        endpoint="grade",
        lat=round(lat, 4),
        lon=round(lon, 4),
        target_date=target.isoformat(),
        recent_days=days,
        cell_id=cell["id"],
    ) as run:
        history, cache_meta, recent = _fetch_history(run, cell, recent_too=True)
        return _cached_response(
            request, run, "grade", cell,
            payloads.grade_key(target, days, after), payloads.recent_token(history, cell["id"]),
            lambda place: payloads.build_grade(cell, target, days, history, recent,
                                            cache_meta, place, run, after=after))


def api_calendar(
    request: Request,
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
    start: str | None = Query(None, description="first date YYYY-MM-DD (overrides months)"),
    end: str | None = Query(None, description="last date YYYY-MM-DD (default = latest available)"),
    months: int = Query(24, ge=1, le=24, description="months back to grade when no start given"),
):
    """Grade every day over a date range (default: last 2 years) for the calendar view.

    A custom [start, end] range is supported and capped at ~2 years per request;
    the frontend splits longer spans into successive 2-year chunks. Grades against
    the already-cached history record (which reaches to ~6 days ago), so no extra
    fetch is needed. The graded payload is cached in the derived store keyed by
    the clamped span and validated by the record's end date, so a repeat span is
    a database read — across restarts — until new archive days arrive. Returns a
    compact per-day shape (see grading.grade_range). New in API v2.
    """
    cell = grid.snap(lat, lon)

    with audit.RunAudit(
        endpoint="calendar",
        lat=round(lat, 4),
        lon=round(lon, 4),
        recent_days=months * 31,  # approximate span graded
        cell_id=cell["id"],
    ) as run:
        history, cache_meta, _ = _fetch_history(run, cell)
        start_ts, end_ts = payloads.cal_span(history, start, end, months)

        def build(place):
            payload = payloads.build_calendar(cell, history, start_ts, end_ts, months, place, run)
            full = not cache_meta.get("cached", False)
            run.set(run_type="full" if full else "partial",
                    history_source="fetch" if full else "cache")
            return payload

        # Compare loads several cells at once, so a transient reverse-geocode miss
        # is most likely here; cache_placeless=False keeps that miss un-persisted.
        return _cached_response(request, run, "calendar", cell,
                                payloads.calendar_key(start_ts, end_ts, months),
                                payloads.history_token(history), build,
                                cache_placeless=False)


def api_day(
    request: Request,
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
    date: str | None = Query(None, description="target date YYYY-MM-DD (default = latest available)"),
):
    """Full percentile breakdown for a single day: the value at every tier boundary
    in that day-of-year's ±7-day window, plus where the observed values land.

    Powers the single-day detail page. Grades against the cached history record;
    for a date newer than the cache it pulls the recent window for the observation
    (those payloads expire hourly — the recent bundle's own cadence — while fully
    archived days stay valid until the record itself advances). New in API v2.
    """
    cell = grid.snap(lat, lon)

    with audit.RunAudit(
        endpoint="day",
        lat=round(lat, 4),
        lon=round(lon, 4),
        cell_id=cell["id"],
    ) as run:
        history, cache_meta, _ = _fetch_history(run, cell)
        last = history["date"].max()
        target = _parse_target_date(date, last)
        run.set(target_date=target.isoformat())

        def build(place):
            payload = payloads.build_day(cell, history, target, place, run)
            full = not cache_meta.get("cached", False)
            run.set(run_type="full" if full else "partial",
                    history_source="fetch" if full else "cache")
            return payload

        return _cached_response(request, run, "day", cell,
                                payloads.day_key(target), payloads.day_token(history, target),
                                build)


def api_score(
    request: Request,
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
):
    """Climate-drift score for a location: how far the last 6 years have moved
    from the full 45-year baseline, per metric, percentile category and season,
    rolled into per-metric and overall scores.

    Derived purely from the archive record, so it is cached until the record's
    tail advances (the hourly top-up); the scoring-math version is baked into the
    cache key so a math change invalidates only score rows. New in API v2.
    """
    cell = grid.snap(lat, lon)

    with audit.RunAudit(endpoint="score", lat=round(lat, 4), lon=round(lon, 4),
                        cell_id=cell["id"]) as run:
        history, cache_meta, _ = _fetch_history(run, cell)

        def build(place):
            payload = payloads.build_score(cell, history, place, run)
            full = not cache_meta.get("cached", False)
            run.set(run_type="full" if full else "partial",
                    history_source="fetch" if full else "cache")
            return payload

        return _cached_response(request, run, "score", cell,
                                payloads.score_key(), payloads.history_token(history), build)


def api_forecast(
    request: Request,
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
    days: int = Query(7, ge=1, le=14, description="how many forecast days ahead to grade"),
):
    """Grade the next `days` of forecast weather against local climatology.

    Same shape as /grade (so the frontend renders it identically), but `recent`
    holds the forward forecast — furthest-out day first. The forecast is fetched
    fresh and cached only ~1 hour (see climate.get_recent_forecast) to track
    updates; the graded payload's validity is tied to that fetch stamp, so it
    expires exactly when a new forecast lands. New in API v2.
    """
    cell = grid.snap(lat, lon)

    with audit.RunAudit(endpoint="forecast", lat=round(lat, 4), lon=round(lon, 4),
                        recent_days=days, cell_id=cell["id"]) as run:
        history, _, fc = _fetch_history(run, cell, recent_too=True, recent_phase="forecast")
        today = datetime.date.today()

        def build(place):
            payload = payloads.build_forecast(cell, days, history, fc, today, place, run)
            run.set(run_type="partial")
            return payload

        return _cached_response(request, run, "forecast", cell,
                                payloads.forecast_key(today, days),
                                payloads.recent_token(history, cell["id"]), build)


def api_cell(
    request: Request,
    lat: float = Query(..., ge=-90, le=90),
    lon: float = Query(..., ge=-180, le=180),
    prefetch: int = Query(0, ge=0, le=1,
                          description="1 = warm-only: never fetch weather upstream; 204 for a cold cell"),
    neighbors: int = Query(0, ge=0, le=1,
                           description="1 = also warm the 8 surrounding cells in the background "
                                       "(warm-only: cold neighbors are skipped, no upstream quota)"),
):
    """One bundle carrying every view's payload for a cell, so the frontend warms
    all views with a single request instead of four.

    Each slice is the EXACT payload its per-view endpoint returns — built by the
    same builders and cached under the same derived-store keys/tokens — paired
    with the etag that endpoint would emit. The client seeds its per-view cache
    from the slices and later revalidates each view individually with
    If-None-Match, so the bundle and the per-view endpoints stay one cache.

    prefetch=1 is the neighbor-warming mode with a hard guarantee: it never
    spends weather-API quota. A cell with no cached archive answers 204 (no
    body), and only the history-derived slices (calendar + latest-day detail)
    are built — grading recent/forecast days needs the hourly upstream bundle.
    Reverse geocoding makes at most one Nominatim call for a never-labeled cell;
    the client staggers neighbor prefetches to respect that service. New in API v2.
    """
    cell = grid.snap(lat, lon)
    today = datetime.date.today()

    with audit.RunAudit(endpoint="cell", lat=round(lat, 4), lon=round(lon, 4),
                        cell_id=cell["id"], prefetch=bool(prefetch)) as run:
        recent = None
        if prefetch:
            history = climate.load_cached_history(cell)
            if history is None or history.is_empty():
                run.set(run_type="cold-skip")
                return Response(status_code=204)
            cache_meta = {"cached": True}
        else:
            history, cache_meta, recent = _fetch_history(run, cell, recent_too=True)

        cid = cell["id"]
        last = history["date"].max()

        if neighbors:
            _enqueue_neighbor_warming(cell)

        with run.phase("reverse_geocode"):
            place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"])

        def slice_for(kind: str, key: str, token: str, build) -> dict:
            """The endpoint's derived row (or build + store it), paired with the
            etag that endpoint would emit for the same request."""
            etag = _etag_for(kind, cid, key, token)
            data = store.get_json(kind, cid, key, token)
            if data is None:
                data = build()
                store.put_payload(kind, cid, key, token, data)
            return {"etag": etag, "data": data}

        slices = {}
        hist_token = payloads.history_token(history)
        # Calendar: the default last-24-months span — what the calendar view (and
        # the cross-view prefetch) requests first.
        start_ts, end_ts = payloads.cal_span(history, None, None, 24)
        slices["calendar"] = slice_for(
            "calendar", payloads.calendar_key(start_ts, end_ts, 24), hist_token,
            lambda: payloads.build_calendar(cell, history, start_ts, end_ts, 24, place, run))

        if prefetch:
            # Latest archived day: its observation comes from history alone, so
            # it's buildable without the (skipped) recent bundle.
            slices["day"] = slice_for(
                "day", payloads.day_key(last), hist_token,
                lambda: payloads.build_day(cell, history, last, place, run))
        else:
            rf_token = payloads.recent_token(history, cid)
            slices["grade"] = slice_for(
                "grade", payloads.grade_key(today, 14, 14), rf_token,
                lambda: payloads.build_grade(cell, today, 14, history, recent, cache_meta, place, run, after=14))
            slices["forecast"] = slice_for(
                "forecast", payloads.forecast_key(today, 7), rf_token,
                lambda: payloads.build_forecast(cell, 7, history, recent, today, place, run))
            slices["day"] = slice_for(
                "day", payloads.day_key(today), payloads.day_token(history, today),
                lambda: payloads.build_day(cell, history, today, place, run, recent=recent))

        # The bundle's identity is the combination of its slices' identities.
        etag = _etag_for("cell", cid, "bundle" + (":p" if prefetch else ""),
                         "|".join(s["etag"] for s in slices.values()))
        if _not_modified(request, etag):
            run.set(run_type="cache", not_modified=True)
            return Response(status_code=304, headers={"ETag": etag})
        run.set(run_type="bundle", slices=sorted(slices))
        payload = {
            "api_version": "v2",
            "cell": cell,
            "place": place,
            "today": today.isoformat(),
            "slices": slices,
        }
        # Not stored as its own derived row — the slices already are.
        return _json_response(_encode(payload), etag)


# --- ops metrics (gated) ---------------------------------------------------
def _metrics_allowed(request: Request) -> bool:
    """Guard the ops metrics: a valid token allows from anywhere (for SSH tunnels);
    with no token it's direct-loopback-only, and any forwarded header (i.e. proxied
    via Caddy) is refused — so prod ops data never leaks through the public domain."""
    token = os.environ.get("THERMOGRAPH_METRICS_TOKEN", "").strip()
    if token:
        supplied = request.headers.get("x-metrics-token") or request.query_params.get("token") or ""
        if hmac.compare_digest(supplied, token):
            return True
    client = request.client.host if request.client else None
    forwarded = any(h in request.headers for h in
                    ("x-forwarded-for", "x-forwarded-host", "x-forwarded-proto"))
    return client in ("127.0.0.1", "::1") and not forwarded


def api_metrics(request: Request):
    """Live in-process traffic counters + worker facts for the ops dashboard."""
    if not _metrics_allowed(request):
        raise HTTPException(status_code=404)   # 404, not 403 — don't reveal the route
    snap = metrics.snapshot()
    snap["warm_queue_depth"] = _warm_queue.qsize()
    snap["warm_seen"] = len(_WARM_SEEN)
    snap["threads"] = threading.active_count()
    snap["thread_names"] = sorted(t.name for t in threading.enumerate())
    return snap


async def api_event(request: Request) -> Response:
    """Record one product event (a tap on "use my location", a digest signup, …).

    Deliberately minimal: the body carries only an allowlisted event name. The
    referrer is read from this request's own Referer header — a client-supplied
    one would be forgeable and buys nothing — and is reduced to a bare domain.
    No cookies, no per-visitor id, no full URLs.

    Always answers 204, whether the event was counted, rejected by the allowlist,
    or dropped by the rate limiter, so probing the endpoint reveals nothing. That
    also suits navigator.sendBeacon, which ignores the response.
    """
    name = ""
    try:
        body = await request.json()
        if isinstance(body, dict):
            name = str(body.get("event") or "")
    except Exception:  # noqa: BLE001 - a junk body is just a no-op event
        pass
    if name:
        await run_in_threadpool(
            metrics.record_event,
            name,
            referer=request.headers.get("referer"),
            host=request.headers.get("host"),
            ip=_client_ip(request),
        )
    return Response(status_code=204)


# --- API versioning --------------------------------------------------------
# Non-backward-compatible additions land in a new version; every version stays
# mounted and served simultaneously. v1 is the original contract (grade/geocode);
# v2 adds the calendar. The unversioned /api/* paths are kept as aliases of v1 so
# existing clients keep working.
v1 = APIRouter(tags=["v1"])
v1.add_api_route("/geocode", api_geocode, methods=["GET"])
v1.add_api_route("/grade", api_grade, methods=["GET"])

v2 = APIRouter(tags=["v2"])
v2.add_api_route("/geocode", api_geocode, methods=["GET"])
v2.add_api_route("/suggest", api_suggest, methods=["GET"])
v2.add_api_route("/place", api_place, methods=["GET"])
v2.add_api_route("/grade", api_grade, methods=["GET"])
v2.add_api_route("/calendar", api_calendar, methods=["GET"])
v2.add_api_route("/day", api_day, methods=["GET"])
v2.add_api_route("/score", api_score, methods=["GET"])
v2.add_api_route("/forecast", api_forecast, methods=["GET"])
v2.add_api_route("/cell", api_cell, methods=["GET"])
v2.add_api_route("/metrics", api_metrics, methods=["GET"], include_in_schema=False)
v2.add_api_route("/event", api_event, methods=["POST"], include_in_schema=False)

app.include_router(v1, prefix=f"{BASE}/api")        # legacy unversioned == v1 (backward compatible)
app.include_router(v1, prefix=f"{BASE}/api/v1")
app.include_router(v2, prefix=f"{BASE}/api/v2")
# SSR content API (repo-split Stage 2), now consumed by the frontend_ssr
# service (Stage 3) instead of anything in this process. See
# backend/api/content_routes.py.
app.include_router(content_routes.router, prefix=f"{BASE}/api/v2")
# Internal control surface for the thermograph-daemon Go binary (the gateway
# bot + recurring jobs that used to run in _lifespan). Deliberately NOT under
# BASE — the daemon hits THERMOGRAPH_API_BASE_INTERNAL directly, same posture
# as /healthz — and never publicly routed (the Caddyfile only forwards
# /api/*, /digest and /discord/interactions here). Registered before the
# catch-all frontend proxy below so /internal/* can't fall through to it.
app.include_router(internal_routes.router)


# --- proxy to the frontend SSR service (repo-split Stage 4) ------------------
_PROXY_EXCLUDED_REQUEST_HEADERS = {"host", "content-length"}
_PROXY_EXCLUDED_RESPONSE_HEADERS = {"content-encoding", "content-length", "transfer-encoding", "connection"}


async def _proxy_to_frontend(request: Request) -> Response:
    """Forward a GET/HEAD request verbatim to the frontend SSR service and relay
    its response back untouched (status, body, headers -- including ETag, so
    frontend's own If-None-Match/304 logic keeps working through this hop).
    Every frontend-owned path (content.register()'s old routes) is read-only,
    so this never needs to forward a request body.

    The primary same-origin mechanism is host Caddy in prod/beta -- it
    path-splits directly to backend vs frontend, so this proxy is never
    actually exercised there. This is the fallback for every environment with
    no Caddy in front (LAN dev, bare-metal run.sh) -- see the repo-split
    plan's Stage 4 section."""
    headers = {k: v for k, v in request.headers.items()
              if k.lower() not in _PROXY_EXCLUDED_REQUEST_HEADERS}
    # Host is deliberately excluded above (it'd otherwise target this internal
    # hop, not the browser-facing address) -- forward it as X-Forwarded-Host
    # instead, so frontend_ssr's own _origin() can still build correct
    # absolute URLs (canonical, og:url, asset_base_url) instead of resolving
    # to its own internal address. Chain through an existing value first, the
    # same precedence _origin() itself uses for the proto.
    headers["x-forwarded-host"] = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc
    headers["x-forwarded-proto"] = request.headers.get("x-forwarded-proto") or request.url.scheme
    upstream = await _frontend_client.request(
        request.method, request.url.path, params=request.url.query, headers=headers)
    resp_headers = {k: v for k, v in upstream.headers.items()
                   if k.lower() not in _PROXY_EXCLUDED_RESPONSE_HEADERS}
    return Response(content=upstream.content, status_code=upstream.status_code, headers=resp_headers)


# The catch-all fallback registered at the very end of this file (after every
# backend route) proxies everything else -- content pages, the IndexNow key
# file, static assets, the SPA shells -- to frontend. See it there for why a
# single fallback replaced what used to be an enumerated path list (repo-split
# Stage 7a).


# --- Discord slash commands (HTTP interactions) ----------------------------
async def discord_interactions(request: Request) -> Response:
    """Discord posts each slash-command interaction here. Verification runs over the
    RAW body, so this must read bytes, not request.json()."""
    raw = await request.body()
    # handle() does real work for /grade (sync parquet/DB reads + grading math),
    # not just the cheap signature check — keep it off the event loop.
    status, payload = await run_in_threadpool(
        discord_interactions_mod.handle,
        raw,
        request.headers.get("x-signature-ed25519", ""),
        request.headers.get("x-signature-timestamp", ""),
    )
    return Response(json.dumps(payload), status_code=status, media_type="application/json")

app.add_api_route(f"{BASE}/discord/interactions", discord_interactions,
                  methods=["POST"], include_in_schema=False)

# --- accounts (fastapi-users) ----------------------------------------------
# Library-provided routers: /auth/login + /auth/logout (cookie session),
# /auth/register (signup), /auth/verify + /auth/request-verify-token (email
# confirmation), and /users/me (session check / profile). All under the same
# v2 prefix as the rest of the API, so the frontend calls them base-relative.
app.include_router(
    users.fastapi_users.get_auth_router(users.auth_backend),
    prefix=f"{BASE}/api/v2/auth", tags=["auth"],
)
app.include_router(
    users.fastapi_users.get_register_router(UserRead, UserCreate),
    prefix=f"{BASE}/api/v2/auth", tags=["auth"],
)
app.include_router(
    users.fastapi_users.get_verify_router(UserRead),
    prefix=f"{BASE}/api/v2/auth", tags=["auth"],
)
app.include_router(
    users.fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix=f"{BASE}/api/v2/users", tags=["users"],
)
# Subscriptions + notifications (authed, user-scoped).
app.include_router(api_accounts.router, prefix=f"{BASE}/api/v2")
# Sign in with / connect an external provider (Discord, Google).
app.include_router(oauth.router, prefix=f"{BASE}/api/v2")
# Discord's legacy /discord/* paths: the callback URL registered in Discord's
# portal, plus what the currently-deployed frontend calls. Mounted AFTER the
# generic router so neither shadows the other — the paths are disjoint, and
# /oauth/{provider}/... would otherwise be a candidate match for /discord/....
app.include_router(discord_link.router, prefix=f"{BASE}/api/v2")


# The un-slashed base redirects to BASE/ so the frontend's relative asset URLs
# resolve correctly. Under a sub-path this keeps the app scoped to BASE (never
# claiming "/", so the domain root stays free for another app via the proxy). When
# BASE is "" the app owns the root: "/" is already the index route below, so there
# is no bare base to redirect (and "" is not a valid route path).
if BASE:
    app.add_api_route(BASE, lambda: RedirectResponse(url=f"{BASE}/"), methods=["GET", "HEAD"], include_in_schema=False)

# Monthly-digest signup (the footer form on every page). Answers the same generic
# message whatever the outcome, so it can't be probed for whether an address is
# already subscribed.
app.add_api_route(f"{BASE}/digest", digest.signup, methods=["POST"], include_in_schema=False)

# --- everything else: proxy to frontend (repo-split Stage 7a) ---------------
# Frontend now owns every page (the SSR content pages, and the interactive
# tool's SPA shells), every static asset (app.js, style.css, manifest,
# favicons, …), and the dynamic IndexNow key file -- backend serves none of
# that itself anymore (it used to read them from a sibling frontend/
# directory; see _proxy_to_frontend's own docstring for why that stops being
# true once backend and frontend are separate repos). A single catch-all,
# registered dead last so every real backend route above already won, proxies
# anything unmatched -- simpler than (and a superset of) the enumerated path
# list this replaced, since static asset filenames aren't enumerable the way
# content pages are. Caddy path-splits directly to whichever service owns a
# path in prod/beta (deploy/Caddyfile), so this is only ever exercised as the
# fallback for environments with no Caddy in front (LAN dev, bare-metal
# run.sh) -- same role _proxy_to_frontend has always had.
app.add_api_route(f"{BASE}/{{path:path}}", _proxy_to_frontend, methods=["GET", "HEAD"], include_in_schema=False)