Traditional Notification Architecture:

+------------+    +----------+
| Approval   |--->| Email    |  API-1
+------------+    +----------+

+------------+    +----------+
| Alerting   |--->| SMS      |  API-2
+------------+    +----------+

+------------+    +----------+
| Collab     |--->| DingTalk |  API-3
+------------+    +----------+

Problems:
- Each module integrates independently
- N modules x M channels = N*M integrations
- No unified template, tracking, or rate limiting

coomia-dip Notification Architecture:

  +----------+  +----------+  +----------+
  | Decision |  | Approval |  | Alert    |
  | Engine   |  | Engine   |  | System   |
  +----+-----+  +----+-----+  +----+-----+
       |              |              |
       v              v              v
  +----------------------------------------+
  |         NotificationEngine              |
  |  +------------+  +------------------+  |
  |  | Template   |  | Preference       |  |
  |  | Engine     |  | Router           |  |
  |  +------------+  +------------------+  |
  |  +------------+  +------------------+  |
  |  | Rate       |  | Delivery         |  |
  |  | Limiter    |  | Tracker          |  |
  |  +------------+  +------------------+  |
  +----+---+---+---+---+---+---+---+---+--+
       |   |   |   |   |   |   |   |   |
       v   v   v   v   v   v   v   v   v
     Email SMS Hook Slack Ding WeCom Fei InApp Push

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class ChannelType(Enum):
    EMAIL = "email"
    SMS = "sms"
    WEBHOOK = "webhook"
    SLACK = "slack"
    DINGTALK = "dingtalk"
    WECOM = "wecom"
    FEISHU = "feishu"
    IN_APP = "in_app"
    PUSH = "push"


@dataclass
class NotificationMessage:
    """Notification message"""
    message_id: str
    channel: ChannelType
    recipient: str
    subject: str
    body: str
    body_html: str | None = None
    priority: str = "normal"
    template_id: str | None = None
    variables: dict[str, Any] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class DeliveryResult:
    """Delivery result"""
    message_id: str
    channel: ChannelType
    status: str          # sent, failed, queued, rate_limited
    provider_id: str = ""
    error: str = ""
    timestamp: datetime = field(default_factory=datetime.utcnow)


class ChannelAdapter(ABC):
    """Channel adapter base class"""

    @abstractmethod
    async def send(self, message: NotificationMessage) -> DeliveryResult:
        ...

    @abstractmethod
    async def send_batch(self, messages: list[NotificationMessage]) -> list[DeliveryResult]:
        ...

    @abstractmethod
    async def health_check(self) -> dict:
        ...

    @abstractmethod
    def channel_type(self) -> ChannelType:
        ...

import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart


class EmailAdapter(ChannelAdapter):
    """Email channel adapter"""

    def __init__(self, smtp_host: str, smtp_port: int,
                 username: str, password: str,
                 from_addr: str):
        self._host = smtp_host
        self._port = smtp_port
        self._username = username
        self._password = password
        self._from = from_addr

    def channel_type(self) -> ChannelType:
        return ChannelType.EMAIL

    async def send(self, message: NotificationMessage) -> DeliveryResult:
        try:
            msg = MIMEMultipart("alternative")
            msg["Subject"] = message.subject
            msg["From"] = self._from
            msg["To"] = message.recipient

            msg.attach(MIMEText(message.body, "plain"))
            if message.body_html:
                msg.attach(MIMEText(message.body_html, "html"))

            await aiosmtplib.send(
                msg,
                hostname=self._host,
                port=self._port,
                username=self._username,
                password=self._password,
                use_tls=True,
            )

            return DeliveryResult(
                message_id=message.message_id,
                channel=ChannelType.EMAIL,
                status="sent",
            )
        except Exception as e:
            return DeliveryResult(
                message_id=message.message_id,
                channel=ChannelType.EMAIL,
                status="failed",
                error=str(e),
            )

    async def send_batch(self, messages: list[NotificationMessage]) -> list[DeliveryResult]:
        return [await self.send(m) for m in messages]

    async def health_check(self) -> dict:
        try:
            smtp = aiosmtplib.SMTP(hostname=self._host, port=self._port)
            await smtp.connect()
            await smtp.quit()
            return {"status": "healthy", "channel": "email"}
        except Exception as e:
            return {"status": "unhealthy", "channel": "email", "error": str(e)}

