Add PWA + Web Push delivery for weather alerts (#95)

Make the app installable and deliver existing alert notifications as OS
push, alongside the in-app bell.

Backend:
- PushSubscription model (per-device endpoint + keys, owned by a user) and
  register/unregister/test endpoints under /api/v2/push, cookie-auth scoped
  to the user like the subscription routes.
- push.py: VAPID key management (env -> data/vapid.json -> generated) and a
  pywebpush send helper that reports gone endpoints for pruning. No DB coupling.
- notify.py: after creating an in-app Notification, dispatch Web Push to the
  user's devices (guarded — a push failure never affects the in-app write;
  endpoints reported gone are pruned).
- Serve the .webmanifest with the correct media type.

Frontend:
- manifest.webmanifest + 192/maskable icons; <link rel="manifest"> on all pages.
- sw.js: push + notificationclick handlers (push-only; no fetch caching, so it
  doesn't fight the existing IndexedDB cache). Registered globally in nav.js in
  secure contexts.
- push-client.js + a "Notifications on this device" toggle and test-send on the
  /alerts page, subscribing through the existing cookie-aware apiFetch.

Push and service workers require a secure context, so this is active over HTTPS
(or http://localhost) and cleanly no-ops on a plain-HTTP LAN origin.

api_accounts.py

@@ -4,23 +4,33 @@ All routes require a logged-in user (the fastapi-users cookie session) and are
 scoped to that user — a row that isn't theirs reads as 404, never 403, so the API
 doesn't leak which ids exist. Mounted under {BASE}/api/v2 alongside the rest of v2.
 """
-from fastapi import APIRouter, Depends, HTTPException, Query
+import os
+import time
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.concurrency import run_in_threadpool
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import climate
 import grid
+import push
 from db import get_async_session
-from models import Notification, Subscription
+from models import Notification, PushSubscription, Subscription
 from schemas import (
     NotificationList,
     NotificationOut,
+    PushSubscriptionIn,
+    PushUnsubscribeIn,
     SubscriptionIn,
     SubscriptionOut,
     SubscriptionPatch,
 )
 from users import current_active_user
 
+_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
+BASE = f"/{_BASE}" if _BASE else ""
+
 router = APIRouter(tags=["accounts"])
 
 
@@ -180,8 +190,6 @@ async def mark_all_read(
     user=Depends(current_active_user),
     session: AsyncSession = Depends(get_async_session),
 ):
-    import time
-
     rows = (
         await session.execute(
             select(Notification).where(
@@ -194,3 +202,106 @@ async def mark_all_read(
         n.read_at = now
     if rows:
         await session.commit()
+
+
+# --- web push -----------------------------------------------------------------
+# The VAPID public key is not a secret and the browser needs it before it can
+# subscribe, so this one route is open (still under BASE). Everything else is
+# user-scoped like the subscription routes above.
+@router.get("/push/vapid-key")
+async def push_vapid_key():
+    return {"key": push.public_key()}
+
+
+@router.post("/push/subscribe", status_code=201)
+async def push_subscribe(
+    body: PushSubscriptionIn,
+    request: Request,
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    """Register (or refresh) this device's Web Push endpoint for the user.
+
+    Keyed on `endpoint` — re-subscribing from the same device updates the keys in
+    place rather than creating a duplicate, and a device that was registered to a
+    different account is re-homed to the current user.
+    """
+    now = time.time()
+    existing = (
+        await session.execute(
+            select(PushSubscription).where(PushSubscription.endpoint == body.endpoint)
+        )
+    ).scalar_one_or_none()
+    if existing is not None:
+        existing.user_id = user.id
+        existing.p256dh = body.keys.p256dh
+        existing.auth = body.keys.auth
+        existing.user_agent = request.headers.get("user-agent")
+        existing.last_used_at = now
+    else:
+        session.add(
+            PushSubscription(
+                user_id=user.id,
+                endpoint=body.endpoint,
+                p256dh=body.keys.p256dh,
+                auth=body.keys.auth,
+                user_agent=request.headers.get("user-agent"),
+                created_at=now,
+                last_used_at=now,
+            )
+        )
+    await session.commit()
+    return {"ok": True}
+
+
+@router.delete("/push/subscribe", status_code=204)
+async def push_unsubscribe(
+    body: PushUnsubscribeIn,
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    """Drop this device's endpoint (called on toggle-off / permission revoked)."""
+    row = (
+        await session.execute(
+            select(PushSubscription).where(
+                PushSubscription.endpoint == body.endpoint,
+                PushSubscription.user_id == user.id,
+            )
+        )
+    ).scalar_one_or_none()
+    if row is not None:
+        await session.delete(row)
+        await session.commit()
+
+
+@router.post("/push/test", status_code=202)
+async def push_test(
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    """Send a canned push to every device registered to the user — the on-device
+    delivery check. Prunes any endpoint the push service reports as gone."""
+    rows = (
+        await session.execute(
+            select(PushSubscription).where(PushSubscription.user_id == user.id)
+        )
+    ).scalars().all()
+    payload = {
+        "title": "Thermograph",
+        "body": "Push notifications are working on this device.",
+        "url": f"{BASE}/alerts",
+        "tag": "thermograph-test",
+    }
+    sent, pruned = 0, 0
+    for row in rows:
+        info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}
+        # pywebpush blocks on network I/O — keep it off the event loop.
+        result = await run_in_threadpool(push.send, info, payload)
+        if result == "ok":
+            sent += 1
+        elif result == "gone":
+            await session.delete(row)
+            pruned += 1
+    if pruned:
+        await session.commit()
+    return {"devices": len(rows), "sent": sent, "pruned": pruned}

app.py

@@ -4,6 +4,7 @@ import datetime
 import functools
 import hashlib
 import json
+import mimetypes
 import os
 import queue
 import threading
@@ -40,6 +41,10 @@ FRONTEND_DIR = os.path.join(os.path.dirname(__file__), "..", "frontend")
 _BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
 BASE = f"/{_BASE}" if _BASE else ""
 
+# The PWA manifest is served as a static file; register its media type so
+# StaticFiles labels it correctly (not in Python's default mimetypes map).
+mimetypes.add_type("application/manifest+json", ".webmanifest")
+
 
 def _weather_fetch_error(e) -> HTTPException:
     """A clean, retryable 503 when weather data is temporarily unavailable; the

models.py

@@ -109,3 +109,35 @@ class Notification(Base):
         Index("idx_notif_user_created", "user_id", "created_at"),
         Index("idx_notif_user_read", "user_id", "read_at"),
     )
+
+
+class PushSubscription(Base):
+    """A single browser/device Web Push registration, owned by a user.
+
+    One row per device (a user with a phone + a laptop has two). The `endpoint`
+    is the push service URL the browser handed us; it's the natural identity, so
+    re-subscribing from the same device updates the keys in place rather than
+    duplicating. Rows are pruned when the push service reports the endpoint gone
+    (404/410) — see notify.py / api_accounts.py.
+    """
+    __tablename__ = "push_subscription"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        GUID, ForeignKey("user.id", ondelete="CASCADE"), nullable=False
+    )
+    # The push service URL (per-device). Unique — it identifies the device.
+    endpoint: Mapped[str] = mapped_column(Text, nullable=False)
+    # The two client keys from PushSubscription.toJSON().keys, used to encrypt the
+    # payload so only this device can read it.
+    p256dh: Mapped[str] = mapped_column(String(200), nullable=False)
+    auth: Mapped[str] = mapped_column(String(100), nullable=False)
+    # Best-effort label for a future "manage devices" view.
+    user_agent: Mapped[str | None] = mapped_column(String(300), nullable=True)
+    created_at: Mapped[float] = mapped_column(Float, nullable=False, default=time.time)
+    last_used_at: Mapped[float | None] = mapped_column(Float, nullable=True)
+
+    __table_args__ = (
+        UniqueConstraint("endpoint", name="uq_push_endpoint"),
+        Index("idx_push_user", "user_id"),
+    )

notify.py

@@ -32,10 +32,14 @@ from sqlalchemy import delete, select
 import climate
 import grading
 import grid
+import push
 from db import sync_session_maker
-from models import AccessToken, Notification, Subscription
+from models import AccessToken, Notification, PushSubscription, Subscription
 from views import OBS_COLS
 
+_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
+BASE = f"/{_BASE}" if _BASE else ""
+
 WEEK_SECONDS = 7 * 86400
 INTERVAL = int(os.environ.get("THERMOGRAPH_NOTIFY_INTERVAL", "900"))  # 15 min default
 SESSION_TTL_SECONDS = int(os.environ.get("THERMOGRAPH_SESSION_TTL_DAYS", "30")) * 86400
@@ -104,6 +108,31 @@ def _compose(sub, event_date, metric, direction, graded):
     return title, body
 
 
+# --- push delivery -----------------------------------------------------------
+def _deep_link(sub: Subscription, event_date: str) -> str:
+    """Where a tapped notification lands — the single-day breakdown for the
+    subscribed cell on the event date."""
+    return f"{BASE}/day#lat={sub.lat:.5f}&lon={sub.lon:.5f}&date={event_date}"
+
+
+def _dispatch_push(session, sub, title, body, event_date) -> None:
+    """Best-effort Web Push of a just-created notification to all of the user's
+    devices. Prunes endpoints the push service reports gone. Runs in the notifier
+    daemon thread (no event loop), so the blocking send is fine here. Never raises
+    — the in-app notification is already committed regardless."""
+    rows = session.execute(
+        select(PushSubscription).where(PushSubscription.user_id == sub.user_id)
+    ).scalars().all()
+    if not rows:
+        return
+    payload = {"title": title, "body": body, "url": _deep_link(sub, event_date),
+               "tag": f"sub-{sub.id}"}
+    for row in rows:
+        info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}
+        if push.send(info, payload) == "gone":
+            session.delete(row)   # committed with the rest of the pass
+
+
 # --- per-cell evaluation -----------------------------------------------------
 def _obs_from_row(row) -> dict:
     return {k: row[k] for k in OBS_COLS if k in row}
@@ -214,6 +243,12 @@ def _process_cell(session, cell_id, subs, today, now, archive_budget):
             continue
         sub.last_notified_at = now
         created += 1
+        # Additionally deliver over Web Push (the in-app row above is the record of
+        # truth for the bell; push is a side-channel that must never break it).
+        try:
+            _dispatch_push(session, sub, title, body, event_date)
+        except Exception:  # noqa: BLE001 - push failures stay isolated from the DB write
+            pass
     session.commit()
     return created
 

push.py

@@ -0,0 +1,121 @@
+"""Web Push (VAPID) — key management and the low-level send.
+
+This is the transport under the notification engine: notify.py decides *what* to
+say and *to whom*; this module signs and delivers it to a browser push service.
+
+Keys are resolved once, in this order (mirrors the self-signed-cert flow in
+run.sh — generated on first use, reused thereafter):
+
+1. env — ``THERMOGRAPH_VAPID_PRIVATE_KEY`` + ``THERMOGRAPH_VAPID_PUBLIC_KEY``,
+   both base64url-raw (the private key is the 32-byte scalar, the public key the
+   65-byte uncompressed point — the standard VAPID interchange format). Set both
+   in production so every worker/deploy signs with the same identity.
+2. ``data/vapid.json`` — a gitignored keypair written on first run.
+3. freshly generated, then written to (2).
+
+The private key signs the VAPID JWT; the public key is the ``applicationServerKey``
+the browser needs at subscribe time (served via GET /push/vapid-key).
+
+Nothing here touches the database — ``send()`` returns a status string and the
+caller (which owns the session) prunes rows the push service says are gone.
+"""
+import base64
+import json
+import logging
+import os
+import threading
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from pywebpush import WebPushException, webpush
+
+log = logging.getLogger("thermograph.push")
+
+_DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data"))
+_VAPID_PATH = os.environ.get("THERMOGRAPH_VAPID_FILE") or os.path.join(_DATA_DIR, "vapid.json")
+# The VAPID "sub" claim — a contact the push service can reach about our traffic.
+_CONTACT = os.environ.get("THERMOGRAPH_VAPID_CONTACT", "mailto:admin@thermograph.app")
+
+_lock = threading.Lock()
+_keys = None  # cached {"private_key": str, "public_key": str} (base64url-raw)
+
+
+def _b64url(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+
+def _generate() -> dict:
+    """A fresh P-256 keypair as base64url-raw: 32-byte private scalar + 65-byte
+    uncompressed public point (the format pywebpush's from_string and the browser
+    applicationServerKey both accept)."""
+    priv = ec.generate_private_key(ec.SECP256R1())
+    scalar = priv.private_numbers().private_value.to_bytes(32, "big")
+    raw_pub = priv.public_key().public_bytes(
+        serialization.Encoding.X962,
+        serialization.PublicFormat.UncompressedPoint,
+    )
+    return {"private_key": _b64url(scalar), "public_key": _b64url(raw_pub)}
+
+
+def _load() -> dict:
+    """Resolve the keypair once (env → file → generate) and cache it."""
+    global _keys
+    if _keys is not None:
+        return _keys
+    with _lock:
+        if _keys is not None:
+            return _keys
+        env_priv = os.environ.get("THERMOGRAPH_VAPID_PRIVATE_KEY")
+        env_pub = os.environ.get("THERMOGRAPH_VAPID_PUBLIC_KEY")
+        if env_priv and env_pub:
+            _keys = {"private_key": env_priv.strip(), "public_key": env_pub.strip()}
+            return _keys
+        try:
+            with open(_VAPID_PATH, encoding="utf-8") as f:
+                data = json.load(f)
+            if data.get("private_key") and data.get("public_key"):
+                _keys = {"private_key": data["private_key"], "public_key": data["public_key"]}
+                return _keys
+        except (OSError, ValueError):
+            pass
+        _keys = _generate()
+        try:
+            os.makedirs(_DATA_DIR, exist_ok=True)
+            with open(_VAPID_PATH, "w", encoding="utf-8") as f:
+                json.dump(_keys, f)
+            os.chmod(_VAPID_PATH, 0o600)
+        except OSError:
+            log.warning("could not persist VAPID keys to %s; using an in-memory pair", _VAPID_PATH)
+        return _keys
+
+
+def public_key() -> str:
+    """The applicationServerKey (base64url raw point) the browser subscribes with."""
+    return _load()["public_key"]
+
+
+def send(subscription_info: dict, payload: dict) -> str:
+    """Deliver one push. Returns 'ok', 'gone' (prune the row), or 'error'.
+
+    `subscription_info` is the browser shape: {endpoint, keys:{p256dh, auth}}.
+    Never raises — a bad send must not break the caller's transaction.
+    """
+    keys = _load()
+    try:
+        webpush(
+            subscription_info=subscription_info,
+            data=json.dumps(payload),
+            vapid_private_key=keys["private_key"],
+            vapid_claims={"sub": _CONTACT},
+            ttl=86400,
+        )
+        return "ok"
+    except WebPushException as e:
+        status = getattr(getattr(e, "response", None), "status_code", None)
+        if status in (404, 410):
+            return "gone"          # endpoint retired — caller should delete it
+        log.warning("web push failed (status=%s): %s", status, e)
+        return "error"
+    except Exception:  # noqa: BLE001 - network/encoding errors must not propagate
+        log.warning("web push send raised", exc_info=True)
+        return "error"

requirements.txt

@@ -6,3 +6,6 @@ numpy==2.2.1
 # Accounts + notification subscriptions (see backend/db.py, users.py, notify.py).
 fastapi-users[sqlalchemy]==15.0.5
 aiosqlite==0.22.1
+# Web Push (VAPID) delivery of notifications (see backend/push.py). Pulls in
+# py-vapid, cryptography, and http-ece.
+pywebpush==2.0.0

schemas.py

@@ -112,3 +112,19 @@ class NotificationOut(BaseModel):
 class NotificationList(BaseModel):
     notifications: list[NotificationOut]
     unread_count: int
+
+
+# --- web push ----------------------------------------------------------------
+class PushKeys(BaseModel):
+    p256dh: str
+    auth: str
+
+
+class PushSubscriptionIn(BaseModel):
+    """Matches the browser's PushSubscription.toJSON() shape."""
+    endpoint: str
+    keys: PushKeys
+
+
+class PushUnsubscribeIn(BaseModel):
+    endpoint: str

tests/test_api_accounts.py

@@ -99,3 +99,55 @@ def test_notifications_empty_and_read_all():
     assert r.status_code == 200
     assert r.json() == {"notifications": [], "unread_count": 0}
     assert c.post(f"{V2}/notifications/read-all").status_code == 204
+
+
+# --- web push ---------------------------------------------------------------
+_SUB = {"endpoint": "https://push.example.com/ep-1", "keys": {"p256dh": "BKEY", "auth": "YXV0aA"}}
+
+
+def test_push_vapid_key_is_open():
+    # The public key is not a secret and the browser needs it before login, so
+    # this route is unauthenticated.
+    r = _client().get(f"{V2}/push/vapid-key")
+    assert r.status_code == 200
+    assert isinstance(r.json()["key"], str) and r.json()["key"]
+
+
+def test_push_requires_auth():
+    c = _client()
+    assert c.post(f"{V2}/push/subscribe", json=_SUB).status_code == 401
+    assert c.post(f"{V2}/push/test").status_code == 401
+
+
+def test_push_subscribe_upsert_test_unsubscribe(monkeypatch):
+    import push
+    sent = []
+    monkeypatch.setattr(push, "send", lambda info, payload: sent.append(info["endpoint"]) or "ok")
+    c = _client()
+    _login(c, "push@example.com")
+
+    assert c.post(f"{V2}/push/subscribe", json=_SUB).status_code == 201
+    # Re-subscribe from the same device (same endpoint, new keys) upserts — one device.
+    assert c.post(f"{V2}/push/subscribe",
+                  json={**_SUB, "keys": {"p256dh": "BNEW", "auth": "bmV3"}}).status_code == 201
+
+    r = c.post(f"{V2}/push/test")
+    assert r.status_code == 202
+    assert r.json() == {"devices": 1, "sent": 1, "pruned": 0}
+    assert sent == [_SUB["endpoint"]]
+
+    assert c.request("DELETE", f"{V2}/push/subscribe",
+                     json={"endpoint": _SUB["endpoint"]}).status_code == 204
+    assert c.post(f"{V2}/push/test").json()["devices"] == 0
+
+
+def test_push_test_prunes_gone_endpoint(monkeypatch):
+    import push
+    monkeypatch.setattr(push, "send", lambda info, payload: "gone")
+    c = _client()
+    _login(c, "push-gone@example.com")
+    assert c.post(f"{V2}/push/subscribe", json=_SUB).status_code == 201
+    r = c.post(f"{V2}/push/test")
+    assert r.json() == {"devices": 1, "sent": 0, "pruned": 1}
+    # the gone endpoint was removed, so a second pass sees no devices
+    assert c.post(f"{V2}/push/test").json()["devices"] == 0

tests/test_notify.py

@@ -11,8 +11,9 @@ import polars as pl
 import climate
 import db
 import notify
-from models import Notification, Subscription, User
-from sqlalchemy import delete, select
+import push
+from models import Notification, PushSubscription, Subscription, User
+from sqlalchemy import delete, func, select
 
 
 def _history() -> pl.DataFrame:
@@ -146,3 +147,67 @@ def test_cached_archive_is_never_refetched(monkeypatch):
     notify.run_pass()
     assert not fetched, "a cached archive must never be re-fetched by the evaluator"
     assert len(_user_notifications(uid)) == 1
+
+
+# --- web push delivery from a pass -------------------------------------------
+def _add_push(uid, endpoint="https://push.example.com/ep"):
+    with db.sync_session_maker() as s:
+        s.add(PushSubscription(user_id=uid, endpoint=endpoint, p256dh="BKEY", auth="YXV0aA"))
+        s.commit()
+
+
+def _push_count():
+    with db.sync_session_maker() as s:
+        return s.execute(select(func.count()).select_from(PushSubscription)).scalar_one()
+
+
+def _cached_extreme(monkeypatch):
+    monkeypatch.setattr(climate, "load_cached_history", lambda cell: _history())
+    monkeypatch.setattr(climate, "get_recent_forecast",
+                        lambda cell: _recent_extreme(datetime.date.today()))
+
+
+def test_push_dispatched_on_new_notification(monkeypatch):
+    uid = _seed_single_subscription()
+    _add_push(uid, "https://push.example.com/ep-a")
+    _cached_extreme(monkeypatch)
+    calls = []
+    monkeypatch.setattr(push, "send", lambda info, payload: calls.append((info, payload)) or "ok")
+
+    notify.run_pass()
+
+    assert len(_user_notifications(uid)) == 1        # in-app row still created
+    assert len(calls) == 1                            # and one push dispatched
+    info, payload = calls[0]
+    assert info["endpoint"] == "https://push.example.com/ep-a"
+    assert payload["title"] and payload["body"]
+    today = datetime.date.today().isoformat()
+    assert payload["url"] == f"/thermograph/day#lat=1.00000&lon=2.00000&date={today}"
+
+
+def test_push_gone_endpoint_is_pruned(monkeypatch):
+    uid = _seed_single_subscription()
+    _add_push(uid, "https://push.example.com/ep-gone")
+    _cached_extreme(monkeypatch)
+    monkeypatch.setattr(push, "send", lambda info, payload: "gone")
+
+    notify.run_pass()
+
+    assert len(_user_notifications(uid)) == 1        # notification still delivered in-app
+    assert _push_count() == 0                         # the dead endpoint was pruned
+
+
+def test_inapp_notification_survives_push_error(monkeypatch):
+    uid = _seed_single_subscription()
+    _add_push(uid, "https://push.example.com/ep-boom")
+    _cached_extreme(monkeypatch)
+
+    def _boom(info, payload):
+        raise RuntimeError("push service exploded")
+
+    monkeypatch.setattr(push, "send", _boom)
+
+    notify.run_pass()   # must not raise
+
+    assert len(_user_notifications(uid)) == 1        # the in-app write is unaffected
+    assert _push_count() == 1                         # a mere error doesn't prune