24 KiB
24 KiB
通知系统实现方案
日期: 2026-06-16
状态: 已确认
版本: v1.0
上游文档: docs/specs/2026-05-15-notification-system-design.md
1. 概述
基于已批准的设计文档 2026-05-15-notification-system-design.md 实现通知系统。本文档聚焦实现层细节（项目结构、调用关系、错误处理、回归点），不重复架构设计。
支持的通知类型：
- 点赞通知 - 用户点赞藏品时由 social service 同步触发
- 系统通知 - admin 后端（运营）手动创建
- 活动通知 - admin 后端创建 activity 时手动或自动广播
列表聚合策略（v1.1 新增）：
- 写入层：每条点赞仍独立写入
notifications表 - 展示层（
GetNotifications）：type=like→ 按target_id聚合返回 1 张卡片（total_count + 前 3 个点赞人预览）type=system/type=activity→ 不聚合，每条独立返回
- 跳转：点击 like 聚合卡直接跳到
/pages/asset/detail?id={target_id}，不需要"展开全部"页面
2. 关键决策（与上游文档的差异/补充）
| 决策项 | 选择 | 理由 |
|---|---|---|
| 失败容忍 | 同步 RPC，失败仅 ERROR 日志，不影响点赞主路径 | 设计文档 §3.3 未明确；但通知丢失比点赞失败代价小 |
| Notification Service 形态 | 独立 Dubbo 服务（端口 20010） | 设计文档 §3.2 明确 |
| admin 集成方式 | admin 后端（FastAPI）共享数据库直接写表 | admin 已在 TopFans-activity-admin/backend，无 Dubbo 客户端 |
| 系统通知接收方 | 单用户 / 按 star 广播 / 全量广播 三种模式均支持 | 运营实际场景需要 |
| 活动通知触发 | ① 手动按钮（POST /api/v1/admin/activities/{id}/broadcast）② 创建 activity 时自动广播 |
用户要求两种都支持 |
| asset 数据获取 | 扩展 GetAssetForRPCResponse 增加 name 和 cover_url 字段 |
复用现有调用，不新增 RPC |
| 防重复唯一约束 | 本次不实现 | 设计文档 §8.2 标注"可选"，依赖客户端去抖 |
| TTL 清理 | 本次不实现 | 设计文档未要求 |
| 列表聚合（v1.1 新增） | 展示层按 target_id 聚合 like；system/activity 不聚合 | 防止热门藏品刷屏通知列表 |
3. 架构与调用关系
┌─────────────────────┐
│ Social Service │
│ LikeAsset 末尾 │──── 同步 RPC (失败仅日志) ────┐
└─────────────────────┘ ▼
┌─────────────────────────┐
│ Notification Service │ 独立 Dubbo 服务
│ port 20010 │
│ - 事务内 INSERT 通知 │
│ - UPSERT 统计 │
└─────────────────────────┘
│
┌───────────────────────────────────────┼───────────────────────┐
▼ ▼ ▼
┌──────────────────┐ ┌──────────────────────┐ ┌──────────────────┐
│ Gateway HTTP 路由 │ │ Asset Service │ │ Admin Backend │
│ /api/v1/notifs/ │ │ GetAssetForRPC │ │ (FastAPI, 共享DB)│
│ 前端调用 │ │ (扩展 name/cover) │ │ 直接写通知表 │
└──────────────────┘ └──────────────────────┘ └──────────────────┘
事务边界：
- notification service 内部：
CreateNotification把 INSERT notifications + UPSERT notification_stats 包在一个事务 - admin 后端：创建 activity + 广播通知包在一个 DB 事务（用 SQLAlchemy
session.begin()）
4. 数据层
4.1 迁移文件
backend/migrations/2026_06_16_001_create_notifications.sql
CREATE TABLE IF NOT EXISTS public.notifications (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL,
star_id BIGINT NOT NULL,
type VARCHAR(20) NOT NULL, -- like / system / activity
title VARCHAR(200) NOT NULL,
content VARCHAR(500),
data JSONB,
is_read BOOLEAN NOT NULL DEFAULT FALSE,
is_deleted BOOLEAN NOT NULL DEFAULT FALSE,
created_at BIGINT NOT NULL, -- 毫秒时间戳
read_at BIGINT
);
CREATE INDEX IF NOT EXISTS idx_notifications_user_type_created
ON public.notifications (user_id, star_id, type, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_notifications_user_unread
ON public.notifications (user_id, star_id, is_read, created_at DESC)
WHERE is_deleted = FALSE;
CREATE TABLE IF NOT EXISTS public.notification_stats (
user_id BIGINT NOT NULL,
star_id BIGINT NOT NULL,
like_unread_count INT NOT NULL DEFAULT 0,
system_unread_count INT NOT NULL DEFAULT 0,
activity_unread_count INT NOT NULL DEFAULT 0,
total_unread_count INT NOT NULL DEFAULT 0,
updated_at BIGINT NOT NULL,
PRIMARY KEY (user_id, star_id)
);
-- 序列起始值预留 10000（符合 CLAUDE.md 数据库规范）
ALTER SEQUENCE notifications_id_seq RESTART WITH 10000;
4.2 admin 后端 ORM 模型
TopFans-activity-admin/backend/models/models.py 新增：
class Notification(Base):
__tablename__ = "notifications"
id = Column(BigInteger, primary_key=True)
user_id = Column(BigInteger, nullable=False, index=True)
star_id = Column(BigInteger, nullable=False)
type = Column(String(20), nullable=False)
title = Column(String(200), nullable=False)
content = Column(String(500))
data = Column(JSON)
is_read = Column(Boolean, default=False, nullable=False)
is_deleted = Column(Boolean, default=False, nullable=False)
created_at = Column(BigInteger, nullable=False)
read_at = Column(BigInteger)
class NotificationStats(Base):
__tablename__ = "notification_stats"
user_id = Column(BigInteger, primary_key=True)
star_id = Column(BigInteger, primary_key=True)
like_unread_count = Column(Integer, default=0, nullable=False)
system_unread_count = Column(Integer, default=0, nullable=False)
activity_unread_count = Column(Integer, default=0, nullable=False)
total_unread_count = Column(Integer, default=0, nullable=False)
updated_at = Column(BigInteger, nullable=False)
5. Proto 定义
5.1 新增 backend/proto/notification.proto
syntax = "proto3";
package topfans.notification;
option go_package = "github.com/topfans/backend/pkg/proto/notification;notification";
import "proto/common.proto";
import "google/api/annotations.proto";
import "google/protobuf/struct.proto";
service NotificationService {
rpc CreateNotification(CreateNotificationRequest) returns (CreateNotificationResponse) {
option (google.api.http) = {
post: "/internal/v1/notifications"
body: "*"
};
}
rpc GetNotifications(GetNotificationsRequest) returns (GetNotificationsResponse) {
option (google.api.http) = { get: "/api/v1/notifications" };
}
rpc GetUnreadCount(GetUnreadCountRequest) returns (GetUnreadCountResponse) {
option (google.api.http) = { get: "/api/v1/notifications/unread-count" };
}
rpc MarkAsRead(MarkAsReadRequest) returns (MarkAsReadResponse) {
option (google.api.http) = { post: "/api/v1/notifications/{id}/read" };
}
rpc MarkAsReadByTarget(MarkAsReadByTargetRequest) returns (MarkAsReadByTargetResponse) {
// 仅 type=like 使用：按 target_id 标记该 user+star+target 下所有未读 like
option (google.api.http) = { post: "/api/v1/notifications/targets/{target_id}/read" };
}
rpc MarkAllAsRead(MarkAllAsReadRequest) returns (MarkAllAsReadResponse) {
option (google.api.http) = { post: "/api/v1/notifications/read-all" };
}
rpc DeleteNotification(DeleteNotificationRequest) returns (DeleteNotificationResponse) {
option (google.api.http) = { delete: "/api/v1/notifications/{id}" };
}
rpc DeleteByTarget(DeleteByTargetRequest) returns (DeleteByTargetResponse) {
// 仅 type=like 使用：按 target_id 软删该 user+star+target 下所有通知
option (google.api.http) = { delete: "/api/v1/notifications/targets/{target_id}" };
}
}
message Notification {
int64 id = 1; int64 user_id = 2; int64 star_id = 3;
string type = 4; string title = 5; string content = 6;
google.protobuf.Struct data = 7;
bool is_read = 8; int64 created_at = 9; int64 read_at = 10;
// 仅 type=like 聚合时返回
bool aggregated = 11; // 是否聚合卡
int32 total_count = 12; // 聚合总数（aggregated=true 时有值）
repeated ActorPreview actors = 13; // 前 3 个点赞人预览
int64 target_id = 14; // 聚合时用 target_id 替代单条 id
}
message ActorPreview {
int64 user_id = 1; string nickname = 2; string avatar = 3;
int64 liked_at = 4;
}
message CreateNotificationRequest {
int64 user_id = 1; int64 star_id = 2;
string type = 3; string title = 4; string content = 5;
google.protobuf.Struct data = 6;
}
message CreateNotificationResponse { topfans.common.BaseResponse base = 1; int64 id = 2; }
message GetNotificationsRequest {
string type = 1; // like / system / activity / 空=全部
string tab = 2; // today / history / 空=全部
int32 page = 3; int32 page_size = 4;
}
message GetNotificationsResponse {
topfans.common.BaseResponse base = 1;
repeated Notification items = 2; int64 total = 3;
int32 page = 4; int32 page_size = 5;
}
message GetUnreadCountRequest {}
message UnreadCount {
int32 like = 1; int32 system = 2; int32 activity = 3; int32 total = 4;
}
message GetUnreadCountResponse {
topfans.common.BaseResponse base = 1;
UnreadCount counts = 2;
}
message MarkAsReadRequest { int64 id = 1; }
message MarkAsReadResponse { topfans.common.BaseResponse base = 1; }
message MarkAllAsReadRequest { string type = 1; } // 空=全部类型
message MarkAllAsReadResponse { topfans.common.BaseResponse base = 1; int32 affected = 2; }
message DeleteNotificationRequest { int64 id = 1; }
message DeleteNotificationResponse { topfans.common.BaseResponse base = 1; }
message MarkAsReadByTargetRequest { int64 target_id = 1; }
message MarkAsReadByTargetResponse { topfans.common.BaseResponse base = 1; int32 affected = 2; }
message DeleteByTargetRequest { int64 target_id = 1; }
message DeleteByTargetResponse { topfans.common.BaseResponse base = 1; int32 affected = 2; }
5.2 扩展 backend/proto/asset.proto
GetAssetForRPCResponse 增加字段：
message GetAssetForRPCResponse {
topfans.common.BaseResponse base = 1;
int64 asset_id = 2;
int64 owner_uid = 3;
int64 star_id = 4;
int32 status = 5;
bool is_active = 6;
string name = 7; // 新增：藏品名称（用于通知）
string cover_url = 8; // 新增：藏品封面（用于通知）
}
asset provider 实现里从 asset_repository 查出 name/cover_url 填入（DB 已有字段，无表结构变更）。
6. 服务目录结构
6.1 Go 端：新建 backend/services/notificationService/
backend/services/notificationService/
├── main.go # Dubbo 启动、DB 初始化
├── configs/config.yaml # 端口 20010
├── provider/notification_provider.go # RPC 接口实现
├── service/notification_service.go # 业务逻辑 + 事务
├── repository/
│ ├── notification_repository.go # notifications CRUD
│ └── notification_stats_repository.go # 统计 UPSERT
├── model/notification.go # 数据模型
└── client/ # （本服务不调用其他服务，留空目录占位）
6.2 Go 端：social service 新增客户端
backend/services/socialService/client/notification_client.go —— Dubbo RPC 客户端封装，参考 asset_client.go 模式。
6.3 Python 端：admin 后端新增
TopFans-activity-admin/backend/
├── models/models.py # 新增 Notification + NotificationStats
├── crud/notification_crud.py # 创建系统通知 / 广播 / 单发
├── handlers/notification.py # 管理 API
├── handlers/activity.py # 改造：创建后自动 broadcast
└── router/__init__.py # 注册 notification 路由
7. 关键逻辑
7.1 CreateNotification 事务（Go 端）
func (s *NotificationService) CreateNotification(ctx, req) (int64, error) {
// 参数校验
if req.UserId <= 0 || req.StarId <= 0 { return 0, ErrInvalidArgument }
if len(req.Type) == 0 || len(req.Title) == 0 { return 0, ErrInvalidArgument }
if len(req.Title) > 200 || len(req.Content) > 500 { return 0, ErrInvalidArgument }
return s.db.Transaction(func(tx *sql.Tx) (int64, error) {
// 1. INSERT notifications
var id int64
err := tx.QueryRowContext(ctx,
`INSERT INTO notifications (user_id, star_id, type, title, content, data, created_at)
VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id`,
req.UserId, req.StarId, req.Type, req.Title, req.Content, dataJSON, now,
).Scan(&id)
if err != nil { return 0, err }
// 2. UPSERT notification_stats（按 type 加未读数）
var col string
switch req.Type {
case "like": col = "like_unread_count"
case "system": col = "system_unread_count"
case "activity": col = "activity_unread_count"
default: return 0, ErrInvalidArgument
}
_, err = tx.ExecContext(ctx, fmt.Sprintf(`
INSERT INTO notification_stats (user_id, star_id, %s, total_unread_count, updated_at)
VALUES ($1, $2, 1, 1, $3)
ON CONFLICT (user_id, star_id) DO UPDATE SET
%s = notification_stats.%s + 1,
total_unread_count = notification_stats.total_unread_count + 1,
updated_at = $3
`, col, col, col), req.UserId, req.StarId, now)
if err != nil { return 0, err }
return id, nil
})
}
7.2 点赞触发通知（social service）
在 services/socialService/service/asset_like_service.go LikeAsset 方法 line 132 之前插入：
// 构造通知 JSON data
data := map[string]interface{}{
"target_type": "asset",
"target_id": assetID,
"actor_id": userID,
"asset_title": getAssetResp.Name, // 新增字段
"asset_cover": getAssetResp.CoverUrl, // 新增字段
"star_id": starID,
}
// 同步查 actor 信息；actor 信息缺失时降级（仅保留 actor_id）
actorName := strconv.FormatInt(userID, 10)
if actorInfo, err := s.userClient.GetUsersByIDs(ctx, []int64{userID}, starID); err == nil {
if info, exists := actorInfo[userID]; exists && info != nil {
data["actor_name"] = info.Nickname
data["actor_avatar"] = info.Avatar
actorName = info.Nickname
}
} else {
logger.Logger.Warn("failed to fetch actor info for notification, degrade to actor_id",
zap.Int64("user_id", userID),
zap.Error(err),
)
}
// 同步调 notification service，失败仅日志
_, notifErr := s.notificationClient.CreateNotification(ctx, ¬ifPb.CreateNotificationRequest{
UserId: getAssetResp.OwnerUid, // 接收方 = asset owner
StarId: starID,
Type: "like",
Title: "新点赞",
Content: fmt.Sprintf("%s 点赞了你的藏品", actorName),
Data: dataStruct,
})
if notifErr != nil {
logger.Logger.Error("failed to create like notification (like itself succeeded)",
zap.Int64("asset_id", assetID),
zap.Int64("actor_id", userID),
zap.Int64("owner_uid", getAssetResp.OwnerUid),
zap.Error(notifErr),
)
}
// 不影响点赞主路径返回值
7.3 admin 创建系统通知（Python 端）
crud/notification_crud.py 暴露：
def create_system_notification(
db: Session,
*,
title: str,
content: str,
target_type: Literal["user", "star", "all"],
target_value: Optional[int], # user_id 或 star_id
data: Optional[dict],
) -> int: # 返回插入条数
"""事务内：解析目标 + INSERT N 条 notifications + UPSERT N 条 notification_stats"""
handlers/notification.py：
POST /api/v1/admin/notificationsbody:{title, content, target_type, target_value, data}→ 调create_system_notification- 复用 admin JWT 鉴权中间件
7.4 GetNotifications 聚合查询（v1.1 新增）
// 列表查询按 type 分支处理
if req.Type == "like" {
// SQL: 按 target_id 分组聚合
SELECT
target_id,
COUNT(*) AS total_count,
MAX(created_at) AS latest_at,
BOOL_AND(is_read) AS is_read, // 仅当全部已读才算已读
json_agg(actor_preview ORDER BY created_at DESC LIMIT 3) AS actors
FROM notifications n
WHERE user_id = ? AND star_id = ?
AND type = 'like' AND is_deleted = FALSE
AND (tab 为 today/history 时加时间条件)
GROUP BY target_id
ORDER BY latest_at DESC
LIMIT ? OFFSET ?
} else {
// system / activity 走原始 SQL
SELECT * FROM notifications WHERE user_id=? AND star_id=? AND type=? AND is_deleted=FALSE
ORDER BY created_at DESC LIMIT ? OFFSET ?
}
返回的 Notification 字段映射：
id= 第一条原始通知的 id（仅作为列表 key，不用于跳转）target_id=data->>'target_id'（前端用于跳转）aggregated= true，total_count= Nactors= 按时间倒序前 3 个 actortitle模板：{actor_names} 等 N 人赞了你的《{asset_title}》（asset_title 从data取出）is_read= 全部已读时为 truecreated_at= 最新一条的时间
7.5 MarkAsReadByTarget / DeleteByTarget
// 标已读：仅作用于 like 类型
func (s *NotificationService) MarkAsReadByTarget(ctx, userID, starID, targetID) (int32, error) {
affected, err := tx.Exec(`
UPDATE notifications
SET is_read = TRUE, read_at = ?
WHERE user_id=? AND star_id=? AND type='like'
AND data->>'target_id' = ? AND is_read = FALSE AND is_deleted = FALSE
`, now, userID, starID, targetID)
// 同步重置 stats: like_unread_count -= affected
tx.Exec(`
UPDATE notification_stats SET
like_unread_count = GREATEST(0, like_unread_count - ?),
total_unread_count = GREATEST(0, total_unread_count - ?),
updated_at = ?
WHERE user_id=? AND star_id=?
`, affected, affected, now, userID, starID)
return affected, nil
}
// 软删：按 target 软删所有 like
func (s *NotificationService) DeleteByTarget(ctx, userID, starID, targetID) (int32, error) {
// 先查有多少未读，删除后未读数 -N
// 然后 UPDATE is_deleted = TRUE
// 然后 stats -=
}
7.6 admin 活动广播
handlers/activity.py 创建 activity 时支持以下两种触发方式（由 admin 创建接口的 auto_notify 字段控制，默认 true）：
# activity 创建接口 POST /api/v1/admin/minting-activities 入参增加 auto_notify: bool = True
def create_minting_activity(payload, db, current_admin):
activity = MintingActivity(**payload.dict(exclude={"auto_notify"}))
db.add(activity); db.flush() # 先拿 id
if payload.auto_notify:
# 同一事务内广播：INSERT N 条 notifications + UPSERT N 条 stats
fans = resolve_recipient_users(db, target_type="all") # 按需可改为 star
notification_crud.create_activity_notification(
db, activity=activity, recipient_users=fans
)
db.commit() # activity + 通知一起提交
POST /api/v1/admin/minting-activities/{id}/broadcast 手动再触发一次（适用于创建时未广播的场景）。
8. 错误处理
| 场景 | 错误码 | 行为 |
|---|---|---|
| 通知不存在 | NotFound |
404 |
| 通知不属于该 user | PermissionDenied |
403 |
| 参数缺失/超长 | InvalidArgument |
400 |
| DB 异常 | Internal |
ERROR 日志（含 userID/starID/type），500 |
| owner_uid=0 | 跳过 + WARN | 不写通知 |
| 点赞主路径调 notification 失败 | 仅 ERROR 日志 | 点赞仍返回成功 |
| 标已读时通知已删 | 静默成功 | 幂等 |
| 删除通知时通知未读 | 同步 --unread_count |
事务内 |
9. 测试计划
| 层 | 类型 | 关键用例 |
|---|---|---|
| notification repository | 单测 | 列表分页、tab=today/history 时间边界、软删除 |
| notification service | 单测 | 事务回滚（故意触发第二条 SQL 失败）、未读数 +1、+0 边界 |
| social asset_like_service | 单测 | Mock notification client 验证 ① 调用参数正确 ② 客户端报错点赞仍成功 |
| admin notification_crud | 单测 | 三种 target_type 产生的 INSERT 行数正确；与 stats UPSERT 一致 |
| admin activity handler | 单测 | 创建 activity 后自动 broadcast；broadcast 失败 activity 创建仍提交 |
| HTTP | curl 手测 | 走通 list/unread/mark/delete |
10. 集成点回归（CLAUDE.md "自审与回归检查"）
| 改动点 | 回归检查 |
|---|---|
proto/asset.proto 加字段 |
query_graph pattern=callers_of target=GetAssetForRPC 确认所有调用方编译通过 |
socialService/service/asset_like_service.go 改 LikeAsset |
跑既有 asset_like_service 单测；检查 asset_likers 缓存仍正确失效 |
| 新建 notificationService | gateway 路由配置新增 /api/v1/notifications/* → notification service |
| admin models 新增表 | admin 既有启动测试通过；migration 不破坏现有表 |
11. 上线顺序
- 应用迁移
2026_06_16_001_create_notifications.sql - 部署 notification service 二进制（暂时无调用方）
- 部署新版 assetService（带 name/cover_url 扩展）
- 部署新版 socialService（带点赞通知调用）
- 部署新版 admin 后端
- gateway 路由配置：
backend/gateway/router/router.go新增notifications := v1.Group("/notifications")并注册全部 HTTP 路径backend/gateway/config/config.yaml新增 notification service backend 配置（Dubbo tri URL，默认tri://localhost:20010）
- 灰度验证：先点赞 → 再系统通知 → 再活动通知
每步独立发布，任何一步可单独回滚。
12. 验收清单
- 迁移文件已应用，notifications 表存在
- notification service 进程启动正常
- 前端 GET /api/v1/notifications?type=like&tab=today 返回今日点赞
- 点赞触发后，被点赞方收到通知（验证 actor_name、asset_title 都正确）
- 未读数 GET /api/v1/notifications/unread-count 返回正确数字
- admin POST 创建单用户系统通知，前端能查到
- admin 按 star 广播指定 star 的所有粉丝收到
- admin 全量广播所有用户收到
- admin 创建 activity 后自动广播
- admin 手动 broadcast API 工作
- 软删除后列表不再返回
- 标已读后未读数对应类型 -1
- 删除未读通知后未读数 -1
- 通知写入失败时点赞主路径返回成功
- 同一 asset 收到 N 个赞 → 列表只显示 1 张聚合卡，含 total_count=N + 前 3 个点赞人预览
- 点击 like 聚合卡 → 跳到
/pages/asset/detail?id={target_id} - MarkAsReadByTarget → 该 target 下所有未读 like 标已读、未读数对应 -N
- DeleteByTarget → 该 target 下所有 like 软删、未读数对应 -N
- system / activity 通知仍按单条 MarkAsRead(id) / DeleteNotification(id) 操作