-- 通知リクエスト (バッチ単位)
CREATE TABLE notification_batches (
    batch_id        UUID PRIMARY KEY,
    template_id     VARCHAR(100) NOT NULL,
    channels        JSONB NOT NULL,          -- ["push", "email"]
    data            JSONB NOT NULL,          -- テンプレート変数
    priority        VARCHAR(10) DEFAULT 'normal',
    scheduled_at    TIMESTAMPTZ,
    created_by      VARCHAR(100) NOT NULL,   -- 送信元サービス
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    status          VARCHAR(20) DEFAULT 'processing'
);
 
-- 個別通知 (ユーザー × チャネル 単位)
CREATE TABLE notifications (
    notification_id UUID PRIMARY KEY,
    batch_id        UUID REFERENCES notification_batches(batch_id),
    user_id         VARCHAR(100) NOT NULL,
    channel         VARCHAR(20) NOT NULL,
    content         JSONB NOT NULL,          -- 展開済みコンテンツ
    priority        VARCHAR(10) NOT NULL,
    status          VARCHAR(20) DEFAULT 'queued',
    -- queued / sending / delivered / failed / dead_lettered
    retry_count     INT DEFAULT 0,
    sent_at         TIMESTAMPTZ,
    delivered_at    TIMESTAMPTZ,
    opened_at       TIMESTAMPTZ,
    clicked_at      TIMESTAMPTZ,
    error_message   TEXT,
    created_at      TIMESTAMPTZ DEFAULT NOW(),
 
    INDEX idx_notifications_user (user_id, created_at DESC),
    INDEX idx_notifications_status (status, channel),
    INDEX idx_notifications_batch (batch_id)
);
 
-- ユーザー通知設定
CREATE TABLE user_notification_preferences (
    user_id         VARCHAR(100) PRIMARY KEY,
    push_enabled    BOOLEAN DEFAULT TRUE,
    email_enabled   BOOLEAN DEFAULT TRUE,
    sms_enabled     BOOLEAN DEFAULT TRUE,
    in_app_enabled  BOOLEAN DEFAULT TRUE,
    quiet_hours     JSONB,  -- {"start": "23:00", "end": "07:00", "timezone": "Asia/Tokyo"}
    categories      JSONB,  -- {"marketing": false, "transaction": true, ...}
    frequency_limit JSONB,  -- カスタムレート制限
    updated_at      TIMESTAMPTZ DEFAULT NOW()
);
 
-- デバイストークン
CREATE TABLE device_tokens (
    token_id        UUID PRIMARY KEY,
    user_id         VARCHAR(100) NOT NULL,
    platform        VARCHAR(20) NOT NULL,    -- ios / android / web
    device_token    TEXT NOT NULL,
    app_version     VARCHAR(20),
    is_active       BOOLEAN DEFAULT TRUE,
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    last_used_at    TIMESTAMPTZ,
 
    UNIQUE(user_id, device_token),
    INDEX idx_device_tokens_user (user_id, is_active)
);
 
-- 通知テンプレート
CREATE TABLE notification_templates (
    template_id     VARCHAR(100) NOT NULL,
    version         INT NOT NULL,
    channel         VARCHAR(20) NOT NULL,
    locale          VARCHAR(10) NOT NULL,    -- ja, en, zh, ...
    subject         TEXT,                     -- メール件名等
    title           TEXT,                     -- プッシュタイトル
    body            TEXT NOT NULL,            -- 本文テンプレート
    metadata        JSONB,                   -- カスタムデータ
    is_active       BOOLEAN DEFAULT TRUE,
    created_at      TIMESTAMPTZ DEFAULT NOW(),
 
    PRIMARY KEY (template_id, version, channel, locale)
);

# コード例 1: 通知サービスのエントリポイント
from fastapi import FastAPI, BackgroundTasks, HTTPException, Depends
from pydantic import BaseModel, validator
from enum import Enum
from typing import Optional
import uuid
import logging
 
app = FastAPI(title="Notification Service", version="2.0")
logger = logging.getLogger(__name__)
 
class Channel(str, Enum):
    PUSH = "push"
    EMAIL = "email"
    SMS = "sms"
    IN_APP = "in_app"
    WEB_PUSH = "web_push"
 
class Priority(str, Enum):
    CRITICAL = "critical"  # システム障害、セキュリティ
    HIGH = "high"          # 決済完了、パスワード変更
    NORMAL = "normal"      # 一般通知
    LOW = "low"            # マーケティング、レコメンド
 
class NotificationRequest(BaseModel):
    user_ids: list[str]
    template_id: str
    channels: list[Channel]
    data: dict
    priority: Priority = Priority.NORMAL
    scheduled_at: Optional[str] = None
    idempotency_key: Optional[str] = None  # べき等性キー
    category: str = "general"               # 通知カテゴリ
 
    @validator('user_ids')
    def validate_user_ids(cls, v):
        if len(v) == 0:
            raise ValueError("user_ids must not be empty")
        if len(v) > 10000:
            raise ValueError("user_ids must not exceed 10000")
        return v
 
class NotificationResponse(BaseModel):
    batch_id: str
    status: str
    total_recipients: int
    accepted: int
    rejected: int
 
