Re-alert Web Push when unread threads receive new updates #572

Merged
timmy merged 1 commits from timmy/571-revision-aware-web-push into main 2026-08-11 13:54:24 +00:00
4 changed files with 237 additions and 38 deletions

View File

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

View File

@ -59,12 +59,12 @@ async def dispatch_unread_updates(
return 0 return 0
try: try:
page = await unread() page = await unread()
thread_ids = { thread_revisions = {
int(item["id"]) int(item["id"]): str(item.get("updated_at") or "")
for item in page.get("items", []) for item in page.get("items", [])
if isinstance(item, dict) and str(item.get("id", "")).isdigit() 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: if session_active is not None:
try: try:
authorized = [await session_active(item.session_id) for item in deliveries] 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(): if ownership_lost.is_set():
return 0 return 0
count = 0 count = 0
digest_pending = set(delivery.digest_ids) digest_pending = set(delivery.digest_revisions)
new_ids = tuple( new_revisions = tuple(
thread_id thread_revision
for thread_id in delivery.thread_ids for thread_revision in delivery.thread_revisions
if thread_id not in digest_pending 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 can_send_digest = True
for thread_id in individual_ids: for thread_id, revision in individual_revisions:
still_owner = await asyncio.to_thread( still_owner = await asyncio.to_thread(
store.acquire_dispatch_lease, store.acquire_dispatch_lease,
owner, owner,
@ -138,10 +143,12 @@ async def dispatch_unread_updates(
# poll instead of paying the endpoint deadline repeatedly. # poll instead of paying the endpoint deadline repeatedly.
break break
await asyncio.to_thread( await asyncio.to_thread(
store.mark_delivered, delivery.session_id, (thread_id,) store.mark_delivered,
delivery.session_id,
((thread_id, revision),),
) )
count += 1 count += 1
if overflow_ids and can_send_digest: if overflow_revisions and can_send_digest:
still_owner = await asyncio.to_thread( still_owner = await asyncio.to_thread(
store.acquire_dispatch_lease, store.acquire_dispatch_lease,
owner, owner,
@ -153,11 +160,11 @@ async def dispatch_unread_updates(
return count return count
payload = json.dumps( 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.", "body": "Tap to review them in Stackchain.",
"route": "#/my-work/updates", "route": "#/my-work/updates",
"tag": "stackchain-update-digest", "tag": "stackchain-update-digest",
"update_count": len(overflow_ids), "update_count": len(overflow_revisions),
}, },
separators=(",", ":"), separators=(",", ":"),
) )
@ -181,11 +188,13 @@ async def dispatch_unread_updates(
await asyncio.to_thread( await asyncio.to_thread(
store.mark_digest_pending, store.mark_digest_pending,
delivery.session_id, delivery.session_id,
overflow_ids, overflow_revisions,
) )
return count return count
await asyncio.to_thread( await asyncio.to_thread(
store.mark_delivered, delivery.session_id, overflow_ids store.mark_delivered,
delivery.session_id,
overflow_revisions,
) )
count += 1 count += 1
return count return count

View File

@ -1,17 +1,45 @@
import json import json
import os import os
import sqlite3 import sqlite3
from collections.abc import Iterable, Mapping
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Iterable
@dataclass(frozen=True) @dataclass(frozen=True)
class PushDelivery: class PushDelivery:
session_id: str session_id: str
subscription: dict subscription: dict
thread_ids: tuple[int, ...] thread_revisions: tuple[tuple[int, str], ...]
digest_ids: tuple[int, ...] = () 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: class PushSubscriptionStore:
@ -32,6 +60,7 @@ class PushSubscriptionStore:
CREATE TABLE IF NOT EXISTS push_deliveries ( CREATE TABLE IF NOT EXISTS push_deliveries (
session_id TEXT NOT NULL, session_id TEXT NOT NULL,
thread_id INTEGER NOT NULL, thread_id INTEGER NOT NULL,
revision TEXT NOT NULL DEFAULT '',
PRIMARY KEY (session_id, thread_id), PRIMARY KEY (session_id, thread_id),
FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id) FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
ON DELETE CASCADE ON DELETE CASCADE
@ -44,12 +73,28 @@ class PushSubscriptionStore:
CREATE TABLE IF NOT EXISTS push_digest_pending ( CREATE TABLE IF NOT EXISTS push_digest_pending (
session_id TEXT NOT NULL, session_id TEXT NOT NULL,
thread_id INTEGER NOT NULL, thread_id INTEGER NOT NULL,
revision TEXT NOT NULL DEFAULT '',
PRIMARY KEY (session_id, thread_id), PRIMARY KEY (session_id, thread_id),
FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id) FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
ON DELETE CASCADE 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) os.chmod(self.path, 0o600)
def _connect(self): def _connect(self):
@ -109,8 +154,10 @@ class PushSubscriptionStore:
"SELECT 1 FROM push_subscriptions WHERE session_id = ?", (session_id,) "SELECT 1 FROM push_subscriptions WHERE session_id = ?", (session_id,)
).fetchone() is not None ).fetchone() is not None
def claim_unseen(self, thread_ids: Iterable[int]) -> list[PushDelivery]: def claim_unseen(
candidates = tuple(sorted({int(value) for value in thread_ids if int(value) > 0})) self, thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]]
) -> list[PushDelivery]:
candidates = _revisions(thread_revisions)
if not candidates: if not candidates:
return [] return []
with self._connect() as connection: with self._connect() as connection:
@ -120,42 +167,92 @@ class PushSubscriptionStore:
deliveries = [] deliveries = []
for session_id, encoded in rows: for session_id, encoded in rows:
delivered = { delivered = {
row[0] row[0]: row[1]
for row in connection.execute( 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,), (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: if unseen:
pending = { pending = {
row[0] row[0]: row[1]
for row in connection.execute( 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,), (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( deliveries.append(
PushDelivery(session_id, json.loads(encoded), unseen, digest_ids) PushDelivery(
session_id,
json.loads(encoded),
unseen,
tuple(digest_revisions),
)
) )
return deliveries 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: with self._connect() as connection:
connection.executemany( connection.executemany(
"INSERT OR IGNORE INTO push_digest_pending(session_id, thread_id) VALUES (?, ?)", """INSERT INTO push_digest_pending(session_id, thread_id, revision)
((session_id, int(thread_id)) for thread_id in thread_ids), 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: def mark_delivered(
values = tuple(int(thread_id) for thread_id in thread_ids) self,
session_id: str,
thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]],
) -> None:
values = _revisions(thread_revisions)
with self._connect() as connection: with self._connect() as connection:
connection.executemany( connection.executemany(
"INSERT OR IGNORE INTO push_deliveries(session_id, thread_id) VALUES (?, ?)", """INSERT INTO push_deliveries(session_id, thread_id, revision)
((session_id, thread_id) for thread_id in values), 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( connection.executemany(
"DELETE FROM push_digest_pending WHERE session_id = ? AND thread_id = ?", "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),
) )

View File

@ -1,6 +1,7 @@
import json import json
import asyncio import asyncio
import os import os
import sqlite3
from types import SimpleNamespace from types import SimpleNamespace
import httpx import httpx
@ -56,6 +57,65 @@ def test_subscription_store_delivers_each_unread_thread_once_per_device(tmp_path
assert second == [] 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 @pytest.mark.anyio
async def test_managed_session_active_applies_configured_idle_deadline(monkeypatch): async def test_managed_session_active_applies_configured_idle_deadline(monkeypatch):
calls = [] 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) 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 @pytest.mark.anyio
async def test_update_burst_sends_bounded_individual_pushes_and_one_private_digest(tmp_path): async def test_update_burst_sends_bounded_individual_pushes_and_one_private_digest(tmp_path):
store = PushSubscriptionStore(tmp_path / "push.sqlite3") store = PushSubscriptionStore(tmp_path / "push.sqlite3")