import httpx


class DingTalkAdapter(ChannelAdapter):
    """DingTalk robot adapter"""

    def __init__(self, webhook_url: str, secret: str | None = None):
        self._url = webhook_url
        self._secret = secret

    def channel_type(self) -> ChannelType:
        return ChannelType.DINGTALK

    async def send(self, message: NotificationMessage) -> DeliveryResult:
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": message.subject,
                "text": message.body,
            },
            "at": {
                "atMobiles": [message.recipient],
                "isAtAll": False,
            },
        }

        async with httpx.AsyncClient() as client:
            url = self._sign_url() if self._secret else self._url
            resp = await client.post(url, json=payload, timeout=10)

            if resp.status_code == 200 and resp.json().get("errcode") == 0:
                return DeliveryResult(
                    message_id=message.message_id,
                    channel=ChannelType.DINGTALK,
                    status="sent",
                )
            else:
                return DeliveryResult(
                    message_id=message.message_id,
                    channel=ChannelType.DINGTALK,
                    status="failed",
                    error=resp.text,
                )

    async def send_batch(self, messages: list[NotificationMessage]) -> list[DeliveryResult]:
        return [await self.send(m) for m in messages]

    async def health_check(self) -> dict:
        return {"status": "healthy", "channel": "dingtalk"}

    def _sign_url(self) -> str:
        import hmac
        import hashlib
        import base64
        import urllib.parse

        timestamp = str(int(datetime.utcnow().timestamp() * 1000))
        string_to_sign = f"{timestamp}\n{self._secret}"
        hmac_code = hmac.new(
            self._secret.encode(), string_to_sign.encode(),
            digestmod=hashlib.sha256
        ).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return f"{self._url}&timestamp={timestamp}&sign={sign}"


class SlackAdapter(ChannelAdapter):
    """Slack channel adapter"""

    def __init__(self, webhook_url: str):
        self._url = webhook_url

    def channel_type(self) -> ChannelType:
        return ChannelType.SLACK

    async def send(self, message: NotificationMessage) -> DeliveryResult:
        payload = {
            "text": message.subject,
            "blocks": [
                {
                    "type": "section",
                    "text": {"type": "mrkdwn", "text": message.body},
                }
            ],
        }

        async with httpx.AsyncClient() as client:
            resp = await client.post(self._url, json=payload, timeout=10)
            status = "sent" if resp.status_code == 200 else "failed"
            return DeliveryResult(
                message_id=message.message_id,
                channel=ChannelType.SLACK,
                status=status,
                error="" if status == "sent" else resp.text,
            )

    async def send_batch(self, messages: list[NotificationMessage]) -> list[DeliveryResult]:
        return [await self.send(m) for m in messages]

    async def health_check(self) -> dict:
        return {"status": "healthy", "channel": "slack"}

from jinja2 import Environment, BaseLoader