class NotificationService:
    """
    通知サービスのコアオーケストレーター。
 
    責務:
    1. リクエストのバリデーション
    2. ユーザー設定に基づくフィルタリング
    3. テンプレートの展開
    4. 重複排除
    5. レート制限の適用
    6. チャネル別キューへの投入
    """
 
    def __init__(self, queue, user_service, template_service,
                 rate_limiter, dedup_service, metrics):
        self.queue = queue
        self.user_service = user_service
        self.template_service = template_service
        self.rate_limiter = rate_limiter
        self.dedup_service = dedup_service
        self.metrics = metrics
 
    async def send(self, request: NotificationRequest) -> NotificationResponse:
        batch_id = str(uuid.uuid4())
        accepted = 0
        rejected = 0
 
        # べき等性チェック
        if request.idempotency_key:
            if await self.dedup_service.is_duplicate(request.idempotency_key):
                logger.info(f"Duplicate request: {request.idempotency_key}")
                raise HTTPException(status_code=409, detail="Duplicate request")
 
        for user_id in request.user_ids:
            try:
                result = await self._process_user(
                    batch_id, user_id, request
                )
                if result:
                    accepted += 1
                else:
                    rejected += 1
            except Exception as e:
                logger.error(f"Error processing user {user_id}: {e}")
                rejected += 1
                self.metrics.increment("notification.processing_error")
 
        # メトリクス記録
        self.metrics.increment("notification.batches_created")
        self.metrics.gauge("notification.batch_size", len(request.user_ids))
 
        return NotificationResponse(
            batch_id=batch_id,
            status="queued",
            total_recipients=len(request.user_ids),
            accepted=accepted,
            rejected=rejected,
        )
 
    async def _process_user(self, batch_id: str, user_id: str,
                             request: NotificationRequest) -> bool:
        """個別ユーザーへの通知処理"""
 
        # 1. ユーザー設定を確認（キャッシュ付き）
        prefs = await self.user_service.get_preferences(user_id)
        if prefs is None:
            logger.warning(f"User not found: {user_id}")
            return False
 
        # 2. カテゴリ設定の確認
        if not prefs.is_category_enabled(request.category):
            return False
 
        # 3. Quiet Hours の確認
        if prefs.is_quiet_hours():
            if request.priority not in (Priority.CRITICAL, Priority.HIGH):
                # 低優先度はスケジュール配信に回す
                await self._schedule_after_quiet_hours(
                    batch_id, user_id, request, prefs
                )
                return True
 
        channels_sent = 0
        for channel in request.channels:
            # 4. ユーザーがこのチャネルを許可しているか
            if not prefs.is_channel_enabled(channel):
                continue
 
            # 5. レート制限チェック
            if not await self.rate_limiter.allow(user_id, channel.value):
                self.metrics.increment("notification.rate_limited",
                                       tags={"channel": channel.value})
                continue
 
            # 6. テンプレートを展開
            locale = prefs.locale or "ja"
            content = await self.template_service.render(
                request.template_id, channel.value,
                request.data, locale=locale
            )
 
            # 7. 通知IDの生成と重複排除
            notification_id = f"{batch_id}:{user_id}:{channel.value}"
            if await self.dedup_service.is_duplicate(notification_id):
                continue
 
            # 8. メッセージキューに投入
            message = {
                "id": notification_id,
                "batch_id": batch_id,
                "user_id": user_id,
                "channel": channel.value,
                "content": content,
                "priority": request.priority.value,
                "category": request.category,
                "created_at": int(time.time()),
            }
 
            # 優先度に応じたキュー選択
            topic = self._select_topic(channel, request.priority)
            await self.queue.publish(topic=topic, message=message)
            channels_sent += 1
 
        return channels_sent > 0
 
    def _select_topic(self, channel: Channel, priority: Priority) -> str:
        """優先度に応じたKafkaトピックを選択"""
        if priority == Priority.CRITICAL:
            return f"notifications.{channel.value}.critical"
        elif priority == Priority.HIGH:
            return f"notifications.{channel.value}.high"
        else:
            return f"notifications.{channel.value}.normal"
 
    async def _schedule_after_quiet_hours(self, batch_id, user_id,
                                           request, prefs):
        """Quiet Hours 後にスケジュール配信"""
        send_at = prefs.get_quiet_hours_end_utc()
        await self.queue.publish(
            topic="notifications.scheduled",
            message={
                "batch_id": batch_id,
                "user_id": user_id,
                "request": request.dict(),
                "scheduled_at": send_at.isoformat(),
            }
        )
 
@app.post("/api/v1/notifications", response_model=NotificationResponse)
async def create_notification(request: NotificationRequest):
    """通知作成 API エンドポイント"""
    service = get_notification_service()  # DI コンテナから取得
    return await service.send(request)
 
@app.get("/api/v1/notifications/{batch_id}/status")
async def get_batch_status(batch_id: str):
    """バッチの配信状況を取得"""
    stats = await get_batch_statistics(batch_id)
    return {
        "batch_id": batch_id,
        "total": stats["total"],
        "delivered": stats["delivered"],
        "failed": stats["failed"],
        "pending": stats["pending"],
        "delivery_rate": stats["delivered"] / max(stats["total"], 1),
    }

# コード例 2: チャネルアダプタの統一インターフェース
import asyncio
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class DeliveryResult:
    """配信結果を統一的に表現"""
    success: bool
    provider_message_id: Optional[str] = None
    error_code: Optional[str] = None
    error_message: Optional[str] = None
    should_retry: bool = False
 
class ChannelAdapter(ABC):
    """
    全チャネルの共通インターフェース (Strategy パターン)。
 
    新しいチャネルを追加する場合は、このクラスを継承して
    send() メソッドを実装するだけでよい。
    Open/Closed Principle (OCP) に従った設計。
    """
 
    @abstractmethod
    async def send(self, user_id: str, content: dict,
                   metadata: dict) -> DeliveryResult:
        pass
 
    @abstractmethod
    async def validate_target(self, user_id: str) -> bool:
        """配信先が有効か確認 (トークン有効性等)"""
        pass
 
class APNsAdapter(ChannelAdapter):
    """Apple Push Notification service アダプタ"""
 
    def __init__(self, team_id: str, key_id: str, private_key: str,
                 bundle_id: str, use_sandbox: bool = False):
        self.team_id = team_id
        self.key_id = key_id
        self.private_key = private_key
        self.bundle_id = bundle_id
        self.base_url = (
            "https://api.sandbox.push.apple.com"
            if use_sandbox else
            "https://api.push.apple.com"
        )
        self._jwt_token = None
        self._jwt_expiry = 0
 
    async def send(self, user_id: str, content: dict,
                   metadata: dict) -> DeliveryResult:
        devices = await get_user_devices(user_id, platform="ios")
        if not devices:
            return DeliveryResult(success=False, error_code="NO_DEVICE")
 
        results = []
        for device in devices:
            result = await self._send_to_device(device, content, metadata)
            results.append(result)
 
            # 無効トークンの場合は即座に無効化
            if result.error_code == "INVALID_TOKEN":
                await invalidate_device_token(device["token_id"])
 
        # 1台でも成功すれば OK
        return DeliveryResult(
            success=any(r.success for r in results),
            should_retry=all(r.should_retry for r in results),
        )
 
    async def _send_to_device(self, device: dict, content: dict,
                                metadata: dict) -> DeliveryResult:
        token = await self._get_jwt_token()
        payload = {
            "aps": {
                "alert": {
                    "title": content.get("title", ""),
                    "body": content.get("body", ""),
                    "subtitle": content.get("subtitle", ""),
                },
                "badge": metadata.get("badge", 1),
                "sound": metadata.get("sound", "default"),
                "category": metadata.get("action_category", ""),
                "mutable-content": 1,  # Notification Service Extension
                "thread-id": metadata.get("thread_id", ""),
            },
            "custom_data": content.get("data", {}),
        }
 
        # HTTP/2 で APNs に送信
        headers = {
            "authorization": f"bearer {token}",
            "apns-topic": self.bundle_id,
            "apns-priority": "10" if metadata.get("priority") == "high" else "5",
            "apns-push-type": "alert",
            "apns-expiration": str(int(time.time()) + 86400),
        }
 
        try:
            response = await self.http_client.post(
                f"{self.base_url}/3/device/{device['device_token']}",
                json=payload,
                headers=headers,
                timeout=10,
            )
 
            if response.status_code == 200:
                return DeliveryResult(
                    success=True,
                    provider_message_id=response.headers.get("apns-id"),
                )
            elif response.status_code == 410:
                return DeliveryResult(
                    success=False,
                    error_code="INVALID_TOKEN",
                    error_message="Device token is no longer active",
                    should_retry=False,
                )
            elif response.status_code in (500, 503):
                return DeliveryResult(
                    success=False,
                    error_code="PROVIDER_ERROR",
                    error_message=f"APNs returned {response.status_code}",
                    should_retry=True,
                )
            else:
                body = response.json()
                return DeliveryResult(
                    success=False,
                    error_code=body.get("reason", "UNKNOWN"),
                    error_message=str(body),
                    should_retry=False,
                )
        except asyncio.TimeoutError:
            return DeliveryResult(
                success=False,
                error_code="TIMEOUT",
                should_retry=True,
            )
 
    async def validate_target(self, user_id: str) -> bool:
        devices = await get_user_devices(user_id, platform="ios")
        return len(devices) > 0
 
