Merge pull request 'Re-alert Web Push when unread threads receive new updates' (#572) from timmy/571-revision-aware-web-push into main

Merge pull request 'Re-alert Web Push when unread threads receive new updates' (#572)

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='<url-safe-public-key>'
 export STACKCHAIN_VAPID_PRIVATE_KEY='<private-key-from-secret-manager>'
 export STACKCHAIN_VAPID_SUBJECT='mailto:ops@example.com'

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

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),
             )

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")