class TemplateEngine:
    """Notification template engine"""

    def __init__(self):
        self._env = Environment(loader=BaseLoader())
        self._templates: dict[str, dict[str, str]] = {}

    def register_template(self, template_id: str,
                           formats: dict[str, str]) -> None:
        """Register a template with multiple formats.

        formats keys: html, markdown, plain, short
        """
        self._templates[template_id] = formats

    def render(self, template_id: str, format_key: str,
               variables: dict[str, Any]) -> str:
        """Render template with variables"""
        templates = self._templates.get(template_id)
        if not templates:
            raise ValueError(f"Template not found: {template_id}")

        template_str = templates.get(format_key, templates.get("plain", ""))
        template = self._env.from_string(template_str)
        return template.render(**variables)

    def render_for_channel(self, template_id: str,
                           channel: ChannelType,
                           variables: dict[str, Any]) -> tuple[str, str | None]:
        """Render template appropriate for channel"""
        format_map = {
            ChannelType.EMAIL: ("plain", "html"),
            ChannelType.SMS: ("short", None),
            ChannelType.SLACK: ("markdown", None),
            ChannelType.DINGTALK: ("markdown", None),
            ChannelType.WECOM: ("markdown", None),
            ChannelType.FEISHU: ("markdown", None),
            ChannelType.IN_APP: ("plain", None),
            ChannelType.PUSH: ("short", None),
            ChannelType.WEBHOOK: ("plain", None),
        }

        body_fmt, html_fmt = format_map.get(channel, ("plain", None))
        body = self.render(template_id, body_fmt, variables)
        html = self.render(template_id, html_fmt, variables) if html_fmt else None
        return body, html


# Template registration example
template_engine = TemplateEngine()
template_engine.register_template("approval_pending", {
    "html": """
    <h2>Approval Request: {{ title }}</h2>
    <p>Applicant: {{ applicant }}</p>
    <p>Amount: {{ amount | currency }}</p>
    <a href="{{ action_url }}">Review Now</a>
    """,
    "markdown": """
    **Approval Request: {{ title }}**
    - Applicant: {{ applicant }}
    - Amount: {{ amount }}
    [Review Now]({{ action_url }})
    """,
    "plain": "Approval: {{ title }} from {{ applicant }} ({{ amount }}). Review: {{ action_url }}",
    "short": "Approval needed: {{ title }} ({{ amount }})",
})

@dataclass
class UserNotificationPreference:
    """User notification preferences"""
    user_id: str
    channels: dict[str, list[ChannelType]]    # category -> preferred channels
    quiet_hours: tuple[int, int] = (22, 8)    # 22:00 - 08:00
    digest_enabled: bool = True
    digest_interval_minutes: int = 60
    language: str = "en"


class PreferenceRouter:
    """Route notifications based on user preferences"""

    def __init__(self, preference_store):
        self._store = preference_store
        self._fallback_chain = [
            ChannelType.IN_APP,
            ChannelType.EMAIL,
            ChannelType.PUSH,
        ]

    async def route(self, user_id: str, category: str,
                     available_channels: set[ChannelType]) -> list[ChannelType]:
        """Determine channels for a notification"""
        pref = await self._store.get(user_id)
        if pref is None:
            return [self._fallback_chain[0]]

        preferred = pref.channels.get(category, self._fallback_chain)
        valid = [ch for ch in preferred if ch in available_channels]

        if not valid:
            for ch in self._fallback_chain:
                if ch in available_channels:
                    return [ch]

        return valid or [ChannelType.IN_APP]

import time
from collections import defaultdict


class NotificationRateLimiter:
    """Per-user, per-channel rate limiting"""

    def __init__(self):
        self._windows: dict[str, list[float]] = defaultdict(list)
        self._limits = {
            ChannelType.EMAIL: (10, 3600),       # 10 per hour
            ChannelType.SMS: (5, 3600),           # 5 per hour
            ChannelType.PUSH: (20, 3600),         # 20 per hour
            ChannelType.DINGTALK: (30, 3600),     # 30 per hour
            ChannelType.SLACK: (30, 3600),        # 30 per hour
            ChannelType.IN_APP: (100, 3600),      # 100 per hour
        }

    def check(self, user_id: str, channel: ChannelType) -> bool:
        """Return True if notification is allowed"""
        key = f"{user_id}:{channel.value}"
        max_count, window_seconds = self._limits.get(channel, (50, 3600))

        now = time.monotonic()
        timestamps = self._windows[key]

        # Remove expired entries
        self._windows[key] = [
            t for t in timestamps if now - t < window_seconds
        ]

        if len(self._windows[key]) >= max_count:
            return False

        self._windows[key].append(now)
        return True


