feat(capture): web-push subscription store + API + delivery on job completion (PR-3, refs #524)

- models/push_subscription.py: push_subscriptions table (id uuid pk,
  user_email indexed, endpoint Text, p256dh/auth Text, user_agent nullable,
  created_at). Registered in main.py AND sync_worker.py so both containers
  create_all the table.
- routers/push.py (wired at /api/push): GET /vapid-public, GET /status,
  POST /subscribe (upsert by user_email+endpoint), DELETE /subscribe
  (query-param or body endpoint; TestClient-safe).
- services/push_service.py: send_push_to_user() — VAPID-signed pywebpush
  delivery, deletes 404/410-gone subs, never raises; logs once and skips
  when VAPID keys are missing.
- capture_job_processor: fires a 'Capture complete' push next to the bell
  Notification at terminal status (target /library).
- config: vapid_private_key/public_key/subject (subject defaults to
  mailto: from Mail_From). requirements.txt + .env.dev/uat.example.
- tests/test_push.py: subscribe upsert-not-duplicate, status, unsubscribe
  idempotent, vapid-public, silent no-op without keys, mocked delivery
  asserting title/body/target_path + gone-sub cleanup.
This commit is contained in:
shridhar.sabat
2026-09-10 14:41:58 +05:30
parent dade2d8a37
commit fec6167037
11 changed files with 578 additions and 0 deletions
+10
View File
@@ -63,5 +63,15 @@ ORG_CHART_ENABLED=false
# WFX_COMPANYID=sapl
# WFX_Token_Name=API Acces
# ── Web Push notifications (issue #524, PR-3) ──
# VAPID keys (base64url, no padding). Generate with:
# python -c "from py_vapid import Vapid02; k=Vapid02(); k.generate_keys(); print(k)"
# Empty = feature off (push silently skipped, one warning). Real keys injected
# server-side, never committed.
VAPID_PRIVATE_KEY=
VAPID_PUBLIC_KEY=
# Optional VAPID "contact" — defaults to a mailto: from Mail_From.
VAPID_SUBJECT=
# ── Build-time only (Dokploy env-only changes won't apply without rebuild) ──
VITE_APP_VERSION=dev-local
+10
View File
@@ -39,5 +39,15 @@ FEATURE_MODEL_ROUTES=
# WFX_COMPANYID=sapl
# WFX_Token_Name=API Acces
# ── Web Push notifications (issue #524, PR-3) ──
# VAPID keys (base64url, no padding). Generate with:
# python -c "from py_vapid import Vapid02; k=Vapid02(); k.generate_keys(); print(k)"
# Empty = feature off (push silently skipped, one warning). Real keys injected
# server-side, never committed.
VAPID_PRIVATE_KEY=
VAPID_PUBLIC_KEY=
# Optional VAPID "contact" — defaults to a mailto: from Mail_From.
VAPID_SUBJECT=
# ── Build-time only (Dokploy env-only changes won't apply without rebuild) ──
VITE_APP_VERSION=uat-local
+14
View File
@@ -166,6 +166,20 @@ class Settings(BaseSettings):
chromadb_path: str = str(_HERE / "data" / "chromadb")
images_path: str = str(_HERE / "data" / "images")
# ── Web Push notifications (issue #524, PR-3) ──
# VAPID keys for signing push messages. Generate a dev pair with
# python -c "from py_vapid import Vapid02; k=Vapid02(); k.generate_keys(); \
# print(k.private_key, k.public_key)"
# (base64url, no padding). Empty keys = feature OFF — the push service
# silently skips sending (one warning) so the worker loop never crashes.
# Env: VAPID_PRIVATE_KEY, VAPID_PUBLIC_KEY.
vapid_private_key: str = ""
vapid_public_key: str = ""
# The "contact" the push service shows the user for this sender — typically
# a mailto: URL. Defaults to the mail-from domain (mail_from alias) so we
# don't need a separate setting on most deployments.
vapid_subject: str = ""
# ── Bulk garment upload grouping (issue #439, phase 2) ──
# Consecutive shots (sorted by capture time) are treated as ONE garment while
# their timestamps are no more than this many minutes apart AND their perceptual
+3
View File
@@ -49,6 +49,7 @@ from models.feed_post import FeedPost # noqa: F401 — registers feed_posts tab
from models.feed_share import FeedShare # noqa: F401 — registers feed_shares table
from models.notification import Notification # noqa: F401 — registers notifications table
from models.capture_job import CaptureJob # noqa: F401 — registers capture_jobs table (async bulk capture, #522)
from models.push_subscription import PushSubscription # noqa: F401 — registers push_subscriptions table (web push, #524)
from models.tna_holiday import TnaHoliday # noqa: F401 — registers tna_holidays table
from services.embedding_service import EmbeddingService
from services.tagging_service import TaggingService
@@ -582,6 +583,8 @@ from routers import llm_routes
app.include_router(llm_routes.router, prefix=PREFIX_ADMIN, tags=["admin-llm-routes"])
from routers import garment_capture
app.include_router(garment_capture.router, prefix="/api/garments-capture", tags=["garments-capture"])
from routers import push
app.include_router(push.router, prefix="/api/push", tags=["push"])
# prefix /api/plm declared on the router
+29
View File
@@ -0,0 +1,29 @@
"""Web-Push subscription store (issue #524, PR-3).
One row per browser/device a user has opted into push notifications on. The
``endpoint`` is the browser's unique push URL; a user may legitimately have
several (desktop + phone + different browsers), so uniqueness is scoped per
``user_email`` — the API upserts by (user_email, endpoint) and the worker can
never push to a stale/unknown user.
Sent at job completion by ``services.push_service.send_push_to_user``.
"""
import uuid
from datetime import datetime, timezone
from sqlalchemy import Column, DateTime, String, Text
from database import Base
class PushSubscription(Base):
__tablename__ = "push_subscriptions"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
user_email = Column(String(255), nullable=False, index=True)
endpoint = Column(Text, nullable=False)
p256dh = Column(Text, nullable=False)
auth = Column(Text, nullable=False)
user_agent = Column(Text, nullable=True)
created_at = Column(DateTime(timezone=True), nullable=False,
default=lambda: datetime.now(timezone.utc))
+1
View File
@@ -25,6 +25,7 @@ pymupdf==1.27.2.3
openpyxl==3.1.5
qrcode==8.0
imapclient==3.1.0
pywebpush==1.14.0 # Web Push VAPID signing + HTTP encryption (issue #524, PR-3)
# Test dependencies
pytest==8.3.4
+120
View File
@@ -0,0 +1,120 @@
"""Web-Push subscription API (issue #524, PR-3).
The browser registers its push endpoint here once the user opts in. The backend
stores it keyed by (user_email, endpoint) so the capture worker can push at job
completion. Also exposes the VAPID public key the browser needs to subscribe,
and a status flag for the UI toggle.
Routes follow the app convention (static before parameterized).
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from fastapi import APIRouter, Body, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from config import settings
from models.push_subscription import PushSubscription
from models.user import User
from routers.auth import current_user, get_db
router = APIRouter()
class PushKeys(BaseModel):
p256dh: str = Field(..., min_length=1)
auth: str = Field(..., min_length=1)
class PushSubscribeBody(BaseModel):
endpoint: str = Field(..., min_length=1)
keys: PushKeys
user_agent: str | None = None
def _domain() -> str:
"""mail-from domain for the VAPID subject default (mirror push_service)."""
if settings.vapid_subject:
return settings.vapid_subject
mail_from = (settings.mail_from or "").strip()
if mail_from:
return f"mailto:{mail_from}"
return "mailto:sapl@saplind.com"
def _vapid_public() -> str | None:
"""The configured VAPID public key, or None when push is disabled."""
return settings.vapid_public_key or None
@router.get("/vapid-public")
def get_vapid_public(user: User = Depends(current_user)):
"""Return the VAPID public key the browser must pass to pushManager.subscribe."""
return {"public_key": _vapid_public()}
@router.get("/status")
def push_status(user: User = Depends(current_user), db: Session = Depends(get_db)):
"""True when the current user has at least one active subscription."""
has = (db.query(PushSubscription)
.filter(PushSubscription.user_email == user.email)
.first())
return {"subscribed": has is not None,
"supported": _vapid_public() is not None}
@router.post("/subscribe")
def subscribe(body: PushSubscribeBody,
user: User = Depends(current_user),
db: Session = Depends(get_db)):
"""Upsert a push subscription by (user_email, endpoint)."""
endpoint = (body.endpoint or "").strip()
if not endpoint:
raise HTTPException(status_code=400, detail={"error": "endpoint required", "code": "EMPTY"})
if _vapid_public() is None:
raise HTTPException(status_code=503,
detail={"error": "Push notifications are not configured on this server",
"code": "VAPID_NOT_CONFIGURED"})
sub = (db.query(PushSubscription)
.filter(PushSubscription.user_email == user.email,
PushSubscription.endpoint == endpoint)
.first())
if sub is None:
sub = PushSubscription(
id=str(uuid.uuid4()),
user_email=user.email,
endpoint=endpoint,
p256dh=body.keys.p256dh,
auth=body.keys.auth,
user_agent=body.user_agent,
created_at=datetime.now(timezone.utc),
)
db.add(sub)
else:
sub.p256dh = body.keys.p256dh
sub.auth = body.keys.auth
if body.user_agent is not None:
sub.user_agent = body.user_agent
db.commit()
return {"ok": True, "id": sub.id}
@router.delete("/subscribe")
def unsubscribe(endpoint: str | None = Query(None),
body: PushSubscribeBody | None = Body(None),
user: User = Depends(current_user),
db: Session = Depends(get_db)):
"""Remove a push subscription by endpoint (query param or body). Idempotent."""
endpoint = (endpoint or (body.endpoint if body else None) or "").strip()
if not endpoint:
raise HTTPException(status_code=400, detail={"error": "endpoint required", "code": "EMPTY"})
removed = (db.query(PushSubscription)
.filter(PushSubscription.user_email == user.email,
PushSubscription.endpoint == endpoint)
.delete(synchronize_session=False))
db.commit()
return {"ok": True, "removed": removed}
@@ -164,6 +164,11 @@ def _notify_terminal(db: Session, job: CaptureJob, n_done: int, n_failed: int) -
kind="capture", event_id=None, target="/library", body=body,
))
db.commit()
# Web push (issue #524, PR-3) — fire-and-forget, never raises. If VAPID
# is unconfigured the service logs once and skips.
from services.push_service import send_push_to_user
send_push_to_user(db, job.owner_email,
title="Capture complete", body=body, target_path="/library")
except Exception as e: # noqa: BLE001
logger.warning("Capture-job notification failed for %s: %s", job.id, e)
try:
+124
View File
@@ -0,0 +1,124 @@
"""Web-Push delivery (issue #524, PR-3).
Sends a VAPID-signed push message to every subscription a user has registered.
Called from the capture-job processor at the terminal-status point so the
notifier fires even when the user has left the screen.
Design rules (mirror the capture worker's "best effort, never crashes"):
* If VAPID keys are missing the feature is OFF — log ONCE and return without
sending (a misconfigured deployment must never break the worker loop).
* A gone subscription (HTTP 404/410 from the push service) is deleted from the
store so we don't hammer a dead endpoint forever.
* Any other failure is logged and swallowed — push is fire-and-forget.
* ``send_push_to_user`` NEVER raises.
"""
from __future__ import annotations
import logging
from sqlalchemy.orm import Session
from config import settings
from models.push_subscription import PushSubscription
logger = logging.getLogger(__name__)
# pywebpush raises this for the gone/expired endpoint cases we treat as delete.
_PUSH_NOT_FOUND_CODES = (404, 410)
_warned_no_vapid = False
def _vapid_ready() -> bool:
"""True when both VAPID keys are configured (feature enabled)."""
return bool(settings.vapid_private_key and settings.vapid_public_key)
def _vapid_subject() -> str:
"""VAPID contact — ``vapid_subject`` or a mailto: from the mail-from alias."""
if settings.vapid_subject:
return settings.vapid_subject
if settings.mail_from:
mail_from = (settings.mail_from or "").strip()
if mail_from:
return f"mailto:{mail_from}"
return "mailto:sapl@saplind.com"
def send_push_to_user(
db: Session,
user_email: str,
title: str,
body: str,
target_path: str,
) -> int:
"""Push ``title``/``body`` (open ``target_path`` on click) to ``user_email``.
Returns the number of push messages successfully dispatched (0 when VAPID is
unconfigured or the user has no subscriptions). Never raises.
"""
global _warned_no_vapid
if not _vapid_ready():
if not _warned_no_vapid:
_warned_no_vapid = True
logger.warning(
"Web Push disabled: VAPID_PRIVATE_KEY / VAPID_PUBLIC_KEY not set — "
"skipping push notifications (set them to enable)."
)
return 0
try:
from pywebpush import WebPushException, webpush
subs = (db.query(PushSubscription)
.filter(PushSubscription.user_email == user_email)
.all())
except Exception as e: # noqa: BLE001
logger.warning("Push: could not load subscriptions for %s: %s", user_email, e)
return 0
if not subs:
return 0
payload = {
"title": title,
"body": body,
"target_path": target_path,
}
sent = 0
for sub in list(subs):
try:
webpush(
subscription_info={
"endpoint": sub.endpoint,
"keys": {"p256dh": sub.p256dh, "auth": sub.auth},
},
data=payload,
vapid_private_key=settings.vapid_private_key,
vapid_claims={"sub": _vapid_subject()},
timeout=10,
)
sent += 1
except WebPushException as e:
status = getattr(e, "response", None)
code = getattr(status, "status_code", None)
if code in _PUSH_NOT_FOUND_CODES:
# Endpoint is gone — stop trying forever.
logger.info("Push: subscription %s is gone (%s) — removing", sub.id, code)
try:
db.delete(sub)
db.commit()
except Exception: # noqa: BLE001
try:
db.rollback()
except Exception: # noqa: BLE001
pass
else:
logger.warning("Push: delivery to %s failed (%s): %s",
sub.id, code, e)
except Exception as e: # noqa: BLE001
logger.warning("Push: unexpected failure to %s: %s", sub.id, e)
return sent
+1
View File
@@ -45,6 +45,7 @@ from models.feed_post import FeedPost # noqa: F401
from models.feed_share import FeedShare # noqa: F401
from models.notification import Notification # noqa: F401
from models.capture_job import CaptureJob # noqa: F401 — registers capture_jobs table (async bulk capture, #522)
from models.push_subscription import PushSubscription # noqa: F401 — registers push_subscriptions table (web push, #524)
logger = logging.getLogger("sync_worker")
+261
View File
@@ -0,0 +1,261 @@
"""Web-Push subscriptions (issue #524, PR-3) — endpoint + service tests.
Covers:
(a) POST /api/push/subscribe upserts a subscription by (user_email, endpoint);
a duplicate subscribe does NOT create a second row;
(b) GET /api/push/status reflects whether the user has >=1 subscription;
(c) DELETE /api/push/subscribe removes the row (idempotent);
(d) GET /api/push/vapid-public returns the configured public key (or null);
(e) send_push_to_user with NO VAPID config silently no-ops (returns 0);
(f) send_push_to_user with VAPID config calls pywebpush with a JSON payload
carrying title/body/target_path, and deletes a 404-gone subscription.
The heavy pywebpush delivery is mocked (no network); the DB dependency + RBAC
identity are overridden to a scratch file-backed SQLite session.
"""
import os
import sys
from pathlib import Path
from unittest.mock import patch
os.environ.setdefault("MISTRAL_API_KEY", "test-only-key")
os.environ.setdefault("DATABASE_URL", "sqlite://")
os.environ.setdefault("MAIL_DRY_RUN", "true")
os.environ.setdefault("SECRET_KEY", "test-only-key")
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database import Base
# Register every table the flow touches before create_all.
import models.user # noqa: F401 (users)
import models.push_subscription # noqa: F401 (push_subscriptions)
import models.notification # noqa: F401 (notifications — processor import)
from models.push_subscription import PushSubscription
from models.user import User
from services.push_service import send_push_to_user
# A well-formed VAPID keypair (only shape matters — delivery is mocked).
PUBLIC_KEY = "IGJYJcheBSFfuKTQgvADUYpyHz7lhJGjqqqPWa-2FxYRzgFUdZ2_5lF95eEWUsKtPCubQYPBedvFuNdg2Ftoss"
PRIVATE_KEY = "Ayun-Qb4lSo9i6-jdESE3cVLlKZQdKMRdNVtK5mkz1M"
@pytest.fixture()
def db(tmp_path):
engine = create_engine(
f"sqlite:///{tmp_path / 'scratch.db'}",
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine, autoflush=False, autocommit=False)
session = Session()
yield session
session.close()
engine.dispose()
@pytest.fixture()
def make_client(db, monkeypatch):
from main import app
import database as database_mod
from routers import auth as auth_mod
def _get_db():
yield db
app.dependency_overrides[database_mod.get_db] = _get_db
app.dependency_overrides[auth_mod.get_db] = _get_db
def _make(user):
if user is not None:
app.dependency_overrides[auth_mod.current_user] = lambda: user
else:
app.dependency_overrides.pop(auth_mod.current_user, None)
return TestClient(app)
yield _make
app.dependency_overrides.clear()
def _user(email: str = "pratik@sonalapparels.in") -> User:
return User(id="usr-push", email=email, name="Pratik", role="Admin",
department_id=None, is_active=True, password_hash="x")
def _body(endpoint: str = "https://push.example.com/ep1"):
return {
"endpoint": endpoint,
"keys": {"p256dh": "p256dh-b64", "auth": "auth-b64"},
"user_agent": "test/1.0",
}
def _del(c, endpoint: str = "https://push.example.com/ep1"):
"""DELETE with the endpoint as a query param (TestClient sends no body here)."""
return c.delete(f"/api/push/subscribe?endpoint={endpoint}")
def _count(db, email: str = "pratik@sonalapparels.in") -> int:
return db.query(PushSubscription).filter(
PushSubscription.user_email == email).count()
# ===========================================================================
# (a) subscribe upserts by (user_email, endpoint)
# ===========================================================================
def test_subscribe_upserts_not_duplicates(make_client, db, monkeypatch):
monkeypatch.setattr("config.settings.vapid_public_key", PUBLIC_KEY)
c = make_client(_user())
r = c.post("/api/push/subscribe", json=_body())
assert r.status_code == 200, r.text
assert r.json()["ok"] is True
assert _count(db) == 1
# Duplicate subscribe (same endpoint) — upsert, NOT a new row.
r2 = c.post("/api/push/subscribe", json=_body())
assert r2.status_code == 200, r2.text
assert _count(db) == 1
# A different endpoint is a genuinely separate device.
c.post("/api/push/subscribe", json=_body("https://push.example.com/ep2"))
assert _count(db) == 2
def test_subscribe_503_when_vapid_not_configured(make_client, db):
c = make_client(_user())
r = c.post("/api/push/subscribe", json=_body())
assert r.status_code == 503, r.text
assert _count(db) == 0
# ===========================================================================
# (b) status reflects whether the user has >=1 subscription
# ===========================================================================
def test_status_reflects_subscription(make_client, db, monkeypatch):
monkeypatch.setattr("config.settings.vapid_public_key", PUBLIC_KEY)
c = make_client(_user())
r0 = c.get("/api/push/status")
assert r0.status_code == 200
assert r0.json()["subscribed"] is False
c.post("/api/push/subscribe", json=_body())
r1 = c.get("/api/push/status")
assert r1.json()["subscribed"] is True
_del(c)
r2 = c.get("/api/push/status")
assert r2.json()["subscribed"] is False
# ===========================================================================
# (c) unsubscribe removes (idempotent)
# ===========================================================================
def test_unsubscribe_removes_and_is_idempotent(make_client, db, monkeypatch):
monkeypatch.setattr("config.settings.vapid_public_key", PUBLIC_KEY)
c = make_client(_user())
c.post("/api/push/subscribe", json=_body())
assert _count(db) == 1
r = _del(c)
assert r.status_code == 200, r.text
assert r.json()["removed"] == 1
assert _count(db) == 0
# Idempotent — deleting again removes nothing but stays 200.
r2 = _del(c)
assert r2.status_code == 200
assert r2.json()["removed"] == 0
def test_unsubscribe_requires_endpoint(make_client, db, monkeypatch):
c = make_client(_user())
r = _del(c, endpoint="")
assert r.status_code == 400, r.text
# ===========================================================================
# (d) vapid-public
# ===========================================================================
def test_vapid_public_endpoint(make_client, db, monkeypatch):
c = make_client(_user())
r = c.get("/api/push/vapid-public")
assert r.status_code == 200
assert r.json()["public_key"] is None
monkeypatch.setattr("config.settings.vapid_public_key", PUBLIC_KEY)
r2 = c.get("/api/push/vapid-public")
assert r2.json()["public_key"] == PUBLIC_KEY
# ===========================================================================
# (e) send_push_to_user silently no-ops without VAPID config
# ===========================================================================
def test_send_no_vapid_silently_noops(db):
assert _count(db) == 0
n = send_push_to_user(db, "pratik@sonalapparels.in", "Capture complete",
"body", "/library")
assert n == 0 # nothing to send even if a subscription existed
# With a subscription present but VAPID unset — still 0, no error.
db.add(PushSubscription(
id="sub-1", user_email="pratik@sonalapparels.in",
endpoint="https://push.example.com/ep1", p256dh="p", auth="a"))
db.commit()
n2 = send_push_to_user(db, "pratik@sonalapparels.in", "Capture complete",
"body", "/library")
assert n2 == 0
# ===========================================================================
# (f) send_push_to_user calls pywebpush with title/body/target_path + cleans 404s
# ===========================================================================
def test_send_calls_pywebpush_with_payload(make_client, db, monkeypatch):
monkeypatch.setattr("config.settings.vapid_private_key", PRIVATE_KEY)
monkeypatch.setattr("config.settings.vapid_public_key", PUBLIC_KEY)
c = make_client(_user())
c.post("/api/push/subscribe", json=_body())
assert _count(db) == 1
captured = {}
def _fake_webpush(subscription_info, data, **kw):
captured["info"] = subscription_info
captured["data"] = data
captured["kw"] = kw
if subscription_info["endpoint"].endswith("gone"):
from pywebpush import WebPushException
class _Resp:
status_code = 410
raise WebPushException("gone", response=_Resp())
with patch("pywebpush.webpush", side_effect=_fake_webpush):
# ok endpoint (already subscribed via API) pushes fine
n = send_push_to_user(db, "pratik@sonalapparels.in", "Capture complete",
"saved 2 garments", "/library")
assert n == 1
payload = captured["data"]
assert payload["title"] == "Capture complete"
assert payload["body"] == "saved 2 garments"
assert payload["target_path"] == "/library"
assert captured["kw"]["vapid_private_key"] == PRIVATE_KEY
assert "sub" in captured["kw"]["vapid_claims"]
# a 410-gone subscription is removed
gone_sub = PushSubscription(
id="sub-gone", user_email="pratik@sonalapparels.in",
endpoint="https://push.example.com/gone", p256dh="p", auth="a")
db.add(gone_sub)
db.commit()
assert _count(db) == 2
send_push_to_user(db, "pratik@sonalapparels.in", "Capture complete",
"x", "/library")
# the API-created sub survives; the gone one is removed
assert _count(db) == 1