cmd/notification-service/
├── main.go                  # 应用入口，初始化 Tracer、加载配置、启动服务
├── config.go                # 配置结构体定义
├── wire.go                  # Wire 依赖注入配置
├── wire_gen.go              # Wire 生成的依赖注入代码
├── internal/
│   ├── domain/              # 领域层（Domain Layer）
│   │   ├── notification.go  # 通知领域模型、仓储接口、值对象
│   │   └── errors.go        # 领域错误定义
│   ├── data/                # 数据访问层（Data Layer）
│   │   ├── data.go          # Data 容器，管理数据库连接
│   │   ├── db.go            # 数据库初始化（PostgreSQL + Redis）
│   │   ├── notification_repo.go  # 通知仓储实现（CRUD）
│   │   └── template_repo.go     # 模板仓储实现
│   ├── biz/                 # 业务逻辑层（Business Layer）
│   │   ├── notification_usecase.go  # 通知用例：发送、重试、状态管理
│   │   └── template_usecase.go      # 模板用例：渲染、管理
│   ├── service/             # 服务层（Service Layer）
│   │   └── notification_service.go  # gRPC/HTTP 服务适配器
│   ├── server/              # 服务器层（Server Layer）
│   │   ├── http.go          # HTTP 服务器（Kratos HTTP）
│   │   └── grpc.go          # gRPC 服务器（Kratos gRPC）
│   ├── infra/               # 基础设施层（Infrastructure Layer）
│   │   ├── providers.go     # 邮件、短信提供者接口
│   │   ├── email/
│   │   │   └── smtp_provider.go  # SMTP 邮件发送实现
│   │   ├── sms/
│   │   │   └── aliyun_provider.go  # 阿里云短信实现
│   │   ├── senders/
│   │   │   ├── email_sender.go    # 邮件发送器
│   │   │   ├── inapp_sender.go    # 站内信发送器
│   │   │   └── webhook_sender.go  # WebHook 发送器
│   │   ├── websocket/
│   │   │   └── manager.go         # WebSocket 连接管理器
│   │   └── kafka/
│   │       └── consumer.go        # Kafka 事件消费器
│   └── scheduler/           # 调度器层（Scheduler Layer）
│       └── task_queue.go    # 基于 Redis 的定时任务队列
└── configs/
    └── notification-service.yaml  # 服务配置文件

