""" Web Push delivery for ModuleDesk. When a server-side sync finds new messages for a tenant, we push a notification to every browser subscription owned by that tenant's users. The service worker (static/sw.js) decides whether to raise an OS notification (no tab open) or hand the event to an already-open tab — so this module just delivers the payload; it does not decide visibility. Push is best-effort and fully optional: if VAPID keys or the pywebpush library are absent, every call here is a silent no-op (logged at debug). Dead subscriptions (HTTP 404/410) are pruned from the DB on send. """ from __future__ import annotations import json import logging from typing import Optional from ..config import config from ..db import session_scope from ..public_models import PushSubscription, User logger = logging.getLogger(__name__) # Only real browser push-relay hosts are accepted as subscription endpoints. # Without this the server could be coerced into POSTing to an arbitrary # (e.g. internal / cloud-metadata) URL by any authenticated user — SSRF. ALLOWED_PUSH_HOSTS = ( "fcm.googleapis.com", # Chrome / Edge (FCM) "updates.push.services.mozilla.com", # Firefox "web.push.apple.com", # Safari "notify.windows.com", # legacy Edge / WNS ) def is_allowed_endpoint(endpoint: str) -> bool: """True only for https URLs whose host is a known push relay service.""" from urllib.parse import urlparse try: parsed = urlparse(endpoint) if parsed.scheme != "https": return False host = (parsed.hostname or "").lower() return any(host == h or host.endswith("." + h) for h in ALLOWED_PUSH_HOSTS) except Exception: return False def push_enabled() -> bool: """True when VAPID keys and pywebpush are both available.""" if not (config.VAPID_PUBLIC_KEY and config.VAPID_PRIVATE_KEY): return False try: import pywebpush # noqa: F401 except Exception: return False return True def _send_one(subscription: "PushSubscription", payload: dict) -> Optional[int]: """ Send a single push. Returns an HTTP status code worth acting on (404/410 → caller should delete the subscription), or None otherwise. """ from pywebpush import webpush, WebPushException try: webpush( subscription_info={ "endpoint": subscription.endpoint, "keys": {"p256dh": subscription.p256dh, "auth": subscription.auth}, }, data=json.dumps(payload), vapid_private_key=config.VAPID_PRIVATE_KEY, vapid_claims={"sub": config.VAPID_CLAIMS_EMAIL}, ttl=600, ) return None except WebPushException as exc: status = getattr(getattr(exc, "response", None), "status_code", None) if status in (404, 410): return status # gone — prune it logger.warning("push: delivery failed (status=%s): %s", status, exc) return None except Exception as exc: # pragma: no cover - defensive logger.warning("push: unexpected delivery error: %s", exc) return None def _fanout(org_id: int, payloads: list[dict]) -> int: """ Deliver each payload to every subscription owned by the users of org_id. Returns the number of pushes attempted; prunes dead subscriptions. """ sent = 0 dead_ids: list[int] = [] # public schema: users + their subscriptions with session_scope() as session: subs = ( session.query(PushSubscription) .join(User, User.id == PushSubscription.user_id) .filter(User.org_id == org_id) .all() ) for sub in subs: if sub.id in dead_ids: continue for payload in payloads: status = _send_one(sub, payload) sent += 1 if status in (404, 410): dead_ids.append(sub.id) break # no point sending the rest to a gone endpoint if dead_ids: session.query(PushSubscription).filter( PushSubscription.id.in_(dead_ids) ).delete(synchronize_session=False) if dead_ids: logger.info("push: pruned %d dead subscription(s)", len(dead_ids)) return sent def notify_org_new_messages(org_id: int, new_count: int) -> int: """ Push a single aggregate "N new messages" notification to every subscription owned by the users of org_id. Returns the number of pushes attempted. No-op when push is disabled or there are no subscriptions. """ if not push_enabled() or not new_count: return 0 payload = { "title": "ModuleDesk", "body": f"{new_count} new message{'s' if new_count != 1 else ''}", "count": new_count, "url": "/inbox", "tag": "sh-new-messages", } return _fanout(org_id, [payload]) # Per-sync-run cap on individual notifications: enough to triage from the # lock screen, but a backfill or a busy night collapses into one summary # instead of flooding the notification tray. MAX_INDIVIDUAL_NOTIFICATIONS = 5 TITLE_MAX = 60 PREVIEW_MAX = 140 def _trim(text: str, limit: int) -> str: text = " ".join((text or "").split()) # collapse whitespace/newlines return text if len(text) <= limit else text[: limit - 1].rstrip() + "…" def notify_org_new_inbound(org_id: int, items: list[dict]) -> int: """ Push one notification PER TICKET with new inbound message(s): trimmed subject as title, a short preview of the latest customer message as body, deep-linking to the ticket. At most MAX_INDIVIDUAL_NOTIFICATIONS are sent individually; the remainder collapses into one aggregate notification. items: dicts from SyncService.last_new_inbound — {ticket_id, thread_id, subject, preview, n_messages, product_name} """ if not push_enabled() or not items: return 0 payloads: list[dict] = [] for item in items[:MAX_INDIVIDUAL_NOTIFICATIONS]: n = item.get("n_messages") or 1 title = _trim(item.get("subject") or "(no subject)", TITLE_MAX) if n > 1: title = f"({n}) {title}" payloads.append({ "title": title, "body": _trim(item.get("preview") or "New customer message", PREVIEW_MAX), "count": n, "url": f"/ticket/{item['ticket_id']}", # Unique tag per ticket so notifications stack instead of # replacing each other; a newer message on the SAME ticket # replaces that ticket's stale notification. "tag": f"sh-msg-{item['ticket_id']}", }) overflow = len(items) - MAX_INDIVIDUAL_NOTIFICATIONS if overflow > 0: payloads.append({ "title": "ModuleDesk", "body": f"…and {overflow} more ticket{'s' if overflow != 1 else ''} with new messages", "count": overflow, "url": "/inbox", "tag": "sh-new-messages", }) return _fanout(org_id, payloads)