""" Notifications on a phone — /v1/push/*: pushed when it can be, fetched when not. A phone registers once and gets a row here. **With a UnifiedPush distributor** it gives an endpoint and a P-256 key, and every notification `create_notification` lets through is sent there, encrypted to the phone (`webpush.py`). **Without one** — nothing to install is the default — the row has no endpoint, and the phone fetches what is new with `POST /v1/push/poll` every quarter of an hour or so. Both are the same rows and the same payload, so a phone can move between them (a distributor installed, removed, refusing) without the hub caring which. **Nothing reaches a phone that was not created**: a muted group and an account with every notification turned off stop at `create_notification`, before this module is reached, so the two switches the person sees are the only two there are. The poll is authenticated by a secret issued with the row, not by a session. It reads notification lines and nothing else, so a phone running in the background holds no token that could do anything more — and a sign-out, which deletes the row, ends it. Who pays (§13.5b): a member's chat costs every other member's phones a push. That fan-out is already bounded where it starts — `chat_notify` is budgeted per node — and here a conversation reaches each phone at most once per `CHAT_COALESCE` seconds: the phone shows one line per group, so the pushes in between would only have replaced it. An account holds `MAX_SUBSCRIPTIONS` rows at most, because each is one outbound request per notification; a row is polled at most once per `POLL_MIN_INTERVAL`. """ import asyncio import hashlib import hmac import logging import math import secrets import time from datetime import UTC, datetime from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub import webpush from meshbay_hub.api.deps import require_user_scope from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import Notification, PushSubscription, User log = logging.getLogger(__name__) router = APIRouter(prefix="/v1/push", tags=["push"]) MAX_SUBSCRIPTIONS = 10 CHAT_COALESCE = 30.0 _COALESCE_ENTRIES = 10_000 POLL_MIN_INTERVAL = 60.0 POLL_LIMIT = 20 # Strong references: asyncio holds a task weakly, and a collected one is a push # that silently never went (CLAUDE.md, "a background task nobody holds"). _tasks: set[asyncio.Task] = set() _last_chat: dict[tuple[str, str], float] = {} _last_poll: dict[str, float] = {} # Replaced by the tests; the real one never raises. _send = webpush.send class SubscriptionIn(BaseModel): # The row this phone already has, to update rather than add one: a phone # with no endpoint has nothing else to be recognised by. id: str | None = Field(default=None, max_length=36) endpoint: str | None = Field(default=None, max_length=webpush.MAX_ENDPOINT) # Lengths bounded before decoding: base64 decoding skips characters outside # its alphabet, so an unbounded string could still decode to 65 bytes. p256dh: str | None = Field(default=None, max_length=128) auth: str | None = Field(default=None, max_length=32) def _hash(secret: str) -> str: return hashlib.sha256(secret.encode()).hexdigest() def _payload(notif: Notification) -> dict: """What a phone is told, pushed or fetched: the hub's own line, never a message.""" return { "v": 1, "id": notif.id, "kind": notif.kind, "title": notif.title, "link": notif.link, "group_id": notif.group_id, "created_at": _iso(notif.created_at), } def _iso(at: datetime) -> str: # SQLite hands back naive datetimes; every one stored here is UTC. return (at if at.tzinfo else at.replace(tzinfo=UTC)).isoformat() @router.post("/subscriptions") async def subscribe( body: SubscriptionIn, current_user: User = Depends(require_user_scope), db: AsyncSession = Depends(get_db), ): """ Register this phone, or update its row: with an endpoint and keys when it has a push distributor, without them when it will fetch instead. Answers the row's id, a fresh secret for `POST /v1/push/poll` (the previous one stops working) and the hub's time, from which the phone counts what is new — what was there before it registered is not news. """ if body.endpoint is not None: if body.p256dh is None or body.auth is None: raise HTTPException(status_code=422, detail="an endpoint needs its keys") try: webpush.check_endpoint(body.endpoint) webpush.check_keys(body.p256dh, body.auth) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) from e sub = None if body.id is not None: sub = await db.get(PushSubscription, body.id) if sub is not None and sub.user_id != current_user.id: sub = None if body.endpoint is not None: same = (await db.execute( select(PushSubscription).where( PushSubscription.user_id == current_user.id, PushSubscription.endpoint == body.endpoint))).scalar_one_or_none() if sub is None: sub = same elif same is not None and same.id != sub.id: # The endpoint moved to this row; the old one would only repeat it. await db.delete(same) await db.flush() if sub is None: held = (await db.execute( select(func.count()).select_from(PushSubscription) .where(PushSubscription.user_id == current_user.id))).scalar() or 0 if held >= MAX_SUBSCRIPTIONS: raise HTTPException(status_code=429, detail="Too many push subscriptions") sub = PushSubscription(user_id=current_user.id) db.add(sub) secret = secrets.token_urlsafe(32) sub.endpoint, sub.p256dh, sub.auth = body.endpoint, body.p256dh, body.auth sub.poll_hash = _hash(secret) await db.commit() return {"id": sub.id, "poll_secret": secret, "now": datetime.now(UTC).isoformat()} class PollIn(BaseModel): id: str = Field(max_length=36) secret: str = Field(max_length=64) since: datetime @router.post("/poll") async def poll(body: PollIn, db: AsyncSession = Depends(get_db)): """ What is new for this phone since `since`: the same payloads a push carries, oldest first, at most twenty. Authenticated by the row's secret rather than a session, so what a phone keeps for running in the background reads notification lines and nothing else. A wrong secret and an unknown row answer the same 404. """ sub = await db.get(PushSubscription, body.id) if (sub is None or sub.poll_hash is None or not hmac.compare_digest(sub.poll_hash, _hash(body.secret))): raise HTTPException(status_code=404, detail="Subscription not found") now = time.monotonic() last = _last_poll.get(sub.id) if last is not None and now - last < POLL_MIN_INTERVAL: raise HTTPException(status_code=429, detail="Polled too often", headers={"Retry-After": str(int(POLL_MIN_INTERVAL - (now - last)) + 1)}) if len(_last_poll) >= _COALESCE_ENTRIES: for k in [k for k, at in _last_poll.items() if now - at >= POLL_MIN_INTERVAL]: del _last_poll[k] _last_poll[sub.id] = now since = body.since if body.since.tzinfo else body.since.replace(tzinfo=UTC) rows = (await db.execute( select(Notification).where( Notification.user_id == sub.user_id, Notification.created_at > since.astimezone(UTC), ).order_by(Notification.created_at.desc()).limit(POLL_LIMIT) )).scalars().all() return {"notifications": [_payload(n) for n in reversed(rows)]} @router.delete("/subscriptions/{subscription_id}") async def unsubscribe( subscription_id: str, current_user: User = Depends(require_user_scope), db: AsyncSession = Depends(get_db), ): """Stop telling one phone anything: turned off there, or signed out of.""" sub = await db.get(PushSubscription, subscription_id) if sub is None or sub.user_id != current_user.id: raise HTTPException(status_code=404, detail="Subscription not found") await db.delete(sub) await db.commit() return {"status": "ok"} def _coalesced(sub_id: str, group_id: str, now: float) -> bool: key = (sub_id, group_id) if now - _last_chat.get(key, -math.inf) < CHAT_COALESCE: return True if len(_last_chat) >= _COALESCE_ENTRIES: for k in [k for k, at in _last_chat.items() if now - at >= CHAT_COALESCE]: del _last_chat[k] _last_chat[key] = now return False async def push_notification(db: AsyncSession, notif: Notification) -> None: """ Send `notif` to the person's phones, off the caller's path. Called by `create_notification` once the row exists, inside the caller's transaction: the subscriptions are read there, the requests leave in a task of their own, so a slow push server delays nobody's request. """ subs = (await db.execute( select(PushSubscription).where(PushSubscription.user_id == notif.user_id, PushSubscription.endpoint.is_not(None)) )).scalars().all() if not subs: return payload = _payload(notif) now = time.monotonic() targets = [ (s.id, webpush.Target(s.endpoint, s.p256dh, s.auth)) for s in subs if not (notif.kind == "chat_message" and notif.group_id and _coalesced(s.id, notif.group_id, now)) ] if not targets: return ttl = webpush.TTL_CHAT if notif.kind == "chat_message" else webpush.TTL_OTHER task = asyncio.get_running_loop().create_task(_deliver(targets, payload, ttl)) _tasks.add(task) task.add_done_callback(_tasks.discard) async def _deliver(targets: list[tuple[str, webpush.Target]], payload: dict, ttl: int) -> None: results = await asyncio.gather(*(_send(t, payload, ttl=ttl) for _, t in targets), return_exceptions=True) gone = [sid for (sid, _), r in zip(targets, results, strict=True) if r == webpush.GONE] if not gone: return # The distributor dropped the registration: pushing there would cost a # request per notification for ever and reach nothing. The row stays, without # its endpoint — the phone still fetches, and registers again when it can. try: from meshbay_hub.db.engine import get_session_factory async with get_session_factory()() as db: await db.execute( update(PushSubscription).where(PushSubscription.id.in_(gone)) .values(endpoint=None, p256dh=None, auth=None)) await db.commit() except Exception as e: log.warning("Could not drop %d gone push subscription(s): %s", len(gone), e) async def drain() -> None: """Wait for every push in flight — for the tests, and for a clean shutdown.""" # Done tasks leave the set from a callback the loop has not run yet, and # awaiting a finished gather never yields to it: wait on the unfinished only. while pending := [t for t in _tasks if not t.done()]: await asyncio.gather(*pending, return_exceptions=True)