From 2f289f1cb638d232758585e6c8eaf8e55410d260 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Wed, 15 Jul 2026 16:21:06 -0700 Subject: [PATCH] Add PWA + Web Push delivery for weather alerts (#95) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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; 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 | 119 ++++++++++++++++++++++++++++++++++-- app.py | 5 ++ models.py | 32 ++++++++++ notify.py | 37 +++++++++++- push.py | 121 +++++++++++++++++++++++++++++++++++++ requirements.txt | 3 + schemas.py | 16 +++++ tests/test_api_accounts.py | 52 ++++++++++++++++ tests/test_notify.py | 69 ++++++++++++++++++++- 9 files changed, 447 insertions(+), 7 deletions(-) create mode 100644 push.py diff --git a/api_accounts.py b/api_accounts.py index 86a3011..09664f2 100644 --- a/api_accounts.py +++ b/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} diff --git a/app.py b/app.py index 86b8df9..5d9bed0 100644 --- a/app.py +++ b/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 diff --git a/models.py b/models.py index 86da3ca..a805803 100644 --- a/models.py +++ b/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"), + ) diff --git a/notify.py b/notify.py index a9e6c2e..8ac6a95 100644 --- a/notify.py +++ b/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 diff --git a/push.py b/push.py new file mode 100644 index 0000000..64ee963 --- /dev/null +++ b/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" diff --git a/requirements.txt b/requirements.txt index 25cae2a..70b76c0 100644 --- a/requirements.txt +++ b/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 diff --git a/schemas.py b/schemas.py index 54eb7f4..1ba3d75 100644 --- a/schemas.py +++ b/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 diff --git a/tests/test_api_accounts.py b/tests/test_api_accounts.py index 69ade16..9209dfb 100644 --- a/tests/test_api_accounts.py +++ b/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 diff --git a/tests/test_notify.py b/tests/test_notify.py index da587b6..9738fcb 100644 --- a/tests/test_notify.py +++ b/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