flowchart TB
    subgraph EventSource["事件源层"]
        Kafka["Kafka Event Bus<br/>事件总线<br/>监听：document.indexed<br/>task.completed<br/>user.registered"]
        HTTPClient["HTTP Client<br/>外部系统主动调用"]
        GRPCClient["gRPC Client<br/>内部服务调用"]
    end

    subgraph APILayer["接入层 API Gateway"]
        HTTPRouter["HTTP Router<br/>Gin Framework<br/>端口: 9005"]
        GRPCServer["gRPC Server<br/>端口: 9505"]
        WSUpgrade["WebSocket Upgrader<br/>实时连接升级"]
    end

    subgraph ServiceLayer["服务层 Service Layer"]
        NotificationService["NotificationService<br/>通知服务接口"]
    end

    subgraph BusinessLayer["业务逻辑层 Business Layer"]
        NotificationUsecase["NotificationUsecase<br/>通知用例<br/>· SendNotification<br/>· SendBulkNotification<br/>· MarkAsRead<br/>· GetUnreadCount"]
        TemplateUsecase["TemplateUsecase<br/>模板用例<br/>· CreateTemplate<br/>· RenderTemplate<br/>· GetTemplate"]
        SubscriptionMgr["SubscriptionManager<br/>订阅管理<br/>· 频率控制<br/>· 静默时段检查"]
    end

    subgraph InfraLayer["基础设施层 Infrastructure"]
        KafkaConsumer["Kafka Consumer<br/>事件消费<br/>· 消息解析<br/>· 路由分发"]
        QueueMgr["Queue Manager<br/>队列管理器"]
        WSManager["WebSocket Manager<br/>连接管理<br/>· 10000 连接池<br/>· 心跳检测"]
        SenderFactory["Sender Factory<br/>发送器工厂"]
    end

    subgraph Queue["通知队列 Notification Queue"]
        PriorityQueue["Priority Queue<br/>Redis ZSet<br/>critical: 0-1s<br/>high: 1-60s<br/>medium: 60-300s<br/>low: 300-1800s"]
        DelayQueue["Delay Queue<br/>Redis ZSet<br/>定时任务扫描<br/>间隔: 1s"]
        RetryQueue["Retry Queue<br/>失败重试<br/>指数退避<br/>最大3次"]
    end

    subgraph Senders["发送器层 Senders"]
        EmailSender["EmailSender<br/>SMTP协议<br/>超时: 30s<br/>支持HTML"]
        SMSSender["SMSSender<br/>多厂商降级<br/>超时: 10s<br/>重试: 2次"]
        WebSocketSender["WebSocketSender<br/>实时推送<br/>超时: 5s<br/>不重试"]
        WebHookSender["WebHookSender<br/>HTTP回调<br/>超时: 10s<br/>重试: 3次"]
        InAppSender["InAppSender<br/>站内信<br/>即时写入DB"]
        PushSender["PushSender<br/>FCM/APNs<br/>批量推送"]
    end

    subgraph Providers["第三方服务 External Providers"]
        SMTP["SMTP Server<br/>Gmail/SendGrid<br/>连接池: 10"]
        AliyunSMS["阿里云短信<br/>主用厂商<br/>QPS: 100"]
        TencentSMS["腾讯云短信<br/>备用厂商<br/>QPS: 100"]
        FCM["Firebase FCM<br/>Android推送"]
        APNs["Apple APNs<br/>iOS推送"]
    end

    subgraph DataLayer["数据层 Data Layer"]
        NotificationRepo["NotificationRepo<br/>通知仓储<br/>CRUD操作"]
        TemplateRepo["TemplateRepo<br/>模板仓储<br/>缓存: 5min"]
        SubscriptionRepo["SubscriptionRepo<br/>订阅仓储"]
    end

    subgraph Storage["存储层 Storage"]
        PostgreSQL["PostgreSQL<br/>主存储<br/>· notifications表<br/>· templates表<br/>· subscriptions表<br/>连接池: 50"]
        Redis["Redis<br/>缓存/队列<br/>· 优先级队列<br/>· 延迟队列<br/>· 未读计数<br/>· 限流计数器<br/>最大内存: 4GB"]
    end

    subgraph Monitoring["监控层 Monitoring"]
        Prometheus["Prometheus<br/>指标采集<br/>· 发送成功率<br/>· 延迟分布<br/>· 队列长度"]
        Grafana["Grafana<br/>可视化监控"]
        Alerting["Alerting<br/>告警规则<br/>· 队列积压>10K<br/>· 失败率>5%<br/>· 延迟P99>10s"]
    end

    %% 事件源到接入层
    Kafka --> KafkaConsumer
    HTTPClient --> HTTPRouter
    GRPCClient --> GRPCServer
    HTTPClient -.升级.-> WSUpgrade

    %% 接入层到服务层
    HTTPRouter --> NotificationService
    GRPCServer --> NotificationService
    KafkaConsumer --> NotificationUsecase

    %% 服务层到业务层
    NotificationService --> NotificationUsecase
    NotificationService --> TemplateUsecase

    %% 业务层内部交互
    NotificationUsecase --> TemplateUsecase
    NotificationUsecase --> SubscriptionMgr

    %% 业务层到基础设施
    NotificationUsecase --> QueueMgr
    NotificationUsecase --> WSManager
    WSUpgrade --> WSManager

    %% 队列管理
    QueueMgr --> PriorityQueue
    QueueMgr --> DelayQueue
    QueueMgr --> RetryQueue

    %% 队列到发送器
    PriorityQueue --> SenderFactory
    RetryQueue --> SenderFactory
    DelayQueue -.定时扫描.-> PriorityQueue

    %% 发送器工厂分发
    SenderFactory --> EmailSender
    SenderFactory --> SMSSender
    SenderFactory --> WebSocketSender
    SenderFactory --> WebHookSender
    SenderFactory --> InAppSender
    SenderFactory --> PushSender

    %% 发送器到外部服务
    EmailSender --> SMTP
    SMSSender --> AliyunSMS
    SMSSender -.降级.-> TencentSMS
    PushSender --> FCM
    PushSender --> APNs

    %% 业务层到数据层
    NotificationUsecase --> NotificationRepo
    TemplateUsecase --> TemplateRepo
    SubscriptionMgr --> SubscriptionRepo
    InAppSender --> NotificationRepo

    %% 数据层到存储
    NotificationRepo --> PostgreSQL
    TemplateRepo --> PostgreSQL
    SubscriptionRepo --> PostgreSQL
    TemplateRepo -.缓存.-> Redis

    QueueMgr --> Redis
    WSManager -.状态.-> Redis

    %% 监控
    NotificationUsecase -.metrics.-> Prometheus
    SenderFactory -.metrics.-> Prometheus
    QueueMgr -.metrics.-> Prometheus
    Prometheus --> Grafana
    Prometheus --> Alerting

    style EventSource fill:#e3f2fd
    style APILayer fill:#fff3e0
    style ServiceLayer fill:#ffe0b2
    style BusinessLayer fill:#fff9c4
    style InfraLayer fill:#f3e5f5
    style Queue fill:#e1bee7
    style Senders fill:#c8e6c9
    style Providers fill:#ffccbc
    style DataLayer fill:#b2dfdb
    style Storage fill:#b2ebf2
    style Monitoring fill:#ffcdd2

