package service import ( "context" "log/slog" "strings" "git.lzcwy.cn/zhengchiliu/zt-closet-manage-backend/global" "git.lzcwy.cn/zhengchiliu/zt-closet-manage-backend/internal/dto" "git.lzcwy.cn/zhengchiliu/zt-closet-manage-backend/internal/mysqldao" "git.lzcwy.cn/zhengchiliu/zt-closet-manage-backend/internal/mysqlmodel" "git.lzcwy.cn/zhengchiliu/zt-closet-manage-backend/internal/redisdao" "git.lzcwy.cn/zhengchiliu/zt-closet-manage-backend/pkg/errcode" ) // broadcastBatchSize 全量广播每批处理的用户数(分批遍历用户 + 分批插入,避免大事务与内存暴涨)。 const broadcastBatchSize = 500 // NotificationService 通知推送服务。 type NotificationService struct { notificationDao *mysqldao.NotificationDao userDao *mysqldao.UserDao } // NewNotificationService 创建通知服务。 func NewNotificationService() *NotificationService { return &NotificationService{ notificationDao: mysqldao.NewNotificationDao(), userDao: mysqldao.NewUserDao(), } } // Send 向单个用户发送系统通知(写 user_notifications + Redis 未读计数自增)。 func (s *NotificationService) Send(ctx context.Context, adminID int64, request *dto.NotificationSendRequest) *errcode.Error { ctx, span := global.Tracer.Start(ctx, "svc.AdminNotificationSend") defer span.End() // 校验接收用户存在。 if _, err := s.userDao.UserGetByID(ctx, request.UserID); err != nil { if isNotFound(err) { return errcode.UserNotFound } return errcode.DatabaseError } notification := buildSystemNotification(request.UserID, request.Title, request.Content) if err := s.notificationDao.NotificationBatchCreate(ctx, []mysqlmodel.UserNotification{notification}); err != nil { return errcode.DatabaseError } s.incrUnread(ctx, request.UserID) return nil } // Broadcast 全量广播系统通知:分批遍历用户写入,返回发送人数。 func (s *NotificationService) Broadcast(ctx context.Context, adminID int64, request *dto.NotificationBroadcastRequest) (*dto.NotificationBroadcastResponse, *errcode.Error) { ctx, span := global.Tracer.Start(ctx, "svc.AdminNotificationBroadcast") defer span.End() title := strings.TrimSpace(request.Title) content := strings.TrimSpace(request.Content) var sent int64 var lastID int64 for { ids, err := s.userDao.UserIDListAll(ctx, lastID, broadcastBatchSize) if err != nil { slog.ErrorContext(ctx, "broadcast fetch user ids failed", "sent", sent, "error", err) return &dto.NotificationBroadcastResponse{SentCount: sent}, errcode.DatabaseError } if len(ids) == 0 { break } batch := make([]mysqlmodel.UserNotification, 0, len(ids)) for _, uid := range ids { batch = append(batch, buildSystemNotification(uid, title, content)) } if err := s.notificationDao.NotificationBatchCreate(ctx, batch); err != nil { slog.ErrorContext(ctx, "broadcast insert batch failed", "sent", sent, "error", err) return &dto.NotificationBroadcastResponse{SentCount: sent}, errcode.DatabaseError } sent += int64(len(ids)) // best-effort 未读计数自增。 for _, uid := range ids { s.incrUnread(ctx, uid) } lastID = ids[len(ids)-1] if len(ids) < broadcastBatchSize { break } } return &dto.NotificationBroadcastResponse{SentCount: sent}, nil } // incrUnread 未读计数自增(best-effort,失败仅告警,不影响主流程)。 func (s *NotificationService) incrUnread(ctx context.Context, userID int64) { if _, err := redisdao.NotificationUnreadIncr(ctx, userID); err != nil { slog.WarnContext(ctx, "incr notification unread failed", "user_id", userID, "error", err) } } // buildSystemNotification 构造系统通知记录(msg_type=3、sender_id=0、未读)。 func buildSystemNotification(userID int64, title, content string) mysqlmodel.UserNotification { return mysqlmodel.UserNotification{ NotificationUUID: genUUID(), UserID: userID, MsgType: NotificationTypeSystem, SenderID: NotificationSenderSystem, Title: strings.TrimSpace(title), Content: strings.TrimSpace(content), IsRead: NotificationUnread, } }