Make unread Web Push use the production notification contract #566

Merged
timmy merged 1 commits from timmy/565-production-push-contract into main 2026-08-11 10:57:49 +00:00
3 changed files with 91 additions and 15 deletions

View File

@ -1739,9 +1739,9 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
headers={"Retry-After": "1"},
) from error
existing_ids = {
int(item["notification_id"])
int(item["id"])
for item in current.get("items", [])
if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit()
if isinstance(item, dict) and str(item.get("id", "")).isdigit()
}
await asyncio.to_thread(
_push_subscription_store.mark_delivered, device_id, existing_ids

View File

@ -59,9 +59,9 @@ async def dispatch_unread_updates(
try:
page = await unread()
thread_ids = {
int(item["notification_id"])
int(item["id"])
for item in page.get("items", [])
if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit()
if isinstance(item, dict) and str(item.get("id", "")).isdigit()
}
deliveries = await asyncio.to_thread(store.claim_unseen, thread_ids)
if session_active is not None:

View File

@ -3,9 +3,10 @@ import asyncio
import os
from types import SimpleNamespace
import httpx
import pytest
from src import dashboard_auth, main
from src import dashboard_auth, gitea_proxy, main
from src.push_notifications import PushConfiguration, dispatch_unread_updates
from src.push_subscription_store import PushSubscriptionStore
@ -127,7 +128,7 @@ async def test_dispatch_sends_one_privacy_safe_deep_link_per_new_thread(tmp_path
async def unread():
return {"items": [
{"notification_id": 42, "repository": "private/repo", "title": "Secret title"},
{"id": 42, "repository": "private/repo", "title": "Secret title"},
]}
async def send(subscription, payload):
@ -147,6 +148,52 @@ 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_production_notification_page_dispatches_one_unread_push(tmp_path):
def upstream(_request):
return httpx.Response(200, json=[{
"id": 42,
"unread": True,
"repository": {"full_name": "private/repo"},
"subject": {
"title": "Secret title",
"type": "Issue",
"html_url": "https://forge.example/private/repo/issues/7",
},
}])
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
try:
page = await gitea_proxy.notification_page(1)
finally:
await gitea_proxy.stop_client()
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 page
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 sent == [{
"title": "New work update",
"body": "Tap to review it in Stackchain.",
"route": "#/my-work/update/42",
"tag": "stackchain-update-42",
"notification_id": 42,
}]
assert "private/repo" not in json.dumps(sent)
assert "Secret title" not in json.dumps(sent)
@pytest.mark.anyio
async def test_concurrent_workers_do_not_dispatch_the_same_update(tmp_path):
path = tmp_path / "push.sqlite3"
@ -161,7 +208,7 @@ async def test_concurrent_workers_do_not_dispatch_the_same_update(tmp_path):
sent = []
async def unread():
return {"items": [{"notification_id": 42}]}
return {"items": [{"id": 42}]}
async def send(_subscription, payload):
sent.append(json.loads(payload)["tag"])
@ -193,7 +240,7 @@ async def test_dispatch_removes_an_expired_push_endpoint(tmp_path):
response = SimpleNamespace(status_code=410)
async def unread():
return {"items": [{"notification_id": 8}]}
return {"items": [{"id": 8}]}
async def send(_subscription, _payload):
raise Expired()
@ -214,7 +261,7 @@ async def test_transient_endpoint_failure_does_not_block_healthy_devices(tmp_pat
sent = []
async def unread():
return {"items": [{"notification_id": 8}]}
return {"items": [{"id": 8}]}
async def send(subscription, _payload):
if subscription["endpoint"].endswith("session-a"):
@ -239,8 +286,8 @@ async def test_transient_failure_stops_that_device_until_the_next_poll(tmp_path)
async def unread():
return {"items": [
{"notification_id": 8},
{"notification_id": 9},
{"id": 8},
{"id": 9},
]}
async def send(subscription, payload):
@ -270,7 +317,7 @@ async def test_timed_out_endpoint_does_not_stall_push_fanout(tmp_path):
sent = []
async def unread():
return {"items": [{"notification_id": 13}]}
return {"items": [{"id": 13}]}
async def send(subscription, _payload):
if subscription["endpoint"].endswith("session-a"):
@ -303,7 +350,7 @@ async def test_dispatches_devices_concurrently_with_a_strict_bound(tmp_path):
release = asyncio.Event()
async def unread():
return {"items": [{"notification_id": 13}]}
return {"items": [{"id": 13}]}
async def send(_subscription, _payload):
nonlocal active, peak
@ -339,7 +386,7 @@ async def test_dispatch_removes_inactive_sessions_without_blocking_active_device
sent = []
async def unread():
return {"items": [{"notification_id": 21}]}
return {"items": [{"id": 21}]}
async def session_active(management_id):
return management_id == "active-device"
@ -366,7 +413,7 @@ async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_regi
sent = []
async def unread():
return {"items": [{"notification_id": 22}]}
return {"items": [{"id": 22}]}
async def registry_unavailable(_management_id):
raise RuntimeError("session registry unavailable")
@ -417,6 +464,35 @@ async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(
assert (await main.push_status(request))["subscribed"] is False
@pytest.mark.anyio
async def test_subscription_baselines_production_notification_ids(tmp_path, monkeypatch):
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
monkeypatch.setattr(main, "_push_subscription_store", store)
monkeypatch.setattr(
main,
"_push_configuration",
lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
)
async def management_id(_session):
return "device-a"
async def current_unread():
return {"items": [{"id": 42}]}
monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
monkeypatch.setattr(main, "notifications", current_unread)
request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
payload = main.PushSubscriptionPayload(
endpoint="https://push.example/device-a",
keys={"p256dh": "public-key", "auth": "auth-secret"},
)
assert await main.subscribe_push(payload, request) == {"subscribed": True}
assert store.claim_unseen({42}) == []
assert store.claim_unseen({42, 43})[0].thread_ids == (43,)
@pytest.mark.anyio
async def test_subscription_fails_closed_when_existing_unread_baseline_is_unavailable(tmp_path, monkeypatch):
store = PushSubscriptionStore(tmp_path / "push.sqlite3")