class QuietHoursChecker:
    """Check if current time falls within quiet hours"""

    @staticmethod
    def is_quiet(quiet_hours: tuple[int, int],
                  timezone_offset: int = 8) -> bool:
        from datetime import datetime, timedelta
        now = datetime.utcnow() + timedelta(hours=timezone_offset)
        hour = now.hour
        start, end = quiet_hours

        if start > end:  # e.g., 22:00 - 08:00
            return hour >= start or hour < end
        else:
            return start <= hour < end

from collections import defaultdict


class DigestAggregator:
    """Aggregate notifications into digests"""

    def __init__(self):
        self._pending: dict[str, list[NotificationMessage]] = defaultdict(list)
        self._last_sent: dict[str, float] = {}

    def add(self, user_id: str, message: NotificationMessage) -> bool:
        """Add message to digest. Returns True if digest should be sent."""
        key = f"{user_id}:{message.metadata.get('category', 'default')}"
        self._pending[key].append(message)

        last = self._last_sent.get(key, 0)
        interval = 3600  # 1 hour default
        if time.monotonic() - last >= interval:
            return True
        return False

    def flush(self, user_id: str, category: str = "default") -> list[NotificationMessage]:
        """Get and clear pending messages for digest"""
        key = f"{user_id}:{category}"
        messages = self._pending.pop(key, [])
        self._last_sent[key] = time.monotonic()
        return messages

    def create_digest(self, messages: list[NotificationMessage]) -> NotificationMessage:
        """Create a single digest message from multiple messages"""
        subjects = [m.subject for m in messages]
        body_parts = [f"- {m.subject}: {m.body[:100]}" for m in messages]

        return NotificationMessage(
            message_id=f"digest-{messages[0].message_id}",
            channel=messages[0].channel,
            recipient=messages[0].recipient,
            subject=f"Digest: {len(messages)} notifications",
            body="\n".join(body_parts),
        )

class FallbackDelivery:
    """Deliver with automatic fallback to backup channels"""

    def __init__(self, adapters: dict[ChannelType, ChannelAdapter]):
        self._adapters = adapters
        self._fallback_chains = {
            ChannelType.EMAIL: [ChannelType.IN_APP, ChannelType.PUSH],
            ChannelType.SMS: [ChannelType.PUSH, ChannelType.IN_APP],
            ChannelType.DINGTALK: [ChannelType.EMAIL, ChannelType.IN_APP],
            ChannelType.SLACK: [ChannelType.EMAIL, ChannelType.IN_APP],
            ChannelType.WECOM: [ChannelType.EMAIL, ChannelType.IN_APP],
            ChannelType.FEISHU: [ChannelType.EMAIL, ChannelType.IN_APP],
            ChannelType.PUSH: [ChannelType.IN_APP],
            ChannelType.IN_APP: [],
        }

    async def deliver(self, message: NotificationMessage) -> DeliveryResult:
        """Attempt delivery with fallback"""
        primary = message.channel
        channels_to_try = [primary] + self._fallback_chains.get(primary, [])

        for channel in channels_to_try:
            adapter = self._adapters.get(channel)
            if adapter is None:
                continue

            result = await adapter.send(
                NotificationMessage(
                    message_id=message.message_id,
                    channel=channel,
                    recipient=message.recipient,
                    subject=message.subject,
                    body=message.body,
                    body_html=message.body_html,
                    priority=message.priority,
                )
            )

            if result.status == "sent":
                return result

        return DeliveryResult(
            message_id=message.message_id,
            channel=primary,
            status="failed",
            error="All channels exhausted",
        )

