diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index 94ad026..31e71d7 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -484,6 +484,172 @@ func (c *ConversationCache) InvalidateUnreadCount(userID, convID string) { c.cache.Delete(UnreadDetailKey(userID, convID)) } +// ============================================================ +// 已读位置缓存（OpenIM 风格 hasReadSeq） +// ============================================================ + +// readSeqExpire 已读位置缓存 TTL +const readSeqExpire = 30 * 24 * time.Hour + +// SetUserReadSeq 设置用户在某会话的已读位置（防回退：仅当新值大于当前值时更新） +func (c *ConversationCache) SetUserReadSeq(ctx context.Context, convID, userID string, seq int64) error { + key := UserReadSeqKey(convID, userID) + + redisCache, ok := c.cache.(*RedisCache) + if !ok { + // 非 Redis 模式，直接设置 + c.cache.Set(key, seq, readSeqExpire) + return nil + } + + rdb := redisCache.GetRedisClient().GetClient() + nKey := normalizeKey(key) + // Lua 脚本：仅当新 seq > 当前值时才更新，防回退 + script := redis.NewScript(` + local current = tonumber(redis.call('GET', KEYS[1])) + if current and current >= tonumber(ARGV[1]) then + return 0 + end + redis.call('SET', KEYS[1], ARGV[1]) + redis.call('EXPIRE', KEYS[1], ARGV[2]) + return 1 + `) + _, err := script.Run(ctx, rdb, []string{nKey}, seq, int64(readSeqExpire.Seconds())).Result() + return err +} + +// GetUserReadSeq 获取用户在某会话的已读位置 +func (c *ConversationCache) GetUserReadSeq(ctx context.Context, convID, userID string) (int64, error) { + key := UserReadSeqKey(convID, userID) + + redisCache, ok := c.cache.(*RedisCache) + if !ok { + raw, ok := c.cache.Get(key) + if !ok { + return 0, ErrKeyNotFound + } + switch v := raw.(type) { + case int64: + return v, nil + case string: + var n int64 + fmt.Sscanf(v, "%d", &n) + return n, nil + } + return 0, ErrKeyNotFound + } + + rdb := redisCache.GetRedisClient().GetClient() + nKey := normalizeKey(key) + val, err := rdb.Get(ctx, nKey).Int64() + if err != nil { + return 0, ErrKeyNotFound + } + return val, nil +} + +// GetUserReadSeqs 批量获取用户在多个会话的已读位置 +func (c *ConversationCache) GetUserReadSeqs(ctx context.Context, userID string, 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.GetUserReadSeq(ctx, convID, userID); 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(UserReadSeqKey(convID, userID)) + } + + 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) + result[convIDs[i]] = seq + } + return result, nil +} + +// GetConvMaxSeq 获取会话的最大 seq（复用 msg_seq 计数器，读取当前值） +func (c *ConversationCache) GetConvMaxSeq(ctx context.Context, convID string) (int64, error) { + seqKey := MessageSeqKey(convID) + + redisCache, ok := c.cache.(*RedisCache) + if !ok { + return 0, ErrKeyNotFound + } + + rdb := redisCache.GetRedisClient().GetClient() + nKey := normalizeKey(seqKey) + val, err := rdb.Get(ctx, nKey).Int64() + if err != nil { + return 0, ErrKeyNotFound + } + return val, nil +} + +// ComputeUnreadCount 计算单个会话未读数：maxSeq - hasReadSeq +func (c *ConversationCache) ComputeUnreadCount(ctx context.Context, convID, userID string) (int64, error) { + maxSeq, err := c.GetConvMaxSeq(ctx, convID) + if err != nil { + return 0, err + } + readSeq, err := c.GetUserReadSeq(ctx, convID, userID) + if err != nil { + // 未读到已读位置，视为 0，返回 maxSeq + return maxSeq, nil + } + unread := maxSeq - readSeq + if unread < 0 { + return 0, nil + } + return unread, nil +} + +// ComputeAllUnreadCount 计算所有会话未读总数：sum(maxSeq - hasReadSeq) +func (c *ConversationCache) ComputeAllUnreadCount(ctx context.Context, userID string, convIDs []string, maxSeqs map[string]int64) (int64, error) { + readSeqs, err := c.GetUserReadSeqs(ctx, userID, convIDs) + if err != nil { + return 0, err + } + + var total int64 + for _, convID := range convIDs { + maxSeq, ok := maxSeqs[convID] + if !ok { + continue + } + readSeq, hasRead := readSeqs[convID] + if !hasRead { + total += maxSeq + continue + } + unread := maxSeq - readSeq + if unread > 0 { + total += unread + } + } + return total, nil +} + // IncrementUnread 递增用户在某会话的未读数（Redis Hash + 总数计数器） func (c *ConversationCache) IncrementUnread(ctx context.Context, userID, convID string) error { hashKey := UnreadHashKey(userID) diff --git a/internal/cache/keys.go b/internal/cache/keys.go index 8ee7d0b..38e0756 100644 --- a/internal/cache/keys.go +++ b/internal/cache/keys.go @@ -29,6 +29,9 @@ const ( PrefixUnreadHash = "unread:hash" PrefixUnreadTotal = "unread:total" + // 已读位置相关（OpenIM 风格：缓存 hasReadSeq，用于 O(1) 未读数计算） + PrefixUserReadSeq = "user_read_seq" + // 用户相关 PrefixUserInfo = "users:info" PrefixUserMe = "users:me" @@ -244,3 +247,13 @@ func MessageIdempotentKey(senderID, clientMsgID string) string { func PushDedupKey(userID, messageID string) string { return fmt.Sprintf("push_dedup:%s:%s", userID, messageID) } + +// UserReadSeqKey 用户已读位置缓存键（OpenIM 风格 SEQ_USER_READ:{convID}:{userID}） +func UserReadSeqKey(convID, userID string) string { + return fmt.Sprintf("%s:%s:%s", PrefixUserReadSeq, convID, userID) +} + +// ConvMaxSeqKey 会话最大 seq 缓存键 +func ConvMaxSeqKey(convID string) string { + return fmt.Sprintf("%s:%s", keyPrefixMsgSeq, convID) +} diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 869560f..d309a4d 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -758,7 +758,7 @@ func (h *MessageHandler) HandleGetConversation(c *gin.Context) { // 获取参与者信息 participants, _ := h.getConversationParticipants(c.Request.Context(), conversationID, userID) - // 获取当前用户的已读位置 + // 获取当前用户的已读位置（优先从 Redis 缓存读取） myLastReadSeq := int64(0) isPinned := false notificationMuted := false @@ -772,7 +772,7 @@ func (h *MessageHandler) HandleGetConversation(c *gin.Context) { } } - // 获取对方用户的已读位置 + // 获取对方用户的已读位置（优先从 Redis 缓存读取） otherLastReadSeq := int64(0) for _, p := range allParticipants { if p.UserID != userID { diff --git a/internal/pkg/jpush/client.go b/internal/pkg/jpush/client.go index 60b57e5..bc401ac 100644 --- a/internal/pkg/jpush/client.go +++ b/internal/pkg/jpush/client.go @@ -93,8 +93,8 @@ type IOSNotif struct { Alert any `json:"alert,omitempty"` Sound any `json:"sound,omitempty"` Badge string `json:"badge,omitempty"` - MutableContent bool `json:"mutable-content,omitempty"` - ContentAvailable bool `json:"content-available,omitempty"` + MutableContent *bool `json:"mutable-content,omitempty"` + ContentAvailable *bool `json:"content-available,omitempty"` Category string `json:"category,omitempty"` ThreadID string `json:"thread-id,omitempty"` Extras map[string]any `json:"extras,omitempty"` diff --git a/internal/repository/message_repo.go b/internal/repository/message_repo.go index 7fc344e..f18f719 100644 --- a/internal/repository/message_repo.go +++ b/internal/repository/message_repo.go @@ -278,27 +278,41 @@ func (r *messageRepository) GetParticipant(conversationID string, userID string) return &participant, nil } -// UpdateLastReadSeq 更新已读位置 +// UpdateLastReadSeq 更新已读位置（防回退：仅当新值大于当前值时才更新） // userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致 func (r *messageRepository) UpdateLastReadSeq(conversationID string, userID string, lastReadSeq int64) error { + // 使用条件更新防止已读位置回退 result := r.db.Model(&model.ConversationParticipant{}). - Where("conversation_id = ? AND user_id = ?", conversationID, userID). + Where("conversation_id = ? AND user_id = ? AND last_read_seq < ?", conversationID, userID, lastReadSeq). Update("last_read_seq", lastReadSeq) if result.Error != nil { return result.Error } - // 如果没有更新任何记录，说明参与者记录不存在，需要插入 + // 如果没有更新任何记录，可能是参与者记录不存在，尝试插入 if result.RowsAffected == 0 { - // 尝试插入新记录（跨数据库 upsert） - err := r.db.Clauses(clause.OnConflict{ + // 检查是否已有记录但 seq 不需要更新 + var existingSeq int64 + err := r.db.Model(&model.ConversationParticipant{}). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + Select("last_read_seq").Scan(&existingSeq).Error + if err == nil && existingSeq >= lastReadSeq { + // 已有记录且 seq 不需要回退，直接返回成功 + return nil + } + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return err + } + + // 参与者记录不存在，插入新记录 + err = r.db.Clauses(clause.OnConflict{ Columns: []clause.Column{ {Name: "conversation_id"}, {Name: "user_id"}, }, DoUpdates: clause.Assignments(map[string]any{ - "last_read_seq": lastReadSeq, + "last_read_seq": gorm.Expr("CASE WHEN last_read_seq < ? THEN ? ELSE last_read_seq END", lastReadSeq, lastReadSeq), "updated_at": gorm.Expr("CURRENT_TIMESTAMP"), }), }).Create(&model.ConversationParticipant{ diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index 460bbde..eb21c44 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -60,6 +60,9 @@ type ChatService interface { // 批量查询（解决 N+1 问题） GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) + + // 已读位置（OpenIM 风格缓存） + GetUserReadSeq(ctx context.Context, conversationID string, userID string) (int64, error) } // chatServiceImpl 聊天服务实现 @@ -355,6 +358,12 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv return nil, fmt.Errorf("failed to save message: %w", err) } + // 发送者自动已读：将发送者的 lastReadSeq 更新到消息 seq + _ = s.repo.UpdateLastReadSeq(conversationID, senderID, message.Seq) + if s.conversationCache != nil { + _ = s.conversationCache.SetUserReadSeq(context.Background(), conversationID, senderID, message.Seq) + } + // 新消息会改变分页结果，先失效分页缓存，避免读到旧列表 if s.conversationCache != nil { s.conversationCache.InvalidateMessagePages(conversationID) @@ -404,7 +413,7 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv if p.UserID == senderID { continue } - if totalUnread, uErr := s.repo.GetAllUnreadCount(p.UserID); uErr == nil { + if totalUnread, uErr := s.GetAllUnreadCount(ctx, p.UserID); uErr == nil { s.publishToUsers([]string{p.UserID}, "conversation_unread", map[string]any{ "conversation_id": conversationID, "total_unread": totalUnread, @@ -590,8 +599,10 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, return fmt.Errorf("failed to update last read seq: %w", err) } - // 2. DB 写入成功后，清除 Redis 未读数 + // 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), @@ -611,17 +622,23 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, if pErr == nil { detailType := "private" groupID := "" - if conv, convErr := s.getConversation(ctx, conversationID); convErr == nil && conv.Type == model.ConversationTypeGroup { + conv, _ := s.getConversation(ctx, conversationID) + if conv != nil && conv.Type == model.ConversationTypeGroup { detailType = "group" if conv.GroupID != nil { groupID = *conv.GroupID } } - targetIDs := make([]string, 0, len(participants)) - for _, p := range participants { - targetIDs = append(targetIDs, p.UserID) + + // 私聊：通知所有参与者（含对方）；群聊：只通知自己 + 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(targetIDs, "message_read", map[string]any{ + s.publishToUsers(readTargets, "message_read", map[string]any{ "detail_type": detailType, "conversation_id": conversationID, "group_id": groupID, @@ -629,7 +646,7 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, "seq": seq, }) } - if totalUnread, uErr := s.repo.GetAllUnreadCount(userID); uErr == nil { + if totalUnread, uErr := s.GetAllUnreadCount(ctx, userID); uErr == nil { s.publishToUsers([]string{userID}, "conversation_unread", map[string]any{ "conversation_id": conversationID, "total_unread": totalUnread, @@ -650,13 +667,15 @@ func (s *chatServiceImpl) GetUnreadCount(ctx context.Context, conversationID str return 0, fmt.Errorf("failed to get participant: %w", err) } - // 优先从 Redis Hash 读取 + // 优先算术计算：maxSeq - hasReadSeq（OpenIM 风格，O(1) 无 DB） if s.conversationCache != nil { - count, err := s.conversationCache.GetUnreadCountFromHash(ctx, userID, conversationID) - if err == 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) } @@ -665,8 +684,24 @@ func (s *chatServiceImpl) GetUnreadCount(ctx context.Context, conversationID str // GetAllUnreadCount 获取所有会话的未读消息总数 func (s *chatServiceImpl) GetAllUnreadCount(ctx context.Context, userID string) (int64, error) { - // 优先从 Redis Hash 读取 + // 优先算术计算：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 + } + } + if len(maxSeqs) > 0 { + 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 } @@ -886,6 +921,55 @@ func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conv // GetUnreadCountBatch 批量获取用户在多个会话中的未读数 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 { + 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 { + continue + } + readSeq, hasRead := readSeqs[convID] + if !hasRead { + result[convID] = maxSeq + } else { + unread := maxSeq - readSeq + if unread < 0 { + unread = 0 + } + 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 { + return result, nil + } + } + } + } + // 降级到 DB return s.repo.GetUnreadCountBatch(ctx, userID, convIDs) } @@ -893,3 +977,18 @@ func (s *chatServiceImpl) GetUnreadCountBatch(ctx context.Context, userID string func (s *chatServiceImpl) GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) { return s.repo.GetLastMessagesBatch(ctx, convIDs) } + +// GetUserReadSeq 获取用户在某会话的已读位置（优先从 Redis 缓存读取） +func (s *chatServiceImpl) GetUserReadSeq(ctx context.Context, conversationID string, userID string) (int64, error) { + if s.conversationCache != nil { + if seq, err := s.conversationCache.GetUserReadSeq(ctx, conversationID, userID); err == nil { + return seq, nil + } + } + // 降级到 DB + participant, err := s.getParticipant(ctx, conversationID, userID) + if err != nil { + return 0, err + } + return participant.LastReadSeq, nil +}