diff --git a/README.md b/README.md index 76b3758..81167bc 100644 --- a/README.md +++ b/README.md @@ -191,7 +191,9 @@ export STACKCHAIN_TRUSTED_PROXY_CIDRS='127.0.0.0/8' # 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. Bursts send -# three individual alerts followed by one private digest that opens Updates. +# three individual alerts followed by one private digest that opens Updates. A later +# comment on an already-delivered thread triggers a fresh alert when Gitea advances +# that thread's updated_at revision; unchanged and older snapshots remain silent. export STACKCHAIN_VAPID_PUBLIC_KEY='' export STACKCHAIN_VAPID_PRIVATE_KEY='' export STACKCHAIN_VAPID_SUBJECT='mailto:ops@example.com' diff --git a/src/push_notifications.py b/src/push_notifications.py index cb6da64..920a935 100644 --- a/src/push_notifications.py +++ b/src/push_notifications.py @@ -59,12 +59,12 @@ async def dispatch_unread_updates( return 0 try: page = await unread() - thread_ids = { - int(item["id"]) + thread_revisions = { + int(item["id"]): str(item.get("updated_at") or "") for item in page.get("items", []) if isinstance(item, dict) and str(item.get("id", "")).isdigit() } - deliveries = await asyncio.to_thread(store.claim_unseen, thread_ids) + deliveries = await asyncio.to_thread(store.claim_unseen, thread_revisions) if session_active is not None: try: authorized = [await session_active(item.session_id) for item in deliveries] @@ -88,16 +88,21 @@ async def dispatch_unread_updates( if ownership_lost.is_set(): return 0 count = 0 - 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 + digest_pending = set(delivery.digest_revisions) + new_revisions = tuple( + thread_revision + for thread_revision in delivery.thread_revisions + if thread_revision not in digest_pending + ) + individual_revisions = new_revisions[ + :max(0, max_individual_notifications) + ] + overflow_revisions = ( + delivery.digest_revisions + + new_revisions[len(individual_revisions):] ) - 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: + for thread_id, revision in individual_revisions: still_owner = await asyncio.to_thread( store.acquire_dispatch_lease, owner, @@ -138,10 +143,12 @@ async def dispatch_unread_updates( # poll instead of paying the endpoint deadline repeatedly. break await asyncio.to_thread( - store.mark_delivered, delivery.session_id, (thread_id,) + store.mark_delivered, + delivery.session_id, + ((thread_id, revision),), ) count += 1 - if overflow_ids and can_send_digest: + if overflow_revisions and can_send_digest: still_owner = await asyncio.to_thread( store.acquire_dispatch_lease, owner, @@ -153,11 +160,11 @@ async def dispatch_unread_updates( return count payload = json.dumps( { - "title": f"{len(overflow_ids)} new work updates", + "title": f"{len(overflow_revisions)} new work updates", "body": "Tap to review them in Stackchain.", "route": "#/my-work/updates", "tag": "stackchain-update-digest", - "update_count": len(overflow_ids), + "update_count": len(overflow_revisions), }, separators=(",", ":"), ) @@ -181,11 +188,13 @@ async def dispatch_unread_updates( await asyncio.to_thread( store.mark_digest_pending, delivery.session_id, - overflow_ids, + overflow_revisions, ) return count await asyncio.to_thread( - store.mark_delivered, delivery.session_id, overflow_ids + store.mark_delivered, + delivery.session_id, + overflow_revisions, ) count += 1 return count diff --git a/src/push_subscription_store.py b/src/push_subscription_store.py index a6b8dd2..ce4edc1 100644 --- a/src/push_subscription_store.py +++ b/src/push_subscription_store.py @@ -1,17 +1,45 @@ import json import os import sqlite3 +from collections.abc import Iterable, Mapping from dataclasses import dataclass from pathlib import Path -from typing import Iterable @dataclass(frozen=True) class PushDelivery: session_id: str subscription: dict - thread_ids: tuple[int, ...] - digest_ids: tuple[int, ...] = () + thread_revisions: tuple[tuple[int, str], ...] + digest_revisions: tuple[tuple[int, str], ...] = () + + @property + def thread_ids(self) -> tuple[int, ...]: + return tuple(thread_id for thread_id, _revision in self.thread_revisions) + + @property + def digest_ids(self) -> tuple[int, ...]: + return tuple(thread_id for thread_id, _revision in self.digest_revisions) + + +def _revisions( + values: Mapping[int, str] | Iterable[int | tuple[int, str]], +) -> tuple[tuple[int, str], ...]: + if isinstance(values, Mapping): + items = list(values.items()) + else: + items = [] + for value in values: + if isinstance(value, tuple): + items.append(value) + else: + items.append((value, "")) + normalized = {} + for thread_id, revision in items: + thread_id = int(thread_id) + if thread_id > 0: + normalized[thread_id] = str(revision or "") + return tuple(sorted(normalized.items())) class PushSubscriptionStore: @@ -32,6 +60,7 @@ class PushSubscriptionStore: CREATE TABLE IF NOT EXISTS push_deliveries ( session_id TEXT NOT NULL, thread_id INTEGER NOT NULL, + revision TEXT NOT NULL DEFAULT '', PRIMARY KEY (session_id, thread_id), FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id) ON DELETE CASCADE @@ -44,12 +73,28 @@ class PushSubscriptionStore: CREATE TABLE IF NOT EXISTS push_digest_pending ( session_id TEXT NOT NULL, thread_id INTEGER NOT NULL, + revision TEXT NOT NULL DEFAULT '', PRIMARY KEY (session_id, thread_id), FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id) ON DELETE CASCADE ); """ ) + delivery_columns = { + row[1] for row in connection.execute("PRAGMA table_info(push_deliveries)") + } + if "revision" not in delivery_columns: + connection.execute( + "ALTER TABLE push_deliveries ADD COLUMN revision TEXT NOT NULL DEFAULT '*'" + ) + digest_columns = { + row[1] + for row in connection.execute("PRAGMA table_info(push_digest_pending)") + } + if "revision" not in digest_columns: + connection.execute( + "ALTER TABLE push_digest_pending ADD COLUMN revision TEXT NOT NULL DEFAULT '*'" + ) os.chmod(self.path, 0o600) def _connect(self): @@ -109,8 +154,10 @@ class PushSubscriptionStore: "SELECT 1 FROM push_subscriptions WHERE session_id = ?", (session_id,) ).fetchone() is not None - def claim_unseen(self, thread_ids: Iterable[int]) -> list[PushDelivery]: - candidates = tuple(sorted({int(value) for value in thread_ids if int(value) > 0})) + def claim_unseen( + self, thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]] + ) -> list[PushDelivery]: + candidates = _revisions(thread_revisions) if not candidates: return [] with self._connect() as connection: @@ -120,42 +167,92 @@ class PushSubscriptionStore: deliveries = [] for session_id, encoded in rows: delivered = { - row[0] + row[0]: row[1] for row in connection.execute( - "SELECT thread_id FROM push_deliveries WHERE session_id = ?", + "SELECT thread_id, revision FROM push_deliveries WHERE session_id = ?", (session_id,), ) } - unseen = tuple(value for value in candidates if value not in delivered) + for thread_id, revision in candidates: + if delivered.get(thread_id) == "*": + connection.execute( + """UPDATE push_deliveries SET revision = ? + WHERE session_id = ? AND thread_id = ?""", + (revision, session_id, thread_id), + ) + delivered[thread_id] = revision + unseen = tuple( + candidate + for candidate in candidates + if ( + candidate[0] not in delivered + or ( + bool(candidate[1]) + and ( + not delivered[candidate[0]] + or candidate[1] > delivered[candidate[0]] + ) + ) + ) + ) if unseen: pending = { - row[0] + row[0]: row[1] for row in connection.execute( - "SELECT thread_id FROM push_digest_pending WHERE session_id = ?", + "SELECT thread_id, revision FROM push_digest_pending WHERE session_id = ?", (session_id,), ) } - digest_ids = tuple(value for value in unseen if value in pending) + digest_revisions = [] + for thread_id, revision in unseen: + if pending.get(thread_id) in {revision, "*"}: + digest_revisions.append((thread_id, revision)) + if pending[thread_id] == "*": + connection.execute( + """UPDATE push_digest_pending SET revision = ? + WHERE session_id = ? AND thread_id = ?""", + (revision, session_id, thread_id), + ) deliveries.append( - PushDelivery(session_id, json.loads(encoded), unseen, digest_ids) + PushDelivery( + session_id, + json.loads(encoded), + unseen, + tuple(digest_revisions), + ) ) return deliveries - def mark_digest_pending(self, session_id: str, thread_ids: Iterable[int]) -> None: + def mark_digest_pending( + self, + session_id: str, + thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]], + ) -> None: + values = _revisions(thread_revisions) 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), + """INSERT INTO push_digest_pending(session_id, thread_id, revision) + VALUES (?, ?, ?) + ON CONFLICT(session_id, thread_id) DO UPDATE SET + revision = excluded.revision""", + ((session_id, thread_id, revision) for thread_id, revision in values), ) - def mark_delivered(self, session_id: str, thread_ids: Iterable[int]) -> None: - values = tuple(int(thread_id) for thread_id in thread_ids) + def mark_delivered( + self, + session_id: str, + thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]], + ) -> None: + values = _revisions(thread_revisions) with self._connect() as connection: connection.executemany( - "INSERT OR IGNORE INTO push_deliveries(session_id, thread_id) VALUES (?, ?)", - ((session_id, thread_id) for thread_id in values), + """INSERT INTO push_deliveries(session_id, thread_id, revision) + VALUES (?, ?, ?) + ON CONFLICT(session_id, thread_id) DO UPDATE SET + revision = excluded.revision""", + ((session_id, thread_id, revision) for thread_id, revision in values), ) connection.executemany( "DELETE FROM push_digest_pending WHERE session_id = ? AND thread_id = ?", - ((session_id, thread_id) for thread_id in values), + ((session_id, thread_id) for thread_id, _revision in values), ) diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 21a5b38..7c9a725 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -1,6 +1,7 @@ import json import asyncio import os +import sqlite3 from types import SimpleNamespace import httpx @@ -56,6 +57,65 @@ def test_subscription_store_delivers_each_unread_thread_once_per_device(tmp_path assert second == [] +def test_subscription_store_reopens_a_delivered_thread_when_its_revision_changes(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"}, + }) + + first = store.claim_unseen({42: "2026-08-11T12:00:00Z"}) + store.mark_delivered("session-a", first[0].thread_revisions) + unchanged = store.claim_unseen({42: "2026-08-11T12:00:00Z"}) + updated = store.claim_unseen({42: "2026-08-11T12:05:00Z"}) + + assert unchanged == [] + assert updated[0].thread_revisions == ((42, "2026-08-11T12:05:00Z"),) + + +def test_subscription_store_does_not_realert_for_an_older_snapshot_revision(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"}, + }) + store.mark_delivered("session-a", {42: "2026-08-11T12:05:00Z"}) + + assert store.claim_unseen({42: "2026-08-11T12:00:00Z"}) == [] + + +def test_subscription_store_migrates_legacy_delivery_without_replaying_it(tmp_path): + path = tmp_path / "push.sqlite3" + subscription = { + "endpoint": "https://push.example/device-a", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + } + with sqlite3.connect(path) as connection: + connection.executescript(""" + CREATE TABLE push_subscriptions ( + session_id TEXT PRIMARY KEY, + endpoint TEXT NOT NULL UNIQUE, + subscription_json TEXT NOT NULL + ); + CREATE TABLE push_deliveries ( + session_id TEXT NOT NULL, + thread_id INTEGER NOT NULL, + PRIMARY KEY (session_id, thread_id) + ); + """) + connection.execute( + "INSERT INTO push_subscriptions VALUES (?, ?, ?)", + ("session-a", subscription["endpoint"], json.dumps(subscription)), + ) + connection.execute("INSERT INTO push_deliveries VALUES (?, ?)", ("session-a", 42)) + + store = PushSubscriptionStore(path) + + assert store.claim_unseen({42: "2026-08-11T12:00:00Z"}) == [] + updated = store.claim_unseen({42: "2026-08-11T12:05:00Z"}) + assert updated[0].thread_revisions == ((42, "2026-08-11T12:05:00Z"),) + + @pytest.mark.anyio async def test_managed_session_active_applies_configured_idle_deadline(monkeypatch): calls = [] @@ -150,6 +210,37 @@ 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_dispatch_realerts_when_the_same_thread_has_a_newer_revision(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"}, + }) + revision = "2026-08-11T12:00:00Z" + sent = [] + + async def unread(): + return {"items": [{ + "id": 42, + "updated_at": revision, + "repository": "private/repo", + "title": "Secret follow-up", + }]} + + async def send(_subscription, payload): + sent.append(json.loads(payload)) + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates(store, config, unread, send) == 1 + assert await dispatch_unread_updates(store, config, unread, send) == 0 + revision = "2026-08-11T12:05:00Z" + assert await dispatch_unread_updates(store, config, unread, send) == 1 + assert [payload["notification_id"] for payload in sent] == [42, 42] + assert "updated_at" not in json.dumps(sent) + assert "Secret follow-up" 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")