class DeliveryTracker:
    """Track notification delivery lifecycle"""

    def __init__(self, store):
        self._store = store

    async def record(self, result: DeliveryResult,
                      message: NotificationMessage) -> None:
        await self._store.save({
            "message_id": result.message_id,
            "channel": result.channel.value,
            "recipient": message.recipient,
            "subject": message.subject,
            "status": result.status,
            "provider_id": result.provider_id,
            "error": result.error,
            "created_at": message.metadata.get("created_at"),
            "delivered_at": result.timestamp.isoformat(),
        })

    async def get_delivery_stats(self, time_range_hours: int = 24) -> dict:
        records = await self._store.query_recent(time_range_hours)
        total = len(records)
        if total == 0:
            return {"total": 0}

        by_status = {}
        by_channel = {}
        for r in records:
            by_status[r["status"]] = by_status.get(r["status"], 0) + 1
            by_channel[r["channel"]] = by_channel.get(r["channel"], 0) + 1

        return {
            "total": total,
            "by_status": by_status,
            "by_channel": by_channel,
            "success_rate": by_status.get("sent", 0) / total,
        }

syntax = "proto3";
package onto.notification.v1;

service NotificationService {
    rpc Send(SendRequest) returns (SendResponse);
    rpc SendBatch(BatchSendRequest) returns (BatchSendResponse);
    rpc GetDeliveryStatus(StatusRequest) returns (StatusResponse);
    rpc GetUserPreferences(PreferenceRequest) returns (PreferenceResponse);
    rpc UpdateUserPreferences(UpdatePreferenceRequest) returns (UpdatePreferenceResponse);
}

message SendRequest {
    string recipient = 1;
    string template_id = 2;
    map<string, string> variables = 3;
    string category = 4;
    string priority = 5;
    string preferred_channel = 6;
}

message SendResponse {
    string message_id = 1;
    string channel = 2;
    string status = 3;
}

class NotificationEngine:
    """Main notification engine orchestrator"""

    def __init__(self, adapters: dict[ChannelType, ChannelAdapter],
                 template_engine: TemplateEngine,
                 preference_router: PreferenceRouter,
                 rate_limiter: NotificationRateLimiter,
                 fallback: FallbackDelivery,
                 tracker: DeliveryTracker):
        self._adapters = adapters
        self._templates = template_engine
        self._router = preference_router
        self._limiter = rate_limiter
        self._fallback = fallback
        self._tracker = tracker

    async def send(self, recipient: str, template_id: str,
                    variables: dict[str, Any],
                    category: str = "general",
                    priority: str = "normal") -> DeliveryResult:
        """Send a notification"""
        # 1. Route to preferred channel
        available = set(self._adapters.keys())
        channels = await self._router.route(recipient, category, available)

        for channel in channels:
            # 2. Rate limiting check
            if not self._limiter.check(recipient, channel):
                continue

            # 3. Render template
            body, html = self._templates.render_for_channel(
                template_id, channel, variables
            )

            message = NotificationMessage(
                message_id=f"ntf-{id(self)}-{channel.value}",
                channel=channel,
                recipient=recipient,
                subject=variables.get("subject", template_id),
                body=body,
                body_html=html,
                priority=priority,
                template_id=template_id,
                variables=variables,
            )

            # 4. Deliver with fallback
            result = await self._fallback.deliver(message)

            # 5. Track
            await self._tracker.record(result, message)

            if result.status == "sent":
                return result

        return DeliveryResult(
            message_id="",
            channel=channels[0] if channels else ChannelType.IN_APP,
            status="rate_limited",
            error="All channels rate limited",
        )

Notification Engine Dashboard (24h):

  Delivery Volume:
  +-------------------------------------------+
  | Email     ============ 4,521              |
  | DingTalk  ========     3,102              |
  | In-App    =======      2,880              |
  | Push      =====        1,950              |
  | SMS       ==             721              |
  | Slack     ==             645              |
  | WeChat    =              412              |
  | Feishu    =              380              |
  | Webhook                  198              |
  +-------------------------------------------+

  Success Rate by Channel:
  Email: 99.2%  SMS: 98.5%  DingTalk: 99.8%  Push: 97.1%

  Rate Limited: 342 (2.3%)
  Fallback Used: 89 (0.6%)
  Digest Merged: 1,204 messages -> 156 digests