class FCMAdapter(ChannelAdapter):
    """Firebase Cloud Messaging アダプタ"""
 
    async def send(self, user_id: str, content: dict,
                   metadata: dict) -> DeliveryResult:
        devices = await get_user_devices(user_id, platform="android")
        if not devices:
            return DeliveryResult(success=False, error_code="NO_DEVICE")
 
        for device in devices:
            message = {
                "message": {
                    "token": device["device_token"],
                    "notification": {
                        "title": content.get("title", ""),
                        "body": content.get("body", ""),
                        "image": content.get("image_url"),
                    },
                    "data": {k: str(v) for k, v in content.get("data", {}).items()},
                    "android": {
                        "priority": "high" if metadata.get("priority") == "high" else "normal",
                        "notification": {
                            "channel_id": metadata.get("android_channel", "default"),
                            "click_action": metadata.get("click_action", ""),
                        },
                    },
                }
            }
 
            try:
                response = await self.http_client.post(
                    f"https://fcm.googleapis.com/v1/projects/{self.project_id}/messages:send",
                    json=message,
                    headers={"authorization": f"Bearer {await self._get_access_token()}"},
                    timeout=10,
                )
 
                if response.status_code == 200:
                    result = response.json()
                    return DeliveryResult(
                        success=True,
                        provider_message_id=result.get("name"),
                    )
                elif response.status_code == 404:
                    await invalidate_device_token(device["token_id"])
                    return DeliveryResult(
                        success=False,
                        error_code="INVALID_TOKEN",
                        should_retry=False,
                    )
                else:
                    return DeliveryResult(
                        success=False,
                        error_code=f"FCM_{response.status_code}",
                        should_retry=response.status_code >= 500,
                    )
            except asyncio.TimeoutError:
                return DeliveryResult(
                    success=False, error_code="TIMEOUT", should_retry=True
                )
 
        return DeliveryResult(success=False, error_code="ALL_FAILED")
 
    async def validate_target(self, user_id: str) -> bool:
        devices = await get_user_devices(user_id, platform="android")
        return len(devices) > 0
 
class EmailAdapter(ChannelAdapter):
    """メール配信アダプタ (SES / SendGrid 対応)"""
 
    async def send(self, user_id: str, content: dict,
                   metadata: dict) -> DeliveryResult:
        email = await get_user_email(user_id)
        if not email:
            return DeliveryResult(success=False, error_code="NO_EMAIL")
 
        ses_message = {
            "Source": metadata.get("from_email", "noreply@example.com"),
            "Destination": {"ToAddresses": [email]},
            "Message": {
                "Subject": {"Data": content.get("subject", "")},
                "Body": {
                    "Html": {"Data": content.get("html_body", "")},
                    "Text": {"Data": content.get("text_body", content.get("body", ""))},
                },
            },
            "Tags": [
                {"Name": "category", "Value": metadata.get("category", "general")},
                {"Name": "batch_id", "Value": metadata.get("batch_id", "")},
            ],
        }
 
        try:
            response = await self.ses_client.send_email(**ses_message)
            return DeliveryResult(
                success=True,
                provider_message_id=response["MessageId"],
            )
        except self.ses_client.exceptions.MessageRejected as e:
            return DeliveryResult(
                success=False,
                error_code="REJECTED",
                error_message=str(e),
                should_retry=False,
            )
        except Exception as e:
            return DeliveryResult(
                success=False,
                error_code="SES_ERROR",
                error_message=str(e),
                should_retry=True,
            )
 
    async def validate_target(self, user_id: str) -> bool:
        email = await get_user_email(user_id)
        return email is not None and await is_email_valid(email)
 
class SMSAdapter(ChannelAdapter):
    """SMS 配信アダプタ (Twilio 対応)"""
 
    async def send(self, user_id: str, content: dict,
                   metadata: dict) -> DeliveryResult:
        phone = await get_user_phone(user_id)
        if not phone:
            return DeliveryResult(success=False, error_code="NO_PHONE")
 
        try:
            message = await self.twilio_client.messages.create_async(
                body=content.get("body", ""),
                from_=self.from_number,
                to=phone,
                status_callback=f"{self.callback_base_url}/sms/status",
            )
            return DeliveryResult(
                success=True,
                provider_message_id=message.sid,
            )
        except Exception as e:
            return DeliveryResult(
                success=False,
                error_code="TWILIO_ERROR",
                error_message=str(e),
                should_retry="temporarily" in str(e).lower(),
            )
 
    async def validate_target(self, user_id: str) -> bool:
        phone = await get_user_phone(user_id)
        return phone is not None

# コード例 3: 通知ワーカーのリトライ戦略
import asyncio
import time
import json
from dataclasses import dataclass
 
@dataclass
class RetryConfig:
    """チャネル別リトライ設定"""
    max_retries: int
    base_delay: float       # 秒
    max_delay: float        # 秒
    backoff_factor: float   # 指数バックオフ係数
 
# チャネルごとの最適なリトライ設定
RETRY_CONFIGS = {
    "push": RetryConfig(max_retries=3, base_delay=1.0, max_delay=60.0, backoff_factor=2.0),
    "email": RetryConfig(max_retries=5, base_delay=5.0, max_delay=300.0, backoff_factor=3.0),
    "sms": RetryConfig(max_retries=2, base_delay=10.0, max_delay=60.0, backoff_factor=2.0),
    "in_app": RetryConfig(max_retries=3, base_delay=1.0, max_delay=30.0, backoff_factor=2.0),
}
 
