feat: Phase 2e-1 统一通知中心 + 我的 TabBar 聚合未读红点
后端（notify 模块） - 新增 notify 模块：DAO/Service/Pusher 接口/Controller/Router/CleanupTask - 数据库 DDL：notify_notifications 表 + 3 索引（user+created/user+is_read/user+category） - 11 种 type 枚举（好友/群聊 9 种 + meeting_* 2 种预留）+ 4 种 category - 跨模块集成：contact 3 处 Pusher（friend_request/accepted/rejected） - 跨模块集成：group 6 处 Pusher（invite/join_request/approved/rejected/kicked/role_changed） - WS handler 断线补偿：连接建立即推送 notify.unread.total - 5 REST API（4 用户 + 1 管理员广播）+ 2 WS 事件（notify.new / notify.unread.total） - 30 天已读通知定时清理（未读永久保留） - Provider/Wire 依赖注入（NotifyPusher、NotifyConnectHook、UserInfoResolver 接口） 前端 - 新增 notify 模块：API/Pinia Store（5 分类分页缓存 + 未读数 + WS 事件）/NotifyItem/通知中心主页 - profile 入口：铃铛 badge + 菜单项 badge + 数字显示 - App.vue/login 初始化 notifyStore WS 监听；logout 调用 notifyStore.reset() 清缓存 - 清理 contact.js/group.js 中散落 toast 与冗余 notify.friend.request/group.join.request 处理 - CustomTabBar 新增 hasDot() 聚合指示器：我的 Tab 显示纯红点（无数字）， 当前聚合 notifyStore.unreadTotal，未来可扩展「资料待完善/安全提醒/新版本」等 文档 - 新增 Phase 2e 整体路线图 docs/plans/2026-04-20-phase2e-design.md - 新增 Phase 2e-1 专用设计 docs/plans/2026-04-20-phase2e-1-design.md（§6.4 TabBar 聚合红点） - 新增 Phase 2e-1 实施计划 docs/plans/2026-04-20-phase2e-1-implementation.plan.md - 新增 E2E 验证报告 test-report-phase2e-1-notification.md（含 Playwright MCP 2 个现场 Bug 修复记录） - 更新 docs/progress/CURRENT_STATUS.md、docs/api/README.md、docs/api/frontend/notify.md - 更新 .cursor/rules/project-context.mdc、docs/plans/2026-02-27-echochat-system-design.md 其他 - .gitignore 排除 .playwright-mcp/ MCP 临时快照 架构决策 - 单端 WS 连接：沿用现有 ws.Hub，多端已读同步推迟到 Phase 2f/二期 - 跨模块依赖：contact/group → notify 严格单向（接口注入模式） - 降级策略：Pusher 先入库后推送；WS 失败不回滚入库；入库失败仅 Warn 不影响业务 Playwright MCP 回归（4 类场景全通） - 实时推送（admin 广播 → 1s 内前端自动插入 + 角标 +1） - Deep-link 跳转（好友申请通知 → contact/request 页） - 批量清零（全部已读按钮） - TabBar 聚合红点（有未读亮/全部已读灭）与 notifyStore.unreadTotal 三层同步 Made-with: Cursor
This commit is contained in:
bujinyuan
420
backend/go-service/app/notify/service/notify_service.go
Normal file
420
backend/go-service/app/notify/service/notify_service.go
Normal file
@@ -0,0 +1,420 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
authModel "github.com/echochat/backend/app/auth/model"
|
||||
"github.com/echochat/backend/app/constants"
|
||||
"github.com/echochat/backend/app/dto"
|
||||
"github.com/echochat/backend/app/notify/dao"
|
||||
"github.com/echochat/backend/app/notify/model"
|
||||
"github.com/echochat/backend/pkg/logs"
|
||||
"github.com/echochat/backend/pkg/ws"
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNotificationNotFound = errors.New("通知不存在")
|
||||
)
|
||||
|
||||
// NotifyService 通知业务服务
|
||||
// 负责：
|
||||
// 1. 面向用户的列表 / 未读数 / 已读操作（REST API 层调用）
|
||||
// 2. 面向上游业务的 Pusher 接口实现（持久化 + WS 推送）
|
||||
// 3. 断线重连时的未读补偿推送
|
||||
type NotifyService struct {
|
||||
notifyDAO *dao.NotificationDAO
|
||||
pubsub *ws.PubSub
|
||||
userResolver UserInfoResolver
|
||||
}
|
||||
|
||||
// NewNotifyService 创建 NotifyService 实例
|
||||
func NewNotifyService(
|
||||
notifyDAO *dao.NotificationDAO,
|
||||
pubsub *ws.PubSub,
|
||||
userResolver UserInfoResolver,
|
||||
) *NotifyService {
|
||||
return &NotifyService{
|
||||
notifyDAO: notifyDAO,
|
||||
pubsub: pubsub,
|
||||
userResolver: userResolver,
|
||||
}
|
||||
}
|
||||
|
||||
// ====== 面向用户的查询与操作 ======
|
||||
|
||||
// GetList 分页查询通知列表
|
||||
func (s *NotifyService) GetList(ctx context.Context, userID int64, req *dto.GetNotificationsRequest) (*dto.NotificationListResponse, error) {
|
||||
funcName := "service.notify_service.GetList"
|
||||
|
||||
types := typesByCategory(req.Category)
|
||||
limit := req.Limit
|
||||
if limit <= 0 {
|
||||
limit = 20
|
||||
}
|
||||
if limit > 100 {
|
||||
limit = 100
|
||||
}
|
||||
|
||||
list, err := s.notifyDAO.List(ctx, &dao.ListRequest{
|
||||
UserID: userID,
|
||||
Types: types,
|
||||
IsRead: req.IsRead,
|
||||
BeforeID: req.BeforeID,
|
||||
Limit: limit + 1, // 多查一条判断是否还有下一页
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
hasMore := false
|
||||
if len(list) > limit {
|
||||
hasMore = true
|
||||
list = list[:limit]
|
||||
}
|
||||
|
||||
items := s.toNotificationDTOs(ctx, list)
|
||||
logs.Debug(ctx, funcName, "查询通知列表",
|
||||
zap.Int64("user_id", userID), zap.Int("count", len(items)), zap.Bool("has_more", hasMore))
|
||||
|
||||
return &dto.NotificationListResponse{
|
||||
List: items,
|
||||
HasMore: hasMore,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// GetUnreadCount 查询未读数（总数 + 按分类）
|
||||
func (s *NotifyService) GetUnreadCount(ctx context.Context, userID int64) (*dto.UnreadCountResponse, error) {
|
||||
countMap, err := s.notifyDAO.CountUnreadByType(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
total := 0
|
||||
byCat := map[string]int{
|
||||
constants.NotifyCategoryFriend: 0,
|
||||
constants.NotifyCategoryGroup: 0,
|
||||
constants.NotifyCategoryMeeting: 0,
|
||||
constants.NotifyCategorySystem: 0,
|
||||
}
|
||||
for t, c := range countMap {
|
||||
total += int(c)
|
||||
byCat[constants.NotifyCategoryOfType(t)] += int(c)
|
||||
}
|
||||
return &dto.UnreadCountResponse{
|
||||
Total: total,
|
||||
ByCategory: byCat,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// MarkRead 标记单条通知为已读
|
||||
func (s *NotifyService) MarkRead(ctx context.Context, userID, id int64) error {
|
||||
affected, err := s.notifyDAO.MarkRead(ctx, id, userID)
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return ErrNotificationNotFound
|
||||
}
|
||||
return err
|
||||
}
|
||||
if affected == 0 {
|
||||
return ErrNotificationNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkAllRead 将用户某分类（或全部）通知标记为已读
|
||||
// category 为空或 all 时清空全部未读
|
||||
func (s *NotifyService) MarkAllRead(ctx context.Context, userID int64, category string) (int64, error) {
|
||||
types := typesByCategory(category)
|
||||
return s.notifyDAO.MarkAllRead(ctx, userID, types)
|
||||
}
|
||||
|
||||
// PushUnreadTotalOnConnect 向刚建立连接的用户推送未读总数补偿
|
||||
// 由 WS handler 的 OnConnected 钩子调用（通过 NotifyConnectHook 接口适配）
|
||||
func (s *NotifyService) PushUnreadTotalOnConnect(ctx context.Context, userID int64) {
|
||||
funcName := "service.notify_service.PushUnreadTotalOnConnect"
|
||||
resp, err := s.GetUnreadCount(ctx, userID)
|
||||
if err != nil {
|
||||
logs.Warn(ctx, funcName, "查询未读数失败", zap.Int64("user_id", userID), zap.Error(err))
|
||||
return
|
||||
}
|
||||
if resp.Total == 0 {
|
||||
return
|
||||
}
|
||||
s.publishWS(ctx, userID, constants.WSEventNotifyUnreadTotal, map[string]interface{}{
|
||||
"total": resp.Total,
|
||||
"by_category": resp.ByCategory,
|
||||
})
|
||||
}
|
||||
|
||||
// ====== Pusher 接口实现 ======
|
||||
|
||||
// Push 写入通知并推送 WS（单用户）
|
||||
// 实现策略：
|
||||
// 1. 同步入库（保证持久化）
|
||||
// 2. 异步 WS 推送（避免阻塞上游业务事务）
|
||||
// 3. 任一步失败不回滚另一步，全部仅记录 Warn 日志
|
||||
func (s *NotifyService) Push(ctx context.Context, payload *PushPayload) {
|
||||
funcName := "service.notify_service.Push"
|
||||
if payload == nil || payload.UserID <= 0 || payload.Type == "" {
|
||||
logs.Warn(ctx, funcName, "无效的推送载荷", zap.Any("payload", payload))
|
||||
return
|
||||
}
|
||||
|
||||
n := s.buildModel(payload)
|
||||
if err := s.notifyDAO.Create(ctx, n); err != nil {
|
||||
logs.Warn(ctx, funcName, "通知入库失败，放弃 WS 推送",
|
||||
zap.Int64("user_id", payload.UserID), zap.String("type", payload.Type), zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
go s.safePushWS(n, payload)
|
||||
}
|
||||
|
||||
// PushBatch 批量写入通知并推送 WS
|
||||
// 适用于管理员广播 / 群操作批量通知场景
|
||||
// 优化：一次 BatchCreate 入库，然后分别 WS 推送
|
||||
func (s *NotifyService) PushBatch(ctx context.Context, payloads []*PushPayload) {
|
||||
funcName := "service.notify_service.PushBatch"
|
||||
if len(payloads) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
models := make([]*model.Notification, 0, len(payloads))
|
||||
valid := make([]*PushPayload, 0, len(payloads))
|
||||
for _, p := range payloads {
|
||||
if p == nil || p.UserID <= 0 || p.Type == "" {
|
||||
continue
|
||||
}
|
||||
models = append(models, s.buildModel(p))
|
||||
valid = append(valid, p)
|
||||
}
|
||||
|
||||
if err := s.notifyDAO.BatchCreate(ctx, models); err != nil {
|
||||
logs.Warn(ctx, funcName, "通知批量入库失败，放弃 WS 推送",
|
||||
zap.Int("count", len(models)), zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
// 入库成功后逐条异步推送
|
||||
for i, n := range models {
|
||||
go s.safePushWS(n, valid[i])
|
||||
}
|
||||
}
|
||||
|
||||
// ====== 内部辅助 ======
|
||||
|
||||
// buildModel 由 PushPayload 构造 Notification 持久化模型
|
||||
// 空 Title 使用类型默认文案；空 TargetType 按业务类型映射为默认值
|
||||
func (s *NotifyService) buildModel(p *PushPayload) *model.Notification {
|
||||
title := p.Title
|
||||
if title == "" {
|
||||
if zh, ok := constants.NotifyTypeMap[p.Type]; ok {
|
||||
title = zh
|
||||
}
|
||||
}
|
||||
targetType := p.TargetType
|
||||
if targetType == "" {
|
||||
targetType = defaultTargetType(p.Type)
|
||||
}
|
||||
return &model.Notification{
|
||||
UserID: p.UserID,
|
||||
Type: p.Type,
|
||||
Title: title,
|
||||
Content: p.Content,
|
||||
Extra: marshalExtra(p.Extra),
|
||||
ActorID: p.ActorID,
|
||||
TargetType: targetType,
|
||||
TargetID: p.TargetID,
|
||||
IsRead: false,
|
||||
}
|
||||
}
|
||||
|
||||
// safePushWS 序列化并通过 Redis Pub/Sub 推送 notify.new 事件
|
||||
// 失败仅记录 Warn，不影响持久化的通知记录
|
||||
func (s *NotifyService) safePushWS(n *model.Notification, p *PushPayload) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
logs.Warn(context.Background(), "service.notify_service.safePushWS",
|
||||
"推送协程 panic", zap.Any("panic", r))
|
||||
}
|
||||
}()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
dtoItem := s.toNotificationDTO(ctx, n)
|
||||
s.publishWS(ctx, n.UserID, constants.WSEventNotifyNew, dtoItem)
|
||||
}
|
||||
|
||||
// publishWS 序列化推送消息并发布到用户频道
|
||||
func (s *NotifyService) publishWS(ctx context.Context, userID int64, event string, data interface{}) {
|
||||
if s.pubsub == nil {
|
||||
return
|
||||
}
|
||||
push := ws.NewPushMessage(event, data)
|
||||
bytes, err := ws.MarshalPush(push)
|
||||
if err != nil {
|
||||
logs.Warn(ctx, "service.notify_service.publishWS", "序列化推送失败",
|
||||
zap.String("event", event), zap.Error(err))
|
||||
return
|
||||
}
|
||||
if err := s.pubsub.Publish(ctx, userID, bytes); err != nil {
|
||||
logs.Warn(ctx, "service.notify_service.publishWS", "WS 推送失败",
|
||||
zap.Int64("user_id", userID), zap.String("event", event), zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// toNotificationDTO 将 model 转为对外 DTO，补全 actor 信息
|
||||
func (s *NotifyService) toNotificationDTO(ctx context.Context, n *model.Notification) dto.NotificationDTO {
|
||||
item := dto.NotificationDTO{
|
||||
ID: n.ID,
|
||||
Type: n.Type,
|
||||
Category: constants.NotifyCategoryOfType(n.Type),
|
||||
Title: n.Title,
|
||||
Content: n.Content,
|
||||
Extra: n.Extra,
|
||||
ActorID: n.ActorID,
|
||||
TargetType: n.TargetType,
|
||||
TargetID: n.TargetID,
|
||||
IsRead: n.IsRead,
|
||||
CreatedAt: n.CreatedAt.Format("2006-01-02 15:04:05"),
|
||||
}
|
||||
if n.ReadAt != nil {
|
||||
item.ReadAt = n.ReadAt.Format("2006-01-02 15:04:05")
|
||||
}
|
||||
|
||||
if n.ActorID != nil && *n.ActorID > 0 && s.userResolver != nil {
|
||||
users, err := s.userResolver.GetUsersByIDs(ctx, []int64{*n.ActorID})
|
||||
if err == nil && len(users) > 0 {
|
||||
item.ActorName = users[0].Nickname
|
||||
item.ActorAvatar = users[0].Avatar
|
||||
}
|
||||
}
|
||||
return item
|
||||
}
|
||||
|
||||
// toNotificationDTOs 批量转换 DTO 并一次性补全 actor 信息，避免 N+1 查询
|
||||
func (s *NotifyService) toNotificationDTOs(ctx context.Context, list []model.Notification) []dto.NotificationDTO {
|
||||
items := make([]dto.NotificationDTO, 0, len(list))
|
||||
if len(list) == 0 {
|
||||
return items
|
||||
}
|
||||
|
||||
actorSet := make(map[int64]struct{})
|
||||
for i := range list {
|
||||
if list[i].ActorID != nil && *list[i].ActorID > 0 {
|
||||
actorSet[*list[i].ActorID] = struct{}{}
|
||||
}
|
||||
}
|
||||
userMap := make(map[int64]*authModel.User)
|
||||
if len(actorSet) > 0 && s.userResolver != nil {
|
||||
ids := make([]int64, 0, len(actorSet))
|
||||
for id := range actorSet {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
users, err := s.userResolver.GetUsersByIDs(ctx, ids)
|
||||
if err == nil {
|
||||
for i := range users {
|
||||
userMap[users[i].ID] = &users[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for i := range list {
|
||||
n := &list[i]
|
||||
item := dto.NotificationDTO{
|
||||
ID: n.ID,
|
||||
Type: n.Type,
|
||||
Category: constants.NotifyCategoryOfType(n.Type),
|
||||
Title: n.Title,
|
||||
Content: n.Content,
|
||||
Extra: n.Extra,
|
||||
ActorID: n.ActorID,
|
||||
TargetType: n.TargetType,
|
||||
TargetID: n.TargetID,
|
||||
IsRead: n.IsRead,
|
||||
CreatedAt: n.CreatedAt.Format("2006-01-02 15:04:05"),
|
||||
}
|
||||
if n.ReadAt != nil {
|
||||
item.ReadAt = n.ReadAt.Format("2006-01-02 15:04:05")
|
||||
}
|
||||
if n.ActorID != nil {
|
||||
if u, ok := userMap[*n.ActorID]; ok {
|
||||
item.ActorName = u.Nickname
|
||||
item.ActorAvatar = u.Avatar
|
||||
}
|
||||
}
|
||||
items = append(items, item)
|
||||
}
|
||||
return items
|
||||
}
|
||||
|
||||
// typesByCategory 把前端的 category 参数转换为 type 列表
|
||||
// all 或空字符串返回 nil（不按类型过滤）
|
||||
func typesByCategory(category string) []string {
|
||||
if category == "" || category == constants.NotifyCategoryAll {
|
||||
return nil
|
||||
}
|
||||
if list, ok := constants.NotifyTypesByCategory[category]; ok {
|
||||
return list
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// defaultTargetType 根据通知类型推导默认业务对象类型
|
||||
// 用于 PushPayload.TargetType 为空时的兜底
|
||||
func defaultTargetType(notifyType string) string {
|
||||
switch constants.NotifyCategoryOfType(notifyType) {
|
||||
case constants.NotifyCategoryFriend:
|
||||
return constants.NotifyTargetUser
|
||||
case constants.NotifyCategoryGroup:
|
||||
return constants.NotifyTargetGroup
|
||||
case constants.NotifyCategoryMeeting:
|
||||
return constants.NotifyTargetMeeting
|
||||
default:
|
||||
return constants.NotifyTargetSystem
|
||||
}
|
||||
}
|
||||
|
||||
// ====== 管理员广播 ======
|
||||
|
||||
// Broadcast 向所有正常用户发送系统公告
|
||||
// 查询活跃用户 ID → 构造 payload → 批量入库 + 推送
|
||||
// 返回受影响用户数
|
||||
func (s *NotifyService) Broadcast(ctx context.Context, operatorID int64, req *dto.BroadcastNotificationRequest) (int, error) {
|
||||
funcName := "service.notify_service.Broadcast"
|
||||
logs.Info(ctx, funcName, "管理员广播",
|
||||
zap.Int64("operator_id", operatorID), zap.String("title", req.Title))
|
||||
|
||||
userIDs, err := s.notifyDAO.ListAllActiveUserIDs(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(userIDs) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
targetType := req.TargetType
|
||||
if targetType == "" {
|
||||
targetType = constants.NotifyTargetSystem
|
||||
}
|
||||
|
||||
payloads := make([]*PushPayload, 0, len(userIDs))
|
||||
for _, uid := range userIDs {
|
||||
payloads = append(payloads, &PushPayload{
|
||||
UserID: uid,
|
||||
Type: constants.NotifyTypeSystemBroadcast,
|
||||
Title: req.Title,
|
||||
Content: req.Content,
|
||||
ActorID: &operatorID,
|
||||
TargetType: targetType,
|
||||
TargetID: req.TargetID,
|
||||
Extra: req.Extra,
|
||||
})
|
||||
}
|
||||
s.PushBatch(ctx, payloads)
|
||||
return len(payloads), nil
|
||||
}
|
||||
Reference in New Issue
Block a user