junhong_cmp_fiber/docs/7月迭代/基础设施/站内消息.md

基础设施：站内消息（Notification）

状态：待评审 被依赖：需求 18/20/21（审批流通知）、需求 22（临期提醒） Phase 2 扩展：企业微信推送、短信通知（预留插拔接口）

---

一、设计原则

• 业务代码不直接写通知表：统一通过通知发布器或领域事件异步处理
• 关键领域事件先写 Outbox：审批、退款、充值等事务内事件由 Outbox Relay 投递 Asynq，禁止事务提交后直接入队
• 通知消费必须幂等：使用稳定的 event_id 和接收人唯一约束，Asynq 重试不得重复生成站内消息
• 不过度抽象：不用 interface，用具体的 NotificationPublisher（后续加渠道 = 在 handler 里加代码）
• 前端轮询：30秒一次 /api/admin/notifications/unread-count，不用 WebSocket

sequenceDiagram
    participant Biz as 业务事务
    participant Outbox as Outbox
    participant Relay as Relay
    participant Worker as Notification Worker
    participant DB as tb_notification
    participant Web as 后台前端

    Biz->>Outbox: 同事务写领域事件
    Relay->>Outbox: 拉取待投递事件
    Relay->>Worker: 至少一次投递
    Worker->>DB: 按 event_id + recipient 幂等插入
    Web->>DB: 经 API 轮询未读数/通知列表
    Web->>DB: 经 API 标记已读

---

二、数据库

-- 迁移文件：YYYYMMDD_create_tb_notification.sql
CREATE TABLE tb_notification (
    id             BIGSERIAL PRIMARY KEY,
    event_id       VARCHAR(64) NOT NULL,          -- 领域事件ID或调用方请求ID，用于幂等
    recipient_id   BIGINT NOT NULL,          -- 用户ID（admin user 或 agent user）
    recipient_type VARCHAR(20) NOT NULL DEFAULT 'admin', -- admin | agent
    type           VARCHAR(50) NOT NULL,     -- 通知类型（见常量定义）
    title          VARCHAR(200) NOT NULL,    -- 标题
    body           TEXT NOT NULL DEFAULT '', -- 纯文本正文
    ref_type       VARCHAR(50),              -- 关联业务类型 approval | recharge | refund | iot_card | device
    ref_id         BIGINT,                   -- 关联业务ID
    is_read        BOOLEAN NOT NULL DEFAULT FALSE,
    read_at        TIMESTAMPTZ,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    expires_at     TIMESTAMPTZ               -- 过期自动不展示（可选，临期提醒用）
);

CREATE INDEX idx_notification_recipient
    ON tb_notification (recipient_id, recipient_type, is_read, created_at DESC);
CREATE INDEX idx_notification_created   ON tb_notification (created_at DESC);
CREATE UNIQUE INDEX uq_notification_event_recipient
    ON tb_notification (event_id, recipient_id, recipient_type);

COMMENT ON TABLE tb_notification IS '站内消息通知表';

---

三、常量定义

// pkg/constants/notification.go

// 通知类型
const (
    NotifyTypeApprovalPending  = "approval.pending"   // 待我审批
    NotifyTypeApprovalDone     = "approval.done"       // 审批完成（申请人收）
    NotifyTypeApprovalRejected = "approval.rejected"   // 审批驳回（申请人收，流程终止）
    NotifyTypeApprovalReturned = "approval.returned"   // 审批退回（申请人收，可修改后重新提交）
    NotifyTypePackageExpiring  = "package.expiring"    // 套餐临期
    NotifyTypeSystemAlert      = "system.alert"        // 系统通知
)

// 通知接收者类型
const (
    NotifyRecipientAdmin = "admin"  // 平台用户
    NotifyRecipientAgent = "agent"  // 代理用户（预留）
)

---

四、Model

// internal/model/notification.go

