From 787461656ed6faf597bf2c355e50406e15046421 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Wed, 22 Jul 2026 12:07:25 -0700 Subject: [PATCH] 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 | 106 +++++++++++++++++++------ tests/notifications/test_discord_dm.py | 45 ++++++----- 2 files changed, 104 insertions(+), 47 deletions(-) diff --git a/notifications/notify.py b/notifications/notify.py index e00955e..f0f9d76 100644 --- a/notifications/notify.py +++ b/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: - """Best-effort Web Push of a just-created notification to all of the user's - devices. Prunes endpoints the push service reports gone. Runs in the notifier - daemon thread (no event loop), so the blocking send is fine here. Never raises - — the in-app notification is already committed regardless.""" +def _push_jobs(session, sub, title, body, event_date) -> list[tuple[int, dict, dict]]: + """Every (subscription-row id, subscription_info, payload) to push to this + user's devices, or [] if none. DB read only — see _send_push_job for the + actual (network) send, which the caller runs off this session entirely.""" 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: - info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}} - if push.send(info, payload) == "gone": - session.delete(row) # committed with the rest of the pass + return [ + (row.id, {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}, payload) + for row in rows + ] -def _dispatch_discord(session, sub, title, body, event_date) -> None: - """Best-effort Discord DM of a just-created notification, when the subscriber - has linked Discord and opted in. Isolated like the push side-channel — the - in-app row is already committed, and push/email remain the fallback for anyone - not reachable on Discord. Runs in the notifier thread (no event loop).""" +def _send_push_job(job: tuple[int, dict, dict]) -> "int | None": + """Best-effort Web Push of one gathered job. Returns the subscription-row id + to prune if the push service reports the endpoint gone, else None. Never + raises. Runs off the notifier thread's send pool — no DB session here.""" + 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 - discord.send_dm(user.discord_id, title, body, _deep_link(sub, event_date)) + return None + 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 - # truth for the bell; push is a side-channel that must never break it). + # Gather who to reach over Web Push and Discord DM (the in-app row above + # 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) - except Exception: # noqa: BLE001 - push failures stay isolated from the DB write + push_jobs.extend(_push_jobs(session, sub, title, body, event_date)) + 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) - except Exception: # noqa: BLE001 - Discord failures stay isolated too + job = _discord_job(session, sub, title, body, event_date) + 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 diff --git a/tests/notifications/test_discord_dm.py b/tests/notifications/test_discord_dm.py index 1152948..555fd8b 100644 --- a/tests/notifications/test_discord_dm.py +++ b/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") - assert len(sent) == 1 and sent[0][0] == "d9" + job = notify._discord_job(_fake_session(user), _sub(), "T", "B", "2026-07-19") + 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._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id="d", discord_dm=False)), - _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 + notify._send_discord_job(("d9", "T", "B", "/day#x")) + assert sent == [("d9", "T", "B", "/day#x")] # --- the on/off toggle + link opt-in -----------------------------------------