feat: pwa notifications

tracker/push.py

@@ -0,0 +1,55 @@
+import json
+import logging
+
+from django.conf import settings
+from django.utils import timezone
+from pywebpush import WebPushException, webpush
+
+from accounts.models import PushSubscription
+
+logger = logging.getLogger(__name__)
+
+
+def push_is_configured() -> bool:
+    return bool(settings.VAPID_PUBLIC_KEY and settings.VAPID_PRIVATE_KEY)
+
+
+def send_to_subscription(subscription: PushSubscription, payload: dict) -> bool:
+    """Push one message. Returns whether it was delivered.
+
+    A 404 or 410 from the push service means the browser threw the subscription
+    away (app uninstalled, notifications revoked); the row is then useless, so we
+    delete it rather than retrying it forever.
+    """
+    if not push_is_configured():
+        logger.warning("Push not configured: set VAPID_PUBLIC_KEY / VAPID_PRIVATE_KEY.")
+        return False
+
+    try:
+        webpush(
+            subscription_info=subscription.as_subscription_info(),
+            data=json.dumps(payload),
+            vapid_private_key=settings.VAPID_PRIVATE_KEY,
+            vapid_claims={"sub": f"mailto:{settings.VAPID_CLAIM_EMAIL}"},
+            timeout=10,
+        )
+    except WebPushException as exc:
+        status = getattr(exc.response, "status_code", None)
+        if status in (404, 410):
+            logger.info("Dropping expired push subscription %s", subscription.pk)
+            subscription.delete()
+        else:
+            logger.error("Push to %s failed: %s", subscription.pk, exc)
+        return False
+
+    subscription.last_success_at = timezone.now()
+    subscription.save(update_fields=["last_success_at"])
+    return True
+
+
+def notify_user(user, payload: dict) -> int:
+    """Push to every device the user has enabled. Returns the number delivered."""
+    return sum(
+        send_to_subscription(subscription, payload)
+        for subscription in user.push_subscriptions.all()
+    )