Coverage for src/lexigram/notification/backends/push/web_push.py: 29%
70 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:26 +0800
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:26 +0800
1"""Web Push (RFC 8030) notification backend using VAPID + encryption."""
3from __future__ import annotations
5import asyncio
6from typing import TYPE_CHECKING
7import uuid
9from lexigram.contracts.core import HealthCheckResult, HealthStatus
10from lexigram.contracts.mailer import MessageDeliveryReceipt
11from lexigram.contracts.notification.types import PushMessage
12from lexigram.logging import get_logger
13from lexigram.notification.exceptions import WebPushNotificationError
14from lexigram.result import Err, Ok, Result
16if TYPE_CHECKING:
17 from lexigram.contracts.notification.errors import NotificationError
19logger = get_logger(__name__)
21try:
22 from pywebpush import WebPushException, webpush # type: ignore[import-untyped]
23except ImportError:
24 WebPushException = Exception
26 def webpush(*args: object, **kwargs: object) -> None:
27 raise ImportError(
28 "pywebpush is not installed. Install lexigram-notification[web-push]."
29 )
32_HTTP_CLIENT_TIMEOUT = 30
35class WebPushChannel:
36 """Web Push (RFC 8030) notification backend.
38 Implements :class:`~lexigram.contracts.notification.protocols.PushChannelProtocol`.
39 Uses VAPID (RFC 8292) for application-level authentication and RFC 8291
40 (aes128gcm) for message encryption.
42 Requires the ``pywebpush`` optional extra.
44 Args:
45 vapid_private_key: VAPID private key (PEM-encoded EC prime256v1).
46 vapid_public_key: VAPID public key (base64url-encoded).
47 vapid_claims_subject: Contact URI for VAPID claims
48 (e.g. ``"mailto:ops@example.com"``).
49 http_timeout: HTTP request timeout in seconds.
50 """
52 def __init__(
53 self,
54 vapid_private_key: str,
55 vapid_public_key: str,
56 vapid_claims_subject: str,
57 http_timeout: int = _HTTP_CLIENT_TIMEOUT,
58 ) -> None:
59 self._vapid_private_key = vapid_private_key
60 self._vapid_public_key = vapid_public_key
61 self._vapid_claims_subject = vapid_claims_subject
62 self._http_timeout = http_timeout
64 async def send(
65 self, message: PushMessage
66 ) -> Result[MessageDeliveryReceipt, NotificationError]:
67 """Send a Web Push notification.
69 ``message.to`` contains a list of **endpoint URLs** (not device tokens).
70 Each endpoint is a separate push subscription.
72 Args:
73 message: The push notification to deliver. ``message.to`` must
74 contain at least one endpoint URL.
76 Returns:
77 Ok(MessageDeliveryReceipt) on success.
78 Err(WebPushNotificationError) for delivery failures.
79 """
80 if not message.to:
81 raise ValueError("WebPushChannel.send() requires at least one endpoint")
83 from lexigram.serialization import dumps_str
85 payload: dict[str, object] = {
86 "title": message.title,
87 "body": message.body,
88 }
89 if message.data:
90 payload["data"] = message.data
91 if message.badge is not None:
92 payload["badge"] = message.badge
93 if message.image:
94 payload["icon"] = message.image
96 endpoint = message.to[0]
98 try:
99 subscription_info: dict[str, object] = {"endpoint": endpoint}
101 keys = (message.data or {}).get("keys")
102 if isinstance(keys, dict):
103 p256dh = keys.get("p256dh", "")
104 auth = keys.get("auth", "")
105 if p256dh and auth:
106 subscription_info["keys"] = {"p256dh": p256dh, "auth": auth}
108 result = webpush(
109 subscription_info=subscription_info,
110 data=dumps_str(payload),
111 vapid_private_key=self._vapid_private_key,
112 vapid_public_key=self._vapid_public_key,
113 vapid_claims={"sub": self._vapid_claims_subject},
114 ttl=message.ttl or 0,
115 timeout=self._http_timeout,
116 )
118 receipt = MessageDeliveryReceipt(
119 message_id=str(uuid.uuid4()),
120 backend="web_push",
121 channel="push",
122 provider_reference=endpoint[:64],
123 )
124 logger.info(
125 "webpush_sent",
126 endpoint_prefix=endpoint[:48],
127 status_code=result.status_code
128 if hasattr(result, "status_code")
129 else None,
130 )
131 return Ok(receipt)
133 except WebPushException as exc:
134 status_code = getattr(exc, "status_code", None)
135 response_body = getattr(exc, "response_body", None)
136 logger.warning(
137 "webpush_send_failed",
138 endpoint_prefix=endpoint[:48],
139 status_code=status_code,
140 response=response_body,
141 )
143 if status_code in (410, 404):
144 return Err(
145 WebPushNotificationError(
146 message=f"Subscription gone (HTTP {status_code})",
147 status_code=status_code,
148 reason="subscription_gone",
149 )
150 )
152 return Err(
153 WebPushNotificationError(
154 message=str(exc),
155 status_code=status_code or 0,
156 reason=response_body or "webpush_error",
157 )
158 )
160 async def send_batch(
161 self, messages: list[PushMessage]
162 ) -> list[Result[MessageDeliveryReceipt, NotificationError]]:
163 """Send multiple Web Push notifications concurrently.
165 Args:
166 messages: Notifications to deliver.
168 Returns:
169 List of Result values, one per message, order preserved.
170 """
171 tasks = [self.send(msg) for msg in messages]
172 return await asyncio.gather(*tasks)
174 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
175 """Check Web Push infrastructure availability.
177 Since Web Push has no global health endpoint, this is a best-effort
178 check that verifies the VAPID key is loadable.
180 Args:
181 timeout: Max seconds to wait (unused, but kept for protocol compat).
183 Returns:
184 HealthCheckResult.
185 """
186 try:
187 from cryptography.hazmat.primitives.asymmetric import ec
188 from py_vapid import Vapid # type: ignore[import-untyped]
190 v = Vapid.from_string(self._vapid_private_key)
191 private_key_obj = v.private_key
192 if not isinstance(private_key_obj, ec.EllipticCurvePrivateKey):
193 return HealthCheckResult(
194 component="web_push",
195 status=HealthStatus.UNHEALTHY,
196 details={"error": "VAPID key is not an EC key"},
197 )
198 return HealthCheckResult(
199 component="web_push",
200 status=HealthStatus.HEALTHY,
201 details={"vapid_key_valid": True},
202 )
203 except Exception as exc: # noqa: BLE001
204 return HealthCheckResult(
205 component="web_push",
206 status=HealthStatus.UNHEALTHY,
207 details={"error": str(exc)},
208 )
211__all__ = ["WebPushChannel"]