service NotificationService {
  rpc SendNotification(SendNotificationRequest) returns (SendNotificationResponse);
  rpc SendBulkNotification(SendBulkNotificationRequest) returns (SendBulkNotificationResponse);
  rpc GetUserNotifications(GetUserNotificationsRequest) returns (GetUserNotificationsResponse);
  rpc MarkAsRead(MarkAsReadRequest) returns (MarkAsReadResponse);
  rpc GetUnreadCount(GetUnreadCountRequest) returns (GetUnreadCountResponse);
}

// 发送单条通知
func (uc *NotificationUsecase) SendNotification(
    ctx context.Context,
    req *NotificationRequest,
) (*Notification, error)

// 批量发送通知
func (uc *NotificationUsecase) SendBulkNotification(
    ctx context.Context,
    tenantID string,
    userIDs []string,
    req *NotificationRequest,
) error

// 标记为已读
func (uc *NotificationUsecase) MarkAsRead(
    ctx context.Context,
    notificationID, userID string,
) error

// 创建模板
func (uc *TemplateUsecase) CreateTemplate(
    ctx context.Context,
    tenantID, name, title, content, channel string,
    variables []string,
) (*Template, error)

// 渲染模板
func (uc *NotificationUsecase) renderTemplate(
    template string,
    variables map[string]interface{},
) (string, error)

// 检查是否可以发送
func (sm *SubscriptionManager) CanSendNow(
    ctx context.Context,
    userID, notificationType string,
) (bool, error)

func (c *Consumer) processMessage(ctx context.Context, message kafka.Message) error {
    // 1. 解析事件
    var event map[string]interface{}
    json.Unmarshal(message.Value, &event)

    // 2. 路由分发
    eventType := event["event_type"].(string)
    switch eventType {
    case "document.indexed":
        return c.handleDocumentIndexed(ctx, event)
    case "user.registered":
        return c.handleUserRegistered(ctx, event)
    // ... 更多事件类型
    }
}

func (w *Worker) pollQueue() {
    for {
        // 1. 从优先级队列弹出（按 score 排序）
        items := redis.ZRangeByScore("notif:queue:priority", 0, now, 0, 10)

        // 2. 批量发送
        for _, item := range items {
            notification := deserialize(item)
            w.sendNotification(notification)
        }

        time.Sleep(100 * time.Millisecond)
    }
}

func (w *Worker) scanDelayQueue() {
    for {
        // 1. 扫描到期任务（score <= now）
        items := redis.ZRangeByScore("notif:queue:delay", 0, now, 0, 100)

        // 2. 转移到 PriorityQueue
        for _, item := range items {
            notification := deserialize(item)
            redis.ZAdd("notif:queue:priority", notification.priority, item)
            redis.ZRem("notif:queue:delay", item)
        }

        time.Sleep(1 * time.Second)
    }
}

