Source code for corvix.config.notifications

"""Notifications configuration model and YAML parsing."""

from __future__ import annotations

from dataclasses import dataclass, field

from corvix.config._utils import (
    _ensure_map,
    _get_bool,
    _get_float,
    _get_int,
    _get_str,
)


@dataclass(slots=True)

[docs]
class BrowserTabTargetConfig:
    """Config for in-tab browser notification delivery."""


[docs]
    enabled: bool = True


[docs]
    max_per_cycle: int = 5


[docs]
    cooldown_seconds: int = 10




@dataclass(slots=True)

[docs]
class WebPushTargetConfig:
    """Config for background Web Push notification delivery (phase 2)."""


[docs]
    enabled: bool = False


[docs]
    vapid_public_key_env: str = "CORVIX_VAPID_PUBLIC_KEY"


[docs]
    vapid_private_key_env: str = "CORVIX_VAPID_PRIVATE_KEY"


[docs]
    subject: str = ""




@dataclass(slots=True)

[docs]
class NotificationsDetectConfig:
    """Controls which records qualify for notification events."""


[docs]
    include_read: bool = False


[docs]
    min_score: float = 0.0




@dataclass(slots=True)

[docs]
class NotificationsConfig:
    """Top-level notifications configuration."""


[docs]
    enabled: bool = True


[docs]
    detect: NotificationsDetectConfig = field(default_factory=NotificationsDetectConfig)


[docs]
    browser_tab: BrowserTabTargetConfig = field(default_factory=BrowserTabTargetConfig)


[docs]
    web_push: WebPushTargetConfig = field(default_factory=WebPushTargetConfig)





[docs]
def _parse_browser_tab(browser_raw: dict[str, object]) -> BrowserTabTargetConfig:
    """Parse and validate browser_tab config section."""
    max_per_cycle = _get_int(browser_raw, "max_per_cycle", 5, "notifications.browser_tab.max_per_cycle")
    if max_per_cycle < 0:
        msg = "Config value 'notifications.browser_tab.max_per_cycle' must be >= 0."
        raise ValueError(msg)
    cooldown_seconds = _get_int(
        browser_raw,
        "cooldown_seconds",
        10,
        "notifications.browser_tab.cooldown_seconds",
    )
    if cooldown_seconds < 0:
        msg = "Config value 'notifications.browser_tab.cooldown_seconds' must be >= 0."
        raise ValueError(msg)
    return BrowserTabTargetConfig(
        enabled=_get_bool(browser_raw, "enabled", True, "notifications.browser_tab.enabled"),
        max_per_cycle=max_per_cycle,
        cooldown_seconds=cooldown_seconds,
    )




[docs]
def _parse_notifications(value: object) -> NotificationsConfig:
    notif = _ensure_map(value, "notifications")
    detect_raw = _ensure_map(notif.get("detect", {}), "notifications.detect")
    browser_raw = _ensure_map(notif.get("browser_tab", {}), "notifications.browser_tab")
    web_push_raw = _ensure_map(notif.get("web_push", {}), "notifications.web_push")
    return NotificationsConfig(
        enabled=_get_bool(notif, "enabled", True, "notifications.enabled"),
        detect=NotificationsDetectConfig(
            include_read=_get_bool(detect_raw, "include_read", False, "notifications.detect.include_read"),
            min_score=_get_float(detect_raw, "min_score", 0.0, "notifications.detect.min_score"),
        ),
        browser_tab=_parse_browser_tab(browser_raw),
        web_push=WebPushTargetConfig(
            enabled=_get_bool(web_push_raw, "enabled", False, "notifications.web_push.enabled"),
            vapid_public_key_env=_get_str(
                web_push_raw,
                "vapid_public_key_env",
                "CORVIX_VAPID_PUBLIC_KEY",
                "notifications.web_push.vapid_public_key_env",
            ),
            vapid_private_key_env=_get_str(
                web_push_raw,
                "vapid_private_key_env",
                "CORVIX_VAPID_PRIVATE_KEY",
                "notifications.web_push.vapid_private_key_env",
            ),
            subject=_get_str(web_push_raw, "subject", "", "notifications.web_push.subject"),
        ),
    )