path: root/packages/meshbay-hub/src/meshbay_hub/api/push.py

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