func (w *Worker) handleFailedNotification(notification *Notification, err error) {
    notification.RetryCount++

    if notification.RetryCount > 3 {
        // 超过最大重试次数，标记失败
        notification.Status = "failed"
        notification.Metadata["error"] = err.Error()
        repo.Update(notification)
        return
    }

    // 计算下次重试时间（指数退避）
    delay := time.Duration(math.Pow(5, float64(notification.RetryCount))) * time.Minute
    retryAt := time.Now().Add(delay)

    // 加入 DelayQueue
    redis.ZAdd("notif:queue:delay", retryAt.Unix(), serialize(notification))
}

type Manager struct {
    // userID -> []*Connection
    connections map[string][]*Connection
    mu          sync.RWMutex
}

func (m *Manager) SendToUser(userID string, message interface{}) error {
    m.mu.RLock()
    connections := m.connections[userID]
    m.mu.RUnlock()

    // 发送到所有连接（用户可能多设备在线）
    for _, conn := range connections {
        select {
        case conn.Send <- message:
        default:
            // 发送通道满，关闭连接
            m.Unregister(conn)
        }
    }
}

func (f *SenderFactory) GetSender(channel NotificationChannel) Sender {
    switch channel {
    case ChannelEmail:
        return f.emailSender
    case ChannelSMS:
        return f.smsSender
    case ChannelWebSocket:
        return f.webSocketSender
    case ChannelWebHook:
        return f.webHookSender
    case ChannelInApp:
        return f.inAppSender
    case ChannelPush:
        return f.pushSender
    }
}

func (s *EmailSender) Send(notification *Notification) error {
    // 1. 构建邮件
    msg := &mail.Message{
        From:    "noreply@voiceassistant.com",
        To:      []string{notification.Metadata["email"]},
        Subject: notification.Title,
        Body:    notification.Content,
        HTML:    true,
    }

    // 2. SMTP 发送（超时 30 秒）
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    return s.smtpClient.Send(ctx, msg)
}

func (s *SMSSender) Send(notification *Notification) error {
    phone := notification.Metadata["phone"]
    content := notification.Content

    // 1. 尝试阿里云短信（主用厂商）
    err := s.aliyunProvider.SendSMS(phone, content)
    if err == nil {
        prometheus.CounterInc("sms_sent_total", "provider", "aliyun")
        return nil
    }

    prometheus.CounterInc("sms_failed_total", "provider", "aliyun", "reason", err.Error())

    // 2. 降级到腾讯云短信（备用厂商）
    err = s.tencentProvider.SendSMS(phone, content)
    if err == nil {
        prometheus.CounterInc("sms_sent_total", "provider", "tencent")
        return nil
    }

    prometheus.CounterInc("sms_failed_total", "provider", "tencent", "reason", err.Error())
    return fmt.Errorf("all SMS providers failed")
}

func (s *WebSocketSender) Send(notification *Notification) error {
    message := map[string]interface{}{
        "type":     "notification",
        "id":       notification.ID,
        "title":    notification.Title,
        "content":  notification.Content,
        "priority": notification.Priority,
        "metadata": notification.Metadata,
        "timestamp": time.Now().Unix(),
    }

    // 超时 5 秒
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    return s.wsManager.SendToUser(ctx, notification.UserID, message)
}

func (s *WebHookSender) Send(notification *Notification) error {
    webhookURL := notification.Metadata["webhook_url"]

    payload := map[string]interface{}{
        "event_type":      notification.Metadata["event_type"],
        "notification_id": notification.ID,
        "title":           notification.Title,
        "content":         notification.Content,
        "user_id":         notification.UserID,
        "tenant_id":       notification.TenantID,
        "timestamp":       time.Now().Unix(),
        "signature":       s.calculateSignature(notification),  // HMAC-SHA256 签名
    }

    // HTTP POST，超时 10 秒
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    req, _ := http.NewRequestWithContext(ctx, "POST", webhookURL, payloadBytes)
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("User-Agent", "VoiceHelper-Notification/1.0")
    req.Header.Set("X-Signature", signature)

    resp, err := s.httpClient.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    // 只接受 2xx 状态码
    if resp.StatusCode < 200 || resp.StatusCode >= 300 {
        return fmt.Errorf("webhook returned %d", resp.StatusCode)
    }

    return nil
}

func (s *InAppSender) Send(notification *Notification) error {
    // 直接写入数据库
    return s.notificationRepo.Create(context.Background(), notification)
}

