CheddahBot/cheddahbot/ntfy.py

"""ntfy.sh push notification sender.

Subscribes to the NotificationBus and routes notifications to ntfy.sh
topics based on category and message-pattern matching.
"""

from __future__ import annotations

import hashlib
import logging
import re
import threading
from dataclasses import dataclass, field
from datetime import date

import httpx

log = logging.getLogger(__name__)


@dataclass
class NtfyChannel:
    """One ntfy topic with routing rules."""

    name: str
    server: str
    topic: str
    categories: list[str]
    include_patterns: list[str] = field(default_factory=list)
    exclude_patterns: list[str] = field(default_factory=list)
    priority: str = "high"
    tags: str = ""

    def accepts(self, message: str, category: str) -> bool:
        """Return True if this channel should receive the notification."""
        if category not in self.categories:
            return False
        if self.exclude_patterns:
            for pat in self.exclude_patterns:
                if re.search(pat, message, re.IGNORECASE):
                    return False
        if self.include_patterns:
            return any(
                re.search(pat, message, re.IGNORECASE)
                for pat in self.include_patterns
            )
        return True  # no include_patterns = accept all matching categories


class NtfyNotifier:
    """Posts notifications to ntfy.sh topics."""

    def __init__(
        self,
        channels: list[NtfyChannel],
        *,
        daily_cap: int = 200,
    ):
        self._channels = [ch for ch in channels if ch.topic]
        self._daily_cap = daily_cap
        self._lock = threading.Lock()
        # dedup: set of hash(channel.name + message) — persists for process lifetime
        self._sent: set[str] = set()
        # daily cap tracking
        self._daily_count = 0
        self._daily_date = ""
        # 429 backoff: date string when rate-limited
        self._rate_limited_until = ""
        if self._channels:
            log.info(
                "ntfy notifier initialized with %d channel(s): %s",
                len(self._channels),
                ", ".join(ch.name for ch in self._channels),
            )

    @property
    def enabled(self) -> bool:
        return bool(self._channels)

    def _today(self) -> str:
        return date.today().isoformat()

    def _check_and_track(self, channel_name: str, message: str) -> bool:
        """Return True if this message should be sent. Updates internal state."""
        today = self._today()

        with self._lock:
            # 429 backoff: skip all sends for rest of day
            if self._rate_limited_until == today:
                return False

            # Reset daily counter on date rollover (but keep dedup memory)
            if self._daily_date != today:
                self._daily_date = today
                self._daily_count = 0
                self._rate_limited_until = ""

            # Daily cap check
            if self._daily_count >= self._daily_cap:
                return False

            # Dedup check — once sent, never send the same message again
            # (until process restart)
            key = hashlib.md5(
                (channel_name + "\0" + message).encode()
            ).hexdigest()
            if key in self._sent:
                log.info(
                    "ntfy dedup: suppressed duplicate to '%s'", channel_name,
                )
                return False

            # All checks passed — record send
            self._sent.add(key)
            self._daily_count += 1

            if self._daily_count == self._daily_cap:
                log.warning(
                    "ntfy daily cap reached (%d). No more sends today.",
                    self._daily_cap,
                )

            return True

    def _mark_rate_limited(self) -> None:
        """Flag that we got a 429 — suppress all sends for rest of day."""
        with self._lock:
            self._rate_limited_until = self._today()
        log.warning("ntfy 429 received. Suppressing all sends for rest of day.")

    def notify(self, message: str, category: str) -> None:
        """Route a notification to matching ntfy channels.

        This is the callback signature expected by NotificationBus.subscribe().
        Each matching channel posts in a daemon thread so the notification
        pipeline is never blocked.
        """
        for channel in self._channels:
            if channel.accepts(message, category):
                if not self._check_and_track(channel.name, message):
                    continue
                t = threading.Thread(
                    target=self._post,
                    args=(channel, message, category),
                    daemon=True,
                )
                t.start()

    def _post(self, channel: NtfyChannel, message: str, category: str) -> None:
        """Send a notification to an ntfy topic. Fire-and-forget."""
        url = f"{channel.server.rstrip('/')}/{channel.topic}"
        headers: dict[str, str] = {
            "Title": f"CheddahBot [{category}]",
            "Priority": channel.priority,
        }
        if channel.tags:
            headers["Tags"] = channel.tags
        try:
            resp = httpx.post(
                url,
                content=message.encode("utf-8"),
                headers=headers,
                timeout=10.0,
            )
            if resp.status_code == 429:
                self._mark_rate_limited()
            elif resp.status_code >= 400:
                log.warning(
                    "ntfy '%s' returned %d: %s",
                    channel.name, resp.status_code, resp.text[:200],
                )
            else:
                log.debug("ntfy notification sent to '%s'", channel.name)
        except httpx.HTTPError as e:
            log.warning("ntfy '%s' failed: %s", channel.name, e)