Pre-warm the SEO content derived-store off-request

The /climate/<city>[/month|/records] pages render from the content
derived-store keyed by content_token (PAYLOAD_VER:CONTENT_VER:archive
last-date), which turns over ~daily when a cell's archive gains a day.
On the first request after each advance the page recomputed a 45-year
payload cold.

Add warm_cities.warm_content: for each curated city, compute the cheap
content token and check whether every content kind (city, 12 months,
records) already has a fresh row; cities wholly fresh are skipped without
loading the archive, so a re-run is near-instant and only cells whose
archive advanced do work. Cache-only like main() — a cell with no cached
archive is skipped, never fetched, so warming spends no upstream quota.
A per-call limit caps cities (re)built so one pass's wall-clock is
bounded; the idempotent skip resumes on the next call.

Ride it on the leader-gated notifier loop next to the homepage sweep:
_maybe_warm_content runs at most every 30 min, capped at 50 cities per
tick, so a tick can't stall the notifier and the ~1000-city set refreshes
across a handful of ticks — well inside the <=1-day staleness the token
already tolerates.

backend/api/payloads.py

@@ -73,6 +73,30 @@ def history_token(history) -> str:
     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:
     """Validity for payloads that also grade the hourly recent/forecast bundle."""
     return f"{history_token(history)}:{climate.recent_stamp(cell_id)}"

backend/data/climate.py

@@ -862,6 +862,30 @@ def recent_stamp(cell_id: str) -> int:
         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:
     """Recent observations + forward forecast, with humidity as absolute humidity
     (g/m³). Thin wrapper over the raw loader (see below)."""

backend/data/climate_store.py

@@ -188,6 +188,29 @@ def recent_synced_at(cell_id: str) -> float:
         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:
     """The stored schema version for a cell's history; 0 if absent."""
     try:

backend/notifications/notify.py

@@ -450,6 +450,37 @@ def _maybe_refresh_homepage() -> None:
         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
 # day, after the feed is refreshed, leader-only like everything in run_loop. No
 # webhook configured => no-op.
@@ -489,6 +520,7 @@ def run_loop():
         try:
             run_pass()
             _maybe_refresh_homepage()
+            _maybe_warm_content()
             _maybe_post_discord()
         except Exception:  # noqa: BLE001 - a bad iteration must never kill the loop
             pass

backend/tests/api/test_payloads.py

@@ -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)}"
 
 
+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):
     monkeypatch.setattr(climate, "recent_stamp", lambda cid: "stamp")
     assert payloads.recent_token(history, "1_2") == f"{payloads.history_token(history)}:stamp"

backend/tests/data/test_climate_store.py

@@ -69,6 +69,17 @@ def test_recent_backend_roundtrips_through_parquet(monkeypatch, tmp_path):
     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 ---------------------------------------------
 
 @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
 
 
+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):
     pg_store.write_history("1_2", _hist_frame(n=2), 1000.0)
     assert pg_store.read_history("1_2") is not None

backend/tests/test_warm_content.py

@@ -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

backend/warm_cities.py

@@ -19,13 +19,21 @@ import os
 import sys
 import time
 
+from api import content_payloads
 from api import homepage
+from api import payloads as cache_ids
 from core import singleton
 from data import cities
 from data import climate
 from data import grid
+from data import store
 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
 # cross-worker) exclusion: the first invocation holds this for its process
 # 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}")
 
 
+# --- 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 not singleton.claim(LOCK_PATH):
         # Not an error: an overlapping run just means a previous deploy's warm is