func (s *PushSender) Send(notification *Notification) error {
    platform := notification.Metadata["platform"]  // android/ios
    deviceToken := notification.Metadata["device_token"]

    if platform == "android" {
        return s.fcmClient.Send(deviceToken, notification)
    } else if platform == "ios" {
        return s.apnsClient.Send(deviceToken, notification)
    }
}

# 1. notification:task_queue - ZSet (按执行时间排序)
#    Score: Unix 时间戳
#    Member: Task ID
ZADD notification:task_queue 1706342400 "task-001"
ZADD notification:task_queue 1706346000 "task-002"

# 2. notification:task_data - Hash (存储任务完整数据)
#    Field: Task ID
#    Value: JSON 序列化的 Task 对象
HSET notification:task_data "task-001" "{\"id\":\"task-001\",\"name\":\"send_notification\",...}"

# 3. notification:processing - Set (正在执行的任务)
#    Member: Task ID
SADD notification:processing "task-001"

# 4. notification:completed - List (已完成任务记录，最多保留 1000 条)
LPUSH notification:completed "task-001"
LTRIM notification:completed 0 999

func (q *TaskQueue) ScheduleTask(ctx context.Context, task *Task) error {
    // 1. 生成任务 ID（如果未提供）
    if task.ID == "" {
        task.ID = uuid.New().String()
    }
    task.Status = TaskStatusScheduled
    task.CreatedAt = time.Now()
    task.UpdatedAt = time.Now()

    // 2. 序列化任务数据
    taskData, err := json.Marshal(task)
    if err != nil {
        return fmt.Errorf("failed to marshal task: %w", err)
    }

    // 3. 保存任务数据到 Hash
    if err := q.redis.HSet(ctx, q.taskDataKey, task.ID, taskData).Err(); err != nil {
        return fmt.Errorf("failed to save task data: %w", err)
    }

    // 4. 添加到 ZSet（按执行时间排序）
    score := float64(task.ScheduledAt.Unix())
    if err := q.redis.ZAdd(ctx, q.taskQueueKey, &redis.Z{
        Score:  score,
        Member: task.ID,
    }).Err(); err != nil {
        return fmt.Errorf("failed to add task to queue: %w", err)
    }

    log.Printf("Task scheduled: id=%s, name=%s, scheduled_at=%s",
        task.ID, task.Name, task.ScheduledAt.Format(time.RFC3339))

    return nil
}

func (q *TaskQueue) worker(ctx context.Context, workerID int) {
    defer q.wg.Done()

    ticker := time.NewTicker(1 * time.Second)  // 每秒扫描一次
    defer ticker.Stop()

    log.Printf("Worker %d started", workerID)

    for {
        select {
        case <-q.stopChan:
            log.Printf("Worker %d stopped", workerID)
            return

        case <-ticker.C:
            // 获取到期任务并执行
            if err := q.processDueTasks(ctx); err != nil {
                log.Printf("Worker %d: error processing tasks: %v", workerID, err)
            }
        }
    }
}

func (q *TaskQueue) processDueTasks(ctx context.Context) error {
    now := time.Now().Unix()

    // 1. 从 ZSet 获取到期任务（score <= now）
    results, err := q.redis.ZRangeByScoreWithScores(ctx, q.taskQueueKey, &redis.ZRangeBy{
        Min:    "-inf",
        Max:    fmt.Sprintf("%d", now),
        Offset: 0,
        Count:  10,  // 每次最多处理 10 个
    }).Result()
    if err != nil {
        return fmt.Errorf("failed to fetch due tasks: %w", err)
    }

    if len(results) == 0 {
        return nil  // 没有到期任务
    }

    // 2. 遍历处理每个任务
    for _, result := range results {
        taskID := result.Member.(string)

        // 3. 获取任务数据
        taskData, err := q.redis.HGet(ctx, q.taskDataKey, taskID).Result()
        if err == redis.Nil {
            // 任务不存在，从队列中移除
            q.redis.ZRem(ctx, q.taskQueueKey, taskID)
            continue
        } else if err != nil {
            log.Printf("Failed to get task data for %s: %v", taskID, err)
            continue
        }

        // 4. 反序列化任务
        var task Task
        if err := json.Unmarshal([]byte(taskData), &task); err != nil {
            log.Printf("Failed to unmarshal task %s: %v", taskID, err)
            continue
        }

        // 5. 检查是否正在处理（防止重复执行）
        exists, err := q.redis.SIsMember(ctx, q.processingKey, taskID).Result()
        if err != nil || exists {
            continue
        }

        // 6. 标记为处理中
        q.redis.SAdd(ctx, q.processingKey, taskID)

        // 7. 异步执行任务
        go q.executeTask(context.Background(), &task)
    }

    return nil
}

