Merge pull request 'Promote main to release: geocode P0 fix + ops tooling + privacy logging' (#53) from main into release
All checks were successful
secrets-guard / encrypted (push) Successful in 9s
shell-lint / shellcheck (push) Successful in 8s
Build + push backend image (Forgejo registry) / build-push (push) Successful in 47s
Deploy backend to prod VPS / deploy (push) Successful in 1m30s

This commit is contained in:
emi 2026-07-24 19:53:19 +00:00
commit 4daa10c76d
20 changed files with 1392 additions and 73 deletions

View file

@ -11,7 +11,7 @@ from fastapi import APIRouter, HTTPException, Request, Response
from api import content_payloads as payloads from api import content_payloads as payloads
from api import sitemap as sitemap_mod from api import sitemap as sitemap_mod
from api.payloads import history_token from api.payloads import content_token
from data import climate from data import climate
from data import cities from data import cities
from data import grid from data import grid
@ -58,13 +58,21 @@ def _cached(request: Request, kind: str, cell_id: str, key: str, token: str, bui
return _json_response(store.put_payload(kind, cell_id, key, token, payload), etag) return _json_response(store.put_payload(kind, cell_id, key, token, payload), etag)
def _resolve_city(slug: str): def _city_cell(slug: str):
"""(city, cell, history) for a slug, or raise 404 (unknown) / 503 (warming) -- """(city, cell) for a slug, or raise 404 — both steps are cheap (a dict
mirrors web/content.py's _resolve_city.""" lookup and a pure grid snap) and, crucially, load no history, so the
derived-store token can be computed and a cache hit served without ever
touching the ~45-year archive."""
city = cities.get(slug) city = cities.get(slug)
if city is None: if city is None:
raise HTTPException(status_code=404, detail="Unknown city.") raise HTTPException(status_code=404, detail="Unknown city.")
cell = grid.snap(city["lat"], city["lon"]) return city, grid.snap(city["lat"], city["lon"])
def _load_history(cell):
"""The cell's ~45-year archive, or raise 503 while it's still warming --
the expensive load that used to run on every request. Now called only from
the content handlers' build() closures, i.e. only on a derived-store miss."""
history = climate.load_cached_history(cell) history = climate.load_cached_history(cell)
if history is None or history.is_empty(): if history is None or history.is_empty():
try: try:
@ -73,7 +81,7 @@ def _resolve_city(slug: str):
history = None history = None
if history is None or history.is_empty(): if history is None or history.is_empty():
raise HTTPException(status_code=503, detail="Climate data is warming up; please retry shortly.") raise HTTPException(status_code=503, detail="Climate data is warming up; please retry shortly.")
return city, cell, history return history
def _origin(request: Request) -> str: def _origin(request: Request) -> str:
@ -105,16 +113,17 @@ def content_home(request: Request):
@router.get("/content/city/{slug}") @router.get("/content/city/{slug}")
def content_city(request: Request, slug: str): def content_city(request: Request, slug: str):
city, cell, history = _resolve_city(slug) city, cell = _city_cell(slug)
recent = None token = content_token(cell["id"])
try:
recent = climate.get_recent_forecast(cell)
except Exception: # noqa: BLE001 - today_vs_normal degrades to None
pass
token = history_token(history)
origin = _origin(request) origin = _origin(request)
def build(): def build():
history = _load_history(cell)
recent = None
try:
recent = climate.get_recent_forecast(cell)
except Exception: # noqa: BLE001 - today_vs_normal degrades to None
pass
return payloads.city_payload(origin, BASE, city, history, recent) return payloads.city_payload(origin, BASE, city, history, recent)
# origin is folded into the cache key (not just slug) because the payload's # origin is folded into the cache key (not just slug) because the payload's
@ -131,10 +140,11 @@ def content_city(request: Request, slug: str):
def content_city_month(request: Request, slug: str, month: str): def content_city_month(request: Request, slug: str, month: str):
if month not in payloads.MONTH_INDEX: if month not in payloads.MONTH_INDEX:
raise HTTPException(status_code=404, detail="Unknown month.") raise HTTPException(status_code=404, detail="Unknown month.")
city, cell, history = _resolve_city(slug) city, cell = _city_cell(slug)
token = history_token(history) token = content_token(cell["id"])
def build(): def build():
history = _load_history(cell)
return payloads.month_payload(BASE, city, history, payloads.MONTH_INDEX[month]) return payloads.month_payload(BASE, city, history, payloads.MONTH_INDEX[month])
return _cached(request, "content-month", cell["id"], f"{slug}:{month}", token, build) return _cached(request, "content-month", cell["id"], f"{slug}:{month}", token, build)
@ -142,11 +152,12 @@ def content_city_month(request: Request, slug: str, month: str):
@router.get("/content/city/{slug}/records") @router.get("/content/city/{slug}/records")
def content_city_records(request: Request, slug: str): def content_city_records(request: Request, slug: str):
city, cell, history = _resolve_city(slug) city, cell = _city_cell(slug)
token = history_token(history) token = content_token(cell["id"])
origin = _origin(request) origin = _origin(request)
def build(): def build():
history = _load_history(cell)
return payloads.records_payload(origin, BASE, city, history) return payloads.records_payload(origin, BASE, city, history)
# See content_city's comment: origin folded into the cache key for the # See content_city's comment: origin folded into the cache key for the

View file

@ -73,6 +73,30 @@ def history_token(history) -> str:
return f"{PAYLOAD_VER}:{hist_end(history)}" return f"{PAYLOAD_VER}:{hist_end(history)}"
# The content-shape version for the SEO content pages (/climate/<city>[/month|
# /records]). Kept separate from PAYLOAD_VER so a content-only shape change need
# not orphan every other kind's cache, and vice-versa. Bump on change.
CONTENT_VER = "c1"
def content_token(cell_id: str) -> str:
"""Validity for the SEO content-page payloads — cheap AND stable.
Unlike history_token, this never loads the ~45-year archive: it keys on the
cell's newest archived DATE (climate.history_max_date — an indexed
``MAX(date)`` on Postgres, a single-column read on the parquet backend), so it
survives the hourly tail top-ups (which only refresh intra-day freshness) and
turns over only when the archive's last day genuinely advances (≈1×/day). That
keeps content pages 1 day stale acceptable for SEO while sparing every
request the full-history load the old token forced even on a cache hit.
Fail-soft: a store/DB error yields a 'none' bucket rather than raising."""
try:
max_date = climate.history_max_date(cell_id)
except Exception: # noqa: BLE001 - token computation must never fail a request
max_date = None
return f"{PAYLOAD_VER}:{CONTENT_VER}:{max_date or 'none'}"
def recent_token(history, cell_id: str) -> str: def recent_token(history, cell_id: str) -> str:
"""Validity for payloads that also grade the hourly recent/forecast bundle.""" """Validity for payloads that also grade the hourly recent/forecast bundle."""
return f"{history_token(history)}:{climate.recent_stamp(cell_id)}" return f"{history_token(history)}:{climate.recent_stamp(cell_id)}"

View file

@ -96,6 +96,12 @@ def classify_inbound(path: str, base: str = "") -> str:
the app's mount prefix (e.g. ``/thermograph`` or ``""``). the app's mount prefix (e.g. ``/thermograph`` or ``""``).
""" """
p = path or "/" p = path or "/"
# The liveness probe (Dockerfile HEALTHCHECK, Caddy health_uri) is by far the
# highest-volume single path in prod (measured ~47% of the access log) and is
# never real traffic — same posture as the metrics check below. Never under
# BASE (see /healthz's own docstring in web/app.py), so check before stripping.
if p.rstrip("/") == "/healthz":
return "health"
# The dashboard polls the metrics endpoint; never count it as traffic, whatever base # The dashboard polls the metrics endpoint; never count it as traffic, whatever base
# prefix it arrives under — e.g. a `/thermograph/api/v2/metrics` probe against a # prefix it arrives under — e.g. a `/thermograph/api/v2/metrics` probe against a
# root-served prod app would otherwise land in "other" and show as an inbound error. # root-served prod app would otherwise land in "other" and show as an inbound error.
@ -118,6 +124,13 @@ def classify_inbound(path: str, base: str = "") -> str:
# category keeps record_inbound from double-counting every interaction. # category keeps record_inbound from double-counting every interaction.
if seg == "event": if seg == "event":
return "event" return "event"
# The SSR content API (backend/api/content_routes.py) is called only by
# the frontend_ssr service's own api_client.py, never by a browser — a
# server-to-server hop, not a page view. Its own category is what lets
# that (measured ~32% of the access log on prod) be excluded from the
# access log below without also hiding real external traffic.
if seg == "content":
return "internal"
return f"api:{seg}" if seg else "api:other" return f"api:{seg}" if seg else "api:other"
if p.endswith((".js", ".css", ".html", ".webmanifest", ".png", ".svg", ".ico", if p.endswith((".js", ".css", ".html", ".webmanifest", ".png", ".svg", ".ico",
".json", ".woff", ".woff2", ".map", ".txt", ".xml")): ".json", ".woff", ".woff2", ".map", ".txt", ".xml")):

View file

@ -862,6 +862,30 @@ def recent_stamp(cell_id: str) -> int:
return 0 return 0
def history_max_date(cell_id: str) -> "str | None":
"""ISO date (YYYY-MM-DD) of the cell's newest cached archived day, or None when
nothing is cached read WITHOUT loading the full multi-decade history.
Backs the content-page cache token (api/payloads.content_token). On Postgres
it's an indexed ``MAX(date)`` over climate_history; on the parquet backend it's
the max of the cached file's date column (a single columnar read via
``scan_parquet``, not a full frame load). Fail-soft: any error reads as None."""
if climate_store.is_postgres():
return climate_store.history_max_date(cell_id)
path = _cache_path(cell_id)
if not os.path.exists(path):
return None
try:
val = pl.scan_parquet(path).select(pl.col("date").max()).collect().item()
if val is None:
return None
if isinstance(val, datetime.datetime): # older files stored date as Datetime
val = val.date()
return val.isoformat()
except Exception: # noqa: BLE001 - a corrupt/absent cache reads as no max date
return None
def get_recent_forecast(cell: dict) -> pl.DataFrame: def get_recent_forecast(cell: dict) -> pl.DataFrame:
"""Recent observations + forward forecast, with humidity as absolute humidity """Recent observations + forward forecast, with humidity as absolute humidity
(g/). Thin wrapper over the raw loader (see below).""" (g/). Thin wrapper over the raw loader (see below)."""
@ -985,10 +1009,13 @@ _REVGEO_CACHE: dict[str, str | None] = {}
_REVGEO_MIN_INTERVAL = 1.1 # seconds between successive Nominatim reverse calls _REVGEO_MIN_INTERVAL = 1.1 # seconds between successive Nominatim reverse calls
_revgeo_last = 0.0 _revgeo_last = 0.0
_REVGEO_QUEUE: "queue.Queue[tuple[float, float, str, concurrent.futures.Future]]" = queue.Queue() _REVGEO_QUEUE: "queue.Queue[tuple]" = queue.Queue()
_REVGEO_WORKER_LOCK = threading.Lock() _REVGEO_WORKER_LOCK = threading.Lock()
_revgeo_worker_started = False _revgeo_worker_started = False
_REVGEO_WAIT_TIMEOUT = 10.0 # seconds a caller waits for ITS OWN request before giving up _REVGEO_WAIT_TIMEOUT = 10.0 # seconds a caller waits for ITS OWN request before giving up
_GEOCODE_WAIT_TIMEOUT = 35.0 # forward /geocode is a deliberate user search, not a map-pan
# enrichment -- worth a longer wait than reverse geocoding's,
# covering the 30s per-request timeout plus queue wait.
def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]: def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]:
@ -1041,31 +1068,46 @@ def _fetch_revgeo_label(lat: float, lon: float) -> str | None:
def _revgeo_worker() -> None: def _revgeo_worker() -> None:
"""Drains _REVGEO_QUEUE one request at a time, forever. The sole caller of """Drains _REVGEO_QUEUE one job at a time, forever — both reverse ("revgeo")
_fetch_revgeo_label, so the ~1/sec pacing below is enforced just by doing the and forward ("geocode") jobs, so the ~1/sec pacing below is enforced just by
work serially no lock needed, since nothing else ever touches Nominatim. doing the work serially on this one thread, no lock needed, since nothing
Re-checks the cache before fetching (a request queued behind an identical one else ever touches Nominatim. Reverse jobs re-check the cache before fetching
is answered from what the earlier request just cached, no duplicate call), (a request queued behind an identical one is answered from what the earlier
and always finishes the fetch + persists the result even if the original request just cached, no duplicate call) and always finish + persist even if
caller already gave up waiting (see reverse_geocode's timeout).""" the original caller already gave up waiting (see reverse_geocode's timeout).
Forward jobs have no cache (see geocode_nominatim) but share the same pacer."""
global _revgeo_last global _revgeo_last
while True: while True:
lat, lon, key, fut = _REVGEO_QUEUE.get() job = _REVGEO_QUEUE.get()
kind = job[0]
try: try:
found, label = reverse_geocode_cached(lat, lon) if kind == "revgeo":
if not found: _, lat, lon, key, fut = job
found, label = reverse_geocode_cached(lat, lon)
if not found:
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
if wait > 0:
time.sleep(wait)
label = _fetch_revgeo_label(lat, lon)
_revgeo_last = time.monotonic()
_REVGEO_CACHE[key] = label
store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL)
if not fut.done():
fut.set_result(label)
else: # "geocode"
_, name, count, fut = job
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last) wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
if wait > 0: if wait > 0:
time.sleep(wait) time.sleep(wait)
label = _fetch_revgeo_label(lat, lon) try:
_revgeo_last = time.monotonic() results = _fetch_geocode_forward(name, count)
_REVGEO_CACHE[key] = label finally:
store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL) _revgeo_last = time.monotonic()
if not fut.done(): if not fut.done():
fut.set_result(label) fut.set_result(results)
except Exception: # noqa: BLE001 - never let a bad request kill the worker except Exception: # noqa: BLE001 - never let a bad request kill the worker
if not fut.done(): if not fut.done():
fut.set_result(None) fut.set_result(None if kind == "revgeo" else [])
finally: finally:
_REVGEO_QUEUE.task_done() _REVGEO_QUEUE.task_done()
@ -1105,44 +1147,28 @@ def reverse_geocode(lat: float, lon: float) -> str | None:
return label return label
_start_revgeo_worker() _start_revgeo_worker()
fut: "concurrent.futures.Future[str | None]" = concurrent.futures.Future() fut: "concurrent.futures.Future[str | None]" = concurrent.futures.Future()
_REVGEO_QUEUE.put((lat, lon, key, fut)) _REVGEO_QUEUE.put(("revgeo", lat, lon, key, fut))
try: try:
return fut.result(timeout=_REVGEO_WAIT_TIMEOUT) return fut.result(timeout=_REVGEO_WAIT_TIMEOUT)
except concurrent.futures.TimeoutError: except concurrent.futures.TimeoutError:
return None return None
def geocode_nominatim(name: str, count: int = 5) -> list[dict]: def _fetch_geocode_forward(name: str, count: int) -> list[dict]:
"""Forward place-name lookup via OpenStreetMap Nominatim (keyless). """The actual Nominatim forward-search HTTP call + result shaping. Called
ONLY from _revgeo_worker (never on a caller's thread), mirroring
Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the _fetch_revgeo_label but unlike that function, a bad response is allowed to
local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population raise here; _revgeo_worker's own except clause turns it into an empty list,
villages, and alternate/native-language spellings so /geocode falls back to same net effect without a second layer of exception-swallowing."""
it on a local miss. It carries no population, so results keep Nominatim's own r = _request(
relevance order; name/admin/country map straight across. "https://nominatim.openstreetmap.org/search",
{"q": name, "format": "jsonv2", "addressdetails": 1,
Shares the reverse geocoder's lock and ~1/sec pacing: both hit the same host, "limit": count, "accept-language": "en"},
so serializing them together keeps total Nominatim traffic under the usage 30,
policy. Only the low-volume /geocode miss path reaches here autocomplete phase="geocode",
(/suggest) is served purely from the local index and never calls out. headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"},
""" )
global _revgeo_last rows = r.json() or []
with _REVGEO_LOCK:
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
if wait > 0:
time.sleep(wait)
try:
r = _request(
"https://nominatim.openstreetmap.org/search",
{"q": name, "format": "jsonv2", "addressdetails": 1,
"limit": count, "accept-language": "en"},
30,
phase="geocode",
headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"},
)
rows = r.json() or []
finally:
_revgeo_last = time.monotonic()
out = [] out = []
for g in rows: for g in rows:
a = g.get("address", {}) or {} a = g.get("address", {}) or {}
@ -1163,3 +1189,28 @@ def geocode_nominatim(name: str, count: int = 5) -> list[dict]:
"population": None, # Nominatim has no population; order is relevance-based "population": None, # Nominatim has no population; order is relevance-based
}) })
return out return out
def geocode_nominatim(name: str, count: int = 5) -> list[dict]:
"""Forward place-name lookup via OpenStreetMap Nominatim (keyless).
Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the
local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population
villages, and alternate/native-language spellings so /geocode falls back to
it on a local miss. It carries no population, so results keep Nominatim's own
relevance order; name/admin/country map straight across.
Shares the reverse geocoder's worker thread and ~1/sec pacing: both hit the
same host, and both jobs are drained serially by the one thread in
_revgeo_worker, so there is exactly one writer of _revgeo_last no lock
needed, same reasoning as reverse_geocode. Only the low-volume /geocode miss
path reaches here autocomplete (/suggest) is served purely from the local
index and never calls out.
"""
_start_revgeo_worker()
fut: "concurrent.futures.Future[list[dict]]" = concurrent.futures.Future()
_REVGEO_QUEUE.put(("geocode", name, count, fut))
try:
return fut.result(timeout=_GEOCODE_WAIT_TIMEOUT) or []
except concurrent.futures.TimeoutError:
return []

View file

@ -188,6 +188,29 @@ def recent_synced_at(cell_id: str) -> float:
return 0.0 return 0.0
def history_max_date(cell_id: str) -> "str | None":
"""The ISO date (YYYY-MM-DD) of the cell's newest archived day, or None when
there is no cached history (or Postgres is off/unreachable).
Backs the content-page cache token (api/payloads.content_token), so it must be
CHEAP an indexed ``MAX(date)`` over the ``(cell_id, date)`` primary key (see
the 0002 migration), never a full-history load. Fail-soft: any error reads as
None, so the caller falls back to a 'none' bucket rather than raising."""
try:
with _conn() as conn:
if conn is None:
return None
row = conn.execute(
"SELECT MAX(date) FROM climate_history WHERE cell_id = %s",
(cell_id,),
).fetchone()
if not row or row[0] is None:
return None
return row[0].isoformat()
except Exception: # noqa: BLE001
return None
def cols_version(cell_id: str) -> int: def cols_version(cell_id: str) -> int:
"""The stored schema version for a cell's history; 0 if absent.""" """The stored schema version for a cell's history; 0 if absent."""
try: try:

View file

@ -85,4 +85,7 @@ if [ "${RUN_MIGRATIONS:-1}" != "0" ]; then
fi fi
echo "==> Starting uvicorn on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)" echo "==> Starting uvicorn on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)"
exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}" # --no-access-log: the app's own request-logging middleware (audit.log_access,
# web/app.py) already writes a structured line per request; uvicorn's own access
# log just duplicated every one of them for no benefit.
exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}" --no-access-log

View file

@ -450,6 +450,37 @@ def _maybe_refresh_homepage() -> None:
pass pass
# Pre-warming the SEO content derived-store rides this loop too, for the same
# reason the homepage sweep does: it's leader-only (the loop runs on the single
# elected leader) and cache-only, so it spends no upstream quota. The content
# token turns over when a cell's archive gains a day (~1x/day), so after each
# advance the first request to a /climate page would otherwise recompute a
# 45-year payload cold; warming rebuilds those rows off-request. It's paced and
# capped per tick (CONTENT_WARM_MAX_CITIES) so one tick can't stall the notifier,
# and the idempotent skip means a tick after everything is fresh is a cheap
# no-op — the ~1000-city set refreshes across a handful of ticks, well inside the
# <=1-day staleness the content token already tolerates.
CONTENT_WARM_INTERVAL = float(os.environ.get("THERMOGRAPH_CONTENT_WARM_INTERVAL", "1800")) # 30 min
CONTENT_WARM_MAX_CITIES = int(os.environ.get("THERMOGRAPH_CONTENT_WARM_MAX_CITIES", "50"))
_last_content_warm = 0.0
def _maybe_warm_content() -> None:
global _last_content_warm
now = time.time()
if now - _last_content_warm < CONTENT_WARM_INTERVAL:
return
_last_content_warm = now
try:
# Local import: warm_cities is a script-style top-level module (mirrors
# api/internal_routes.py's warm-cities job), so only the leader pays its
# import cost, and it's kept off notify.py's import graph.
import warm_cities
warm_cities.warm_content(limit=CONTENT_WARM_MAX_CITIES)
except Exception: # noqa: BLE001 - warming is best-effort; pages self-heal on request
pass
# The daily "most unusual right now" post to Discord rides this loop too: once per # The daily "most unusual right now" post to Discord rides this loop too: once per
# day, after the feed is refreshed, leader-only like everything in run_loop. No # day, after the feed is refreshed, leader-only like everything in run_loop. No
# webhook configured => no-op. # webhook configured => no-op.
@ -489,6 +520,7 @@ def run_loop():
try: try:
run_pass() run_pass()
_maybe_refresh_homepage() _maybe_refresh_homepage()
_maybe_warm_content()
_maybe_post_discord() _maybe_post_discord()
except Exception: # noqa: BLE001 - a bad iteration must never kill the loop except Exception: # noqa: BLE001 - a bad iteration must never kill the loop
pass pass

View file

@ -0,0 +1,130 @@
"""Route tests for the SSR content JSON API (api/content_routes.py).
The point of these is the latency fix: a derived-store cache HIT must be O(1)
it must NOT load the ~45-year archive. So the weather layer is faked with call
counters, and the core assertion is that a second identical request loads the
history zero more times than the first. ETag/304 revalidation and the 404/payload
shapes are covered alongside, against a fresh per-test store.
"""
import pytest
from fastapi.testclient import TestClient
from web import app as appmod
from data import climate
from data import cities
BASE = "/thermograph/api/v2/content"
CITY = {
"slug": "testville",
"name": "Testville",
"admin1": "Washington",
"country": "United States",
"country_code": "US",
"lat": 47.6062,
"lon": -122.3321,
"population": 100000,
}
@pytest.fixture
def counts():
return {"load_cached_history": 0, "get_history": 0}
@pytest.fixture
def client(monkeypatch, history, recent, counts, tmp_store):
"""TestClient with the weather + city layer faked and the history loaders
wrapped in counters. tmp_store gives each test a fresh derived store so the
miss->hit sequence is deterministic."""
def load_cached_history(cell):
counts["load_cached_history"] += 1
return history.clone()
def get_history(cell):
counts["get_history"] += 1
return history.clone(), {"cached": True, "cache_age_days": 3}
monkeypatch.setattr(climate, "load_cached_history", load_cached_history)
monkeypatch.setattr(climate, "get_history", get_history)
monkeypatch.setattr(climate, "get_recent_forecast", lambda cell: recent.clone())
# Stable token: content_token() -> payloads -> climate.history_max_date.
monkeypatch.setattr(climate, "history_max_date", lambda cell_id: "2026-07-01")
monkeypatch.setattr(climate, "reverse_geocode", lambda lat, lon: "Testville, Washington")
# Only "testville" is a known city; anything else -> 404.
monkeypatch.setattr(cities, "get", lambda slug: CITY if slug == "testville" else None)
return TestClient(appmod.app)
# --- the regression guard: a cache hit loads no history ----------------------
def test_cache_hit_does_not_load_history(client, counts):
"""First request misses -> builds -> loads history exactly once. A second
identical request is served from the derived store and loads history ZERO
more times. This is the whole point of the fix."""
r1 = client.get(f"{BASE}/city/testville")
assert r1.status_code == 200
assert counts["load_cached_history"] == 1
assert counts["get_history"] == 0 # cached load was non-empty, no live fetch
r2 = client.get(f"{BASE}/city/testville")
assert r2.status_code == 200
assert r2.content == r1.content # replayed from the store
assert counts["load_cached_history"] == 1 # unchanged: no history load on the hit
assert counts["get_history"] == 0
def test_304_revalidation_loads_no_history(client, counts):
"""If-None-Match on a stored payload -> 304 with no history load."""
r1 = client.get(f"{BASE}/city/testville")
assert r1.status_code == 200
etag = r1.headers["etag"]
before = counts["load_cached_history"]
r304 = client.get(f"{BASE}/city/testville", headers={"If-None-Match": etag})
assert r304.status_code == 304
assert r304.headers["etag"] == etag
assert counts["load_cached_history"] == before # 304 loaded nothing
def test_month_and_records_hits_skip_history_load(client, counts):
"""Same O(1)-hit guarantee for the month and records handlers."""
for path in ("month/july", "records"):
counts["load_cached_history"] = 0
assert client.get(f"{BASE}/city/testville/{path}").status_code == 200
assert counts["load_cached_history"] == 1
assert client.get(f"{BASE}/city/testville/{path}").status_code == 200
assert counts["load_cached_history"] == 1 # second call is a store hit
# --- payload shapes ----------------------------------------------------------
def test_city_payload_shape(client):
body = client.get(f"{BASE}/city/testville").json()
assert body["city"]["slug"] == "testville"
assert body["canonical_path"] == "/climate/testville"
assert len(body["months"]) == 12
assert body["today_vs_normal"] is not None # recent forecast folded in
def test_month_payload_shape(client):
body = client.get(f"{BASE}/city/testville/month/july").json()
assert body["month_slug"] == "july"
assert body["canonical_path"] == "/climate/testville/july"
def test_records_payload_shape(client):
body = client.get(f"{BASE}/city/testville/records").json()
assert body["canonical_path"] == "/climate/testville/records"
# --- 404s --------------------------------------------------------------------
def test_unknown_slug_is_404(client, counts):
for path in ("city/nope", "city/nope/month/july", "city/nope/records"):
assert client.get(f"{BASE}/{path}").status_code == 404
assert counts["load_cached_history"] == 0 # never reached the archive
def test_unknown_month_is_404(client):
assert client.get(f"{BASE}/city/testville/month/smarch").status_code == 404

View file

@ -39,6 +39,34 @@ def test_cache_identity_formats_are_pinned(history):
assert payloads.history_token(history) == f"{payloads.PAYLOAD_VER}:{payloads.hist_end(history)}" assert payloads.history_token(history) == f"{payloads.PAYLOAD_VER}:{payloads.hist_end(history)}"
def test_content_token_is_stable_across_tail_topups(monkeypatch):
"""The content token keys on the archive's newest DATE, not its row count/mtime,
so hourly tail top-ups that don't change the max date leave it unchanged — while
a genuinely-advanced last day turns it over."""
date_box = {"v": "2026-06-15"}
monkeypatch.setattr(climate, "history_max_date", lambda cid: date_box["v"])
first = payloads.content_token("1_2")
assert first == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:2026-06-15"
# Simulated hourly top-ups: same max date -> identical token every time.
assert payloads.content_token("1_2") == first
assert payloads.content_token("1_2") == first
# The archive's last day advances -> the token turns over.
date_box["v"] = "2026-06-16"
assert payloads.content_token("1_2") != first
assert payloads.content_token("1_2") == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:2026-06-16"
def test_content_token_fail_soft_buckets_to_none(monkeypatch):
"""A store/DB error (or an uncached cell) never raises — it buckets to 'none'."""
monkeypatch.setattr(climate, "history_max_date", lambda cid: None)
assert payloads.content_token("1_2") == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:none"
def boom(cid):
raise RuntimeError("store down")
monkeypatch.setattr(climate, "history_max_date", boom)
assert payloads.content_token("1_2") == f"{payloads.PAYLOAD_VER}:{payloads.CONTENT_VER}:none"
def test_recent_token_composes_history_and_stamp(history, monkeypatch): def test_recent_token_composes_history_and_stamp(history, monkeypatch):
monkeypatch.setattr(climate, "recent_stamp", lambda cid: "stamp") monkeypatch.setattr(climate, "recent_stamp", lambda cid: "stamp")
assert payloads.recent_token(history, "1_2") == f"{payloads.history_token(history)}:stamp" assert payloads.recent_token(history, "1_2") == f"{payloads.history_token(history)}:stamp"

View file

@ -40,6 +40,16 @@ def test_classify_inbound_categories():
# dashboard probing /thermograph/... against a root-served (base="") prod app. # dashboard probing /thermograph/... against a root-served (base="") prod app.
assert c("/thermograph/api/v2/metrics", "") == "metrics" assert c("/thermograph/api/v2/metrics", "") == "metrics"
assert c("/api/v2/metrics", "") == "metrics" assert c("/api/v2/metrics", "") == "metrics"
# The SSR content API (backend/api/content_routes.py) is a server-to-server
# hop from frontend_ssr's own api_client.py, never a browser -- its own
# category is what lets it be excluded from the access log without also
# hiding real external traffic under the generic "api:*" bucket.
assert c("/thermograph/api/v2/content/hub", "/thermograph") == "internal"
assert c("/api/v2/content/city/tokyo", "") == "internal"
# The liveness probe: never under BASE, whatever base the app happens to be
# mounted at (it's registered at a fixed path -- see web/app.py's healthz).
assert c("/healthz") == "health"
assert c("/healthz", "/thermograph") == "health"
# pages, static, seo # pages, static, seo
assert c("/thermograph/", "/thermograph") == "page" assert c("/thermograph/", "/thermograph") == "page"
assert c("/thermograph/calendar", "/thermograph") == "page" assert c("/thermograph/calendar", "/thermograph") == "page"

View file

@ -615,3 +615,104 @@ def test_reverse_geocode_timeout_returns_none_without_blocking_the_caller(monkey
elapsed = time_mod.monotonic() - t0 elapsed = time_mod.monotonic() - t0
assert label is None assert label is None
assert elapsed < 0.2 # returned near the wait timeout, not after the 0.3s fetch assert elapsed < 0.2 # returned near the wait timeout, not after the 0.3s fetch
# --- forward geocode: shares the reverse-geocode worker, not a second lock -----
# Regression coverage for the NameError _REVGEO_LOCK bug (geocode_nominatim
# referenced a lock that was removed when reverse geocoding moved to the
# worker/queue design, so every forward lookup 502'd in production). These
# exercise the real queue/worker plumbing rather than mocking geocode_nominatim
# itself away, so a reintroduced bare `with _REVGEO_LOCK:` or any other
# not-actually-defined-name bug fails loudly here instead of shipping unseen.
def test_geocode_nominatim_resolves_via_the_worker_thread(monkeypatch):
monkeypatch.setattr(
climate, "_fetch_geocode_forward",
lambda name, count: [{"name": name, "admin1": None, "country": "Testland",
"country_code": "TL", "lat": 1.0, "lon": 2.0,
"population": None}],
)
results = climate.geocode_nominatim("Nowheresville")
assert results[0]["name"] == "Nowheresville"
assert results[0]["country"] == "Testland"
def test_geocode_nominatim_timeout_returns_empty_list(monkeypatch):
"""Same shape as reverse_geocode's timeout test: a caller waits only up to
_GEOCODE_WAIT_TIMEOUT, not as long as the fetch itself takes."""
import time as time_mod
monkeypatch.setattr(climate, "_GEOCODE_WAIT_TIMEOUT", 0.05)
def slow_fetch(name, count):
time_mod.sleep(0.3)
return [{"name": "Too Slow"}]
monkeypatch.setattr(climate, "_fetch_geocode_forward", slow_fetch)
t0 = time_mod.monotonic()
results = climate.geocode_nominatim("anywhere")
elapsed = time_mod.monotonic() - t0
assert results == []
assert elapsed < 0.2
def test_geocode_nominatim_a_bad_fetch_degrades_to_empty_list_not_a_crash(monkeypatch):
"""_fetch_geocode_forward is allowed to raise (matches _fetch_revgeo_label's
contract loosely -- the worker's except clause is the actual safety net);
confirm a raising fetch never reaches the caller as an exception."""
def boom(name, count):
raise RuntimeError("Nominatim is down")
monkeypatch.setattr(climate, "_fetch_geocode_forward", boom)
assert climate.geocode_nominatim("anywhere") == []
def test_geocode_nominatim_shares_the_reverse_geocode_pacer(monkeypatch):
"""Forward and reverse jobs are drained by the SAME worker thread off the
SAME queue, so a forward call advances _revgeo_last exactly like a reverse
one does -- this is what makes a second lock unnecessary."""
monkeypatch.setattr(climate, "_revgeo_last", 0.0)
monkeypatch.setattr(climate, "_fetch_geocode_forward", lambda name, count: [])
before = climate._revgeo_last
climate.geocode_nominatim("anywhere")
assert climate._revgeo_last > before
def test_fetch_geocode_forward_parses_nominatim_response(monkeypatch):
"""Executes _fetch_geocode_forward's real body (the function the NameError
bug prevented from ever running) against a stubbed HTTP transport -- not a
monkeypatch of geocode_nominatim itself, unlike the API-layer tests."""
class _FakeResp:
def json(self):
return [
{"name": "West Seattle", "lat": "47.57", "lon": "-122.38",
"display_name": "West Seattle, Seattle, King County, Washington, United States",
"address": {"suburb": "West Seattle", "city": "Seattle",
"state": "Washington", "country": "United States",
"country_code": "us"}},
# No `name`, no recognized address component -- falls back to the
# head of display_name, exercising that branch too.
{"lat": "51.5", "lon": "-0.1",
"display_name": "Some Unnamed Place, Greater London, England",
"address": {"country": "United Kingdom", "country_code": "gb"}},
]
captured = {}
def fake_request(url, params, timeout, *, phase, headers=None, attempts=climate.MAX_ATTEMPTS):
captured["url"] = url
captured["params"] = params
captured["phase"] = phase
return _FakeResp()
monkeypatch.setattr(climate, "_request", fake_request)
results = climate._fetch_geocode_forward("west seattle", 5)
assert captured["url"] == "https://nominatim.openstreetmap.org/search"
assert captured["params"]["q"] == "west seattle"
assert captured["phase"] == "geocode"
assert results[0] == {
"name": "West Seattle", "admin1": "Washington", "country": "United States",
"country_code": "US", "lat": 47.57, "lon": -122.38, "population": None,
}
assert results[1]["name"] == "Some Unnamed Place"
assert results[1]["country_code"] == "GB"

View file

@ -69,6 +69,17 @@ def test_recent_backend_roundtrips_through_parquet(monkeypatch, tmp_path):
assert hit is not None and hit[0].height == 3 assert hit is not None and hit[0].height == 3
def test_history_max_date_reads_parquet_tail_without_full_load(monkeypatch, tmp_path):
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
cell_id = "9_10"
assert climate.history_max_date(cell_id) is None # nothing cached -> None
climate._write_history_backed(cell_id, _hist_frame(n=5)) # dates 1995-01-01..05
assert climate.history_max_date(cell_id) == "1995-01-05"
# A tail top-up that appends a newer day advances the max date.
climate._write_history_backed(cell_id, _hist_frame(n=7))
assert climate.history_max_date(cell_id) == "1995-01-07"
# --- gated Postgres integration --------------------------------------------- # --- gated Postgres integration ---------------------------------------------
@pytest.fixture @pytest.fixture
@ -140,6 +151,15 @@ def test_pg_recent_stamp_advances_only_on_write(pg_store):
assert pg_store.history_synced_at("1_2") == 6000.0 assert pg_store.history_synced_at("1_2") == 6000.0
def test_pg_history_max_date(pg_store):
assert pg_store.history_max_date("1_2") is None # no rows -> None
pg_store.write_history("1_2", _hist_frame(n=5), 1000.0) # dates 1995-01-01..05
assert pg_store.history_max_date("1_2") == "1995-01-05"
# Another cell's rows must not leak into this cell's max.
pg_store.write_history("3_4", _hist_frame(n=9), 1000.0)
assert pg_store.history_max_date("1_2") == "1995-01-05"
def test_pg_cols_version_gate(pg_store): def test_pg_cols_version_gate(pg_store):
pg_store.write_history("1_2", _hist_frame(n=2), 1000.0) pg_store.write_history("1_2", _hist_frame(n=2), 1000.0)
assert pg_store.read_history("1_2") is not None assert pg_store.read_history("1_2") is not None

View file

@ -0,0 +1,131 @@
"""warm_cities.warm_content -- the off-request pre-warm that rebuilds the SEO
content derived-store (api/content_routes.py's content-city / content-month /
content-records rows) so the first request after each daily archive advance hits
a warm cache instead of a cold 45-year recompute.
The invariants under test mirror the ones content_routes.py relies on:
* it populates all three kinds under the exact (kind, key, token) the routes use;
* a city with no cached archive is skipped and never triggers an upstream fetch
(warming must not spend quota);
* it is idempotent -- a second run with nothing changed writes zero payloads.
"""
import warm_cities
from api import content_payloads
from api import payloads as cache_ids
from data import cities as cities_mod
from data import climate
from data import grid
from data import store
# A fixed archive last-date -> a stable content token across a test's runs, so the
# idempotency check exercises the "already fresh for this token" fast path.
_MAX_DATE = "2026-07-20"
def _two_cities():
"""Two real curated cities (so the payload builders get the fields they read),
the first with a cached archive, the second without."""
everyone = cities_mod.all_cities()
return everyone[0], everyone[1]
def _wire(monkeypatch, city_with_history, history):
"""Point warm_content's cache-only reads at fixtures and make any upstream
fetch a hard failure, so a test proves warming never reaches the network."""
have_cell = grid.snap(city_with_history["lat"], city_with_history["lon"])["id"]
monkeypatch.setattr(climate, "history_max_date", lambda cell_id: _MAX_DATE)
def fake_cached_history(cell):
return history if cell["id"] == have_cell else None
monkeypatch.setattr(climate, "load_cached_history", fake_cached_history)
monkeypatch.setattr(climate, "load_cached_recent_forecast", lambda cell: None)
def _no_upstream(*a, **k): # pragma: no cover - only fires on a regression
raise AssertionError("warm_content must not fetch upstream")
monkeypatch.setattr(climate, "get_history", _no_upstream)
monkeypatch.setattr(climate, "get_recent_forecast", _no_upstream)
def test_warms_three_kinds_with_matching_key_and_token(tmp_store, monkeypatch, history):
warm, cold = _two_cities()
_wire(monkeypatch, warm, history)
monkeypatch.setattr(cities_mod, "all_cities", lambda: [warm, cold])
result = warm_cities.warm_content(origin="https://thermograph.org")
origin = "https://thermograph.org"
cell_id = grid.snap(warm["lat"], warm["lon"])["id"]
token = cache_ids.content_token(cell_id)
slug = warm["slug"]
# content-city + content-records under {slug}:{origin}, 12 months under {slug}:{month}
assert store.get_json("content-city", cell_id, f"{slug}:{origin}", token) is not None
assert store.get_json("content-records", cell_id, f"{slug}:{origin}", token) is not None
for month in content_payloads.MONTH_INDEX:
assert store.get_json("content-month", cell_id, f"{slug}:{month}", token) is not None
# 1 city + 1 records + 12 months = 14 payloads for the one city with history.
assert result["built"] == 14
assert result["warmed"] == 1
def test_skips_city_without_cached_history_and_never_fetches(tmp_store, monkeypatch, history):
warm, cold = _two_cities()
_wire(monkeypatch, warm, history)
monkeypatch.setattr(cities_mod, "all_cities", lambda: [warm, cold])
result = warm_cities.warm_content(origin="https://thermograph.org")
# The archive-less city produced no rows (and _no_upstream never fired, or the
# call above would have raised).
cold_cell = grid.snap(cold["lat"], cold["lon"])["id"]
token = cache_ids.content_token(cold_cell)
assert store.get_json("content-city", cold_cell, f"{cold['slug']}:https://thermograph.org", token) is None
assert result["empty"] == 1
def test_idempotent_second_run_writes_nothing(tmp_store, monkeypatch, history):
warm, cold = _two_cities()
_wire(monkeypatch, warm, history)
monkeypatch.setattr(cities_mod, "all_cities", lambda: [warm, cold])
# Spy on put_payload while still persisting, so run 2 sees run 1's rows.
calls = []
real_put = store.put_payload
def spy(kind, cell_id, key, token, payload, cache=True):
calls.append((kind, cell_id, key, token))
return real_put(kind, cell_id, key, token, payload, cache)
monkeypatch.setattr(store, "put_payload", spy)
first = warm_cities.warm_content(origin="https://thermograph.org")
assert len(calls) == 14 # one full city warmed
assert first["built"] == 14
calls.clear()
second = warm_cities.warm_content(origin="https://thermograph.org")
assert calls == [] # nothing rebuilt: every kind still fresh
assert second["built"] == 0
assert second["warmed"] == 0
assert second["skipped"] == 1 # the warm city skipped cheaply
assert second["empty"] == 1 # the archive-less city still empty
def test_limit_caps_cities_built_per_call(tmp_store, monkeypatch, history):
# Two cities, both with a cached archive; limit=1 should build exactly one.
a, b = _two_cities()
have = {grid.snap(a["lat"], a["lon"])["id"], grid.snap(b["lat"], b["lon"])["id"]}
monkeypatch.setattr(climate, "history_max_date", lambda cell_id: _MAX_DATE)
monkeypatch.setattr(climate, "load_cached_history",
lambda cell: history if cell["id"] in have else None)
monkeypatch.setattr(climate, "load_cached_recent_forecast", lambda cell: None)
monkeypatch.setattr(cities_mod, "all_cities", lambda: [a, b])
result = warm_cities.warm_content(limit=1, origin="https://thermograph.org")
assert result["warmed"] == 1
assert result["built"] == 14

View file

@ -0,0 +1,87 @@
"""The request-logging middleware (web/app.py's revalidate_static): what gets
counted, what gets written to the access log, and what gets truncated before
it's persisted. Companion to core/test_metrics.py's classify_inbound tests --
these exercise the actual ASGI request path, not just the classifier function.
"""
import pytest
from fastapi.testclient import TestClient
from web import app as appmod
metrics = appmod.metrics # the exact module object app.py's own calls use --
audit = appmod.audit # see the note on module-reload isolation below.
@pytest.fixture
def client():
return TestClient(appmod.app)
def _patch_log_access(monkeypatch):
calls = []
monkeypatch.setattr(audit, "log_access", lambda record: calls.append(record))
return calls
def test_healthz_is_never_audited(monkeypatch, client):
"""/healthz is ~47% of prod's daily access log; it must never even reach
audit.log_access (not just be cheap once there -- see classify_inbound's
"health" category and the exclusion in revalidate_static)."""
calls = _patch_log_access(monkeypatch)
r = client.get("/healthz")
assert r.status_code == 200
assert calls == []
def test_internal_category_is_never_audited(monkeypatch, client):
"""The SSR content API's own category ("internal") is excluded from the
access log the same way -- it's a server-to-server hop (frontend_ssr's
api_client.py calling backend), not a page view."""
calls = _patch_log_access(monkeypatch)
monkeypatch.setattr(metrics, "classify_inbound", lambda path, base: "internal")
r = client.get("/thermograph/api/version")
assert r.status_code == 200
assert calls == []
def test_access_log_ip_is_truncated_not_raw(monkeypatch, client):
calls = _patch_log_access(monkeypatch)
r = client.get("/thermograph/api/version",
headers={"X-Forwarded-For": "203.0.113.77, 10.0.0.1"})
assert r.status_code == 200
assert len(calls) == 1
assert calls[0]["ip"] == "203.0.113.0" # /24, not the exact address
assert calls[0]["ip"] != "203.0.113.77"
def test_loggable_ip_truncation():
assert appmod._loggable_ip("203.0.113.77") == "203.0.113.0"
assert appmod._loggable_ip("2001:db8:1234:5678::1") == "2001:db8:1234::"
assert appmod._loggable_ip(None) is None
assert appmod._loggable_ip("") == ""
assert appmod._loggable_ip("not-an-ip") == "not-an-ip" # best-effort, never raises
def test_rate_limiter_still_sees_the_full_precision_ip(monkeypatch, client):
"""_client_ip (which feeds both the access log and the event rate limiter)
is never itself truncated -- only what audit.log_access persists is. A
/24-truncated rate-limit key would let one abuser exhaust a whole NAT'd
office's quota, which is exactly what must NOT happen here."""
seen = {}
monkeypatch.setattr(metrics, "record_event", lambda name, **kw: seen.update(kw))
r = client.post("/thermograph/api/v2/event", json={"event": "home.locate"},
headers={"X-Forwarded-For": "203.0.113.77"})
assert r.status_code == 204
assert seen["ip"] == "203.0.113.77"
def test_event_route_records_end_to_end(client):
"""Full HTTP round trip through the real ASGI route -- confirms the write
path backend/api/event -> metrics.record_event -> the counters store works
end to end (the existing suite only ever called record_event directly)."""
before = metrics.snapshot()["events_total"]
r = client.post("/thermograph/api/v2/event", json={"event": "home.locate"})
assert r.status_code == 204
after = metrics.snapshot()
assert after["events_total"] == before + 1
assert after["events"]["home.locate"]["direct"] # no Referer sent -> "direct"

View file

@ -19,13 +19,21 @@ import os
import sys import sys
import time import time
from api import content_payloads
from api import homepage from api import homepage
from api import payloads as cache_ids
from core import singleton from core import singleton
from data import cities from data import cities
from data import climate from data import climate
from data import grid from data import grid
from data import store
import paths import paths
# The deployment base path, derived exactly as api/content_routes.py does so the
# breadcrumb hrefs / jsonld URLs the warmer bakes match what a live request bakes.
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""
# core/singleton.py's flock guard, reused here for cross-*process* (not # core/singleton.py's flock guard, reused here for cross-*process* (not
# cross-worker) exclusion: the first invocation holds this for its process # cross-worker) exclusion: the first invocation holds this for its process
# lifetime; a second one (an overlapping deploy) fails the non-blocking flock # lifetime; a second one (an overlapping deploy) fails the non-blocking flock
@ -67,6 +75,100 @@ def main(limit: int | None = None, pace: float = 2.0) -> None:
print(f"homepage feed: FAILED {e}") print(f"homepage feed: FAILED {e}")
# --- content derived-store pre-warm -------------------------------------------
# The SEO /climate/<city>[/month|/records] pages render their JSON from the
# derived store (api/content_routes.py), keyed by the cheap+stable content token
# (payloads.content_token -> PAYLOAD_VER:CONTENT_VER:archive_last_date). That token
# turns over only when a cell's archive gains a new day (~1x/day), so on the first
# request after each daily advance the page would otherwise recompute a 45-year
# payload cold. Pre-warming rebuilds those rows off-request — cache-only, so it
# never spends upstream quota — so users and crawlers land on a warm cache.
def _content_specs(slug: str, origin: str) -> "list[tuple[str, str, str, int | None]]":
"""The (store-kind, store-key, builder-tag, month_idx) tuples for one city,
matching api/content_routes.py's kinds/keys exactly: content-city and
content-records key on ``{slug}:{origin}`` (the origin is folded in because the
payloads' jsonld.url is origin-qualified), content-month keys on
``{slug}:{month}`` for each of the 12 months."""
specs = [
("content-city", f"{slug}:{origin}", "city", None),
("content-records", f"{slug}:{origin}", "records", None),
]
for month, idx in content_payloads.MONTH_INDEX.items():
specs.append(("content-month", f"{slug}:{month}", "month", idx))
return specs
def _build_content(tag: str, city: dict, history, recent, origin: str, month_idx: "int | None") -> dict:
if tag == "city":
return content_payloads.city_payload(origin, BASE, city, history, recent)
if tag == "records":
return content_payloads.records_payload(origin, BASE, city, history)
return content_payloads.month_payload(BASE, city, history, month_idx)
def warm_content(limit: int | None = None, origin: str | None = None, pace: float = 0.05) -> dict:
"""Rebuild any missing/stale content-page payloads in the derived store.
For each curated city, compute the current content token (cheap no history
load) and check whether every content kind (city, 12 months, records) already
has a row for that token. Cities that are wholly fresh are skipped *without*
loading the archive, so a re-run after everything is warm is near-instant and
only cells whose archive genuinely advanced (new token) do work the same
idempotent contract as ``main``.
Cache-only, exactly like ``main``: the history and recent/forecast bundles are
read from the cache; a cell with no cached archive is skipped rather than
fetched, so warming never spends a single upstream request.
``limit`` caps the number of cities actually (re)built this call (not scanned),
so a caller can bound one pass's wall-clock and rely on the idempotent skip to
resume from where it left off on the next call. Returns a small counts dict.
"""
if origin is None:
origin = os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org").rstrip("/")
built = warmed = skipped = empty = failed = 0
for c in cities.all_cities():
if limit is not None and warmed >= limit:
break
slug = c["slug"]
cell = grid.snap(c["lat"], c["lon"])
cell_id = cell["id"]
token = cache_ids.content_token(cell_id)
specs = _content_specs(slug, origin)
# Cheap idempotency gate: which kinds are missing/stale for this token?
# get_payload is a cheap keyed lookup; loading the 45-year archive is not,
# so we defer that until we know at least one kind actually needs building.
stale = [s for s in specs if store.get_payload(s[0], cell_id, s[1], token) is None]
if not stale:
skipped += 1
continue
history = climate.load_cached_history(cell)
if history is None or history.is_empty():
# No cached archive yet -> nothing to build from, and warming must not
# fetch upstream. The cell self-heals on its first live request.
empty += 1
continue
# today_vs_normal on the city payload needs the recent/forecast bundle;
# read it cache-only (never fetch) — None just drops that one block.
recent = climate.load_cached_recent_forecast(cell)
did_work = False
for kind, key, tag, month_idx in stale:
try:
payload = _build_content(tag, c, history, recent, origin, month_idx)
store.put_payload(kind, cell_id, key, token, payload)
built += 1
did_work = True
except Exception: # noqa: BLE001 - one bad payload must not abort the sweep
failed += 1
if did_work:
warmed += 1
if pace:
time.sleep(pace)
return {"warmed": warmed, "built": built, "skipped": skipped, "empty": empty, "failed": failed}
if __name__ == "__main__": if __name__ == "__main__":
if not singleton.claim(LOCK_PATH): if not singleton.claim(LOCK_PATH):
# Not an error: an overlapping run just means a previous deploy's warm is # Not an error: an overlapping run just means a previous deploy's warm is

View file

@ -3,6 +3,7 @@ import contextlib
import datetime import datetime
import hashlib import hashlib
import hmac import hmac
import ipaddress
import json import json
import os import os
import queue import queue
@ -263,13 +264,34 @@ def api_version():
def _client_ip(request) -> "str | None": def _client_ip(request) -> "str | None":
"""The real client IP: the left-most X-Forwarded-For hop when a proxy fronts us """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).""" (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") xff = request.headers.get("x-forwarded-for")
if xff: if xff:
return xff.split(",")[0].strip() return xff.split(",")[0].strip()
return request.client.host if request.client else None 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") @app.middleware("http")
async def revalidate_static(request, call_next): async def revalidate_static(request, call_next):
"""Serve the frontend with no-cache (NOT no-store): browsers may keep a copy """Serve the frontend with no-cache (NOT no-store): browsers may keep a copy
@ -288,11 +310,15 @@ async def revalidate_static(request, call_next):
# loop so one request's instrumentation can't stall every other request # loop so one request's instrumentation can't stall every other request
# this worker is serving concurrently. # this worker is serving concurrently.
await run_in_threadpool(metrics.record_inbound, cat, response.status_code) await run_in_threadpool(metrics.record_inbound, cat, response.status_code)
# Retain per-request client IPs for later analysis; skip static assets and the # Retain per-request client IPs for later analysis; skip static assets, the
# dashboard's own metrics polling to keep the log to real, meaningful traffic. # dashboard's own metrics polling, the liveness probe (health -- by far the
if cat not in ("static", "metrics", "event"): # 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, { await run_in_threadpool(audit.log_access, {
"ip": _client_ip(request), "method": request.method, "ip": _loggable_ip(_client_ip(request)), "method": request.method,
"path": path, "status": response.status_code, "cat": cat}) "path": path, "status": response.status_code, "cat": cat})
# Threshold-gated slow-request line: RunAudit only times the 7 graded # Threshold-gated slow-request line: RunAudit only times the 7 graded
# endpoints, so this is the only latency signal for auth/account/content # endpoints, so this is the only latency signal for auth/account/content

View file

@ -0,0 +1,133 @@
# Iceberg: implementing the agent-queryable service
Handoff spec for whoever builds the Iceberg layer. The Postgres equivalent is
already shipped (`infra/ops/dbq.sh`) — **mirror it**. This document is the
contract, the environment facts you'll trip over, and how the result gets
verified.
## Definition of done
`infra/ops/iceberg.sh` exists and behaves like `dbq.sh`:
```sh
infra/ops/iceberg.sh prod -c "select count(*) from <namespace>.<table>"
infra/ops/iceberg.sh dev -c "show tables"
echo "select 1" | infra/ops/iceberg.sh beta -f -
```
An agent (or operator) can query Iceberg in any environment, read-only, with no
new network exposure and no interactive steps.
## The contract (non-negotiable)
1. **Same interface shape as `dbq.sh`** — `iceberg.sh <dev|beta|prod> [flags]
"<sql>"`. Extra flags pass through to the engine; **stdin is forwarded** so
`-f -` works. Non-interactive, deterministic, safe to call from CI.
2. **Read-only enforced by the system, not by convention.** A dedicated
read-only identity — not an admin/superuser credential. You must be able to
*demonstrate* a refused write (see Acceptance).
3. **No new network endpoints.** Run the engine where the data is already
reachable (containerised), rather than publishing ports or opening firewall
rules. See the prod constraint below — this is not negotiable, it's physics.
4. **Env-keyed dispatch, names resolved at call time.** Prod is Docker Swarm;
task container names change on **every redeploy**. Resolve via
`docker ps --filter name=…`; never hardcode a container name.
5. **No secrets in the repo** or in workflow files.
## Environment facts you need
- Monorepo `emi/thermograph` (domain dirs: `backend/ frontend/ infra/
observability/`). Ops tooling lives in `infra/ops/`.
- Environments: **dev** = LAN compose stack on the dev machine; **beta** =
`75.119.132.91`; **prod** = `169.58.46.181`. SSH as `agent` with
`~/.ssh/thermograph_agent_ed25519` (passwordless sudo on both boxes).
- **Prod runs Docker Swarm.** Its app database sits on the overlay network
`thermograph_internal` (`10.0.2.0/24`), which **the prod host itself cannot
route to**. Consequence, learned the hard way: `ssh -L` tunnelling works for
beta but is *impossible* for prod. Any design that depends on a tunnel or a
published port will fail on prod — that's why `dbq.sh` execs into the
container instead. Assume the same for anything you deploy there.
- The overlay is `attachable=true`, and the **dev machine is a Swarm worker** in
the prod cluster (NodeAddr `10.10.0.3`) — so `docker run --network
thermograph_internal …` from the dev box is a plausible route to prod-side
services. Untested; verify before relying on it.
- WireGuard mesh: prod `10.10.0.1`, beta `10.10.0.2`, dev `10.10.0.3`.
- `rclone` and `age` are already installed on prod and beta.
- Reference implementation to copy: **`infra/ops/dbq.sh`** + `infra/ops/README.md`.
## Storage — almost certainly your warehouse
Contabo Object Storage (S3-compatible), already provisioned and in use:
- Endpoint `https://eu2.contabostorage.com` · bucket `era5-thermograph` ·
region `default` · ~2 TB · **private** (keep it that way).
- ⚠️ **Contabo requires PATH-STYLE addressing.** Set
`force_path_style=true` / `s3.path-style-access=true` (whatever your engine's
FileIO calls it) and `region=default`. Virtual-host-style addressing **fails**.
This is validated, not theoretical.
- ⚠️ **The `backups/` prefix is in use** by the nightly backup jobs (prod DB +
Forgejo). Put Iceberg data under its own prefix (e.g. `iceberg/` or
`warehouse/`). Do not write outside your prefix.
- Credentials already exist — **reuse them, don't mint new ones silently**:
- SOPS vault: `infra/deploy/secrets/{prod,beta}.yaml` as `THERMOGRAPH_S3_*`
(rendered to `/etc/thermograph.env` at deploy by
`infra/deploy/render-secrets.sh`).
- Forgejo Actions secrets: `S3_ENDPOINT`, `S3_BUCKET`, `S3_ACCESS_KEY`,
`S3_SECRET_KEY`.
- If read-only S3 keys are needed for the query identity, ask the operator to
mint a scoped keypair rather than reusing the read-write one.
## Secrets handling
- Host-side: SOPS + age. Recipient `age1xx4dzs0dxlwvkv9sjuqzsphl7lfrxannkfken374yu2qvvcte9sqzktqt2`;
the private key is on each host at `/etc/thermograph/age.key` and on the
operator's machine at `~/.config/sops/age/keys.txt`.
- CI-side: Forgejo Actions secrets.
- ⚠️ **beta gotcha:** `/etc/thermograph.env` on beta is `agent:agent 0640` — the
`deploy` user **cannot read it**. If your service or job runs as `deploy` on
beta, pull credentials from Actions secrets instead, or change the ownership
deliberately and say so.
## Decisions you must make — and report back
1. **Catalog**: REST (Lakekeeper / Nessie / Polaris), JDBC-on-Postgres, Hive, or
filesystem/hadoop — and where it runs per environment.
2. **Engine**: DuckDB + iceberg extension, Trino, Spark, or pyiceberg. Prefer the
lightest that satisfies the contract; it must run containerised and headless.
3. **Warehouse location**: bucket + prefix, and whether environments are isolated
by separate prefixes, namespaces, or buckets.
4. **Read-only mechanism**: scoped S3 keys? catalog RBAC? engine restricted mode?
State which, and how a write is refused.
5. **Where the engine runs** for each env (local container on dev, on the box for
beta/prod, or attached to the overlay) — consistent with constraint #3.
## Acceptance criteria
The work is accepted when all of these are demonstrated with pasted output:
- [ ] `infra/ops/iceberg.sh <env> -c "<select>"` returns rows for **each**
environment that has data (explicitly state any env that doesn't yet).
- [ ] **A write/DDL attempt is refused by the system**, with the error pasted.
The Postgres bar to match:
`create table` on prod → `ERROR: permission denied for schema public`.
- [ ] **No new exposure**: `sudo ss -ltnp` on prod and beta shows no new
listening port; no new ufw rules; no Swarm `EndpointSpec` publish added.
State this explicitly.
- [ ] Non-interactive: extra flags pass through, stdin (`-f -`) works.
- [ ] Prod container/task resolution is dynamic (survives a redeploy).
- [ ] `infra/ops/README.md` updated with an Iceberg section: usage examples and
the *why* (same style as the Postgres rationale).
## Anti-goals
- ❌ Publishing catalog/engine/DB ports publicly, or adding firewall holes.
- ❌ Any design requiring a persistent SSH tunnel (**prod cannot be tunnelled**).
- ❌ Querying with an admin/superuser/read-write credential.
- ❌ Secrets committed to the repo or pasted into workflow files.
- ❌ Writing outside your own S3 prefix (`backups/` is live and load-bearing).
## Questions to route back to the operator
Anything that would require: minting new cloud credentials, opening a port,
changing firewall/Swarm configuration, or writing to a shared prefix. Don't
improvise on those — ask.

126
infra/ops/README.md Normal file
View file

@ -0,0 +1,126 @@
# ops/ — operator + agent query tooling
Read-only query access to Thermograph data, uniform across environments.
## `dbq.sh` — Postgres, any environment
```sh
infra/ops/dbq.sh dev "select count(*) from climate_history"
infra/ops/dbq.sh prod -tA -c "select max(date) from climate_history"
infra/ops/dbq.sh beta -c '\dt'
echo "select 1" | infra/ops/dbq.sh prod -f -
```
Environments: `dev` (LAN compose stack on the dev machine), `beta`
(75.119.132.91), `prod` (169.58.46.181). Extra args pass straight through to
`psql` (`-tA`, `-x`, `--csv`, `-f`, …); stdin is forwarded.
### Why it execs into the container instead of using a connection string
None of the databases are exposed over TCP — each listens only on its private
docker network. Prod's is a Swarm **overlay** (`10.0.2.0/24`) that the host
cannot route to, so `ssh -L` works for beta but is *impossible* for prod;
publishing 5432 would mean new ufw rules plus a Swarm endpoint change on
production. Running `psql` inside the db container works identically in all
three environments with no ports, no tunnels, and no infra changes.
The prod container name is a Swarm task name that changes on every redeploy, so
it is resolved at call time via `docker ps --filter name=…`, never hardcoded.
### Safety
Queries connect as **`thermograph_ro`** — a `NOSUPERUSER` role granted only
`pg_read_all_data`. Read-only is enforced by Postgres, not by convention:
```
$ infra/ops/dbq.sh prod -c "create table t(i int)"
ERROR: permission denied for schema public
```
The app's own `thermograph` role is a superuser and is deliberately **not** used
by this tool. If the role is ever missing (fresh database), recreate it with:
```sql
CREATE ROLE thermograph_ro LOGIN;
GRANT CONNECT ON DATABASE thermograph TO thermograph_ro;
GRANT pg_read_all_data TO thermograph_ro;
```
## `iceberg.sh` — the Iceberg lake, any environment
```sh
infra/ops/iceberg.sh dev -c "show tables"
infra/ops/iceberg.sh prod -c "select count(*) from era5_daily"
infra/ops/iceberg.sh beta -json -c "select * from era5_daily limit 3"
echo "select 1" | infra/ops/iceberg.sh prod -f -
```
Same environments and argument shape as `dbq.sh` (spec:
[`ICEBERG-HANDOFF.md`](./ICEBERG-HANDOFF.md)). Extra args pass straight
through to the duckdb CLI (`-json`, `-csv`, `-line`, `-c`, `-f`, …); with
`-f -`, stdin is forwarded and read as piped SQL; a single bare argument is
treated as `-c` SQL.
### Why an ephemeral DuckDB container instead of a query service
The lake is **one shared warehouse** in Contabo object storage
(`s3://era5-thermograph/iceberg`), reachable from every environment with the
same credentials — so unlike Postgres there is no per-env database to reach,
and the environments differ only in *where the engine runs*: a throwaway
`docker run` of a small DuckDB image (duckdb CLI + `httpfs`/`iceberg`
extensions pre-installed) on the target box — locally for LAN dev, over SSH
for beta/prod. The image is built on the host the first time it's needed from
a Dockerfile embedded in the script; its tag is a hash of that Dockerfile, so
editing the script rolls every host forward on the next call.
A container per query means no daemon, no listening port, no tunnel (prod's
overlay cannot be tunnelled), no firewall or Swarm changes — `sudo ss -ltnp`
is identical before and after a query. It also never execs into an app
container and resolves no container names, so prod redeploys can't break it.
There is no catalog service to run or expose: an Iceberg table's current
state is fully described by the newest metadata JSON under
`<warehouse>/<table>/metadata/`. `iceberg.sh` resolves that at call time
(numeric metadata version, newest wins) and exposes each table directory as a
view of the same name — which is also how it keeps reading fresh snapshots of
a table that is still being loaded.
Credentials are resolved at call time and never stored in the repo: the
target host's `/etc/thermograph.env` (`THERMOGRAPH_LAKE_S3_*`) wins if
present; otherwise the calling machine decrypts the SOPS vault
(`infra/deploy/secrets/prod.yaml`) and hands the two values to the remote
shell over stdin — never argv, so never visible in `ps`, and never written to
a file on the target.
### Safety
Every session locks itself down *before* user SQL runs:
```sql
SET allowed_directories=['s3://era5-thermograph/iceberg/', 's3://era5-thermograph/era5/'];
SET enable_external_access=false;
SET lock_configuration=true;
```
Both prefixes are required for reads: `era5_daily` was built with pyiceberg
`add_files` over the existing hive parquet, so its metadata lives under
`iceberg/` while its data files stay in place under `era5/daily/`. DuckDB
itself then refuses everything else — the live `backups/` prefix, all local
files — and the lockdown cannot be SET away by query text:
```
$ infra/ops/iceberg.sh prod -c "COPY (SELECT 1) TO 's3://era5-thermograph/backups/x.csv'"
Permission Error: Cannot access file "s3://era5-thermograph/backups/x.csv" - file system operations are disabled by configuration
$ infra/ops/iceberg.sh dev -c "SET enable_external_access=true"
Invalid Input Error: Cannot change configuration option "enable_external_access" - the configuration has been locked
```
**Known gap:** the only provisioned keypair for the bucket is read-write, so
a raw `COPY TO` targeting a path *inside* the two allowed prefixes is not
refused — the DuckDB iceberg extension cannot write Iceberg tables, but it
could still clobber raw objects under `iceberg/` or `era5/`. Path-based
restriction cannot both allow reading a prefix and forbid writing it; closing
the gap needs a scoped read-only keypair (the object-storage equivalent of
`thermograph_ro`). When the operator mints one, render it as
`THERMOGRAPH_LAKE_S3_*` on the hosts or swap it into the vault —
`iceberg.sh` needs no code change.

68
infra/ops/dbq.sh Executable file
View file

@ -0,0 +1,68 @@
#!/usr/bin/env bash
# dbq -- run READ-ONLY SQL against any Thermograph Postgres (LAN dev / beta / prod).
#
# infra/ops/dbq.sh dev "select count(*) from climate_history"
# infra/ops/dbq.sh prod -tA "select max(date) from climate_history"
# infra/ops/dbq.sh beta -c '\dt'
# echo "select 1" | infra/ops/dbq.sh prod -f -
#
# Why exec-into-the-container instead of a connection string:
# none of the databases are exposed over TCP. Each listens only on its private
# docker network -- and prod's is a Swarm *overlay* (10.0.2.0/24) that the host
# itself cannot route to, so `ssh -L` works for beta but is impossible for prod.
# Publishing 5432 would mean new firewall + Swarm endpoint changes on production.
# Running psql *inside* the db container works identically in all three
# environments with no ports, no tunnels and no infra changes -- locally for LAN
# dev, over SSH for beta/prod.
#
# Safety: always connects as `thermograph_ro`, a NOSUPERUSER role granted only
# pg_read_all_data. Read-only is enforced by Postgres itself, not by convention,
# so a stray INSERT/DDL fails with "permission denied" even against prod. (The
# app's own `thermograph` role is a superuser -- deliberately not used here.)
#
# Any extra arguments are passed straight through to psql, so -tA, -c, -f, -x,
# --csv etc. all work. Stdin is forwarded, so `-f -` reads piped SQL.
set -euo pipefail
KEYFILE="${THERMOGRAPH_AGENT_KEY:-$HOME/.ssh/thermograph_agent_ed25519}"
DB_USER="${THERMOGRAPH_DB_QUERY_USER:-thermograph_ro}"
DB_NAME="${THERMOGRAPH_DB_NAME:-thermograph}"
usage() {
sed -n '2,9p' "$0" | sed 's/^# \{0,1\}//' >&2
echo "environments: dev | beta | prod" >&2
exit 2
}
env_name="${1:-}"
[ -n "$env_name" ] || usage
shift
case "$env_name" in
dev) filter=thermograph-dev-db ; ssh_target= ;;
beta) filter=thermograph-db-1 ; ssh_target=agent@75.119.132.91 ;;
prod) filter=thermograph_db ; ssh_target=agent@169.58.46.181 ;;
-h|--help) usage ;;
*) echo "dbq: unknown environment '$env_name' (want: dev|beta|prod)" >&2; exit 2 ;;
esac
[ $# -gt 0 ] || usage
# Quote the psql arguments so they survive the remote shell intact.
psql_args=$(printf '%q ' "$@")
# The container name is resolved on the target host at call time: prod's Swarm
# task name changes on every redeploy, so it can never be hardcoded.
remote=$(cat <<REMOTE
cid=\$(docker ps -qf name=${filter} | head -1)
[ -n "\$cid" ] || { echo "dbq: no running db container matching '${filter}'" >&2; exit 1; }
exec docker exec -i "\$cid" psql -U ${DB_USER} -d ${DB_NAME} -v ON_ERROR_STOP=1 ${psql_args}
REMOTE
)
if [ -z "$ssh_target" ]; then
exec bash -c "$remote"
else
exec ssh -i "$KEYFILE" -o StrictHostKeyChecking=no -o ConnectTimeout=15 \
"$ssh_target" "$remote"
fi

200
infra/ops/iceberg.sh Executable file
View file

@ -0,0 +1,200 @@
#!/usr/bin/env bash
# iceberg -- run READ-ONLY SQL against the Thermograph Iceberg lake (LAN dev / beta / prod).
#
# infra/ops/iceberg.sh dev -c "show tables"
# infra/ops/iceberg.sh prod -c "select count(*) from era5_daily"
# infra/ops/iceberg.sh beta -json -c "select * from era5_daily limit 3"
# echo "select 1" | infra/ops/iceberg.sh prod -f -
#
# Why an ephemeral container instead of a query service: the lake is one shared
# warehouse in Contabo object storage (s3://era5-thermograph/iceberg), readable
# from every environment, so the environments differ only in WHERE the engine
# runs -- a throwaway `docker run` of a small DuckDB image on the target box.
# No daemon, no listening port, no tunnel (prod's overlay cannot be tunnelled),
# and no container names to resolve, so prod redeploys cannot break it.
#
# There is no catalog service: a table's current state is its newest metadata
# JSON under <warehouse>/<table>/metadata/, resolved at call time and exposed
# as a view named after the table directory.
#
# Safety: before user SQL runs, the session pins external access to the lake
# prefixes (iceberg/ metadata + era5/ data files) and locks the configuration
# -- anything else, notably backups/ and all local files, is refused by
# DuckDB itself, and the lockdown cannot be SET away. The provisioned S3
# keypair is read-write, so a write *inside* those prefixes is still possible
# until a scoped read-only keypair exists (see infra/ops/README.md).
#
# Credentials are resolved at call time, never stored here: the target host's
# /etc/thermograph.env (THERMOGRAPH_LAKE_S3_*) wins if present; otherwise the
# calling machine decrypts the SOPS vault and feeds the two values to the
# remote shell over stdin (never argv, so never visible in `ps`).
#
# Any extra arguments are passed straight through to the duckdb CLI, so -c,
# -json, -csv, -line, -f etc. all work. With `-f -`, stdin is forwarded and
# read as piped SQL. A single bare argument is treated as `-c` SQL.
set -euo pipefail
KEYFILE="${THERMOGRAPH_AGENT_KEY:-$HOME/.ssh/thermograph_agent_ed25519}"
usage() {
sed -n '2,8p' "$0" | sed 's/^# \{0,1\}//' >&2
echo "environments: dev | beta | prod" >&2
exit 2
}
env_name="${1:-}"
[ -n "$env_name" ] || usage
shift
case "$env_name" in
dev) ssh_target= ;;
beta) ssh_target=agent@75.119.132.91 ;;
prod) ssh_target=agent@169.58.46.181 ;;
-h|--help) usage ;;
*) echo "iceberg: unknown environment '$env_name' (want: dev|beta|prod)" >&2; exit 2 ;;
esac
[ $# -gt 0 ] || usage
# The engine image: duckdb CLI with the httpfs + iceberg extensions installed
# at build time, so a query needs no downloads. Built on the target host the
# first time it's needed; the tag is a hash of this Dockerfile, so editing it
# here rolls every host forward automatically on the next call.
dockerfile=$(cat <<'DOCKERFILE'
FROM debian:bookworm-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends ca-certificates curl \
&& rm -rf /var/lib/apt/lists/*
RUN curl -fsSL https://github.com/duckdb/duckdb/releases/download/v1.5.5/duckdb_cli-linux-amd64.gz \
| gunzip > /usr/local/bin/duckdb \
&& chmod +x /usr/local/bin/duckdb
RUN duckdb -c "INSTALL httpfs; INSTALL iceberg;"
COPY <<'ENTRY' /usr/local/bin/iceberg-query
#!/bin/sh
# Entrypoint of the thermograph-iceberg image; run by iceberg.sh, not directly.
set -eu
: "${LAKE_S3_ACCESS_KEY:?}" "${LAKE_S3_SECRET_KEY:?}"
WAREHOUSE="${LAKE_WAREHOUSE:-s3://era5-thermograph/iceberg}"
ENDPOINT="${LAKE_S3_ENDPOINT:-eu2.contabostorage.com}"
# Prefixes a query may touch. era5/ is needed because era5_daily was built
# with add_files over the existing hive parquet: its metadata lives in the
# warehouse but its data files stay in place under era5/daily/.
ALLOWED="${LAKE_ALLOWED_DIRS:-$WAREHOUSE/,s3://era5-thermograph/era5/}"
esc() { printf %s "$1" | sed "s/'/''/g"; }
SECRET="CREATE SECRET lake (TYPE s3, KEY_ID '$(esc "$LAKE_S3_ACCESS_KEY")', SECRET '$(esc "$LAKE_S3_SECRET_KEY")', ENDPOINT '$ENDPOINT', REGION 'default', URL_STYLE 'path');"
# One view per table dir, over its newest metadata JSON (numeric version wins,
# so v10 beats v9 and 00010-... beats 00009-...). grep drops statement noise
# like CREATE SECRET's "true" row; an empty warehouse yields no views.
VIEWS=$(duckdb -batch -noheader -list \
-cmd "LOAD httpfs; $SECRET" \
-c "SELECT 'CREATE VIEW \"' || tbl || '\" AS SELECT * FROM iceberg_scan(''' || file || ''');'
FROM (SELECT file,
split_part(replace(file, '$WAREHOUSE/', ''), '/', 1) AS tbl,
row_number() OVER (
PARTITION BY split_part(replace(file, '$WAREHOUSE/', ''), '/', 1)
ORDER BY coalesce(try_cast(regexp_extract(parse_filename(file), '[0-9]+') AS BIGINT), -1) DESC,
file DESC) AS rn
FROM glob('$WAREHOUSE/*/metadata/*.metadata.json'))
WHERE rn = 1;" | grep '^CREATE VIEW ' || true)
# Lock the session down BEFORE user SQL: external access is pinned to the
# allowed prefixes, local files are unreachable, and the config can't be
# un-SET. Views bind lazily, so reads still work under the lockdown.
allowed_sql=
oifs=$IFS; IFS=,
for d in $ALLOWED; do allowed_sql="$allowed_sql${allowed_sql:+, }'$d'"; done
IFS=$oifs
INIT="LOAD httpfs; LOAD iceberg; $SECRET $VIEWS
SET allowed_directories=[$allowed_sql];
SET enable_external_access=false;
SET lock_configuration=true;"
# A single bare argument is the SQL itself (dbq.sh calling convention).
if [ $# -eq 1 ] && [ "${1#-}" = "$1" ]; then set -- -c "$1"; fi
# duckdb has no "-f - means stdin" convention; hand it the real thing.
i=0; n=$#; prev=
while [ "$i" -lt "$n" ]; do
a=$1; shift
if [ "$prev" = "-f" ] && [ "$a" = "-" ]; then a=/dev/stdin; fi
prev=$a
set -- "$@" "$a"
i=$((i+1))
done
# The .output pair silences the init's own result rows (CREATE SECRET prints
# a "true"); errors still reach stderr, and user SQL output is untouched.
exec duckdb -batch -cmd ".output /dev/null" -cmd "$INIT" -cmd ".output" "$@"
ENTRY
RUN chmod +x /usr/local/bin/iceberg-query
DOCKERFILE
)
tag="thermograph-iceberg:$(printf '%s' "$dockerfile" | sha256sum | cut -c1-12)"
b64=$(printf '%s' "$dockerfile" | base64 -w0)
# Quote the duckdb arguments so they survive the remote shell intact.
args=$(printf '%q ' "$@")
# Credentials: env override first, then a call-time sops decrypt of the vault
# on the calling machine. Either may come up empty here -- the remote side
# still gets a chance to fill them from its own /etc/thermograph.env.
ak="${THERMOGRAPH_LAKE_S3_ACCESS_KEY:-}"
sk="${THERMOGRAPH_LAKE_S3_SECRET_KEY:-}"
if [ -z "$ak" ] || [ -z "$sk" ]; then
repo_root=$(cd -- "$(dirname -- "$0")/../.." && pwd)
vault="${THERMOGRAPH_LAKE_VAULT:-$repo_root/infra/deploy/secrets/prod.yaml}"
sops_bin=$(command -v sops || echo "$HOME/.local/bin/sops")
if [ -r "$vault" ] && [ -x "$sops_bin" ]; then
ak=$("$sops_bin" -d --extract '["THERMOGRAPH_LAKE_S3_ACCESS_KEY"]' "$vault" 2>/dev/null || true)
sk=$("$sops_bin" -d --extract '["THERMOGRAPH_LAKE_S3_SECRET_KEY"]' "$vault" 2>/dev/null || true)
fi
fi
# The engine image is (re)built on the target host only when its tag is
# missing, then run with the credentials passed via environment -- docker
# inherits them from the remote shell, which read them from stdin, so they
# never appear on a command line. The rest of stdin flows through to duckdb.
remote=$(cat <<REMOTE
set -eu
IFS= read -r ak; IFS= read -r sk
if [ -r /etc/thermograph.env ]; then
hak=\$(sed -n 's/^THERMOGRAPH_LAKE_S3_ACCESS_KEY=//p' /etc/thermograph.env | tail -n1)
hsk=\$(sed -n 's/^THERMOGRAPH_LAKE_S3_SECRET_KEY=//p' /etc/thermograph.env | tail -n1)
if [ -n "\$hak" ] && [ -n "\$hsk" ]; then ak=\$hak sk=\$hsk; fi
fi
if [ -z "\$ak" ] || [ -z "\$sk" ]; then
echo "iceberg: no lake credentials (vault decrypt failed on the caller and no THERMOGRAPH_LAKE_S3_* in /etc/thermograph.env)" >&2
exit 1
fi
export LAKE_S3_ACCESS_KEY="\$ak" LAKE_S3_SECRET_KEY="\$sk"
docker image inspect $tag >/dev/null 2>&1 \
|| echo $b64 | base64 -d | docker build -q -t $tag - >/dev/null
exec docker run --rm -i -e LAKE_S3_ACCESS_KEY -e LAKE_S3_SECRET_KEY $tag iceberg-query $args
REMOTE
)
# Stdin is forwarded only for the documented `-f -` form. Anything else gets
# just the credential lines -- unconditionally forwarding an idle stdin would
# leave the local pipeline waiting on input the query never reads.
want_stdin=false
prev=
for a in "$@"; do
if [ "$prev" = "-f" ] && [ "$a" = "-" ]; then want_stdin=true; fi
prev=$a
done
feed() {
printf '%s\n%s\n' "$ak" "$sk" || true
if $want_stdin; then cat || true; fi
}
if [ -z "$ssh_target" ]; then
feed | bash -c "$remote"
else
feed | ssh -i "$KEYFILE" -o StrictHostKeyChecking=no -o ConnectTimeout=15 \
"$ssh_target" "$remote"
fi