class NotificationWorker:
    """
    Kafka からメッセージを消費して通知を配信するワーカー。
 
    リトライ戦略:
    - 指数バックオフ + ジッター
    - チャネルごとに最大リトライ回数を設定
    - 最大リトライ超過時は DLQ (Dead Letter Queue) に移動
    - DLQ のメッセージはアラート + 手動対応
    """
 
    def __init__(self, channel: str, adapter: ChannelAdapter,
                 consumer, dlq_producer, metrics):
        self.channel = channel
        self.adapter = adapter
        self.consumer = consumer
        self.dlq_producer = dlq_producer
        self.metrics = metrics
        self.retry_config = RETRY_CONFIGS.get(
            channel, RetryConfig(3, 1.0, 60.0, 2.0)
        )
 
    async def run(self):
        """メインループ: Kafka からメッセージを消費し続ける"""
        async for message in self.consumer:
            try:
                await self._process_message(json.loads(message.value))
                await self.consumer.commit()
            except Exception as e:
                logger.error(f"Unexpected error: {e}", exc_info=True)
                self.metrics.increment("worker.unexpected_error")
 
    async def _process_message(self, message: dict):
        """メッセージの処理とリトライ"""
        notification_id = message["id"]
        retry_count = message.get("retry_count", 0)
        start_time = time.time()
 
        # 配信実行
        result = await self.adapter.send(
            user_id=message["user_id"],
            content=message["content"],
            metadata={
                "priority": message.get("priority"),
                "batch_id": message.get("batch_id"),
                "category": message.get("category"),
            },
        )
 
        # メトリクス記録
        latency = time.time() - start_time
        self.metrics.histogram(
            "notification.delivery_latency",
            latency,
            tags={"channel": self.channel, "success": str(result.success)},
        )
 
        if result.success:
            # 配信成功
            await self._record_delivery(notification_id, result)
            self.metrics.increment(
                "notification.delivered",
                tags={"channel": self.channel},
            )
        elif result.should_retry and retry_count < self.retry_config.max_retries:
            # リトライ
            delay = self._calculate_delay(retry_count)
            message["retry_count"] = retry_count + 1
            await self._schedule_retry(message, delay)
            self.metrics.increment(
                "notification.retry",
                tags={"channel": self.channel, "attempt": str(retry_count + 1)},
            )
        else:
            # DLQ に移動
            await self._send_to_dlq(message, result)
            self.metrics.increment(
                "notification.dead_lettered",
                tags={"channel": self.channel},
            )
 
    def _calculate_delay(self, retry_count: int) -> float:
        """指数バックオフ + ジッターでリトライ間隔を計算"""
        import random
        delay = min(
            self.retry_config.base_delay * (self.retry_config.backoff_factor ** retry_count),
            self.retry_config.max_delay,
        )
        # ジッター: 0.5x ~ 1.5x のランダム変動
        jitter = delay * (0.5 + random.random())
        return jitter
 
    async def _schedule_retry(self, message: dict, delay: float):
        """リトライキューにスケジュール投入"""
        await asyncio.sleep(delay)  # 簡易実装。本番では遅延キューを使用
        topic = f"notifications.{self.channel}.retry"
        await self.dlq_producer.send(topic, json.dumps(message).encode())
 
    async def _send_to_dlq(self, message: dict, result: DeliveryResult):
        """Dead Letter Queue に移動"""
        dlq_message = {
            **message,
            "dlq_reason": result.error_code,
            "dlq_error": result.error_message,
            "dlq_timestamp": int(time.time()),
        }
        topic = f"notifications.{self.channel}.dlq"
        await self.dlq_producer.send(topic, json.dumps(dlq_message).encode())
 
        # アラート発行
        if message.get("priority") in ("critical", "high"):
            await send_alert(
                f"High priority notification failed: {message['id']}",
                severity="warning",
            )
 
    async def _record_delivery(self, notification_id: str,
                                result: DeliveryResult):
        """配信結果をDBに記録"""
        await update_notification_status(
            notification_id=notification_id,
            status="delivered",
            provider_message_id=result.provider_message_id,
            delivered_at=int(time.time()),
        )

# コード例 4: スライディングウィンドウによるレート制限
import time
import redis.asyncio as redis
 
class NotificationRateLimiter:
    """
    ユーザーあたり・チャネルあたりの通知レート制限。
 
    WHY: レート制限が必要な理由
    1. ユーザー体験の保護: 過剰な通知はアプリのアンインストールを招く
    2. プロバイダの制限遵守: APNs/FCM にもレート制限がある
    3. コスト管理: SMS は1通あたり課金されるため
    4. 法的遵守: CAN-SPAM 法、特定電子メール法等
 
    アルゴリズム: スライディングウィンドウ (Redis Sorted Set)
    - Fixed Window の「境界問題」を回避
    - O(log N) の計算量で正確なカウント
    - Redis のアトミック操作でレースコンディション防止
    """
 
    # デフォルトのレート制限設定
    DEFAULT_LIMITS = {
        "push":     {"per_hour": 10, "per_day": 50},
        "email":    {"per_hour": 3,  "per_day": 10},
        "sms":      {"per_hour": 2,  "per_day": 5},
        "in_app":   {"per_hour": 20, "per_day": 100},
        "web_push": {"per_hour": 8,  "per_day": 40},
    }
 
    # 優先度別の制限倍率
    PRIORITY_MULTIPLIERS = {
        "critical": float('inf'),  # 制限なし
        "high": 5.0,
        "normal": 1.0,
        "low": 0.5,
    }
 
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
 
    async def allow(self, user_id: str, channel: str,
                    priority: str = "normal") -> bool:
        """通知を送信してよいか判定する"""
        # Critical は常に許可
        if priority == "critical":
            return True
 
        # ユーザーカスタム制限があれば適用
        custom_limits = await self._get_custom_limits(user_id, channel)
        limits = custom_limits or self.DEFAULT_LIMITS.get(
            channel, {"per_hour": 10, "per_day": 50}
        )
 
        # 優先度による倍率適用
        multiplier = self.PRIORITY_MULTIPLIERS.get(priority, 1.0)
        effective_hour = int(limits["per_hour"] * multiplier)
        effective_day = int(limits["per_day"] * multiplier)
 
        now = time.time()
 
        # Lua スクリプトでアトミックにチェック＆記録
        result = await self._check_and_record(
            user_id, channel, now, effective_hour, effective_day
        )
        return result
 
    LUA_RATE_CHECK = """
    local hour_key = KEYS[1]
    local day_key = KEYS[2]
    local now = tonumber(ARGV[1])
    local hour_limit = tonumber(ARGV[2])
    local day_limit = tonumber(ARGV[3])
 
    -- 古いエントリを削除
    redis.call('ZREMRANGEBYSCORE', hour_key, 0, now - 3600)
    redis.call('ZREMRANGEBYSCORE', day_key, 0, now - 86400)
 
    -- 現在のカウントを確認
    local hour_count = redis.call('ZCARD', hour_key)
    local day_count = redis.call('ZCARD', day_key)
 
    if hour_count >= hour_limit or day_count >= day_limit then
        return 0  -- 拒否
    end
 
    -- カウントを記録
    local member = now .. ':' .. math.random(1000000)
    redis.call('ZADD', hour_key, now, member)
    redis.call('ZADD', day_key, now, member)
    redis.call('EXPIRE', hour_key, 3600)
    redis.call('EXPIRE', day_key, 86400)
 
    return 1  -- 許可
    """
 
    async def _check_and_record(self, user_id: str, channel: str,
                                 now: float, hour_limit: int,
                                 day_limit: int) -> bool:
        """Lua スクリプトでアトミックにレート制限チェック"""
        hour_key = f"ratelimit:{user_id}:{channel}:hour"
        day_key = f"ratelimit:{user_id}:{channel}:day"
 
        result = await self.redis.eval(
            self.LUA_RATE_CHECK,
            2,  # KEYS の数
            hour_key, day_key,
            now, hour_limit, day_limit,
        )
        return bool(result)
 
    async def _get_custom_limits(self, user_id: str,
                                  channel: str) -> dict | None:
        """ユーザーカスタムのレート制限設定を取得"""
        key = f"user_prefs:{user_id}"
        data = await self.redis.hget(key, "frequency_limit")
        if data:
            import json
            limits = json.loads(data)
            return limits.get(channel)
        return None
 
    async def get_remaining(self, user_id: str, channel: str) -> dict:
        """残りのクォータを取得（API レスポンス用）"""
        limits = self.DEFAULT_LIMITS.get(channel, {"per_hour": 10, "per_day": 50})
        now = time.time()
 
        hour_key = f"ratelimit:{user_id}:{channel}:hour"
        day_key = f"ratelimit:{user_id}:{channel}:day"
 
        pipe = self.redis.pipeline()
        pipe.zrangebyscore(hour_key, now - 3600, now)
        pipe.zrangebyscore(day_key, now - 86400, now)
        hour_entries, day_entries = await pipe.execute()
 
        return {
            "hour": {
                "limit": limits["per_hour"],
                "remaining": max(0, limits["per_hour"] - len(hour_entries)),
                "reset": int(now) + 3600,
            },
            "day": {
                "limit": limits["per_day"],
                "remaining": max(0, limits["per_day"] - len(day_entries)),
                "reset": int(now) + 86400,
            },
        }

