"""Notification channels: LogChannel + WebhookChannel. Protocol-based: NotificationChannel defines send(kind, text, data) -> bool. LogChannel writes to notification_log (delivered=true always). WebhookChannel POSTs to NOTIFY_WEBHOOK_URL (2xx=delivered, else error row). send_notification(kind, text, data) is the public entry point used by the scheduler and email watch modules. """ from __future__ import annotations import json import logging import os from typing import Any, Protocol from app.db import execute, fetch_all logger = logging.getLogger(__name__) class NotificationChannel(Protocol): """Protocol for notification delivery channels.""" def send(self, kind: str, text: str, data: dict[str, Any]) -> bool: """Send a notification. Returns True if delivered, False otherwise.""" ... def _normalize_log_row(row: dict[str, Any]) -> dict[str, Any]: payload = row.get("payload") if payload is not None and not isinstance(payload, dict): payload = json.loads(payload) if isinstance(payload, str) else payload return { "id": str(row["id"]), "channel": row["channel"], "kind": row["kind"], "payload": payload or {}, "delivered": row["delivered"], "error": row.get("error"), "created_at": row["created_at"].isoformat() if row.get("created_at") else None, } class LogChannel: """Writes notification entries to notification_log with delivered=true.""" channel_name: str = "log" def send(self, kind: str, text: str, data: dict[str, Any]) -> bool: row = execute( """ INSERT INTO notification_log (channel, kind, payload, delivered, error) VALUES (%s, %s, %s, true, NULL) RETURNING * """, (self.channel_name, kind, json.dumps({"text": text, **data})), ) if row is None: logger.error("LogChannel: insert notification_log failed") return False logger.info("LogChannel: delivered kind=%s", kind) return True class WebhookChannel: """POSTs notification payload to NOTIFY_WEBHOOK_URL. On 2xx response: writes notification_log with delivered=true. On non-2xx or exception: writes notification_log with delivered=false and error. """ channel_name: str = "webhook" def __init__(self, url: str | None = None) -> None: self.url = url or os.environ.get("NOTIFY_WEBHOOK_URL", "") def send(self, kind: str, text: str, data: dict[str, Any]) -> bool: payload: dict[str, Any] = {"kind": kind, "text": text, "data": data} error: str | None = None delivered = False if not self.url: error = "NOTIFY_WEBHOOK_URL not configured" logger.warning("WebhookChannel: %s", error) else: try: import httpx resp = httpx.post(self.url, json=payload, timeout=10) if 200 <= resp.status_code < 300: delivered = True else: error = f"HTTP {resp.status_code}: {resp.text[:200]}" logger.warning("WebhookChannel: %s", error) except Exception as exc: error = str(exc) logger.warning("WebhookChannel: exception: %s", error) row = execute( """ INSERT INTO notification_log (channel, kind, payload, delivered, error) VALUES (%s, %s, %s, %s, %s) RETURNING * """, ( self.channel_name, kind, json.dumps(payload), delivered, error, ), ) if row is None: logger.error("WebhookChannel: insert notification_log failed") return delivered # --- Channel registry --- _channels: list[NotificationChannel] | None = None def get_channels() -> list[NotificationChannel]: """Return the list of active notification channels. LogChannel is always included. WebhookChannel is included when NOTIFY_WEBHOOK_URL is set. """ global _channels if _channels is not None: return _channels channels: list[NotificationChannel] = [LogChannel()] webhook_url = os.environ.get("NOTIFY_WEBHOOK_URL", "").strip() if webhook_url: channels.append(WebhookChannel()) _channels = channels return channels def set_channels(channels: list[NotificationChannel] | None) -> None: """Override channel list (for testing).""" global _channels _channels = channels def reset_channels() -> None: """Reset to default (for testing).""" global _channels _channels = None def send_notification(kind: str, text: str, data: dict[str, Any] | None = None) -> None: """Send a notification via all active channels. kind: e.g. 'daily_digest', 'email_suggestion' text: human-readable notification text data: structured payload dict """ chs = get_channels() payload = data or {} for ch in chs: try: ch.send(kind, text, payload) except Exception: logger.exception("send_notification: channel %s failed", type(ch).__name__) # --- Repository helpers --- def list_notification_log(limit: int = 50) -> list[dict[str, Any]]: """Return recent notification_log rows, newest first.""" rows = fetch_all( """ SELECT * FROM notification_log ORDER BY created_at DESC LIMIT %s """, (limit,), ) return [_normalize_log_row(r) for r in rows]