From 5e0683cd4a2a9b30fc45377331a78ae3aa4bf73b Mon Sep 17 00:00:00 2001 From: timmy Date: Tue, 11 Aug 2026 12:57:07 +0000 Subject: [PATCH] feat: send bounded push digests for update bursts (Closes #569) --- README.md | 5 +- frontend/service-worker.js | 15 +++++ src/main.py | 5 ++ src/push_notifications.py | 61 +++++++++++++++++- src/push_subscription_store.py | 34 +++++++++- tests/test_push_notifications.py | 104 +++++++++++++++++++++++++++++++ tests/test_service_worker.py | 26 ++++++++ 7 files changed, 245 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 270ddc3..76b3758 100644 --- a/README.md +++ b/README.md @@ -190,7 +190,8 @@ export STACKCHAIN_TRUSTED_PROXY_CIDRS='127.0.0.0/8' # Optional Web Push. Generate a VAPID key pair outside the repo and inject it. # The feature stays disabled unless all three values are present. Privacy-safe update # alerts offer Mark read and Tomorrow; Tomorrow syncs the unread item to Later at -# 09:00 in the device's local timezone without opening the dashboard. +# 09:00 in the device's local timezone without opening the dashboard. Bursts send +# three individual alerts followed by one private digest that opens Updates. export STACKCHAIN_VAPID_PUBLIC_KEY='' export STACKCHAIN_VAPID_PRIVATE_KEY='' export STACKCHAIN_VAPID_SUBJECT='mailto:ops@example.com' @@ -201,6 +202,8 @@ export STACKCHAIN_VAPID_SUBJECT='mailto:ops@example.com' export STACKCHAIN_PUSH_POLL_SECONDS=30 export STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS=10 export STACKCHAIN_PUSH_MAX_CONCURRENCY=8 +# Maximum individual alerts per device and poll before one digest covers the rest. +export STACKCHAIN_PUSH_MAX_INDIVIDUAL_NOTIFICATIONS=3 export STACKCHAIN_PUSH_LEASE_SECONDS=60 export STACKCHAIN_PUSH_DB='/var/lib/stackchain-dashboard/push-subscriptions.sqlite3' uvicorn src.main:app --host 127.0.0.1 --port 8000 diff --git a/frontend/service-worker.js b/frontend/service-worker.js index c039ad3..6c0b935 100644 --- a/frontend/service-worker.js +++ b/frontend/service-worker.js @@ -278,6 +278,21 @@ self.addEventListener('push', event => { const route = String(payload.route || ''); const tag = String(payload.tag || ''); const notificationId = Number(payload.notification_id); + const updateCount = Number(payload.update_count); + if ( + route === '#/my-work/updates' + && tag === 'stackchain-update-digest' + && Number.isSafeInteger(updateCount) + && updateCount > 0 + && updateCount <= 50 + ) { + event.waitUntil(self.registration.showNotification(updateCount + ' new work updates', { + body: 'Tap to review them in Stackchain.', + tag, + data: {route}, + })); + return; + } if (!/^#\/my-work\/update\/\d+$/.test(route) || !/^stackchain-update-\d+$/.test(tag)) return; const options = { body: 'Tap to review it in Stackchain.', diff --git a/src/main.py b/src/main.py index 73794db..5198239 100644 --- a/src/main.py +++ b/src/main.py @@ -90,6 +90,10 @@ async def _push_poll_loop() -> None: max_concurrency = max( 1, int(os.getenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "8")) ) + max_individual_notifications = max( + 0, + int(os.getenv("STACKCHAIN_PUSH_MAX_INDIVIDUAL_NOTIFICATIONS", "3")), + ) while True: await asyncio.sleep(interval) try: @@ -101,6 +105,7 @@ async def _push_poll_loop() -> None: lease_seconds=lease_seconds, send_timeout_seconds=send_timeout, max_concurrency=max_concurrency, + max_individual_notifications=max_individual_notifications, ) except asyncio.CancelledError: raise diff --git a/src/push_notifications.py b/src/push_notifications.py index fa7ddd8..cb6da64 100644 --- a/src/push_notifications.py +++ b/src/push_notifications.py @@ -44,6 +44,7 @@ async def dispatch_unread_updates( lease_seconds: float = 60.0, send_timeout_seconds: float = 10.0, max_concurrency: int = 8, + max_individual_notifications: int = 3, ) -> int: if not configuration.enabled: return 0 @@ -87,7 +88,16 @@ async def dispatch_unread_updates( if ownership_lost.is_set(): return 0 count = 0 - for thread_id in delivery.thread_ids: + digest_pending = set(delivery.digest_ids) + new_ids = tuple( + thread_id + for thread_id in delivery.thread_ids + if thread_id not in digest_pending + ) + individual_ids = new_ids[:max(0, max_individual_notifications)] + overflow_ids = delivery.digest_ids + new_ids[len(individual_ids):] + can_send_digest = True + for thread_id in individual_ids: still_owner = await asyncio.to_thread( store.acquire_dispatch_lease, owner, @@ -123,7 +133,7 @@ async def dispatch_unread_updates( await asyncio.to_thread( store.delete_session, delivery.session_id ) - break + can_send_digest = False # Leave this device's transient failures unseen for a later # poll instead of paying the endpoint deadline repeatedly. break @@ -131,6 +141,53 @@ async def dispatch_unread_updates( store.mark_delivered, delivery.session_id, (thread_id,) ) count += 1 + if overflow_ids and can_send_digest: + still_owner = await asyncio.to_thread( + store.acquire_dispatch_lease, + owner, + now=time.time(), + lease_seconds=lease_seconds, + ) + if not still_owner: + ownership_lost.set() + return count + payload = json.dumps( + { + "title": f"{len(overflow_ids)} new work updates", + "body": "Tap to review them in Stackchain.", + "route": "#/my-work/updates", + "tag": "stackchain-update-digest", + "update_count": len(overflow_ids), + }, + separators=(",", ":"), + ) + try: + if send is None: + operation = send_web_push( + delivery.subscription, payload, configuration + ) + else: + operation = send(delivery.subscription, payload) + await asyncio.wait_for(operation, timeout=send_timeout_seconds) + except Exception as error: + status = getattr( + getattr(error, "response", None), "status_code", None + ) + if status in {404, 410}: + await asyncio.to_thread( + store.delete_session, delivery.session_id + ) + else: + await asyncio.to_thread( + store.mark_digest_pending, + delivery.session_id, + overflow_ids, + ) + return count + await asyncio.to_thread( + store.mark_delivered, delivery.session_id, overflow_ids + ) + count += 1 return count counts = await asyncio.gather( diff --git a/src/push_subscription_store.py b/src/push_subscription_store.py index b60f925..a6b8dd2 100644 --- a/src/push_subscription_store.py +++ b/src/push_subscription_store.py @@ -11,6 +11,7 @@ class PushDelivery: session_id: str subscription: dict thread_ids: tuple[int, ...] + digest_ids: tuple[int, ...] = () class PushSubscriptionStore: @@ -40,6 +41,13 @@ class PushSubscriptionStore: owner TEXT NOT NULL, expires_at REAL NOT NULL ); + CREATE TABLE IF NOT EXISTS push_digest_pending ( + session_id TEXT NOT NULL, + thread_id INTEGER NOT NULL, + PRIMARY KEY (session_id, thread_id), + FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id) + ON DELETE CASCADE + ); """ ) os.chmod(self.path, 0o600) @@ -120,12 +128,34 @@ class PushSubscriptionStore: } unseen = tuple(value for value in candidates if value not in delivered) if unseen: - deliveries.append(PushDelivery(session_id, json.loads(encoded), unseen)) + pending = { + row[0] + for row in connection.execute( + "SELECT thread_id FROM push_digest_pending WHERE session_id = ?", + (session_id,), + ) + } + digest_ids = tuple(value for value in unseen if value in pending) + deliveries.append( + PushDelivery(session_id, json.loads(encoded), unseen, digest_ids) + ) return deliveries + def mark_digest_pending(self, session_id: str, thread_ids: Iterable[int]) -> None: + with self._connect() as connection: + connection.executemany( + "INSERT OR IGNORE INTO push_digest_pending(session_id, thread_id) VALUES (?, ?)", + ((session_id, int(thread_id)) for thread_id in thread_ids), + ) + def mark_delivered(self, session_id: str, thread_ids: Iterable[int]) -> None: + values = tuple(int(thread_id) for thread_id in thread_ids) with self._connect() as connection: connection.executemany( "INSERT OR IGNORE INTO push_deliveries(session_id, thread_id) VALUES (?, ?)", - ((session_id, int(thread_id)) for thread_id in thread_ids), + ((session_id, thread_id) for thread_id in values), + ) + connection.executemany( + "DELETE FROM push_digest_pending WHERE session_id = ? AND thread_id = ?", + ((session_id, thread_id) for thread_id in values), ) diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 6aebc32..21a5b38 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -85,6 +85,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch raise asyncio.CancelledError monkeypatch.setenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "3") + monkeypatch.setenv("STACKCHAIN_PUSH_MAX_INDIVIDUAL_NOTIFICATIONS", "4") monkeypatch.setattr(main.asyncio, "sleep", no_wait) monkeypatch.setattr(main, "dispatch_unread_updates", stop_after_capture) @@ -93,6 +94,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch assert captured["session_active"] is dashboard_auth.managed_session_active assert captured["max_concurrency"] == 3 + assert captured["max_individual_notifications"] == 4 def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path): @@ -148,6 +150,108 @@ async def test_dispatch_sends_one_privacy_safe_deep_link_per_new_thread(tmp_path assert "Secret title" not in json.dumps(sent) +@pytest.mark.anyio +async def test_update_burst_sends_bounded_individual_pushes_and_one_private_digest(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + store.upsert("session-a", { + "endpoint": "https://push.example/device-a", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + sent = [] + + async def unread(): + return {"items": [ + { + "id": notification_id, + "repository": "private/repo", + "title": f"Secret update {notification_id}", + } + for notification_id in range(41, 46) + ]} + + async def send(_subscription, payload): + sent.append(json.loads(payload)) + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + delivered = await dispatch_unread_updates( + store, + config, + unread, + send, + max_individual_notifications=2, + ) + + assert delivered == 3 + assert [payload["tag"] for payload in sent] == [ + "stackchain-update-41", + "stackchain-update-42", + "stackchain-update-digest", + ] + assert sent[-1] == { + "title": "3 new work updates", + "body": "Tap to review them in Stackchain.", + "route": "#/my-work/updates", + "tag": "stackchain-update-digest", + "update_count": 3, + } + assert "private/repo" not in json.dumps(sent) + assert "Secret update" not in json.dumps(sent) + assert await dispatch_unread_updates( + store, + config, + unread, + send, + max_individual_notifications=2, + ) == 0 + + +@pytest.mark.anyio +async def test_failed_digest_retries_only_overflow_after_individual_checkpoints(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + store.upsert("session-a", { + "endpoint": "https://push.example/device-a", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + attempts = [] + + async def unread(): + return {"items": [{"id": notification_id} for notification_id in range(51, 56)]} + + async def fail_digest(_subscription, payload): + decoded = json.loads(payload) + attempts.append(decoded["tag"]) + if decoded["tag"] == "stackchain-update-digest": + raise RuntimeError("push provider unavailable") + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates( + store, + config, + unread, + fail_digest, + max_individual_notifications=2, + ) == 2 + assert store.claim_unseen(range(51, 56))[0].thread_ids == (53, 54, 55) + + async def succeed(_subscription, payload): + attempts.append(json.loads(payload)["tag"]) + + assert await dispatch_unread_updates( + store, + config, + unread, + succeed, + max_individual_notifications=2, + ) == 1 + assert attempts == [ + "stackchain-update-51", + "stackchain-update-52", + "stackchain-update-digest", + "stackchain-update-digest", + ] + assert store.claim_unseen(range(51, 56)) == [] + + @pytest.mark.anyio async def test_production_notification_page_dispatches_one_unread_push(tmp_path): def upstream(_request): diff --git a/tests/test_service_worker.py b/tests/test_service_worker.py index 5a9be1b..57dd1be 100644 --- a/tests/test_service_worker.py +++ b/tests/test_service_worker.py @@ -417,6 +417,32 @@ def test_inbound_push_renders_generic_update_and_tap_opens_existing_workflow(): assert "must-not-render" not in json.dumps(result["notifications"]) +def test_update_digest_push_opens_unread_inbox_without_item_actions_or_private_copy(): + result = run_worker_scenario( + """ + await dispatchPush({ + title:'must-not-render', body:'private details must-not-render', + tag:'stackchain-update-digest', route:'#/my-work/updates', update_count:7, + }); + await dispatchNotificationClick('#/my-work/updates'); + process.stdout.write(JSON.stringify(state)); +""" + ) + + assert result["notifications"] == [{ + "title": "7 new work updates", + "options": { + "body": "Tap to review them in Stackchain.", + "tag": "stackchain-update-digest", + "data": {"route": "#/my-work/updates"}, + }, + }] + assert result["opened"] == [ + "https://forge.example/dashboard/#/my-work/updates" + ] + assert "must-not-render" not in json.dumps(result["notifications"]) + + def test_push_mark_read_action_confirms_authenticated_mutation_without_opening_app(): result = run_worker_scenario( """ -- 2.43.0