Add a worker scheduler for city warming and IndexNow pings (#242)

warm_cities.py and indexnow.py are one-shot scripts run manually at deploy
today - nothing keeps the curated city set topped up, or pings IndexNow,
between deploys.

Add notifications/scheduler.py: an in-process APScheduler instance driving
two recurring jobs (city warming, daily; IndexNow --if-changed, every 6h by
default), gated by the same leader election as the notifier - only the
process that already won leadership starts it, so the jobs run in exactly
one place under multi-host Swarm rather than once per replica. Neither job
runs immediately on start: warm_cities already runs at deploy time, so an
immediate duplicate on every worker boot/restart would be wasted work.

Extract indexnow.submit_if_changed() from the CLI's --if-changed branch so
the scheduler and the command line share the exact same skip-when-unchanged
logic instead of two copies drifting apart. The CLI's plain (non-flag) path
is unchanged.

APScheduler over a Postgres-native queue: exactly one process ever runs
this, no fan-out/backpressure/dead-letter need, no new infrastructure.

indexnow.py

@@ -133,6 +133,23 @@ def _write_state(sig: str) -> None:
         log.warning("could not persist IndexNow state to %s", _STATE_PATH)
 
 
+def submit_if_changed(base_url: str | None = None) -> dict | None:
+    """Submit every indexable page, but only when the URL set changed since the
+    last successful submission — the shared logic behind the CLI's --if-changed
+    flag and the worker scheduler's periodic ping (notifications/scheduler.py), so
+    a code-only deploy or an unremarkable scheduled tick never resubmits. Returns
+    the submit_all() result, or None when skipped (unchanged, or a partial/failed
+    submission that leaves the prior state in place so the next attempt retries)."""
+    base = base_url or os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org")
+    sig = url_signature()
+    if sig == _read_state():
+        return None
+    result = submit_all(base)
+    if result["total"] and result["submitted"] == result["total"]:
+        _write_state(sig)   # only remember a fully-successful submission
+    return result
+
+
 if __name__ == "__main__":
     import sys
 
@@ -145,14 +162,16 @@ if __name__ == "__main__":
     base = (positional[0] if positional else "") or os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org")
     logging.basicConfig(level=logging.INFO)
 
-    sig = url_signature()
+    print(f"IndexNow key: {key()}  (served at {base.rstrip('/')}/{key()}.txt)")
-    if if_changed and sig == _read_state():
+    if if_changed:
+        result = submit_if_changed(base)
+        if result is None:
             print("IndexNow: URL set unchanged since last submission — skipping.")
             sys.exit(0)
-
+    else:
-    print(f"IndexNow key: {key()}  (served at {base.rstrip('/')}/{key()}.txt)")
         result = submit_all(base)
     print(f"Submitted {result['submitted']}/{result['total']} URLs "
           f"in {result['batches']} batch(es); statuses={result['statuses']}")
-    if result["total"] and result["submitted"] == result["total"]:
+    if result["total"] and result["submitted"] == result["total"] and not if_changed:
+        sig = url_signature()
         _write_state(sig)   # only remember a fully-successful submission

notifications/scheduler.py

@@ -0,0 +1,68 @@
+"""Recurring worker-tier jobs: keep the curated city set warm and ping IndexNow
+on a schedule, so neither needs a human to run it after every deploy anymore.
+
+Only the process that already won leader election for the notifier runs this
+too (gated by the same check in web/app.py's _lifespan) — APScheduler has no
+cross-host coordination of its own, so running it unguarded on every worker
+replica would repeat every job once per replica, same reason the notifier
+itself needs a leader.
+
+APScheduler, not a Postgres-native queue: there is exactly one process that
+ever runs this, no fan-out/backpressure/dead-letter need, and it needs no new
+infrastructure. procrastinate/pgqueuer are the documented upgrade path if that
+changes — see docs/architecture/repo-topology-and-infrastructure.md §7.
+"""
+import os
+
+from apscheduler.schedulers.background import BackgroundScheduler
+
+# How often the curated city set is (re-)warmed. warm_cities.py's own docstring
+# frames it as a deploy-time script ("Run at/after deploy") because a warm cache
+# mostly just stays warm; scheduling it periodically instead catches a newly
+# added city, or one that fell out of cache, without waiting for the next
+# deploy. Daily is plenty — every already-cached city is a cheap skip.
+WARM_CITIES_INTERVAL_HOURS = int(os.environ.get("THERMOGRAPH_WARM_CITIES_INTERVAL_HOURS", "24") or 24)
+
+# How often to check whether the indexable URL set changed and, if so, ping
+# IndexNow. submit_if_changed() is a cheap no-op when nothing changed, so this
+# can run more often than warm_cities without spending anything on most ticks.
+INDEXNOW_INTERVAL_HOURS = int(os.environ.get("THERMOGRAPH_INDEXNOW_INTERVAL_HOURS", "6") or 6)
+
+_scheduler: "BackgroundScheduler | None" = None
+
+
+def _warm_cities_job() -> None:
+    import warm_cities   # local import: a script-style top-level module — only
+                          # pay its import cost on the process that runs the job
+    warm_cities.main()
+
+
+def _indexnow_job() -> None:
+    import indexnow
+    indexnow.submit_if_changed()
+
+
+def start() -> None:
+    """Start the scheduler (idempotent, mirroring notify.start()). The caller is
+    responsible for the leader-election gate — see web/app.py's _lifespan."""
+    global _scheduler
+    if _scheduler is not None:
+        return
+    sched = BackgroundScheduler(daemon=True)
+    # No next_run_time override: the default first run is one interval from now,
+    # not immediately — warm_cities already runs at deploy time (warm_cities.py /
+    # the deploy hook), so an immediate duplicate run on every worker boot/restart
+    # would just be wasted work.
+    sched.add_job(_warm_cities_job, "interval", hours=WARM_CITIES_INTERVAL_HOURS,
+                 id="warm_cities")
+    sched.add_job(_indexnow_job, "interval", hours=INDEXNOW_INTERVAL_HOURS,
+                 id="indexnow")
+    sched.start()
+    _scheduler = sched
+
+
+def stop() -> None:
+    global _scheduler
+    if _scheduler is not None:
+        _scheduler.shutdown(wait=False)
+        _scheduler = None

requirements.txt

@@ -22,3 +22,6 @@ jinja2==3.1.6
 # Structured SSR copy (glossary, static-page SEO meta) — see
 # backend/web/content_loader.py and content/.
 PyYAML==6.0.3
+# Recurring worker-tier jobs — city warming, IndexNow pings (see
+# notifications/scheduler.py). In-process; no broker/queue needed at this scale.
+apscheduler==3.11.3

tests/notifications/test_scheduler.py

@@ -0,0 +1,79 @@
+"""Tests for the recurring worker-tier jobs (scheduler.py): job registration,
+cadence from env, idempotent start/stop, and that each job calls the right
+underlying function. No real hours-long wait — inspects APScheduler's own job
+registry rather than letting anything actually fire."""
+import datetime
+
+import pytest
+
+from notifications import scheduler
+
+
+@pytest.fixture(autouse=True)
+def _clean_scheduler():
+    """Every test starts from "not running" and leaves nothing behind, even on
+    failure — scheduler._scheduler is module-global state shared across tests."""
+    scheduler.stop()
+    yield
+    scheduler.stop()
+
+
+def test_start_registers_both_jobs_with_configured_intervals(monkeypatch):
+    monkeypatch.setattr(scheduler, "WARM_CITIES_INTERVAL_HOURS", 24)
+    monkeypatch.setattr(scheduler, "INDEXNOW_INTERVAL_HOURS", 6)
+    scheduler.start()
+    jobs = {j.id: j for j in scheduler._scheduler.get_jobs()}
+    assert set(jobs) == {"warm_cities", "indexnow"}
+    assert jobs["warm_cities"].trigger.interval == datetime.timedelta(hours=24)
+    assert jobs["indexnow"].trigger.interval == datetime.timedelta(hours=6)
+    assert jobs["warm_cities"].func is scheduler._warm_cities_job
+    assert jobs["indexnow"].func is scheduler._indexnow_job
+
+
+def test_start_does_not_run_a_job_immediately():
+    """warm_cities already runs at deploy time; an immediate duplicate run on
+    every worker boot/restart would just be wasted work — the first scheduled
+    run must be one interval away, not "now"."""
+    scheduler.start()
+    now = datetime.datetime.now(datetime.timezone.utc)
+    for job in scheduler._scheduler.get_jobs():
+        assert job.next_run_time > now
+
+
+def test_start_is_idempotent(monkeypatch):
+    scheduler.start()
+    first = scheduler._scheduler
+    calls = []
+    monkeypatch.setattr(scheduler.BackgroundScheduler, "__init__",
+                        lambda self, **kw: calls.append(1) or None)
+    scheduler.start()
+    assert scheduler._scheduler is first
+    assert calls == []   # a second BackgroundScheduler was never constructed
+
+
+def test_stop_shuts_down_and_clears_state():
+    scheduler.start()
+    assert scheduler._scheduler is not None
+    scheduler.stop()
+    assert scheduler._scheduler is None
+
+
+def test_stop_without_start_is_a_safe_no_op():
+    scheduler.stop()   # must not raise
+    assert scheduler._scheduler is None
+
+
+def test_warm_cities_job_calls_warm_cities_main(monkeypatch):
+    import warm_cities
+    calls = []
+    monkeypatch.setattr(warm_cities, "main", lambda: calls.append(1))
+    scheduler._warm_cities_job()
+    assert calls == [1]
+
+
+def test_indexnow_job_calls_submit_if_changed(monkeypatch):
+    import indexnow
+    calls = []
+    monkeypatch.setattr(indexnow, "submit_if_changed", lambda: calls.append(1))
+    scheduler._indexnow_job()
+    assert calls == [1]

tests/test_indexnow.py

@@ -0,0 +1,59 @@
+"""submit_if_changed: the shared skip-when-unchanged logic behind the CLI's
+--if-changed flag and the worker scheduler's periodic ping (no network —
+submit_all/url_signature are stubbed)."""
+import indexnow
+
+
+def _ok(total=3):
+    return {"submitted": total, "total": total, "batches": 1, "statuses": [200]}
+
+
+def test_skips_when_url_set_unchanged(monkeypatch, tmp_path):
+    monkeypatch.setattr(indexnow, "_STATE_PATH", str(tmp_path / "state.txt"))
+    monkeypatch.setattr(indexnow, "url_signature", lambda: "same-sig")
+    (tmp_path / "state.txt").write_text("same-sig")
+
+    def boom(*a, **k):
+        raise AssertionError("must not submit when the URL set is unchanged")
+
+    monkeypatch.setattr(indexnow, "submit_all", boom)
+    assert indexnow.submit_if_changed("https://example.org") is None
+
+
+def test_submits_and_persists_state_when_changed(monkeypatch, tmp_path):
+    state = tmp_path / "state.txt"
+    monkeypatch.setattr(indexnow, "_STATE_PATH", str(state))
+    monkeypatch.setattr(indexnow, "url_signature", lambda: "new-sig")
+    state.write_text("old-sig")
+
+    calls = []
+    monkeypatch.setattr(indexnow, "submit_all", lambda base, **kw: calls.append(base) or _ok())
+    result = indexnow.submit_if_changed("https://example.org")
+    assert calls == ["https://example.org"]
+    assert result["submitted"] == result["total"] == 3
+    assert state.read_text() == "new-sig"
+
+
+def test_partial_failure_does_not_persist_state(monkeypatch, tmp_path):
+    """A partial/failed submission must leave the prior state in place so the
+    next attempt retries, rather than silently giving up on the missed URLs."""
+    state = tmp_path / "state.txt"
+    monkeypatch.setattr(indexnow, "_STATE_PATH", str(state))
+    monkeypatch.setattr(indexnow, "url_signature", lambda: "new-sig")
+    state.write_text("old-sig")
+
+    monkeypatch.setattr(indexnow, "submit_all",
+                        lambda base, **kw: {"submitted": 1, "total": 3, "batches": 1, "statuses": [429]})
+    result = indexnow.submit_if_changed("https://example.org")
+    assert result["submitted"] < result["total"]
+    assert state.read_text() == "old-sig"
+
+
+def test_default_base_url_falls_back_to_env(monkeypatch, tmp_path):
+    monkeypatch.setattr(indexnow, "_STATE_PATH", str(tmp_path / "state.txt"))
+    monkeypatch.setattr(indexnow, "url_signature", lambda: "sig")
+    monkeypatch.setenv("THERMOGRAPH_BASE_URL", "https://from-env.example")
+    calls = []
+    monkeypatch.setattr(indexnow, "submit_all", lambda base, **kw: calls.append(base) or _ok())
+    indexnow.submit_if_changed()
+    assert calls == ["https://from-env.example"]

web/app.py

@@ -27,6 +27,7 @@ from accounts import db
 from data import grid
 from core import metrics
 from notifications import notify
+from notifications import scheduler
 import paths
 from data import places
 from core import singleton
@@ -159,8 +160,12 @@ async def _lifespan(app):
     # N-replica web tier can scale without also scaling notifier instances.
     if _should_run_notifier():
         notify.start()
+        # Same leader gate as the notifier above — the recurring worker jobs
+        # (city warming, IndexNow) must likewise run in exactly one process.
+        scheduler.start()
     yield
     notify.stop()
+    scheduler.stop()
 
 
 app = FastAPI(title="Thermograph", version="0.2.0", lifespan=_lifespan)