// Notification 站内消息模型
type Notification struct {
    ID            uint       `gorm:"column:id;primaryKey" json:"id"`
    EventID       string     `gorm:"column:event_id;type:varchar(64);not null;uniqueIndex:uq_notification_event_recipient,priority:1" json:"event_id"`
    RecipientID   uint       `gorm:"column:recipient_id;not null;index;uniqueIndex:uq_notification_event_recipient,priority:2" json:"recipient_id"`
    RecipientType string     `gorm:"column:recipient_type;type:varchar(20);not null;default:'admin';uniqueIndex:uq_notification_event_recipient,priority:3" json:"recipient_type"`
    Type          string     `gorm:"column:type;type:varchar(50);not null" json:"type"`
    Title         string     `gorm:"column:title;type:varchar(200);not null" json:"title"`
    Body          string     `gorm:"column:body;type:text;not null;default:''" json:"body"`
    RefType       *string    `gorm:"column:ref_type;type:varchar(50)" json:"ref_type,omitempty"`
    RefID         *uint      `gorm:"column:ref_id" json:"ref_id,omitempty"`
    IsRead        bool       `gorm:"column:is_read;not null;default:false" json:"is_read"`
    ReadAt        *time.Time `gorm:"column:read_at" json:"read_at,omitempty"`
    CreatedAt     time.Time  `gorm:"column:created_at;not null" json:"created_at"`
    ExpiresAt     *time.Time `gorm:"column:expires_at" json:"expires_at,omitempty"`
}

func (Notification) TableName() string { return "tb_notification" }

---

五、发布侧：NotificationPublisher

非事务性、允许调用方直接发起的通知使用发布器异步入队。审批等领域事件不直接调用该发布器，而由 TaskCreated、ProcessApproved 等事件处理器转换为通知载荷。

领域事件通知使用领域事件自身的 event_id；非领域事件调用方使用稳定的业务请求 ID。网络重试或 Asynq 重试时禁止重新生成 ID。

// internal/infrastructure/messaging/notification_publisher.go

// SendPayload 发送通知的参数
type SendPayload struct {
    EventID       string   // 领域事件ID或调用方请求ID，同一次重试必须保持不变
    RecipientIDs  []uint   // 接收人ID列表
    RecipientType string   // admin | agent
    Type          string   // 通知类型常量
    Title         string   // 标题
    Body          string   // 正文
    RefType       string   // 关联业务类型（可选）
    RefID         uint     // 关联业务ID（可选）
    ExpiresAt     *time.Time // 过期时间（可选）
}

// NotificationPublisher 通知发布器（非接口，简单具体实现）
type NotificationPublisher struct {
    queueClient *queue.Client
    logger      *zap.Logger
}

// Publish 发布通知（异步）
// 返回错误供调用方或 Asynq Handler 决定重试，禁止吞掉关键通知错误。
func (p *NotificationPublisher) Publish(ctx context.Context, payload SendPayload) error {
    if payload.EventID == "" {
        return errors.New(errors.CodeInvalidParam, "通知事件ID不能为空")
    }
    if err := p.queueClient.EnqueueTask(ctx, constants.TaskTypeNotification, payload); err != nil {
        p.logger.Error("通知入队失败", zap.Error(err), zap.String("type", payload.Type))
        return err
    }
    return nil
}

审批事件处理器调用示例（节点激活后通知候选审批人）：

return h.notifyPublisher.Publish(ctx, notification.SendPayload{
    EventID:       event.EventID,
    RecipientIDs:  []uint{nextApproverID},
    RecipientType: constants.NotifyRecipientAdmin,
    Type:          constants.NotifyTypeApprovalPending,
    Title:         "您有一条待审批记录",
    Body:          fmt.Sprintf("代理「%s」提交了充值申请，请及时审批。", shopName),
    RefType:       "approval",
    RefID:         processInstanceID,
})

审批事务中只写 TaskCreated Outbox 事件。即使 Redis/Asynq 暂时不可用，事件仍保留在数据库，由 Relay 重试；通知处理失败则由 Asynq 重试当前任务。

---

六、消费侧：Asynq Task Handler

// internal/task/notification_handler.go

// HandleNotification 处理通知发送任务
func (h *NotificationHandler) HandleNotification(ctx context.Context, t *asynq.Task) error {
    var payload notification.SendPayload
    if err := sonic.Unmarshal(t.Payload(), &payload); err != nil {
        return fmt.Errorf("反序列化通知载荷失败: %w", err)
    }

    // 批量写入 tb_notification
    records := make([]model.Notification, 0, len(payload.RecipientIDs))
    for _, uid := range payload.RecipientIDs {
        ref_type := (*string)(nil)
        ref_id := (*uint)(nil)
        if payload.RefType != "" {
            ref_type = &payload.RefType
        }
        if payload.RefID != 0 {
            ref_id = &payload.RefID
        }
        records = append(records, model.Notification{
            EventID:        payload.EventID,
            RecipientID:   uid,
            RecipientType: payload.RecipientType,
            Type:          payload.Type,
            Title:         payload.Title,
            Body:          payload.Body,
            RefType:       ref_type,
            RefID:         ref_id,
            ExpiresAt:     payload.ExpiresAt,
        })
    }

    if err := h.db.WithContext(ctx).
        Clauses(clause.OnConflict{DoNothing: true}).
        CreateInBatches(records, 100).Error; err != nil {
        return fmt.Errorf("批量写入通知失败: %w", err)
    }

    // Phase 2：在此处加企微/短信调用
    // for _, sender := range h.extraSenders { sender.Send(ctx, payload) }

    return nil
}