func (q *TaskQueue) executeTask(ctx context.Context, task *Task) {
    defer func() {
        // 执行完毕，从 processing Set 移除
        q.redis.SRem(ctx, q.processingKey, task.ID)
    }()

    log.Printf("Executing task: id=%s, name=%s", task.ID, task.Name)

    // 1. 更新状态为运行中
    task.Status = TaskStatusRunning
    now := time.Now()
    task.ExecutedAt = &now
    task.UpdatedAt = now
    q.saveTask(ctx, task)

    // 2. 获取任务处理器
    q.mu.RLock()
    handler, exists := q.handlers[task.Name]
    q.mu.RUnlock()

    if !exists {
        task.Status = TaskStatusFailed
        task.Error = fmt.Sprintf("no handler registered for task: %s", task.Name)
        log.Printf("Task failed: %s", task.Error)
        q.saveTask(ctx, task)
        q.redis.ZRem(ctx, q.taskQueueKey, task.ID)
        return
    }

    // 3. 执行任务
    err := handler(ctx, task)

    completed := time.Now()
    task.CompletedAt = &completed
    task.UpdatedAt = completed

    // 4. 处理执行结果
    if err != nil {
        task.RetryCount++
        task.Error = err.Error()

        // 检查是否需要重试
        if task.RetryCount < task.MaxRetries {
            task.Status = TaskStatusScheduled
            // 指数退避重试：2^n 分钟后重试
            retryDelay := time.Duration(1<<uint(task.RetryCount)) * time.Minute
            task.ScheduledAt = time.Now().Add(retryDelay)

            log.Printf("Task failed, will retry (attempt %d/%d) after %v: %s",
                task.RetryCount, task.MaxRetries, retryDelay, task.ID)

            // 重新调度
            q.redis.ZAdd(ctx, q.taskQueueKey, &redis.Z{
                Score:  float64(task.ScheduledAt.Unix()),
                Member: task.ID,
            })
        } else {
            task.Status = TaskStatusFailed
            log.Printf("Task failed after %d retries: %s", task.MaxRetries, task.ID)
            q.redis.ZRem(ctx, q.taskQueueKey, task.ID)
        }
    } else {
        task.Status = TaskStatusCompleted
        log.Printf("Task completed: id=%s, name=%s", task.ID, task.Name)

        // 从队列中移除
        q.redis.ZRem(ctx, q.taskQueueKey, task.ID)

        // 如果是重复任务，调度下一次执行
        if task.Type == TaskTypeRecurring && task.Interval > 0 {
            nextTask := &Task{
                Type:        task.Type,
                Name:        task.Name,
                Payload:     task.Payload,
                ScheduledAt: time.Now().Add(task.Interval),
                Interval:    task.Interval,
                MaxRetries:  task.MaxRetries,
            }
            q.ScheduleTask(ctx, nextTask)
        }

        // 记录到已完成列表（最多保留 1000 条）
        q.redis.LPush(ctx, q.completedKey, task.ID)
        q.redis.LTrim(ctx, q.completedKey, 0, 999)
    }

    // 5. 保存任务状态
    q.saveTask(ctx, task)
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant UseCase as NotificationUsecase
    participant Queue as TaskQueue
    participant Redis as Redis
    participant Worker as Worker Goroutine
    participant Handler as TaskHandler

    Note over Client,Handler: 任务调度阶段
    Client->>UseCase: SendNotification(scheduled_at=明天8点)
    UseCase->>Queue: ScheduleTask(task)
    Queue->>Queue: 生成 Task ID
    Queue->>Redis: HSET task_data task_id task_json
    Redis-->>Queue: ok
    Queue->>Redis: ZADD task_queue timestamp task_id
    Redis-->>Queue: ok
    Queue-->>UseCase: task_id
    UseCase-->>Client: notification created (status=pending)

    Note over Worker,Handler: Worker 轮询阶段（每秒）
    Worker->>Redis: ZRANGEBYSCORE task_queue -inf now 0 10
    Redis-->>Worker: [task_id1, task_id2]
    Worker->>Redis: HGET task_data task_id1
    Redis-->>Worker: task_json
    Worker->>Worker: Unmarshal JSON to Task
    Worker->>Redis: SISMEMBER processing task_id1
    Redis-->>Worker: false (未在执行中)
    Worker->>Redis: SADD processing task_id1
    Redis-->>Worker: ok
    Worker->>Worker: go executeTask(task)

    Note over Worker,Handler: 任务执行阶段
    Worker->>Redis: HSET task_data task_id1 (status=running)
    Worker->>Handler: handler(ctx, task)

    alt 执行成功
        Handler-->>Worker: nil
        Worker->>Redis: ZREM task_queue task_id1
        Worker->>Redis: LPUSH completed task_id1
        Worker->>Redis: LTRIM completed 0 999
        Worker->>Redis: HSET task_data task_id1 (status=completed)
    else 执行失败，需要重试
        Handler-->>Worker: error
        Worker->>Worker: retryCount++
        Worker->>Worker: retryDelay = 2^retryCount min
        Worker->>Redis: ZADD task_queue (now+retryDelay) task_id1
        Worker->>Redis: HSET task_data task_id1 (status=scheduled, retry_count=1)
    else 执行失败，超过最大重试次数
        Handler-->>Worker: error
        Worker->>Worker: retryCount > maxRetries
        Worker->>Redis: ZREM task_queue task_id1
        Worker->>Redis: HSET task_data task_id1 (status=failed)
    end

    Worker->>Redis: SREM processing task_id1
    Redis-->>Worker: ok

type Connection struct {
    UserID   string              // 用户 ID
    TenantID string              // 租户 ID
    Conn     *websocket.Conn     // Gorilla WebSocket 连接
    Send     chan []byte         // 发送通道（缓冲 256）
}

type Manager struct {
    // 连接映射：userID -> []*Connection
    connections map[string][]*Connection
    mu          sync.RWMutex

    // 注册连接通道（缓冲 100）
    register chan *Connection

    // 注销连接通道（缓冲 100）
    unregister chan *Connection
}

func (m *Manager) Register(conn *Connection) {
    // 1. 启动读取协程（处理客户端消息、心跳）
    go conn.readPump(m)

    // 2. 启动写入协程（发送服务端消息）
    go conn.writePump()

    // 3. 注册到管理器
    m.register <- conn
}

func (m *Manager) addConnection(conn *Connection) {
    m.mu.Lock()
    defer m.mu.Unlock()

    // 初始化用户连接列表
    if m.connections[conn.UserID] == nil {
        m.connections[conn.UserID] = make([]*Connection, 0)
    }

    // 追加连接
    m.connections[conn.UserID] = append(m.connections[conn.UserID], conn)
}

func (m *Manager) SendToUser(userID string, message interface{}) error {
    // 1. 读锁获取用户连接列表
    m.mu.RLock()
    connections := m.connections[userID]
    m.mu.RUnlock()

    if len(connections) == 0 {
        return nil  // 用户不在线，静默失败
    }

    // 2. 序列化消息
    data, err := json.Marshal(message)
    if err != nil {
        return err
    }

    // 3. 发送到所有连接（多设备）
    for _, conn := range connections {
        select {
        case conn.Send <- data:
            // 成功发送到 Send 通道
        default:
            // Send 通道满（256 条），关闭连接
            m.Unregister(conn)
        }
    }

    return nil
}

func (conn *Connection) readPump(manager *Manager) {
    defer func() {
        // 连接断开，注销
        manager.Unregister(conn)
    }()

    for {
        _, message, err := conn.Conn.ReadMessage()
        if err != nil {
            break  // 连接断开或读取错误
        }

        // 处理客户端消息（如心跳 ping）
        _ = message
    }
}

func (conn *Connection) writePump() {
    for {
        select {
        case message, ok := <-conn.Send:
            if !ok {
                // Send 通道关闭，发送 Close 消息
                conn.Conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }

            // 发送消息
            if err := conn.Conn.WriteMessage(websocket.TextMessage, message); err != nil {
                return  // 发送失败，退出
            }
        }
    }
}

sequenceDiagram
    autonumber
    participant Client as 客户端（浏览器）
    participant HTTPServer as HTTP Server
    participant Manager as WebSocketManager
    participant ReadPump as readPump Goroutine
    participant WritePump as writePump Goroutine
    participant UseCase as NotificationUsecase
    participant Sender as WebSocketSender

    Note over Client,Sender: 连接建立阶段
    Client->>HTTPServer: GET /ws?user_id=123&token=xxx
    HTTPServer->>HTTPServer: validateToken(token, user_id)
    HTTPServer->>HTTPServer: upgrader.Upgrade(HTTP → WebSocket)
    HTTPServer->>Manager: Register(connection)
    Manager->>ReadPump: go conn.readPump(manager)
    Manager->>WritePump: go conn.writePump()
    Manager->>Manager: connections[user_id].append(conn)
    Manager-->>HTTPServer: registered
    HTTPServer-->>Client: 101 Switching Protocols

    Note over Client,Sender: 心跳保持阶段
    loop 每 30 秒
        Client->>ReadPump: WebSocket: ping
        ReadPump->>ReadPump: ReadMessage()
        ReadPump->>Client: WebSocket: pong
    end

    Note over Client,Sender: 消息推送阶段
    UseCase->>UseCase: SendNotification(channel=websocket)
    UseCase->>Sender: Send(notification)
    Sender->>Sender: 构建消息 JSON
    Sender->>Manager: SendToUser(user_id, message)
    Manager->>Manager: connections = m.connections[user_id]
    Manager->>Manager: json.Marshal(message)
    Manager->>WritePump: conn.Send <- data
    WritePump->>WritePump: <-conn.Send (读取消息)
    WritePump->>Client: conn.WriteMessage(TextMessage, data)
    Client->>Client: 收到通知，显示弹窗

    Note over Client,Sender: 连接断开阶段
    Client->>ReadPump: WebSocket: close
    ReadPump->>ReadPump: ReadMessage() error
    ReadPump->>Manager: Unregister(conn)
    Manager->>Manager: removeConnection(conn)
    Manager->>Manager: close(conn.Send)
    Manager->>Manager: conn.Conn.Close()
    WritePump->>WritePump: <-conn.Send (通道关闭)
    WritePump->>Client: conn.WriteMessage(CloseMessage)
    WritePump->>WritePump: return (退出)
    ReadPump->>ReadPump: return (退出)

classDiagram
    class Notification {
        +string ID
        +string TenantID
        +string UserID
        +NotificationChannel Channel
        +NotificationPriority Priority
        +string Title
        +string Content
        +NotificationStatus Status
        +time.Time SentAt
        +time.Time ReadAt
        +map Metadata
        +time.Time CreatedAt
        +MarkAsSending()
        +MarkAsSent()
        +MarkAsFailed(error string)
        +MarkAsRead()
    }

    class Template {
        +string ID
        +string TenantID
        +string Name
        +string Title
        +string Content
        +NotificationChannel Channel
        +[]string Variables
        +bool IsActive
        +time.Time CreatedAt
        +Render(variables map) (title, content, error)
        +Activate()
        +Deactivate()
    }

    class Subscription {
        +string UserID
        +string NotificationType
        +bool Enabled
        +Frequency Frequency
        +[]TimeRange QuietHours
        +CanSendNow() bool
        +IncrementCount()
        +ResetDailyCount()
    }

    class NotificationChannel {
        <<enumeration>>
        EMAIL
        SMS
        WEBSOCKET
        WEBHOOK
        IN_APP
        PUSH
    }

    class NotificationPriority {
        <<enumeration>>
        LOW = 0
        MEDIUM = 5
        HIGH = 8
        CRITICAL = 10
    }

    class NotificationStatus {
        <<enumeration>>
        PENDING
        SENDING
        SENT
        FAILED
    }

    class Frequency {
        <<enumeration>>
        IMMEDIATE
        HOURLY
        DAILY
        WEEKLY
    }

    Notification --> NotificationChannel
    Notification --> NotificationPriority
    Notification --> NotificationStatus
    Template --> NotificationChannel
    Subscription --> Frequency