Port 3504dd6 Send notifier push/Discord DMs concurrently instead of one at a time from monorepo (drift reconciliation)

Claude-Session: https://claude.ai/code/session_01RdARHDJaYC1wSQRTYM6t3Z

notifications/notify.py

@@ -21,6 +21,7 @@ cached archive yet is fetched ONCE (budget-capped per pass) so a freshly-subscri
 city starts working; the recent/forecast bundle refreshes on its own hourly cadence.
 A forecast-fetch failure just skips the cell, so a pass never dies on a rate limit.
 """
+import concurrent.futures
 import datetime
 import os
 import threading
@@ -119,35 +120,74 @@ def _deep_link(sub: Subscription, event_date: str) -> str:
     return f"{BASE}/day#lat={sub.lat:.5f}&lon={sub.lon:.5f}&date={event_date}"
 
 
-def _dispatch_push(session, sub, title, body, event_date) -> None:
+def _push_jobs(session, sub, title, body, event_date) -> list[tuple[int, dict, dict]]:
-    """Best-effort Web Push of a just-created notification to all of the user's
+    """Every (subscription-row id, subscription_info, payload) to push to this
-    devices. Prunes endpoints the push service reports gone. Runs in the notifier
+    user's devices, or [] if none. DB read only — see _send_push_job for the
-    daemon thread (no event loop), so the blocking send is fine here. Never raises
+    actual (network) send, which the caller runs off this session entirely."""
-    — the in-app notification is already committed regardless."""
     rows = session.execute(
         select(PushSubscription).where(PushSubscription.user_id == sub.user_id)
     ).scalars().all()
     if not rows:
-        return
+        return []
     payload = {"title": title, "body": body, "url": _deep_link(sub, event_date),
                "tag": f"sub-{sub.id}"}
-    for row in rows:
+    return [
-        info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}
+        (row.id, {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}, payload)
-        if push.send(info, payload) == "gone":
+        for row in rows
-            session.delete(row)   # committed with the rest of the pass
+    ]
 
 
-def _dispatch_discord(session, sub, title, body, event_date) -> None:
+def _send_push_job(job: tuple[int, dict, dict]) -> "int | None":
-    """Best-effort Discord DM of a just-created notification, when the subscriber
+    """Best-effort Web Push of one gathered job. Returns the subscription-row id
-    has linked Discord and opted in. Isolated like the push side-channel — the
+    to prune if the push service reports the endpoint gone, else None. Never
-    in-app row is already committed, and push/email remain the fallback for anyone
+    raises. Runs off the notifier thread's send pool — no DB session here."""
-    not reachable on Discord. Runs in the notifier thread (no event loop)."""
+    row_id, info, payload = job
+    return row_id if push.send(info, payload) == "gone" else None
+
+
+def _discord_job(session, sub, title, body, event_date) -> "tuple[str, str, str, str] | None":
+    """The (discord_id, title, body, link) to DM for this subscription, or None
+    when the subscriber hasn't linked Discord or opted in. DB read only — see
+    _send_discord_job for the actual send."""
     if not discord.dm_enabled():
-        return
+        return None
     user = session.get(User, sub.user_id)
     if not user or not user.discord_id or not user.discord_dm:
-        return
+        return None
-    discord.send_dm(user.discord_id, title, body, _deep_link(sub, event_date))
+    return (user.discord_id, title, body, _deep_link(sub, event_date))
+
+
+def _send_discord_job(job: tuple[str, str, str, str]) -> None:
+    """Best-effort Discord DM of one gathered job. Never raises (send_dm itself
+    doesn't). Runs off the notifier thread's send pool — no DB session here."""
+    discord_id, title, body, link = job
+    discord.send_dm(discord_id, title, body, link)
+
+
+# How many push/Discord sends the notifier fans out at once. These are pure
+# network I/O with no DB session involved (see _push_jobs/_discord_job above),
+# so a small thread pool is all that's needed — one bad weather event touching
+# hundreds of subscribers no longer stretches a pass out send-by-send.
+SEND_WORKERS = int(os.environ.get("THERMOGRAPH_NOTIFY_SEND_WORKERS", "8"))
+
+
+def _flush_sends(push_jobs, discord_jobs) -> list[int]:
+    """Deliver every gathered push/Discord job concurrently. Returns the
+    push-subscription row ids the push service reported gone, for the caller
+    to prune. Blocks until every job has finished (or errored) — a pass still
+    waits for delivery to complete, it just no longer does so serially."""
+    with concurrent.futures.ThreadPoolExecutor(max_workers=SEND_WORKERS) as pool:
+        futures = [pool.submit(_send_push_job, job) for job in push_jobs]
+        futures += [pool.submit(_send_discord_job, job) for job in discord_jobs]
+        gone = []
+        for f in futures:
+            try:
+                result = f.result()
+            except Exception:  # noqa: BLE001 - one bad send must not lose the rest
+                continue
+            if result is not None:
+                gone.append(result)
+    return gone
 
 
 # --- per-cell evaluation -----------------------------------------------------
@@ -227,6 +267,8 @@ def _process_cell(session, cell_id, subs, today, now, archive_budget):
 
     grade_cache = {}
     created = 0
+    push_jobs = []
+    discord_jobs = []
     for sub in subs:
         # Weekly cap: one notification per subscription per 7 days.
         if sub.last_notified_at and now - sub.last_notified_at < WEEK_SECONDS:
@@ -260,18 +302,32 @@ def _process_cell(session, cell_id, subs, today, now, archive_budget):
             continue
         sub.last_notified_at = now
         created += 1
-        # Additionally deliver over Web Push (the in-app row above is the record of
+        # Gather who to reach over Web Push and Discord DM (the in-app row above
-        # truth for the bell; push is a side-channel that must never break it).
+        # is the record of truth for the bell; these are side-channels that must
+        # never break it) — the actual sends happen after commit, see below.
         try:
-            _dispatch_push(session, sub, title, body, event_date)
+            push_jobs.extend(_push_jobs(session, sub, title, body, event_date))
-        except Exception:  # noqa: BLE001 - push failures stay isolated from the DB write
+        except Exception:  # noqa: BLE001 - gathering must not break the DB write
             pass
-        # And, independently, a Discord DM for linked opted-in users.
         try:
-            _dispatch_discord(session, sub, title, body, event_date)
+            job = _discord_job(session, sub, title, body, event_date)
-        except Exception:  # noqa: BLE001 - Discord failures stay isolated too
+            if job is not None:
+                discord_jobs.append(job)
+        except Exception:  # noqa: BLE001 - gathering must not break the DB write
             pass
     session.commit()
+
+    # Deliver every push/Discord job gathered above concurrently, only now that
+    # the notifications they're for are actually committed. A single cell can
+    # carry hundreds of subscribers during a broad weather event, so this also
+    # avoids sending one at a time and stretching the pass past its interval.
+    if push_jobs or discord_jobs:
+        gone = _flush_sends(push_jobs, discord_jobs)
+        if gone:
+            with sync_session_maker() as cleanup:
+                cleanup.execute(delete(PushSubscription).where(PushSubscription.id.in_(gone)))
+                cleanup.commit()
+
     return created
 
 

tests/notifications/test_discord_dm.py

@@ -88,35 +88,36 @@ def _sub():
     return types.SimpleNamespace(user_id="u1", lat=47.6, lon=-122.3)
 
 
-def test_dispatch_sends_for_linked_opted_in_user(monkeypatch):
+def test_discord_job_gathers_linked_opted_in_user(monkeypatch):
     monkeypatch.setattr(discord, "dm_enabled", lambda: True)
-    sent = []
-    monkeypatch.setattr(discord, "send_dm", lambda *a, **k: sent.append(a) or True)
     user = types.SimpleNamespace(discord_id="d9", discord_dm=True)
-    notify._dispatch_discord(_fake_session(user), _sub(), "T", "B", "2026-07-19")
+    job = notify._discord_job(_fake_session(user), _sub(), "T", "B", "2026-07-19")
-    assert len(sent) == 1 and sent[0][0] == "d9"
+    assert job is not None and job[0] == "d9"
 
 
-def test_dispatch_skips_unlinked_or_opted_out(monkeypatch):
+def test_discord_job_skips_unlinked_or_opted_out(monkeypatch):
     monkeypatch.setattr(discord, "dm_enabled", lambda: True)
+    # linked but opted out
+    assert notify._discord_job(
+        _fake_session(types.SimpleNamespace(discord_id="d", discord_dm=False)),
+        _sub(), "T", "B", "2026-07-19") is None
+    # not linked
+    assert notify._discord_job(
+        _fake_session(types.SimpleNamespace(discord_id=None, discord_dm=True)),
+        _sub(), "T", "B", "2026-07-19") is None
+
+
+def test_discord_job_noop_when_bot_unconfigured(monkeypatch):
+    monkeypatch.setattr(discord, "dm_enabled", lambda: False)
+    user = types.SimpleNamespace(discord_id="d", discord_dm=True)
+    assert notify._discord_job(_fake_session(user), _sub(), "T", "B", "2026-07-19") is None
+
+
+def test_send_discord_job_calls_send_dm(monkeypatch):
     sent = []
     monkeypatch.setattr(discord, "send_dm", lambda *a, **k: sent.append(a) or True)
-    # linked but opted out
+    notify._send_discord_job(("d9", "T", "B", "/day#x"))
-    notify._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id="d", discord_dm=False)),
+    assert sent == [("d9", "T", "B", "/day#x")]
-                             _sub(), "T", "B", "2026-07-19")
-    # not linked
-    notify._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id=None, discord_dm=True)),
-                             _sub(), "T", "B", "2026-07-19")
-    assert sent == []
-
-
-def test_dispatch_noop_when_bot_unconfigured(monkeypatch):
-    monkeypatch.setattr(discord, "dm_enabled", lambda: False)
-    called = []
-    monkeypatch.setattr(discord, "send_dm", lambda *a, **k: called.append(1))
-    notify._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id="d", discord_dm=True)),
-                             _sub(), "T", "B", "2026-07-19")
-    assert called == []           # never even looked the user up beyond the gate
 
 
 # --- the on/off toggle + link opt-in -----------------------------------------