"""Deliver pending notifications from the durable outbox.""" import asyncio import logging from dataclasses import dataclass from datetime import datetime, timedelta from typing import Callable, Protocol import httpx from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import Settings, settings from app.models import AlertEvent, NotificationOutbox, NotificationStatus logger = logging.getLogger("pingwatch.notification_dispatcher") # A small count caps one WeCom text request while still aggregating alert bursts. MAX_EVENTS_PER_MESSAGE = 5 _MESSAGE_SEPARATOR = "\n\n---\n\n" _TOKEN_EXPIRY_MARGIN_SECONDS = 60 _HTTP_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=5.0) _TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" _MESSAGE_URL = "https://qyapi.weixin.qq.com/cgi-bin/message/send" @dataclass(frozen=True) class DeliveryResult: """Sanitized delivery outcome returned by a notification transport.""" success: bool error: str | None = None retryable: bool = False @dataclass(frozen=True) class DispatchSummary: """Counts produced by one bounded dispatcher pass.""" sent: int = 0 rescheduled: int = 0 failed: int = 0 class TextNotificationClient(Protocol): """Minimal transport boundary used by the persistent dispatcher.""" async def send_text(self, content: str) -> DeliveryResult: """Deliver one text message without exposing credentials.""" @dataclass(frozen=True) class _TokenResult: token: str | None error: str | None = None retryable: bool = False class WeComClient: """Minimal WeCom application client with an in-process token cache.""" def __init__( self, runtime_settings: Settings | None = None, *, http_client: httpx.AsyncClient | None = None, now: Callable[[], datetime] | None = None, ): self._settings = runtime_settings or settings self._http = http_client or httpx.AsyncClient(timeout=_HTTP_TIMEOUT) self._owns_http_client = http_client is None self._now = now or datetime.now self._access_token: str | None = None self._access_token_expires_at: datetime | None = None self._token_lock = asyncio.Lock() async def aclose(self) -> None: """Close only the HTTP client created by this service instance.""" if self._owns_http_client: await self._http.aclose() async def send_text(self, content: str) -> DeliveryResult: """Deliver text while returning only sanitized failure evidence.""" if not self._settings.wecom_notification_enabled: return DeliveryResult(False, "delivery disabled", False) token_result = await self._get_access_token() if token_result.token is None: return DeliveryResult( False, token_result.error, token_result.retryable, ) payload = { "msgtype": "text", "agentid": self._settings.WECOM_AGENT_ID, "text": {"content": content}, "safe": 0, } to_party = self._settings.WECOM_TO_PARTY.strip() if to_party: payload["toparty"] = to_party else: payload["touser"] = "@all" try: response = await self._http.post( _MESSAGE_URL, params={"access_token": token_result.token}, json=payload, ) except (httpx.TimeoutException, httpx.TransportError): logger.warning("WeCom message transport failure") return DeliveryResult(False, "transport error", True) if response.status_code == 429 or response.status_code >= 500: logger.warning( "WeCom message transient HTTP failure status=%s", response.status_code, ) return DeliveryResult(False, f"HTTP {response.status_code}", True) if response.status_code >= 400: logger.warning( "WeCom message HTTP failure status=%s", response.status_code, ) return DeliveryResult(False, f"HTTP {response.status_code}", False) try: data = response.json() if not isinstance(data, dict): raise TypeError("response JSON must be an object") error_code = int(data.get("errcode", -1)) except (TypeError, ValueError): logger.warning("WeCom message returned invalid JSON") return DeliveryResult(False, "invalid response", True) if error_code == 0: return DeliveryResult(True) if error_code in {40014, 42001}: self._access_token = None self._access_token_expires_at = None retryable = error_code in {-1, 40014, 42001, 45009} logger.warning("WeCom message rejected errcode=%s", error_code) return DeliveryResult( False, f"WeCom errcode {error_code}", retryable, ) async def _get_access_token(self) -> _TokenResult: """Return a cached token or refresh it once under a process-local lock.""" if self._token_is_valid(): return _TokenResult(self._access_token) async with self._token_lock: if self._token_is_valid(): return _TokenResult(self._access_token) try: response = await self._http.get( _TOKEN_URL, params={ "corpid": self._settings.WECOM_CORP_ID, "corpsecret": self._settings.WECOM_APP_SECRET, }, ) except (httpx.TimeoutException, httpx.TransportError): logger.warning("WeCom token transport failure") return _TokenResult(None, "token transport error", True) if response.status_code == 429 or response.status_code >= 500: logger.warning( "WeCom token transient HTTP failure status=%s", response.status_code, ) return _TokenResult(None, f"token HTTP {response.status_code}", True) if response.status_code >= 400: logger.warning( "WeCom token HTTP failure status=%s", response.status_code, ) return _TokenResult(None, f"token HTTP {response.status_code}", False) try: data = response.json() if not isinstance(data, dict): raise TypeError("response JSON must be an object") error_code = int(data.get("errcode", -1)) access_token = data.get("access_token") expires_in = int(data.get("expires_in", 0)) except (TypeError, ValueError): logger.warning("WeCom token endpoint returned invalid JSON") return _TokenResult(None, "invalid token response", True) if error_code != 0 or not access_token or expires_in <= 0: logger.warning("WeCom token request rejected errcode=%s", error_code) return _TokenResult( None, f"WeCom token errcode {error_code}", error_code == -1, ) self._access_token = str(access_token) cache_seconds = max(0, expires_in - _TOKEN_EXPIRY_MARGIN_SECONDS) self._access_token_expires_at = self._now() + timedelta( seconds=cache_seconds ) return _TokenResult(self._access_token) def _token_is_valid(self) -> bool: return ( self._access_token is not None and self._access_token_expires_at is not None and self._now() < self._access_token_expires_at ) class NotificationDispatcher: """Claim due outbox rows and persist their delivery outcome.""" def __init__( self, client: TextNotificationClient, runtime_settings: Settings | None = None, ): self._client = client self._settings = runtime_settings or settings async def dispatch_due( self, db: AsyncSession, now: datetime, ) -> DispatchSummary: """Attempt each currently due message once without committing the session.""" rows = list( ( await db.execute( select(NotificationOutbox) .where(NotificationOutbox.status == NotificationStatus.pending) .where( or_( NotificationOutbox.next_attempt_at.is_(None), NotificationOutbox.next_attempt_at <= now, ) ) .order_by(NotificationOutbox.created_at, NotificationOutbox.id) ) ) .scalars() .all() ) event_ids = [row.alert_event_id for row in rows] events = {} if event_ids: events = { event.id: event for event in ( ( await db.execute( select(AlertEvent).where(AlertEvent.id.in_(event_ids)) ) ) .scalars() .all() ) } sent = rescheduled = failed = 0 for start in range(0, len(rows), MAX_EVENTS_PER_MESSAGE): batch = rows[start : start + MAX_EVENTS_PER_MESSAGE] for row in batch: row.status = NotificationStatus.sending row.locked_at = now try: result = await self._client.send_text( _MESSAGE_SEPARATOR.join(row.message_content for row in batch) ) except Exception as exc: # Transport implementations must not strand rows. logger.warning( "Notification client raised type=%s", type(exc).__name__, ) result = DeliveryResult(False, "notification client error", True) for row in batch: row.attempt_count += 1 event = events.get(row.alert_event_id) if result.success: row.status = NotificationStatus.sent row.sent_at = now row.next_attempt_at = None row.last_error = None sent += 1 elif ( result.retryable and row.attempt_count < self._settings.wecom_notification_max_attempts ): delay = self._settings.wecom_retry_base_seconds * ( 2 ** (row.attempt_count - 1) ) row.status = NotificationStatus.pending row.next_attempt_at = now + timedelta(seconds=delay) row.last_error = (result.error or "delivery failed")[:512] rescheduled += 1 else: row.status = NotificationStatus.failed row.next_attempt_at = None row.last_error = (result.error or "delivery failed")[:512] failed += 1 if event is not None: event.notification_sent = result.success event.notification_attempts = row.attempt_count event.last_notification_error = row.last_error await db.flush() return DispatchSummary(sent=sent, rescheduled=rescheduled, failed=failed)