---

七、API 设计

7.1 未读数量（前端轮询用，30秒一次）

GET /api/admin/notifications/unread-count

响应：

{ "code": 0, "data": { "count": 5 } }

7.2 通知列表

GET /api/admin/notifications?is_read=false&type=approval.pending&page=1&page_size=20

请求参数：

type NotificationListRequest struct {
    IsRead *bool  `query:"is_read" description:"是否已读（不传=全部）"`
    Type   string `query:"type"    description:"通知类型过滤（可选）"`
    Page   int    `query:"page"    description:"页码"`
    PageSize int  `query:"page_size" description:"每页数量（最大50）"`
}

响应：

{
  "code": 0,
  "data": {
    "list": [
      {
        "id": 1,
        "type": "approval.pending",
        "title": "您有一条待审批记录",
        "body": "代理「XX店」提交了充值申请，请及时审批。",
        "ref_type": "approval",
        "ref_id": 42,
        "is_read": false,
        "created_at": "2026-07-11T10:00:00Z"
      }
    ],
    "total": 3,
    "page": 1,
    "page_size": 20
  }
}

7.3 标记已读

PUT /api/admin/notifications/{id}/read

7.4 全部标记已读

PUT /api/admin/notifications/read-all

可传 type 过滤（只把某类型全部已读）：

{ "type": "approval.pending" }

DTO

// internal/model/dto/notification_dto.go

type NotificationListRequest struct {
    IsRead   *bool  `query:"is_read"`
    Type     string `query:"type"`
    Page     int    `query:"page" default:"1"`
    PageSize int    `query:"page_size" default:"20"`
}

type NotificationItem struct {
    ID        uint       `json:"id"`
    Type      string     `json:"type" description:"通知类型"`
    Title     string     `json:"title"`
    Body      string     `json:"body"`
    RefType   *string    `json:"ref_type,omitempty"`
    RefID     *uint      `json:"ref_id,omitempty"`
    IsRead    bool       `json:"is_read"`
    ReadAt    *time.Time `json:"read_at,omitempty"`
    CreatedAt time.Time  `json:"created_at"`
}

type UnreadCountResponse struct {
    Count int64 `json:"count"`
}

type ReadAllRequest struct {
    Type string `json:"type" description:"通知类型（为空则全部已读）"`
}

---

八、前端对接

顶部导航栏铃铛

组件挂载 → 轮询 GET /api/admin/notifications/unread-count（30秒一次）
→ count > 0 时铃铛显示红点 + 数字
→ 点击铃铛 → 弹出通知抽屉 or 跳转 /notifications 页面
→ 打开时调 GET /api/admin/notifications?is_read=false
→ 点击某条通知 → PUT /api/admin/notifications/{id}/read → 根据 ref_type+ref_id 跳转对应业务页

跳转逻辑（ref_type）

ref_type	跳转页面
approval	/approvals/instances/{ref_id} 审批详情
recharge	/agent-recharges/{ref_id} 充值单详情
refund	/refunds/{ref_id} 退款单详情
iot_card	/iot-cards/{ref_id} IoT卡详情（临期提醒）
device	/devices/{ref_id} 设备详情（临期提醒）

通知列表页（/notifications）

筛选：通知类型（下拉）、已读状态 操作：全部已读按钮 列表字段：类型、标题、时间、已读状态 点击行：跳转关联业务详情

前端页面不可见时暂停未读数轮询，恢复可见时立即刷新。通知正文按纯文本渲染；未来需要富文本时使用受控模板和统一净化，禁止直接渲染业务方提交的 HTML。

---

九、Phase 2 扩展预留

当需要接入企微通知时，只需在 HandleNotification 里追加：

// 企微通知（Phase 2）
if h.wecomClient != nil {
    for _, uid := range payload.RecipientIDs {
        wecomOpenID := h.userStore.GetWecomOpenID(ctx, uid)
        h.wecomClient.SendMessage(wecomOpenID, payload.Title, payload.Body)
    }
}

业务代码零修改，Handler 里加一段即可。