diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index bfd6999..b8b349d 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -606,6 +606,48 @@ func (c *ConversationCache) GetConvMaxSeq(ctx context.Context, convID string) (i return val, nil } +// GetConvMaxSeqBatch 批量获取多个会话的 maxSeq(MGet 减少网络往返) +func (c *ConversationCache) GetConvMaxSeqBatch(ctx context.Context, convIDs []string) (map[string]int64, error) { + if len(convIDs) == 0 { + return nil, nil + } + + redisCache, ok := c.cache.(*RedisCache) + if !ok { + result := make(map[string]int64, len(convIDs)) + for _, convID := range convIDs { + if seq, err := c.GetConvMaxSeq(ctx, convID); err == nil { + result[convID] = seq + } + } + return result, nil + } + + rdb := redisCache.GetRedisClient().GetClient() + keys := make([]string, len(convIDs)) + for i, convID := range convIDs { + keys[i] = normalizeKey(MessageSeqKey(convID)) + } + + vals, err := rdb.MGet(ctx, keys...).Result() + if err != nil { + return nil, err + } + + result := make(map[string]int64, len(convIDs)) + for i, val := range vals { + if val == nil { + continue + } + var seq int64 + fmt.Sscanf(fmt.Sprint(val), "%d", &seq) + if seq > 0 { + result[convIDs[i]] = seq + } + } + return result, nil +} + // ComputeUnreadCount 计算单个会话未读数:maxSeq - hasReadSeq func (c *ConversationCache) ComputeUnreadCount(ctx context.Context, convID, userID string) (int64, error) { maxSeq, err := c.GetConvMaxSeq(ctx, convID) @@ -650,128 +692,34 @@ func (c *ConversationCache) ComputeAllUnreadCount(ctx context.Context, userID st return total, nil } -// IncrementUnread 递增用户在某会话的未读数(Redis Hash + 总数计数器) +// IncrementUnread 递增用户在某会话的未读数(已废弃:改用 seq 算术计算) +// Deprecated: Unread count is now computed via arithmetic (maxSeq - readSeq). func (c *ConversationCache) IncrementUnread(ctx context.Context, userID, convID string) error { - hashKey := UnreadHashKey(userID) - totalKey := UnreadTotalKey(userID) - - _, err := c.cache.HIncrBy(ctx, hashKey, convID, 1) - if err != nil { - return fmt.Errorf("hincrby unread failed: %w", err) - } - - _, err = c.cache.Incr(ctx, totalKey) - if err != nil { - return fmt.Errorf("incr unread total failed: %w", err) - } - - _ = c.cache.Expire(ctx, hashKey, c.settings.UnreadTTL*10) - _ = c.cache.Expire(ctx, totalKey, c.settings.UnreadTTL*10) return nil } -// ClearUnread 清零用户在某会话的未读数(原子操作,防止与 IncrementUnread 竞态) +// ClearUnread 清零用户在某会话的未读数(已废弃:改用 seq 算术计算) +// Deprecated: No longer needed. Unread is computed from seq arithmetic. func (c *ConversationCache) ClearUnread(ctx context.Context, userID, convID string) error { - hashKey := UnreadHashKey(userID) - totalKey := UnreadTotalKey(userID) - - redisCache, ok := c.cache.(*RedisCache) - if !ok { - // miniredis/内存模式:保持原逻辑(非原子,但无并发竞态风险) - currentStr, err := c.cache.HGet(ctx, hashKey, convID) - if err != nil { - currentStr = "0" - } - current := int64(0) - if v, parseErr := fmt.Sscanf(currentStr, "%d", ¤t); parseErr != nil || v != 1 { - current = 0 - } - if err := c.cache.HSet(ctx, hashKey, convID, "0"); err != nil { - return fmt.Errorf("hset clear unread failed: %w", err) - } - if current > 0 { - c.cache.IncrBySeq(ctx, totalKey, -current) - } - return nil - } - - rdb := redisCache.GetRedisClient().GetClient() - nHashKey := normalizeKey(hashKey) - nTotalKey := normalizeKey(totalKey) - - script := redis.NewScript(` - local current = tonumber(redis.call('HGET', KEYS[1], ARGV[1])) - if current == nil or current <= 0 then - return 0 - end - redis.call('HSET', KEYS[1], ARGV[1], '0') - local newTotal = tonumber(redis.call('INCRBY', KEYS[2], -current)) - if newTotal < 0 then - redis.call('SET', KEYS[2], '0') - end - return current - `) - _, err := script.Run(ctx, rdb, []string{nHashKey, nTotalKey}, convID).Result() - if err != nil { - return fmt.Errorf("lua clear unread failed: %w", err) - } return nil } -// GetUnreadCountFromHash 从 Redis Hash 获取单个会话未读数 +// GetUnreadCountFromHash 从 Redis Hash 获取单个会话未读数(已废弃) +// Deprecated: Use ComputeUnreadCount instead. func (c *ConversationCache) GetUnreadCountFromHash(ctx context.Context, userID, convID string) (int64, error) { - hashKey := UnreadHashKey(userID) - val, err := c.cache.HGet(ctx, hashKey, convID) - if err != nil { - return 0, err - } - var count int64 - fmt.Sscanf(val, "%d", &count) - return count, nil + return 0, ErrKeyNotFound } -// GetAllUnreadCounts 从 Redis Hash 获取用户所有会话的未读数 +// GetAllUnreadCounts 从 Redis Hash 获取用户所有会话的未读数(已废弃) +// Deprecated: Use ComputeAllUnreadCount instead. func (c *ConversationCache) GetAllUnreadCounts(ctx context.Context, userID string) (map[string]int64, error) { - hashKey := UnreadHashKey(userID) - all, err := c.cache.HGetAll(ctx, hashKey) - if err != nil { - return nil, err - } - result := make(map[string]int64, len(all)) - for k, v := range all { - var count int64 - fmt.Sscanf(v, "%d", &count) - result[k] = count - } - return result, nil + return nil, ErrKeyNotFound } -// GetTotalUnread 从 Redis 获取用户总未读数 +// GetTotalUnread 从 Redis 获取用户总未读数(已废弃) +// Deprecated: Use ComputeAllUnreadCount instead. func (c *ConversationCache) GetTotalUnread(ctx context.Context, userID string) (int64, error) { - totalKey := UnreadTotalKey(userID) - - redisCache, ok := c.cache.(*RedisCache) - if ok { - val, err := redisCache.GetRedisClient().GetClient().Get(ctx, normalizeKey(totalKey)).Int64() - if err != nil { - return 0, nil - } - return val, nil - } - - raw, ok := c.cache.Get(totalKey) - if !ok { - return 0, nil - } - switch v := raw.(type) { - case int64: - return v, nil - case string: - var n int64 - fmt.Sscanf(v, "%d", &n) - return n, nil - } - return 0, nil + return 0, ErrKeyNotFound } // GetNextSeq 获取会话的下一个 seq 值(原子递增) diff --git a/internal/dto/dto.go b/internal/dto/dto.go index 06adc47..9e60901 100644 --- a/internal/dto/dto.go +++ b/internal/dto/dto.go @@ -437,6 +437,23 @@ type MarkReadRequest struct { LastReadSeq int64 `json:"last_read_seq" binding:"required"` // 已读到的seq位置 } +// BatchMarkReadItem 批量标记已读单项 +type BatchMarkReadItem struct { + ConversationID string `json:"conversation_id" binding:"required"` + LastReadSeq int64 `json:"last_read_seq" binding:"required"` +} + +// BatchMarkReadRequest 批量标记已读请求 +type BatchMarkReadRequest struct { + Conversations []BatchMarkReadItem `json:"conversations" binding:"required"` +} + +// BatchMarkReadResponse 批量标记已读响应 +type BatchMarkReadResponse struct { + SuccessCount int `json:"success_count"` + TotalCount int `json:"total_count"` +} + // SetConversationPinnedRequest 设置会话置顶请求 type SetConversationPinnedRequest struct { ConversationID string `json:"conversation_id" binding:"required"` @@ -481,6 +498,18 @@ type ConversationUnreadCountResponse struct { UnreadCount int64 `json:"unread_count"` } +// SyncDataItem 同步数据单项(轻量级,仅包含 seq 和时间) +type SyncDataItem struct { + ConversationID string `json:"id"` + MaxSeq int64 `json:"max_seq"` + LastMessageAt string `json:"last_message_at"` +} + +// SyncDataResponse 同步数据响应 +type SyncDataResponse struct { + Conversations []SyncDataItem `json:"conversations"` +} + // MessageListResponse 消息列表响应 type MessageListResponse struct { Messages []*MessageResponse `json:"messages"` diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index d309a4d..6c37558 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -565,6 +565,26 @@ func (h *MessageHandler) GetUnreadCount(c *gin.Context) { }) } +// HandleGetSyncData 获取同步元数据(轻量级 seq + 时间) +// GET /api/v1/conversations/sync-data +func (h *MessageHandler) HandleGetSyncData(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + items, err := h.chatService.GetSyncData(c.Request.Context(), userID) + if err != nil { + response.InternalServerError(c, "failed to get sync data") + return + } + + response.Success(c, &dto.SyncDataResponse{ + Conversations: items, + }) +} + // GetConversationUnreadCount 获取单个会话的未读数 // GET /api/conversations/:id/unread/count func (h *MessageHandler) GetConversationUnreadCount(c *gin.Context) { @@ -900,6 +920,33 @@ func (h *MessageHandler) HandleMarkRead(c *gin.Context) { response.SuccessWithMessage(c, "marked as read", nil) } +// HandleMarkReadAll 批量标记所有会话已读 +// POST /api/v1/conversations/read-all +func (h *MessageHandler) HandleMarkReadAll(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + var req dto.BatchMarkReadRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, err.Error()) + return + } + + successCount, err := h.chatService.MarkAsReadBatch(c.Request.Context(), userID, req.Conversations) + if err != nil { + response.InternalServerError(c, "failed to mark all as read") + return + } + + response.Success(c, &dto.BatchMarkReadResponse{ + SuccessCount: successCount, + TotalCount: len(req.Conversations), + }) +} + // HandleSetConversationPinned 设置会话置顶 // PUT /api/v1/conversations/:id/pinned func (h *MessageHandler) HandleSetConversationPinned(c *gin.Context) { diff --git a/internal/router/router.go b/internal/router/router.go index 4421c3c..dad91c3 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -426,9 +426,11 @@ func (r *Router) setupRoutes() { conversations.GET("/:id/messages/cursor", r.messageHandler.GetMessagesByCursor) // 消息列表游标分页 conversations.POST("/:id/messages", middleware.RequireVerified(r.userRepo), r.messageHandler.HandleSendMessage) // 发送消息 conversations.POST("/:id/read", r.messageHandler.HandleMarkRead) // 标记已读 + conversations.POST("/read-all", r.messageHandler.HandleMarkReadAll) // 批量标记已读 conversations.PUT("/:id/pinned", r.messageHandler.HandleSetConversationPinned) // 置顶设置 conversations.PUT("/:id/notification_muted", r.messageHandler.HandleSetConversationNotificationMuted) // 免打扰设置 conversations.GET("/unread/count", r.messageHandler.GetUnreadCount) // 未读数 + conversations.GET("/sync-data", r.messageHandler.HandleGetSyncData) // 同步元数据 conversations.POST("/:id/typing", r.messageHandler.HandleTyping) // 输入状态 conversations.DELETE("/:id/self", r.messageHandler.HandleDeleteConversationForSelf) // 删除会话(仅自己) diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index 4500b21..ff86d43 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -41,9 +41,13 @@ type ChatService interface { // 已读管理 MarkAsRead(ctx context.Context, conversationID string, userID string, seq int64) error + MarkAsReadBatch(ctx context.Context, userID string, items []dto.BatchMarkReadItem) (int, error) GetUnreadCount(ctx context.Context, conversationID string, userID string) (int64, error) GetAllUnreadCount(ctx context.Context, userID string) (int64, error) + // 消息同步 + GetSyncData(ctx context.Context, userID string) ([]dto.SyncDataItem, error) + // 消息扩展功能 RecallMessage(ctx context.Context, messageID string, userID string) error DeleteMessage(ctx context.Context, messageID string, userID string) error @@ -388,20 +392,8 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv participants, err := s.getParticipants(ctx, conversationID) if err == nil { // 先更新未读计数器,再推送 WS 事件,确保事件中的 total_unread 已包含新消息 + // 注意:已移除 IncrementUnread,未读数完全由 maxSeq - readSeq 算术计算 if s.conversationCache != nil { - for _, p := range participants { - if p.UserID == senderID { - continue - } - if incrErr := s.conversationCache.IncrementUnread(context.Background(), p.UserID, conversationID); incrErr != nil { - zap.L().Warn("increment unread from redis hash failed", - zap.String("userID", p.UserID), - zap.String("convID", conversationID), - zap.Error(incrErr), - ) - } - } - keys := make([]string, 0, len(participants)*2) for _, p := range participants { keys = append(keys, @@ -602,15 +594,7 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, // 2. 更新 Redis hasReadSeq 缓存(防回退) if s.conversationCache != nil { _ = s.conversationCache.SetUserReadSeq(ctx, conversationID, userID, seq) - // 清除旧式 Hash 未读数缓存 - if clearErr := s.conversationCache.ClearUnread(ctx, userID, conversationID); clearErr != nil { - zap.L().Warn("clear unread from redis hash failed", - zap.String("userID", userID), - zap.String("convID", conversationID), - zap.Error(clearErr), - ) - } - // 失效参与者缓存,下次读取时会从 DB 加载最新数据 + // 清除旧// 失效参与者缓存 s.conversationCache.InvalidateParticipant(conversationID, userID) // 失效未读数缓存(旧 cache-aside 键) s.conversationCache.InvalidateUnreadCount(userID, conversationID) @@ -656,9 +640,66 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, return nil } -// GetUnreadCount 获取指定会话的未读消息数(带缓存) +// MarkAsReadBatch 批量标记多个会话已读 +func (s *chatServiceImpl) MarkAsReadBatch(ctx context.Context, userID string, items []dto.BatchMarkReadItem) (int, error) { + successCount := 0 + + for _, item := range items { + _, err := s.getParticipant(ctx, item.ConversationID, userID) + if err != nil { + continue + } + + if err := s.repo.UpdateLastReadSeq(item.ConversationID, userID, item.LastReadSeq); err != nil { + continue + } + + if s.conversationCache != nil { + _ = s.conversationCache.SetUserReadSeq(ctx, item.ConversationID, userID, item.LastReadSeq) + s.conversationCache.InvalidateParticipant(item.ConversationID, userID) + s.conversationCache.InvalidateUnreadCount(userID, item.ConversationID) + } + + successCount++ + } + + if s.conversationCache != nil { + s.conversationCache.InvalidateConversationList(userID) + } + + // 为每个成功的会话发送 read 通知 + for i := 0; i < successCount && i < len(items); i++ { + item := items[i] + participants, pErr := s.getParticipants(ctx, item.ConversationID) + if pErr == nil { + conv, _ := s.getConversation(ctx, item.ConversationID) + readTargets := []string{userID} + if conv != nil && conv.Type == model.ConversationTypePrivate { + readTargets = make([]string, 0, len(participants)) + for _, p := range participants { + readTargets = append(readTargets, p.UserID) + } + } + s.publishToUsers(readTargets, "message_read", map[string]any{ + "conversation_id": item.ConversationID, + "user_id": userID, + "seq": item.LastReadSeq, + }) + } + } + + // 发送总未读数通知 + if totalUnread, uErr := s.GetAllUnreadCount(ctx, userID); uErr == nil { + s.publishToUsers([]string{userID}, "conversation_unread", map[string]any{ + "total_unread": totalUnread, + }) + } + + return successCount, nil +} + +// GetUnreadCount 获取指定会话的未读消息数(纯算术:maxSeq - readSeq) func (s *chatServiceImpl) GetUnreadCount(ctx context.Context, conversationID string, userID string) (int64, error) { - // 验证用户是否是会话参与者 _, err := s.getParticipant(ctx, conversationID, userID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { @@ -667,45 +708,29 @@ func (s *chatServiceImpl) GetUnreadCount(ctx context.Context, conversationID str return 0, fmt.Errorf("failed to get participant: %w", err) } - // 优先算术计算:maxSeq - hasReadSeq(OpenIM 风格,O(1) 无 DB) if s.conversationCache != nil { if count, err := s.conversationCache.ComputeUnreadCount(ctx, conversationID, userID); err == nil { return count, nil } - // 降级到 Redis Hash - if count, err := s.conversationCache.GetUnreadCountFromHash(ctx, userID, conversationID); err == nil { - return count, nil - } - return s.conversationCache.GetUnreadCount(ctx, userID, conversationID) } - return s.repo.GetUnreadCount(conversationID, userID) } -// GetAllUnreadCount 获取所有会话的未读消息总数 +// GetAllUnreadCount 获取所有会话的未读消息总数(批量 MGet maxSeq,纯算术) func (s *chatServiceImpl) GetAllUnreadCount(ctx context.Context, userID string) (int64, error) { - // 优先算术计算:sum(maxSeq - hasReadSeq)(OpenIM 风格) if s.conversationCache != nil { if convs, _, err := s.GetConversationList(ctx, userID, 1, 200); err == nil && len(convs) > 0 { convIDs := make([]string, len(convs)) - maxSeqs := make(map[string]int64, len(convs)) for i, conv := range convs { convIDs[i] = conv.ID - if maxSeq, err := s.conversationCache.GetConvMaxSeq(ctx, conv.ID); err == nil { - maxSeqs[conv.ID] = maxSeq - } } - // 只有全部会话的 msg_seq 都命中时才用算术结果,避免部分缺失导致低估 - if len(maxSeqs) == len(convIDs) { + maxSeqs, err := s.conversationCache.GetConvMaxSeqBatch(ctx, convIDs) + if err == nil && len(maxSeqs) == len(convIDs) { if total, err := s.conversationCache.ComputeAllUnreadCount(ctx, userID, convIDs, maxSeqs); err == nil { return total, nil } } } - // 降级到 Redis Hash - if total, err := s.conversationCache.GetTotalUnread(ctx, userID); err == nil { - return total, nil - } } return s.repo.GetAllUnreadCount(userID) } @@ -920,23 +945,18 @@ func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conv return message, nil } -// GetUnreadCountBatch 批量获取用户在多个会话中的未读数 +// GetUnreadCountBatch 批量获取用户在多个会话中的未读数(批量 MGet) func (s *chatServiceImpl) GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) { - // 优先算术计算:maxSeq - hasReadSeq(OpenIM 风格) if s.conversationCache != nil && len(convIDs) > 0 { - maxSeqs := make(map[string]int64, len(convIDs)) - for _, convID := range convIDs { - if maxSeq, err := s.conversationCache.GetConvMaxSeq(ctx, convID); err == nil { - maxSeqs[convID] = maxSeq - } - } - if len(maxSeqs) > 0 { + maxSeqs, err := s.conversationCache.GetConvMaxSeqBatch(ctx, convIDs) + if err == nil && len(maxSeqs) > 0 { readSeqs, err := s.conversationCache.GetUserReadSeqs(ctx, userID, convIDs) if err == nil { result := make(map[string]int64, len(convIDs)) for _, convID := range convIDs { maxSeq, ok := maxSeqs[convID] if !ok { + result[convID] = 0 continue } readSeq, hasRead := readSeqs[convID] @@ -950,27 +970,12 @@ func (s *chatServiceImpl) GetUnreadCountBatch(ctx context.Context, userID string result[convID] = unread } } - // 补齐缺失的会话(未命中缓存时归零) - for _, convID := range convIDs { - if _, exists := result[convID]; !exists { - result[convID] = 0 - } - } - // 只有全部会话都算出来了才回返回,否则降级 - allComputed := true - for _, convID := range convIDs { - if _, exists := maxSeqs[convID]; !exists { - allComputed = false - break - } - } - if allComputed { + if len(maxSeqs) == len(convIDs) { return result, nil } } } } - // 降级到 DB return s.repo.GetUnreadCountBatch(ctx, userID, convIDs) } @@ -993,3 +998,33 @@ func (s *chatServiceImpl) GetUserReadSeq(ctx context.Context, conversationID str } return participant.LastReadSeq, nil } + +// GetSyncData 获取用户所有会话的同步元数据(seq + 时间,轻量级) +func (s *chatServiceImpl) GetSyncData(ctx context.Context, userID string) ([]dto.SyncDataItem, error) { + convs, _, err := s.GetConversationList(ctx, userID, 1, 200) + if err != nil { + return nil, fmt.Errorf("failed to get conversation list: %w", err) + } + + if len(convs) == 0 { + return nil, nil + } + + convIDs := make([]string, len(convs)) + for i, conv := range convs { + convIDs[i] = conv.ID + } + + maxSeqs, _ := s.conversationCache.GetConvMaxSeqBatch(ctx, convIDs) + + items := make([]dto.SyncDataItem, 0, len(convs)) + for _, conv := range convs { + item := dto.SyncDataItem{ + ConversationID: conv.ID, + MaxSeq: maxSeqs[conv.ID], + LastMessageAt: conv.UpdatedAt.Format(time.RFC3339), + } + items = append(items, item) + } + return items, nil +}