# コード例 5: べき等性キーによる重複通知の防止
import hashlib
import json
 
class DeduplicationService:
    """
    同一通知の重複送信を防止するサービス。
 
    WHY: なぜ重複排除が必要か
 
    1. イベント駆動アーキテクチャでの at-least-once 配信
       → 同じイベントが複数回処理される可能性がある
    2. Kafka コンシューマーのリバランス
       → オフセットコミット前にリバランスが発生すると再処理
    3. ユーザーの重複操作
       → 「送信」ボタンの二重クリック
    4. マイクロサービス間のリトライ
       → HTTP タイムアウトでリトライ → 実際は成功していた
 
    実装方式の比較:
    +------------------+----------+---------+--------+
    | 方式              | メモリ   | 精度    | 速度   |
    +------------------+----------+---------+--------+
    | Redis SETNX      | 中       | 高      | 高     | ← 採用
    | Bloom Filter      | 低       | 近似    | 最高   |
    | DB Unique Key    | 高       | 最高    | 低     |
    +------------------+----------+---------+--------+
    """
 
    def __init__(self, redis_client, ttl: int = 86400):
        self.redis = redis_client
        self.ttl = ttl  # デフォルト24時間
 
    async def is_duplicate(self, key: str) -> bool:
        """このキーが既に処理済みかどうかを確認する"""
        dedup_key = f"dedup:{self._hash_key(key)}"
        result = await self.redis.set(dedup_key, "1", nx=True, ex=self.ttl)
        return result is None  # None = 既に存在 = 重複
 
    async def is_duplicate_content(self, user_id: str, channel: str,
                                    content_hash: str,
                                    window_seconds: int = 300) -> bool:
        """
        同一内容の通知が短時間に送信されていないか確認。
 
        例: 同じ「注文完了」通知が5分以内に2回送られるのを防ぐ
        """
        key = f"dedup:content:{user_id}:{channel}:{content_hash}"
        result = await self.redis.set(key, "1", nx=True, ex=window_seconds)
        return result is None
 
    @staticmethod
    def compute_content_hash(content: dict) -> str:
        """通知内容のハッシュを計算"""
        serialized = json.dumps(content, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(serialized.encode()).hexdigest()[:16]
 
    async def mark_sent(self, notification_id: str, channel: str):
        """通知の送信完了を記録する（配信追跡用）"""
        key = f"sent:{notification_id}:{channel}"
        await self.redis.set(key, "1", ex=self.ttl)
 
    async def was_sent(self, notification_id: str, channel: str) -> bool:
        """通知が既に送信済みか確認（リトライ時の冪等性チェック用）"""
        key = f"sent:{notification_id}:{channel}"
        return await self.redis.exists(key) > 0
 
    @staticmethod
    def _hash_key(key: str) -> str:
        """長いキーをハッシュ化してRedisメモリを節約"""
        if len(key) > 64:
            return hashlib.sha256(key.encode()).hexdigest()
        return key

# コード例 6: 多言語・多チャネル対応テンプレートサービス
from jinja2 import Environment, BaseLoader, TemplateSyntaxError
from typing import Optional
import json
 
class NotificationTemplateService:
    """
    通知テンプレートの管理と展開を行うサービス。
 
    設計ポイント:
    1. テンプレートはDBで管理し、キャッシュで高速化
    2. チャネルごとに異なるテンプレートを用意
    3. 多言語対応 (locale ベースのフォールバック)
    4. テンプレートのバージョン管理 (ロールバック可能)
    5. A/B テスト用のバリアント管理
    """
 
    def __init__(self, db, cache, metrics):
        self.db = db
        self.cache = cache
        self.metrics = metrics
        self.env = Environment(loader=BaseLoader())
 
    async def render(self, template_id: str, channel: str,
                      data: dict, locale: str = "ja",
                      variant: Optional[str] = None) -> dict:
        """テンプレートを展開して配信コンテンツを生成"""
 
        # 1. キャッシュからテンプレートを取得
        cache_key = f"template:{template_id}:{channel}:{locale}"
        if variant:
            cache_key += f":{variant}"
 
        template_data = await self.cache.get(cache_key)
 
        if not template_data:
            # 2. DB から取得（フォールバック付き）
            template_data = await self._load_template(
                template_id, channel, locale, variant
            )
            if template_data:
                await self.cache.set(cache_key, json.dumps(template_data), ex=3600)
            else:
                self.metrics.increment("template.not_found")
                raise TemplateNotFoundError(
                    f"Template not found: {template_id}/{channel}/{locale}"
                )
        else:
            template_data = json.loads(template_data)
 
        # 3. Jinja2 でレンダリング
        try:
            rendered = {}
            for key, template_str in template_data.items():
                template = self.env.from_string(template_str)
                rendered[key] = template.render(**data)
            return rendered
        except TemplateSyntaxError as e:
            self.metrics.increment("template.render_error")
            raise TemplateRenderError(f"Template render failed: {e}")
 
    async def _load_template(self, template_id: str, channel: str,
                               locale: str, variant: Optional[str]) -> dict:
        """
        DB からテンプレートを取得。フォールバックチェーン:
        1. 指定 locale + variant
        2. 指定 locale (variant なし)
        3. デフォルト locale (en)
        """
        for try_locale in [locale, "en"]:
            result = await self.db.fetch_one(
                """
                SELECT title, subject, body, metadata
                FROM notification_templates
                WHERE template_id = :template_id
                  AND channel = :channel
                  AND locale = :locale
                  AND is_active = TRUE
                ORDER BY version DESC
                LIMIT 1
                """,
                {
                    "template_id": template_id,
                    "channel": channel,
                    "locale": try_locale,
                },
            )
            if result:
                template = {}
                if result["title"]:
                    template["title"] = result["title"]
                if result["subject"]:
                    template["subject"] = result["subject"]
                if result["body"]:
                    template["body"] = result["body"]
                return template
 
        return None
 
    async def create_template(self, template_id: str, channel: str,
                                locale: str, content: dict,
                                created_by: str) -> int:
        """テンプレートの新バージョンを作成"""
        # 現在の最新バージョンを取得
        current = await self.db.fetch_one(
            "SELECT MAX(version) as ver FROM notification_templates "
            "WHERE template_id = :tid AND channel = :ch AND locale = :lo",
            {"tid": template_id, "ch": channel, "lo": locale},
        )
        new_version = (current["ver"] or 0) + 1
 
        await self.db.execute(
            """
            INSERT INTO notification_templates
            (template_id, version, channel, locale, title, subject, body, metadata)
            VALUES (:tid, :ver, :ch, :lo, :title, :subject, :body, :meta)
            """,
            {
                "tid": template_id,
                "ver": new_version,
                "ch": channel,
                "lo": locale,
                "title": content.get("title"),
                "subject": content.get("subject"),
                "body": content.get("body"),
                "meta": json.dumps({"created_by": created_by}),
            },
        )
 
        # キャッシュ無効化
        cache_key = f"template:{template_id}:{channel}:{locale}"
        await self.cache.delete(cache_key)
 
        return new_version
 
class TemplateNotFoundError(Exception):
    pass
 
class TemplateRenderError(Exception):
    pass

# コード例 7: 同種通知のインテリジェント集約
import asyncio
import json
from collections import defaultdict
 
class NotificationAggregator:
    """
    同種の通知をバッチ集約するサービス。
 
    例: 「Aがいいねしました」「Bがいいねしました」「Cがいいねしました」
    → 「A、B、他1人がいいねしました」
 
    設計:
    - Redis List をバッファとして使用
    - 一定時間(5分)または一定件数(10件)でフラッシュ
    - ユーザー × カテゴリ単位で集約
    """
 
    AGGREGATION_WINDOW = 300  # 5分
    MAX_BATCH_SIZE = 10       # 最大集約数
 
    def __init__(self, redis_client, notification_service):
        self.redis = redis_client
        self.notification_service = notification_service
 
    async def add(self, user_id: str, category: str,
                  notification: dict) -> bool:
        """
        通知をバッファに追加。閾値に達したらフラッシュ。
 
        Returns:
            True: バッファに追加（後でまとめて送信）
            False: 集約対象外（即時送信）
        """
        # 集約対象のカテゴリか確認
        if category not in self.AGGREGATABLE_CATEGORIES:
            return False
 
        key = f"batch:{user_id}:{category}"
 
        pipe = self.redis.pipeline()
        pipe.rpush(key, json.dumps(notification))
        pipe.expire(key, self.AGGREGATION_WINDOW)
        pipe.llen(key)
        results = await pipe.execute()
 
        current_count = results[2]
 
        # 最大集約数に達したら即フラッシュ
        if current_count >= self.MAX_BATCH_SIZE:
            await self.flush(user_id, category)
 
        return True
 
    async def flush(self, user_id: str, category: str):
        """バッファ内の通知をまとめて送信"""
        key = f"batch:{user_id}:{category}"
 
        # アトミックに取得＆削除
        pipe = self.redis.pipeline()
        pipe.lrange(key, 0, -1)
        pipe.delete(key)
        results = await pipe.execute()
 
        raw_notifications = results[0]
        if not raw_notifications:
            return
 
        notifications = [json.loads(n) for n in raw_notifications]
 
        # 集約メッセージの生成
        aggregated_content = self._aggregate_content(category, notifications)
 
        # 集約通知を送信
        await self.notification_service.send_aggregated(
            user_id=user_id,
            content=aggregated_content,
            original_count=len(notifications),
        )
 
    def _aggregate_content(self, category: str,
                            notifications: list) -> dict:
        """通知内容を集約してサマリーを生成"""
        if category == "like":
            actors = [n["data"]["actor_name"] for n in notifications]
            if len(actors) <= 3:
                names = "、".join(actors)
            else:
                names = f"{actors[0]}、{actors[1]}、他{len(actors)-2}人"
            return {
                "title": "いいね通知",
                "body": f"{names}があなたの投稿にいいねしました",
                "data": {
                    "type": "like_aggregated",
                    "count": len(notifications),
                    "actor_ids": [n["data"]["actor_id"] for n in notifications],
                },
            }
        elif category == "follow":
            actors = [n["data"]["actor_name"] for n in notifications]
            if len(actors) <= 3:
                names = "、".join(actors)
            else:
                names = f"{actors[0]}、{actors[1]}、他{len(actors)-2}人"
            return {
                "title": "フォロー通知",
                "body": f"{names}があなたをフォローしました",
                "data": {
                    "type": "follow_aggregated",
                    "count": len(notifications),
                },
            }
        else:
            return {
                "title": f"{len(notifications)}件の通知",
                "body": f"{category}に関する{len(notifications)}件の新着通知があります",
            }
 
    AGGREGATABLE_CATEGORIES = {"like", "follow", "comment", "mention"}
 
    async def flush_all_expired(self):
        """
        定期実行: 期限切れバッファの一括フラッシュ。
        Cron ジョブで1分ごとに実行する。
        """
        cursor = 0
        while True:
            cursor, keys = await self.redis.scan(
                cursor, match="batch:*", count=1000
            )
            for key in keys:
                ttl = await self.redis.ttl(key)
                if ttl < 0 or ttl < self.AGGREGATION_WINDOW * 0.1:
                    # TTL が残り少ないバッファをフラッシュ
                    parts = key.decode().split(":")
                    if len(parts) == 3:
                        await self.flush(parts[1], parts[2])
            if cursor == 0:
                break

# コード例 8: 配信イベント追跡パイプライン
from datetime import datetime
 
class DeliveryTracker:
    """
    通知の配信ライフサイクルを追跡する。
 
    イベントフロー:
    SENT → DELIVERED → OPENED → CLICKED
                    → BOUNCED (バウンス)
                    → COMPLAINED (苦情)
    """
 
    def __init__(self, event_store, analytics_producer):
        self.event_store = event_store
        self.analytics_producer = analytics_producer
 
    async def track_event(self, notification_id: str, event_type: str,
                           metadata: dict = None):
        """配信イベントを記録"""
        event = {
            "notification_id": notification_id,
            "event_type": event_type,  # sent, delivered, opened, clicked, ...
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": metadata or {},
        }
 
        # 1. イベントストアに記録 (永続化)
        await self.event_store.append(event)
 
        # 2. 分析用 Kafka トピックに送信
        await self.analytics_producer.send(
            topic="notification-events",
            value=event,
        )
 
        # 3. リアルタイムメトリクス更新
        metrics.increment(
            f"notification.{event_type}",
            tags={"channel": metadata.get("channel", "unknown")},
        )
 
    async def track_email_webhook(self, webhook_data: dict):
        """
        メールプロバイダ (SES/SendGrid) からの Webhook を処理。
 
        SES の場合:
        - Delivery: 配信成功
        - Bounce: バウンス (無効アドレス)
        - Complaint: 苦情 (スパム報告)
        - Open: 開封 (トラッキングピクセル)
        - Click: クリック (リンクリダイレクト)
        """
        event_type_map = {
            "Delivery": "delivered",
            "Bounce": "bounced",
            "Complaint": "complained",
            "Open": "opened",
            "Click": "clicked",
        }
 
        ses_event = webhook_data.get("eventType", "")
        mapped_type = event_type_map.get(ses_event)
 
        if mapped_type:
            notification_id = webhook_data.get("mail", {}).get(
                "messageId", "unknown"
            )
            await self.track_event(
                notification_id=notification_id,
                event_type=mapped_type,
                metadata={
                    "channel": "email",
                    "provider": "ses",
                    "raw_event": ses_event,
                },
            )
 
            # バウンスの場合はメールアドレスを無効化
            if mapped_type == "bounced":
                bounce_type = webhook_data.get("bounce", {}).get("bounceType")
                if bounce_type == "Permanent":
                    email = webhook_data["mail"]["destination"][0]
                    await disable_email_address(email)
 
    async def get_delivery_stats(self, batch_id: str) -> dict:
        """バッチの配信統計を取得"""
        events = await self.event_store.get_by_batch(batch_id)
 
        stats = {
            "total": 0,
            "sent": 0,
            "delivered": 0,
            "opened": 0,
            "clicked": 0,
            "bounced": 0,
            "complained": 0,
            "failed": 0,
        }
 
        for event in events:
            event_type = event["event_type"]
            if event_type in stats:
                stats[event_type] += 1
            stats["total"] = max(stats["total"], stats["sent"])
 
        # レートの計算
        if stats["sent"] > 0:
            stats["delivery_rate"] = stats["delivered"] / stats["sent"]
            stats["open_rate"] = stats["opened"] / stats["delivered"] if stats["delivered"] > 0 else 0
            stats["click_rate"] = stats["clicked"] / stats["opened"] if stats["opened"] > 0 else 0
 
        return stats

# NG: ユーザーのアクションごとに即座に通知を送る
# 結果: ユーザーが通知をオフにしてアプリをアンインストール
 
class BadNotificationHandler:
    async def on_like(self, post_id: str, liker_id: str):
        # 個別にプッシュ通知を送信
        await send_push(
            user_id=post.author_id,
            title="いいね",
            body=f"{liker.name}さんがいいねしました"
        )
        # 10:00 「Aさんがいいねしました」
        # 10:01 「Bさんがいいねしました」
        # 10:02 「Cさんがいいねしました」
        # ... ユーザーは通知をオフにする
 
 
# OK: インテリジェントなバッチングと集約
class GoodNotificationHandler:
    def __init__(self, aggregator: NotificationAggregator):
        self.aggregator = aggregator
 
    async def on_like(self, post_id: str, liker_id: str):
        # 集約バッファに追加
        await self.aggregator.add(
            user_id=post.author_id,
            category="like",
            notification={
                "data": {
                    "actor_id": liker_id,
                    "actor_name": liker.name,
                    "post_id": post_id,
                },
            },
        )
        # 5分後にまとめて送信:
        # 「A、B、他3人があなたの投稿にいいねしました」
 
    # さらに:
    # - レート制限で1時間あたりの上限を設定
    # - ユーザーの活動時間帯に合わせて配信
    # - 重要度に応じた配信頻度の調整

# NG: 1つのキューとワーカーで全チャネルを処理
class BadWorker:
    async def process(self, message):
        channel = message["channel"]
        if channel == "push":
            await self.send_push(message)      # 高速 (< 100ms)
        elif channel == "email":
            await self.send_email(message)     # 中速 (< 1s)
        elif channel == "sms":
            await self.send_sms(message)       # 低速 (1-5s)
            # SMS の遅延が Push/Email をブロック!
 
        # 問題:
        # 1. SMS プロバイダ障害 → 全チャネルの配信が停止
        # 2. チャネルごとのスケーリングが不可能
        # 3. 優先度の高いPushが低優先度SMSの後に待機
 
 
# OK: チャネルごとに独立したキューとワーカー
class GoodArchitecture:
    """
    チャネル分離アーキテクチャ:
 
    Kafka Topics:
    - notifications.push.critical   → PushWorker (4 instances)
    - notifications.push.normal     → PushWorker (2 instances)
    - notifications.email.normal    → EmailWorker (3 instances)
    - notifications.sms.normal      → SMSWorker (1 instance)
 
    利点:
    1. 障害分離: SMS が落ちても Push/Email は継続
    2. 独立スケーリング: Push は高スループット、SMS は低スループット
    3. 優先度制御: critical トピックに多くのワーカーを割り当て
    4. 監視の粒度: チャネル別にメトリクスを取得
    """
    pass

# NG: レート制限なしで通知を受け付ける
@app.post("/api/v1/notifications")
async def bad_create_notification(request: NotificationRequest):
    # バリデーションのみで即座にキューに投入
    for user_id in request.user_ids:
        for channel in request.channels:
            await queue.publish({"user_id": user_id, "channel": channel, ...})
    return {"status": "queued"}
 
    # 問題:
    # 1. バグのあるサービスが無限ループで通知を送信
    # 2. 全ユーザーに大量通知 → UX 崩壊
    # 3. SMS コストが爆発 (1通 $0.05 × 数百万通 = ...)
    # 4. APNs/FCM のレート制限に抵触 → IP がブロックされる
 
 
# OK: 多層レート制限
@app.post("/api/v1/notifications")
async def good_create_notification(request: NotificationRequest):
    # 1. API レベル: 呼び出し元サービスのレート制限
    caller = get_caller_service(request)
    if not api_rate_limiter.allow(caller.id):
        raise HTTPException(429, "API rate limit exceeded")
 
    # 2. ユーザーレベル: ユーザーあたりのレート制限
    for user_id in request.user_ids:
        for channel in request.channels:
            if not user_rate_limiter.allow(user_id, channel):
                continue  # スキップ (ログに記録)
 
    # 3. グローバルレベル: システム全体の安全弁
    if not global_rate_limiter.allow():
        raise HTTPException(503, "System overloaded")
 
    return {"status": "queued"}

# NG: 送信したら終わり、結果を確認しない
class BadSender:
    async def send_push(self, message):
        try:
            await apns_client.send(message)
            # 成功扱い。実際は APNs に到達しただけで
            # デバイスに届いたかは不明
        except Exception:
            pass  # エラーも無視
 
 
# OK: 配信結果を追跡し、フィードバックループを形成
class GoodSender:
    async def send_push(self, message):
        result = await apns_client.send(message)
 
        # 配信結果を記録
        await delivery_tracker.track_event(
            notification_id=message["id"],
            event_type="sent" if result.success else "failed",
            metadata={"provider_id": result.provider_message_id},
        )
 
        # 無効トークンの処理
        if result.error_code == "INVALID_TOKEN":
            await device_token_service.invalidate(message["device_token"])
 
        # 失敗時のリトライ
        if not result.success and result.should_retry:
            await retry_queue.publish(message)
 
        # メトリクス更新
        metrics.increment("push.sent", tags={"success": str(result.success)})

# 要件:
# 1. GET /api/v1/users/{user_id}/notification-preferences
#    → ユーザーの通知設定を取得
# 2. PUT /api/v1/users/{user_id}/notification-preferences
#    → 通知設定を更新
# 3. チャネル別のオン/オフ設定
# 4. カテゴリ別の設定（marketing, transaction, social）
# 5. Quiet Hours の設定
 
# スケルトンコード:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
 
app = FastAPI()
 
class QuietHours(BaseModel):
    enabled: bool = False
    start: str = "23:00"  # HH:MM
    end: str = "07:00"
    timezone: str = "Asia/Tokyo"
 
class ChannelSettings(BaseModel):
    push: bool = True
    email: bool = True
    sms: bool = True
    in_app: bool = True
 
class CategorySettings(BaseModel):
    marketing: bool = False
    transaction: bool = True
    social: bool = True
    security: bool = True  # セキュリティは常にTrue推奨
 
class NotificationPreferences(BaseModel):
    channels: ChannelSettings = ChannelSettings()
    categories: CategorySettings = CategorySettings()
    quiet_hours: QuietHours = QuietHours()
 
@app.get("/api/v1/users/{user_id}/notification-preferences")
async def get_preferences(user_id: str):
    # TODO: DBからユーザー設定を取得
    # TODO: キャッシュの活用
    # TODO: デフォルト値の設定
    pass
 
@app.put("/api/v1/users/{user_id}/notification-preferences")
async def update_preferences(user_id: str, prefs: NotificationPreferences):
    # TODO: バリデーション
    # TODO: DBに保存
    # TODO: キャッシュの無効化
    # TODO: 変更イベントの発行
    pass

GET /api/v1/users/user-123/notification-preferences
 
{
  "channels": {
    "push": true,
    "email": true,
    "sms": false,
    "in_app": true
  },
  "categories": {
    "marketing": false,
    "transaction": true,
    "social": true,
    "security": true
  },
  "quiet_hours": {
    "enabled": true,
    "start": "23:00",
    "end": "07:00",
    "timezone": "Asia/Tokyo"
  }
}

# 要件:
# 1. Push → Email → SMS の順にフォールバック
# 2. 各チャネルで配信後、一定時間以内に開封確認がなければ次へ
# 3. タイムアウト設定はチャネルごとにカスタマイズ可能
# 4. 最終手段 (SMS) が失敗したらアラート
 
# スケルトンコード:
class FallbackDeliveryService:
    """
    マルチチャネルフォールバック配信サービス。
 
    フロー:
    1. Push 送信 → 30秒待ち → 開封確認あり → 完了
                                開封確認なし → 2へ
    2. Email 送信 → 5分待ち → 開封確認あり → 完了
                               開封確認なし → 3へ
    3. SMS 送信 → 完了 (SMS は開封確認なし)
    """
 
    FALLBACK_CHAIN = [
        {"channel": "push", "timeout_seconds": 30},
        {"channel": "email", "timeout_seconds": 300},
        {"channel": "sms", "timeout_seconds": 0},  # 最終手段
    ]
 
    async def deliver_with_fallback(self, user_id: str,
                                     notification: dict) -> str:
        # TODO: FALLBACK_CHAIN に沿って順番に配信
        # TODO: 各チャネルでタイムアウト後に次のチャネルへ
        # TODO: 全チャネル失敗時のアラート
        # TODO: 配信結果のログ記録
        pass

# 要件:
# 1. チャネル別の配信成功率（日次/週次/月次）
# 2. カテゴリ別の開封率・クリック率
# 3. 時間帯別の配信パフォーマンス分析
# 4. A/B テスト結果の統計分析
# 5. ユーザーセグメント別のエンゲージメント分析
# 6. データは ClickHouse に格納されていると仮定
 
# スケルトンコード:
from datetime import date, timedelta
 
class NotificationAnalytics:
    """通知分析サービス"""
 
    async def get_channel_performance(self, start_date: date,
                                       end_date: date) -> dict:
        """
        チャネル別の配信パフォーマンスを取得。
 
        返り値の例:
        {
          "push": {"sent": 5000000, "delivered": 4750000,
                   "opened": 2375000, "delivery_rate": 0.95,
                   "open_rate": 0.50},
          "email": {"sent": 3000000, "delivered": 2400000,
                    "opened": 720000, "delivery_rate": 0.80,
                    "open_rate": 0.30},
          ...
        }
        """
        # TODO: ClickHouse からデータを集計
        # TODO: 日次/週次/月次の粒度で返す
        # TODO: 前期間との比較
        pass
 
    async def get_ab_test_results(self, test_id: str) -> dict:
        """
        A/B テスト結果の統計分析。
 
        返り値の例:
        {
          "test_id": "test-001",
          "variants": {
            "A": {"sent": 50000, "opened": 15000, "clicked": 3000,
                  "open_rate": 0.30, "click_rate": 0.20},
            "B": {"sent": 50000, "opened": 18000, "clicked": 4500,
                  "open_rate": 0.36, "click_rate": 0.25},
          },
          "winner": "B",
          "confidence": 0.97,
          "p_value": 0.003,
        }
        """
        # TODO: 統計検定 (カイ二乗検定 or Z検定) の実装
        # TODO: 統計的有意性の判定
        pass
 
    async def get_hourly_heatmap(self, channel: str,
                                  days: int = 7) -> list:
        """
        時間帯別の配信パフォーマンスヒートマップデータ。
 
        返り値: 24時間 × 7日 のマトリクス
        """
        # TODO: 時間帯別の開封率を計算
        # TODO: 最適送信時間帯の推定
        pass

GET /api/v1/analytics/channel-performance?start=2024-01-01&end=2024-01-31
 
{
  "period": {"start": "2024-01-01", "end": "2024-01-31"},
  "channels": {
    "push": {
      "sent": 155000000,
      "delivered": 147250000,
      "opened": 73625000,
      "clicked": 22087500,
      "delivery_rate": 0.95,
      "open_rate": 0.50,
      "click_rate": 0.15,
      "trend": "+2.3%"
    },
    "email": {
      "sent": 93000000,
      "delivered": 74400000,
      "opened": 22320000,
      "clicked": 6696000,
      "delivery_rate": 0.80,
      "open_rate": 0.30,
      "click_rate": 0.09,
      "trend": "-1.1%"
    }
  }
}

# トークン管理の実装例
async def handle_apns_response(device_token: str, status_code: int,
                                response_body: dict):
    if status_code == 410:  # Gone
        await db.execute(
            "UPDATE device_tokens SET is_active = FALSE "
            "WHERE device_token = :token",
            {"token": device_token},
        )
        # キャッシュも無効化
        user_id = await get_user_by_token(device_token)
        await cache.delete(f"devices:{user_id}")
    elif status_code == 400 and response_body.get("reason") == "BadDeviceToken":
        # 不正なトークン形式
        await db.execute(
            "DELETE FROM device_tokens WHERE device_token = :token",
            {"token": device_token},
        )