app/push.py

"""Notifications Web Push (VAPID).

Les clés VAPID sont générées au premier démarrage et stockées dans la table
`settings`, donc elles survivent aux redémarrages tant que /data est monté.
pywebpush s'appuie sur `requests` (bloquant) : les envois partent dans un
thread pool pour ne pas figer l'event loop asyncio.
"""

from __future__ import annotations

import asyncio
import base64
import json
import logging
from concurrent.futures import ThreadPoolExecutor

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
from pywebpush import WebPushException, webpush

from . import config, db

log = logging.getLogger("push")

# Peu de threads : les envois sont courts et le tier gratuit n'a que 2 vCPU.
_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="push")


def _b64(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")


def _ensure_keys() -> tuple[str, str]:
    """Renvoie (clé privée b64url, clé publique b64url), en les créant au besoin."""
    priv = db.get_setting("vapid_private")
    pub = db.get_setting("vapid_public")
    if priv and pub:
        return priv, pub

    key = ec.generate_private_key(ec.SECP256R1())
    private_raw = key.private_numbers().private_value.to_bytes(32, "big")
    public_raw = key.public_key().public_bytes(
        encoding=Encoding.X962, format=PublicFormat.UncompressedPoint
    )
    priv, pub = _b64(private_raw), _b64(public_raw)
    db.set_setting("vapid_private", priv)
    db.set_setting("vapid_public", pub)
    log.info("Nouvelles clés VAPID générées.")
    return priv, pub


def public_key() -> str:
    """Clé publique à passer à `pushManager.subscribe` côté navigateur."""
    return _ensure_keys()[1]


def _send_one(sub_id: int, endpoint: str, p256dh: str, auth_secret: str, payload: str) -> None:
    priv, _ = _ensure_keys()
    try:
        webpush(
            subscription_info={
                "endpoint": endpoint,
                "keys": {"p256dh": p256dh, "auth": auth_secret},
            },
            data=payload,
            vapid_private_key=priv,
            vapid_claims={"sub": config.VAPID_SUBJECT},
            ttl=86400,
            timeout=10,
        )
    except WebPushException as exc:
        code = getattr(exc.response, "status_code", None)
        # 404/410 : l'abonnement n'existe plus côté navigateur, on le purge.
        if code in (404, 410):
            db.execute("DELETE FROM push_subscriptions WHERE id = ?", (sub_id,))
            log.info("Abonnement push %s expiré, supprimé.", sub_id)
        else:
            log.warning("Échec d'envoi push (%s) : %s", code, exc)
    except Exception as exc:  # réseau, DNS, etc.
        log.warning("Erreur push inattendue : %s", exc)


async def notify(phones: list[str], payload: dict) -> None:
    """Envoie une notification à tous les appareils des numéros donnés."""
    if not phones:
        return
    placeholders = ",".join("?" * len(phones))
    rows = db.query(
        f"SELECT id, endpoint, p256dh, auth FROM push_subscriptions WHERE phone IN ({placeholders})",
        phones,
    )
    if not rows:
        return
    body = json.dumps(payload, ensure_ascii=False)
    loop = asyncio.get_running_loop()
    for row in rows:
        loop.run_in_executor(
            _executor, _send_one, row["id"], row["endpoint"], row["p256dh"], row["auth"], body
        )


def save_subscription(phone: str, subscription: dict) -> None:
    endpoint = subscription.get("endpoint", "")
    keys = subscription.get("keys") or {}
    p256dh, auth_secret = keys.get("p256dh", ""), keys.get("auth", "")
    if not (endpoint and p256dh and auth_secret):
        raise ValueError("Abonnement push incomplet.")
    db.execute(
        "INSERT INTO push_subscriptions(phone, endpoint, p256dh, auth, created_at) "
        "VALUES(?, ?, ?, ?, ?) "
        "ON CONFLICT(endpoint) DO UPDATE SET phone = excluded.phone, "
        "p256dh = excluded.p256dh, auth = excluded.auth",
        (phone, endpoint, p256dh, auth_secret, db.now_ms()),
    )


def delete_subscription(endpoint: str) -> None:
    db.execute("DELETE FROM push_subscriptions WHERE endpoint = ?", (endpoint,))


def shutdown() -> None:
    _executor.shutdown(wait=False, cancel_futures=True)