81135c1be5
All checks were successful
secrets-guard / encrypted (pull_request) Successful in 6s
PR build (required check) / changes (pull_request) Successful in 11s
shell-lint / shellcheck (pull_request) Successful in 8s
PR build (required check) / validate-observability (pull_request) Has been skipped
PR build (required check) / build-frontend (pull_request) Successful in 57s
PR build (required check) / build-backend (pull_request) Successful in 1m25s
PR build (required check) / gate (pull_request) Successful in 1s
Adds Google as a second identity provider. Rather than a second copy of the Discord flow, the flow itself moves to accounts/oauth.py and both providers become configurations of it — account resolution is the security-critical part, and a parallel hand-rolled copy is where a subtle divergence becomes a takeover. A third provider is now a PROVIDERS entry and nothing else. Identity moves to an oauth_account table keyed (provider, subject), so a login resolves on the provider's own stable id rather than an email that can be changed or reassigned. user.discord_id deliberately stays: it is the DM delivery address notify.py reads, not merely an identity, and the Discord link keeps writing it. discord_only becomes oauth_only, and the lockout guard now refuses to unlink the *last* provider from a password-less account rather than singling out Discord — with two linked, either may go. Migration 0005 renames the flag and backfills an oauth_account row for every existing discord_id. Without that backfill an already-linked user would stop being recognised at login and would silently get a second account on their next sign-in. It also drops a vestigial discord_only that 0003 re-adds on a database whose 0001 already built oauth_only from current metadata, which otherwise left fresh and upgraded databases with different schemas. Compatibility, since the frontend deploys independently of this service: /discord/link/callback keeps answering because that URL is registered in Discord's developer portal, the other /discord/* routes stay because the deployed frontend calls them, the callback still emits ?discord= alongside ?oauth=, and UserRead still carries discord_only as a computed alias. Google needs THERMOGRAPH_GOOGLE_CLIENT_ID/_CLIENT_SECRET and its own registered redirect URI; unset, it reports disabled and shows no UI.
1027 lines
50 KiB
Python
1027 lines
50 KiB
Python
"""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)
|