diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index 0c9b095..d654074 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -221,6 +221,7 @@ type MessageRepository interface { GetMessages(convID string, page, pageSize int) ([]*model.Message, int64, error) GetMessagesAfterSeq(convID string, afterSeq int64, limit int) ([]*model.Message, error) GetMessagesBeforeSeq(convID string, beforeSeq int64, limit int) ([]*model.Message, error) + GetNextSeq(convID string) (int64, error) } // ============================================================ @@ -588,22 +589,29 @@ func (c *ConversationCache) GetUserReadSeqs(ctx context.Context, userID string, return result, nil } -// GetConvMaxSeq 获取会话的最大 seq(复用 msg_seq 计数器,读取当前值) +// GetConvMaxSeq 获取会话的最大 seq(优先 Redis,降级 DB) func (c *ConversationCache) GetConvMaxSeq(ctx context.Context, convID string) (int64, error) { seqKey := MessageSeqKey(convID) redisCache, ok := c.cache.(*RedisCache) - if !ok { - return 0, ErrKeyNotFound + if ok { + rdb := redisCache.GetRedisClient().GetClient() + nKey := normalizeKey(seqKey) + val, err := rdb.Get(ctx, nKey).Int64() + if err == nil { + return val, nil + } } - rdb := redisCache.GetRedisClient().GetClient() - nKey := normalizeKey(seqKey) - val, err := rdb.Get(ctx, nKey).Int64() - if err != nil { - return 0, ErrKeyNotFound + // Redis 未命中或不可用,降级到 DB + if c.repo != nil { + conv, err := c.repo.GetConversationByID(convID) + if err == nil && conv.LastSeq > 0 { + return conv.LastSeq, nil + } } - return val, nil + + return 0, ErrKeyNotFound } // GetConvMaxSeqBatch 批量获取多个会话的 maxSeq(MGet 减少网络往返) @@ -637,6 +645,12 @@ func (c *ConversationCache) GetConvMaxSeqBatch(ctx context.Context, convIDs []st result := make(map[string]int64, len(convIDs)) for i, val := range vals { if val == nil { + // Redis 未命中,降级到 DB + if c.repo != nil { + if conv, err := c.repo.GetConversationByID(convIDs[i]); err == nil && conv.LastSeq > 0 { + result[convIDs[i]] = conv.LastSeq + } + } continue } var seq int64 @@ -699,10 +713,17 @@ func (c *ConversationCache) GetNextSeq(ctx context.Context, convID string) (int6 newSeq, err := c.cache.Incr(ctx, seqKey) if err != nil { + // Redis 不可用时降级到 DB,拿到值后回写 Redis 以便下次命中缓存 + if c.msgRepo != nil { + if dbSeq, dbErr := c.msgRepo.GetNextSeq(convID); dbErr == nil { + c.cache.Set(seqKey, dbSeq, c.settings.SeqTTL) + return dbSeq, nil + } + } return 0, fmt.Errorf("redis incr seq failed: %w", err) } - // 首次调用时 Redis key 不存在,INCR 从 0 返回 1 + // 首次调用时 Redis key 不存在,INCR 从 0 返回 1,需要从 DB 补齐 if newSeq == 1 && c.repo != nil { conv, err := c.repo.GetConversationByID(convID) if err != nil { @@ -710,17 +731,17 @@ func (c *ConversationCache) GetNextSeq(ctx context.Context, convID string) (int6 } if conv.LastSeq > 0 { - // INCR 已返回 1,需要补齐 delta 使总值为 last_seq + 1 alignedSeq, err := c.cache.IncrBySeq(ctx, seqKey, conv.LastSeq) if err != nil { return conv.LastSeq + 1, nil // 降级返回 DB 值 } newSeq = alignedSeq } - - _ = c.cache.Expire(ctx, seqKey, c.settings.SeqTTL) } + // 每次 INCR 后刷新 TTL,防止 seq 计数器过期导致从头开始 + _ = c.cache.Expire(ctx, seqKey, c.settings.SeqTTL) + return newSeq, nil } diff --git a/internal/cache/layered_cache.go b/internal/cache/layered_cache.go index 3bc160f..284c6c0 100644 --- a/internal/cache/layered_cache.go +++ b/internal/cache/layered_cache.go @@ -352,8 +352,13 @@ func (c *LayeredCache) ZCard(ctx context.Context, key string) (int64, error) { return c.redis.ZCard(ctx, key) } +// IsEnabled 返回 Redis 是否可用 +func (c *LayeredCache) IsEnabled() bool { + return c.enabled && c.redis != nil +} + func (c *LayeredCache) Incr(ctx context.Context, key string) (int64, error) { - if !c.enabled || c.redis == nil { + if !c.IsEnabled() { return 0, ErrKeyNotFound } c.local.Delete(key) diff --git a/internal/cache/repository_adapter.go b/internal/cache/repository_adapter.go index fe5c723..569b793 100644 --- a/internal/cache/repository_adapter.go +++ b/internal/cache/repository_adapter.go @@ -74,3 +74,8 @@ func (a *MessageRepositoryAdapter) CreateMessage(msg *model.Message) error { func (a *MessageRepositoryAdapter) UpdateConversationLastSeq(convID string, seq int64) error { return a.repo.UpdateConversationLastSeq(convID, seq) } + +// GetNextSeq 实现 MessageRepository 接口 +func (a *MessageRepositoryAdapter) GetNextSeq(convID string) (int64, error) { + return a.repo.GetNextSeq(convID) +} diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index da5b683..de1e278 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -348,13 +348,9 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv if s.conversationCache != nil { seq, err := s.conversationCache.GetNextSeq(ctx, conversationID) if err != nil { - zap.L().Warn("redis get next seq failed, falling back to DB", - zap.String("convID", conversationID), - zap.Error(err), - ) - } else { - message.Seq = seq + return nil, fmt.Errorf("failed to get next seq for conversation %s: %w", conversationID, err) } + message.Seq = seq } // 使用事务创建消息并更新seq @@ -916,13 +912,9 @@ func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conv if s.conversationCache != nil { seq, err := s.conversationCache.GetNextSeq(ctx, conversationID) if err != nil { - zap.L().Warn("redis get next seq failed, falling back to DB", - zap.String("convID", conversationID), - zap.Error(err), - ) - } else { - message.Seq = seq + return nil, fmt.Errorf("failed to get next seq for conversation %s: %w", conversationID, err) } + message.Seq = seq } if err := s.repo.CreateMessageWithSeq(message); err != nil { diff --git a/internal/service/group_service.go b/internal/service/group_service.go index 87565d3..ae98157 100644 --- a/internal/service/group_service.go +++ b/internal/service/group_service.go @@ -474,13 +474,13 @@ func (s *groupService) broadcastMemberJoinNotice(groupID string, targetUserID st if s.conversationCache != nil { seq, seqErr := s.conversationCache.GetNextSeq(context.Background(), conv.ID) if seqErr != nil { - zap.L().Warn("redis get next seq failed, falling back to DB", + zap.L().Warn("get next seq failed, skipping system message", zap.String("convID", conv.ID), zap.Error(seqErr), ) - } else { - msg.Seq = seq + return } + msg.Seq = seq } if err := s.messageRepo.CreateMessageWithSeq(msg); err != nil { zap.L().Warn("保存入群提示消息失败", @@ -1399,13 +1399,13 @@ func (s *groupService) MuteMember(userID string, groupID string, targetUserID st if s.conversationCache != nil { seq, seqErr := s.conversationCache.GetNextSeq(context.Background(), conv.ID) if seqErr != nil { - zap.L().Warn("redis get next seq failed, falling back to DB", + zap.L().Warn("get next seq failed, skipping system message", zap.String("convID", conv.ID), zap.Error(seqErr), ) - } else { - msg.Seq = seq + return nil } + msg.Seq = seq } if err := s.messageRepo.CreateMessageWithSeq(msg); err != nil { diff --git a/internal/service/message_service.go b/internal/service/message_service.go index 6da0fd1..71201de 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "time" "with_you/internal/cache" @@ -90,13 +91,9 @@ func (s *MessageService) SendMessage(ctx context.Context, senderID, receiverID s if s.conversationCache != nil { seq, seqErr := s.conversationCache.GetNextSeq(context.Background(), conv.ID) if seqErr != nil { - zap.L().Warn("redis get next seq failed, falling back to DB", - zap.String("convID", conv.ID), - zap.Error(seqErr), - ) - } else { - msg.Seq = seq + return nil, fmt.Errorf("failed to get next seq for conversation %s: %w", conv.ID, seqErr) } + msg.Seq = seq } // 使用事务创建消息并更新seq