diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index cd60906..34d7b0c 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -9,6 +9,7 @@ import ( "github.com/redis/go-redis/v9" "with_you/internal/model" + redisPkg "with_you/internal/pkg/redis" "go.uber.org/zap" ) @@ -31,6 +32,14 @@ redis.call('SETNX', seqKey, initVal) return redis.call('INCR', seqKey) `) +// syncSeqLua 写透缓存：确保 Redis seq >= DB seq，防止冷启动后读到过期值 +var syncSeqLua = redis.NewScript(` +local current = tonumber(redis.call('GET', KEYS[1])) +if current == nil or current < tonumber(ARGV[1]) then + redis.call('SET', KEYS[1], ARGV[1]) +end +`) + // CachedConversation 带缓存元数据的会话 type CachedConversation struct { Data *model.Conversation // 实际数据 @@ -713,86 +722,58 @@ func (c *ConversationCache) ComputeAllUnreadCount(ctx context.Context, userID st return total, nil } -// GetNextSeq 获取会话的下一个 seq 值（原子递增） -// 启用 seq 预分配时使用本地缓冲区 + Redis Lua 分配；否则使用旧 INCR 逻辑 +// GetNextSeq 获取会话的下一个 seq 值（单次 Redis INCR，原子无竞态） func (c *ConversationCache) GetNextSeq(ctx context.Context, convID string) (int64, error) { - // 优先使用 seq 预分配 + // 优先使用 SeqBufferManager（已简化为单次 INCR） if c.seqBufferMgr != nil && c.seqBufferMgr.IsEnabled() { return c.seqBufferMgr.GetNextSeq(ctx, convID, false) } - // 降级到旧逻辑：使用 Lua 脚本原子执行 INCR + 首次初始化对齐，避免竞态 seqKey := MessageSeqKey(convID) - // 尝试用 Lua 脚本原子递增（key 存在时 INCR，不存在时返回 0 触发冷启动） + // 尝试 Redis INCR（key 存在时递增，不存在时返回 0 触发冷启动） if lc, ok := c.cache.(*LayeredCache); ok && lc.IsEnabled() { rdb := lc.GetRedisClient() if rdb != nil { result, err := nextSeqLua.Run(ctx, rdb.GetClient(), []string{seqKey}).Int64() if err == nil { if result > 0 { - // key 已存在，INCR 成功 _ = c.cache.Expire(ctx, seqKey, c.settings.SeqTTL) return result, nil } - // result == 0: 冷启动，key 不存在，需从 DB 初始化 - if c.repo != nil { - conv, dbErr := c.repo.GetConversationByID(convID) - if dbErr != nil { - return 0, fmt.Errorf("failed to load conversation for seq init: %w", dbErr) - } - initVal := conv.LastSeq // SETNX 使用 last_seq，INCR 返回 last_seq + 1 - initResult, initErr := initSeqLua.Run(ctx, rdb.GetClient(), []string{seqKey}, initVal).Int64() - if initErr != nil { - // 初始化失败，降级到非原子路径 - zap.L().Warn("initSeqLua failed, falling back", zap.Error(initErr)) - } else { - _ = c.cache.Expire(ctx, seqKey, c.settings.SeqTTL) - return initResult, nil - } - } + // 冷启动：从 DB 加载 last_seq，SETNX+INCR 原子初始化 + return c.initSeqFromDB(ctx, convID, seqKey, rdb) } - // Lua 脚本失败，降级到非原子路径 } } - // 非原子降级路径（Redis 不可用或 Lua 不支持） - newSeq, err := c.cache.Incr(ctx, seqKey) - if err != nil { - // Redis 不可用时降级到 DB - 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，需要从 DB 补齐 - // 注意：此路径存在竞态窗口，优先使用上面的 Lua 原子路径 - if newSeq == 1 && c.repo != nil { - conv, err := c.repo.GetConversationByID(convID) - if err != nil { - return 0, fmt.Errorf("failed to load conversation for seq init: %w", err) - } - - if conv.LastSeq > 0 { - alignedSeq, err := c.cache.IncrBySeq(ctx, seqKey, conv.LastSeq) - if err != nil { - return conv.LastSeq + 1, nil // 降级返回 DB 值 - } - newSeq = alignedSeq + // Redis 不可用，降级到 DB（SELECT FOR UPDATE 保证原子性） + if c.msgRepo != nil { + if dbSeq, dbErr := c.msgRepo.GetNextSeq(convID); dbErr == nil { + return dbSeq, nil } } - - // 每次 INCR 后刷新 TTL，防止 seq 计数器过期导致从头开始 - _ = c.cache.Expire(ctx, seqKey, c.settings.SeqTTL) - - return newSeq, nil + return 0, fmt.Errorf("seq allocation failed for %s: no redis or db available", convID) } -// GetNextSeqWithGroup 获取会话的下一个 seq 值（支持群聊标识，用于 seq 预分配步长区分） +// initSeqFromDB 冷启动：从 DB 加载 last_seq，用 SETNX+INCR 原子初始化 +func (c *ConversationCache) initSeqFromDB(ctx context.Context, convID, seqKey string, rdb *redisPkg.Client) (int64, error) { + if c.repo == nil { + return 0, fmt.Errorf("conversation repository not configured for seq init") + } + conv, dbErr := c.repo.GetConversationByID(convID) + if dbErr != nil { + return 0, fmt.Errorf("load conversation for seq init: %w", dbErr) + } + result, initErr := initSeqLua.Run(ctx, rdb.GetClient(), []string{seqKey}, conv.LastSeq).Int64() + if initErr != nil { + return 0, fmt.Errorf("initSeqLua failed: %w", initErr) + } + _ = c.cache.Expire(ctx, seqKey, c.settings.SeqTTL) + return result, nil +} + +// GetNextSeqWithGroup 获取会话的下一个 seq 值（isGroup 参数保留兼容，简化后不区分步长） func (c *ConversationCache) GetNextSeqWithGroup(ctx context.Context, convID string, isGroup bool) (int64, error) { if c.seqBufferMgr != nil && c.seqBufferMgr.IsEnabled() { return c.seqBufferMgr.GetNextSeq(ctx, convID, isGroup) @@ -800,6 +781,22 @@ func (c *ConversationCache) GetNextSeqWithGroup(ctx context.Context, convID stri return c.GetNextSeq(ctx, convID) } +// SyncConvSeq 写透缓存：将 DB 刚分配的 seq 同步到 Redis，确保 Redis seq >= DB seq +func (c *ConversationCache) SyncConvSeq(ctx context.Context, convID string, seq int64) { + seqKey := MessageSeqKey(convID) + + if lc, ok := c.cache.(*LayeredCache); ok && lc.IsEnabled() { + rdb := lc.GetRedisClient() + if rdb != nil { + _ = syncSeqLua.Run(ctx, rdb.GetClient(), []string{seqKey}, seq).Err() + } + } + + if c.seqBufferMgr != nil && c.seqBufferMgr.rdb != nil { + _ = syncSeqLua.Run(ctx, c.seqBufferMgr.rdb, []string{seqKey}, seq).Err() + } +} + // ============================================================ // 消息缓存方法 // ============================================================ diff --git a/internal/cache/seq_buffer.go b/internal/cache/seq_buffer.go index 11037bc..cdb47de 100644 --- a/internal/cache/seq_buffer.go +++ b/internal/cache/seq_buffer.go @@ -3,7 +3,6 @@ package cache import ( "context" "fmt" - "sync" "time" "github.com/redis/go-redis/v9" @@ -12,139 +11,27 @@ import ( redisPkg "with_you/internal/pkg/redis" ) -// SeqBufferConfig seq 预分配配置 +const seqKeyTTL = 24 * time.Hour + +// SeqBufferConfig seq 预分配配置（保留结构以兼容配置文件，实际不再使用缓冲） type SeqBufferConfig struct { - Enabled bool `mapstructure:"enabled"` - PrivateBufferSize int `mapstructure:"private_buffer_size"` // 私聊预分配步长 - GroupBufferSize int `mapstructure:"group_buffer_size"` // 群聊预分配步长 - LockTimeoutMs int `mapstructure:"lock_timeout_ms"` // 分布式锁超时(ms) - MaxRetries int `mapstructure:"max_retries"` // 冷启动重试次数 + Enabled bool `mapstructure:"enabled"` + PrivateBufferSize int `mapstructure:"private_buffer_size"` + GroupBufferSize int `mapstructure:"group_buffer_size"` + LockTimeoutMs int `mapstructure:"lock_timeout_ms"` + MaxRetries int `mapstructure:"max_retries"` } -// localSeqBuffer 本地 seq 缓冲区 -type localSeqBuffer struct { - mu sync.Mutex - curr int64 // 当前已分配到的 seq - last int64 // 本地缓冲区的上界 -} - -// SeqBufferManager seq 预分配管理器 +// SeqBufferManager seq 分配管理器（简化版：每条消息一个 Redis INCR，无本地缓冲） type SeqBufferManager struct { - rdb *redis.Client // 原始 Redis 客户端（用于执行 Lua 脚本） - config SeqBufferConfig - repo ConversationRepository - msgRepo MessageRepository - buffers sync.Map // convID -> *localSeqBuffer + rdb *redis.Client + config SeqBufferConfig + repo ConversationRepository + msgRepo MessageRepository } -// mallocLua 预分配 Lua 跟踪脚本 -// KEYS[1] = seq_buf:{convID} (Hash: CURR/LAST/LOCK/TIME) -// KEYS[2] = msg_seq:{convID} (String, 兼容旧 INCR key) -// ARGV[1] = bufferSize -// ARGV[2] = lockTimeoutMs (毫秒) -// ARGV[3] = current timestamp (毫秒) -// -// 返回: -// status=1, startSeq, endSeq — 成功分配 -// status=2 — 冷启动（key 不存在） -// status=0 — 获取锁失败 -var mallocLua = redis.NewScript(` -local bufKey = KEYS[1] -local seqKey = KEYS[2] -local bufSize = tonumber(ARGV[1]) -local lockMs = tonumber(ARGV[2]) -local nowMs = tonumber(ARGV[3]) - --- 尝试获取分布式锁 -local lockVal = redis.call('HGET', bufKey, 'LOCK') -if lockVal then - local lockTs = tonumber(lockVal) - if lockTs > 0 and (nowMs - lockTs) < lockMs then - return {0, 0, 0} - end -end -redis.call('HSET', bufKey, 'LOCK', nowMs) -redis.call('HSET', bufKey, 'TIME', nowMs) - -local curr = tonumber(redis.call('HGET', bufKey, 'CURR')) -local last = tonumber(redis.call('HGET', bufKey, 'LAST')) - -if curr == nil or last == nil then - -- 冷启动：key 不存在 - redis.call('HDEL', bufKey, 'LOCK') - return {2, 0, 0} -end - -if curr < last then - -- 本地缓冲区仍有余量，直接分配一个 seq - local newCurr = redis.call('HINCRBY', bufKey, 'CURR', 1) - redis.call('HDEL', bufKey, 'LOCK') - return {1, newCurr, last} -end - --- 缓冲区耗尽，扩展 LAST -local newLast = last + bufSize -redis.call('HSET', bufKey, 'LAST', newLast) --- 推进 CURR 防止下一个调用者获得相同的 start seq -redis.call('HSET', bufKey, 'CURR', curr + 1) --- 同步更新 msg_seq（兼容旧 INCR key，确保 READ 的值不回退） -local oldSeq = tonumber(redis.call('GET', seqKey)) -if oldSeq and oldSeq > newLast then - newLast = oldSeq + bufSize - redis.call('HSET', bufKey, 'LAST', newLast) -end -redis.call('SET', seqKey, newLast) -redis.call('HDEL', bufKey, 'LOCK') -return {1, curr + 1, newLast} -`) - -// initLua 冷启动初始化 Lua 跟踪脚本 -// KEYS[1] = seq_buf:{convID} -// KEYS[2] = msg_seq:{convID} -// ARGV[1] = startSeq (从 DB 获取的 conv.LastSeq) -// ARGV[2] = bufferSize -// ARGV[3] = lockTimeoutMs -// ARGV[4] = current timestamp (毫秒) -var initLua = redis.NewScript(` -local bufKey = KEYS[1] -local seqKey = KEYS[2] -local startSeq = tonumber(ARGV[1]) -local bufSize = tonumber(ARGV[2]) -local lockMs = tonumber(ARGV[3]) -local nowMs = tonumber(ARGV[4]) - --- 创建 seq_buf Hash -redis.call('HSET', bufKey, 'CURR', startSeq) -redis.call('HSET', bufKey, 'LAST', startSeq + bufSize) -redis.call('HSET', bufKey, 'LOCK', 0) -redis.call('HSET', bufKey, 'TIME', nowMs) - --- 同步更新 msg_seq（兼容旧 INCR key） -local oldSeq = tonumber(redis.call('GET', seqKey)) -if oldSeq and oldSeq > (startSeq + bufSize) then - redis.call('HSET', bufKey, 'LAST', oldSeq + bufSize) - redis.call('SET', seqKey, oldSeq + bufSize) -else - redis.call('SET', seqKey, startSeq + bufSize) -end - -return {1, startSeq + 1, startSeq + bufSize} -`) - -// NewSeqBufferManager 创建 seq 预分配管理器 +// NewSeqBufferManager 创建 seq 分配管理器 func NewSeqBufferManager(rdb *redis.Client, cfg SeqBufferConfig, repo ConversationRepository, msgRepo MessageRepository) *SeqBufferManager { - if cfg.PrivateBufferSize <= 0 { - cfg.PrivateBufferSize = 50 - } - if cfg.GroupBufferSize <= 0 { - cfg.GroupBufferSize = 200 - } - if cfg.LockTimeoutMs <= 0 { - cfg.LockTimeoutMs = 5000 - } - if cfg.MaxRetries <= 0 { - cfg.MaxRetries = 3 - } return &SeqBufferManager{ rdb: rdb, config: cfg, @@ -153,160 +40,67 @@ func NewSeqBufferManager(rdb *redis.Client, cfg SeqBufferConfig, repo Conversati } } -// GetNextSeq 获取下一个 seq（从本地缓冲区分配，耗尽时向 Redis 预分配新区间） -func (m *SeqBufferManager) GetNextSeq(ctx context.Context, convID string, isGroup bool) (int64, error) { - bufSize := m.config.PrivateBufferSize - if isGroup { - bufSize = m.config.GroupBufferSize - } - - // 1. 检查本地缓冲区 - bufVal, ok := m.buffers.Load(convID) - if ok { - buf := bufVal.(*localSeqBuffer) - buf.mu.Lock() - if buf.curr < buf.last { - seq := buf.curr + 1 - buf.curr = seq - buf.mu.Unlock() - return seq, nil - } - buf.mu.Unlock() - } - - // 2. 本地缓冲区耗尽或不存在，向 Redis 预分配 - startSeq, endSeq, err := m.allocateFromRedis(ctx, convID, bufSize) - if err != nil { - return 0, fmt.Errorf("allocate from redis: %w", err) - } - - // 3. 更新本地缓冲区（CAS 防止并发分配时丢失区间） - newBuf := &localSeqBuffer{curr: startSeq, last: endSeq} - for { - existing, loaded := m.buffers.LoadOrStore(convID, newBuf) - if !loaded { - // 首次插入，直接使用 - return startSeq, nil - } - // 已有缓冲区，检查是否被其他 goroutine 刷新 - oldBuf := existing.(*localSeqBuffer) - oldBuf.mu.Lock() - if oldBuf.curr < oldBuf.last { - // 旧缓冲区仍有余量，直接使用（丢弃本次分配，避免区间泄漏） - seq := oldBuf.curr + 1 - oldBuf.curr = seq - oldBuf.mu.Unlock() - return seq, nil - } - // 旧缓冲区耗尽，替换为新分配的区间 - m.buffers.Store(convID, newBuf) - oldBuf.mu.Unlock() - return startSeq, nil - } -} - -// allocateFromRedis 向 Redis 预分配 seq 区间 -func (m *SeqBufferManager) allocateFromRedis(ctx context.Context, convID string, bufferSize int) (startSeq, endSeq int64, err error) { - bufKey := SeqBufferKey(convID) +// GetNextSeq 获取下一个 seq（单次 Redis INCR，原子且无竞态） +func (m *SeqBufferManager) GetNextSeq(ctx context.Context, convID string, _ bool) (int64, error) { seqKey := MessageSeqKey(convID) - nowMs := time.Now().UnixMilli() - for i := 0; i < m.config.MaxRetries; i++ { - result, err := mallocLua.Run(ctx, m.rdb, []string{bufKey, seqKey}, bufferSize, m.config.LockTimeoutMs, nowMs).Int64Slice() - if err != nil { - // Redis 不可用，降级到 DB - zap.L().Warn("seq malloc lua failed, falling back to DB", zap.String("convID", convID), zap.Error(err)) - if m.msgRepo != nil { - if dbSeq, dbErr := m.msgRepo.GetNextSeq(convID); dbErr == nil { - return dbSeq, dbSeq, nil - } - } - return 0, 0, fmt.Errorf("lua script error: %w", err) - } - - status := result[0] - switch status { - case 1: - // 成功分配 - return result[1], result[2], nil - case 2: - // 冷启动，需要从 DB 初始化 - startSeq, endSeq, initErr := m.initFromDB(ctx, convID, bufferSize) - if initErr != nil { - return 0, 0, fmt.Errorf("init from db: %w", initErr) - } - return startSeq, endSeq, nil - case 0: - // 获取锁失败，短暂等待后重试 - time.Sleep(time.Duration(50) * time.Millisecond) - continue + // 1. 尝试 INCR（key 存在时直接递增，原子操作） + result, err := nextSeqLua.Run(ctx, m.rdb, []string{seqKey}).Int64() + if err == nil { + if result > 0 { + m.rdb.Expire(ctx, seqKey, seqKeyTTL) + return result, nil } + // result == 0: 冷启动，需从 DB 初始化 + return m.initAndIncr(ctx, convID, seqKey) } - // 所有重试失败，降级到 DB - zap.L().Warn("seq malloc retries exhausted, falling back to DB", zap.String("convID", convID)) + // Redis 不可用，降级到 DB + zap.L().Warn("seq INCR failed, falling back to DB", zap.String("convID", convID), zap.Error(err)) if m.msgRepo != nil { if dbSeq, dbErr := m.msgRepo.GetNextSeq(convID); dbErr == nil { - return dbSeq, dbSeq, nil + return dbSeq, nil } } - return 0, 0, fmt.Errorf("malloc retries exhausted") + return 0, fmt.Errorf("seq allocation failed for %s: %w", convID, err) } -// initFromDB 冷启动时从 DB 加载 LastSeq 并初始化 Redis seq_buf Hash -func (m *SeqBufferManager) initFromDB(ctx context.Context, convID string, bufferSize int) (startSeq, endSeq int64, err error) { +// initAndIncr 冷启动：从 DB 加载 last_seq，然后用 SETNX+INCR 原子初始化 +func (m *SeqBufferManager) initAndIncr(ctx context.Context, convID, seqKey string) (int64, error) { if m.repo == nil { - return 0, 0, fmt.Errorf("conversation repository not configured") + return 0, fmt.Errorf("conversation repository not configured for seq init") } conv, err := m.repo.GetConversationByID(convID) if err != nil { - return 0, 0, fmt.Errorf("load conversation: %w", err) + return 0, fmt.Errorf("load conversation for seq init: %w", err) } - dbLastSeq := conv.LastSeq - if dbLastSeq <= 0 { - dbLastSeq = 0 - } - - bufKey := SeqBufferKey(convID) - seqKey := MessageSeqKey(convID) - nowMs := time.Now().UnixMilli() - - result, err := initLua.Run(ctx, m.rdb, []string{bufKey, seqKey}, dbLastSeq, bufferSize, m.config.LockTimeoutMs, nowMs).Int64Slice() + initVal := conv.LastSeq + result, err := initSeqLua.Run(ctx, m.rdb, []string{seqKey}, initVal).Int64() if err != nil { - return 0, 0, fmt.Errorf("init lua script error: %w", err) + return 0, fmt.Errorf("initSeqLua failed: %w", err) } - zap.L().Info("seq buffer cold-start initialized", - zap.String("convID", convID), - zap.Int64("dbLastSeq", dbLastSeq), - zap.Int64("startSeq", result[1]), - zap.Int64("endSeq", result[2]), - ) - - return result[1], result[2], nil + m.rdb.Expire(ctx, seqKey, seqKeyTTL) + return result, nil } -// InvalidateBuffer 清除本地缓冲区（会话 seq 被外部重置时调用） -func (m *SeqBufferManager) InvalidateBuffer(convID string) { - m.buffers.Delete(convID) -} +// InvalidateBuffer no-op（保留接口兼容） +func (m *SeqBufferManager) InvalidateBuffer(convID string) {} -// IsEnabled 是否启用预分配 +// IsEnabled 是否启用 func (m *SeqBufferManager) IsEnabled() bool { return m.config.Enabled && m.rdb != nil } // NewSeqBufferManagerFromCache 从 ConversationCache 的 Cache 接口创建 SeqBufferManager -// 当 wire 中不方便注入 redis.Client 时，可通过 Cache 接口获取底层 Redis 连接 func NewSeqBufferManagerFromCache(cacheBackend Cache, cfg SeqBufferConfig, repo ConversationRepository, msgRepo MessageRepository) *SeqBufferManager { redisCache, ok := cacheBackend.(*RedisCache) if !ok { zap.L().Warn("SeqBuffer: cache is not RedisCache, seq pre-allocation disabled") return &SeqBufferManager{config: cfg, repo: repo, msgRepo: msgRepo} } - rdb := redisCache.GetRedisClient().GetClient() return NewSeqBufferManager(rdb, cfg, repo, msgRepo) } @@ -319,4 +113,4 @@ func NewSeqBufferManagerFromRedisPkg(redisPkgClient *redisPkg.Client, cfg SeqBuf } rdb := redisPkgClient.GetClient() return NewSeqBufferManager(rdb, cfg, repo, msgRepo) -} \ No newline at end of file +} diff --git a/internal/repository/message_repo.go b/internal/repository/message_repo.go index 3ec4fc9..72c8384 100644 --- a/internal/repository/message_repo.go +++ b/internal/repository/message_repo.go @@ -431,40 +431,43 @@ func (r *messageRepository) GetNextSeq(conversationID string) (int64, error) { return nextSeq, err } -// CreateMessageWithSeq 创建消息并更新seq（事务操作） -// msg.Seq 应由调用方通过 Redis INCR 预分配 +// CreateMessageWithSeq 创建消息并在事务内原子分配 seq +// DB 是 seq 的唯一真相来源：SELECT FOR UPDATE 锁行 → last_seq+1 → 写入消息 → 更新 last_seq +// msg.Seq 会在事务内被设置，调用方无需预先分配 func (r *messageRepository) CreateMessageWithSeq(msg *model.Message) error { return r.db.Transaction(func(tx *gorm.DB) error { - // 创建消息（seq 已由调用方预分配） + // 1. 锁定会话行，读取当前 last_seq + var conv model.Conversation + query := tx.Where("id = ?", msg.ConversationID) + if tx.Dialector.Name() == "postgres" { + query = query.Clauses(clause.Locking{Strength: "UPDATE"}) + } + if err := query.First(&conv).Error; err != nil { + return fmt.Errorf("lock conversation for seq: %w", err) + } + + // 2. 原子分配 seq + msg.Seq = conv.LastSeq + 1 + + // 3. 创建消息 if err := tx.Create(msg).Error; err != nil { return err } - // 更新会话last_seq：仅当新 seq 更大时才更新，防止并发写入导致 last_seq 回退 - result := tx.Model(&model.Conversation{}). - Where("id = ? AND last_seq < ?", msg.ConversationID, msg.Seq). + // 4. 更新会话 last_seq 和 last_msg_time + if err := tx.Model(&model.Conversation{}). + Where("id = ?", msg.ConversationID). Updates(map[string]any{ "last_seq": msg.Seq, "last_msg_time": gorm.Expr("CURRENT_TIMESTAMP"), - }) - if result.Error != nil { - return result.Error - } - // 当 last_seq >= msg.Seq 时（并发写入已更新），仍需刷新 last_msg_time - if err := tx.Model(&model.Conversation{}). - Where("id = ? AND last_seq >= ?", msg.ConversationID, msg.Seq). - Update("last_msg_time", gorm.Expr("CURRENT_TIMESTAMP")).Error; err != nil { + }).Error; err != nil { return err } - // 新消息到达后，自动恢复被"仅自己删除"的会话 - if err := tx.Model(&model.ConversationParticipant{}). + // 5. 新消息到达后，自动恢复被"仅自己删除"的会话 + return tx.Model(&model.ConversationParticipant{}). Where("conversation_id = ?", msg.ConversationID). - Update("hidden_at", nil).Error; err != nil { - return err - } - - return nil + Update("hidden_at", nil).Error }) } diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index f20cc02..899701c 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -479,20 +479,18 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv Status: model.MessageStatusNormal, } - // 从 Redis 获取下一个 seq - if s.conversationCache != nil { - seq, err := s.conversationCache.GetNextSeqWithGroup(ctx, conversationID, conv.Type == model.ConversationTypeGroup) - if err != nil { - return nil, fmt.Errorf("failed to get next seq for conversation %s: %w", conversationID, err) - } - message.Seq = seq - } + // seq 由 CreateMessageWithSeq 在事务内原子分配 // 使用事务创建消息并更新seq if err := s.repo.CreateMessageWithSeq(message); err != nil { return nil, fmt.Errorf("failed to save message: %w", err) } + // 写透 seq 到 Redis，保持缓存与 DB 一致 + if s.conversationCache != nil { + s.conversationCache.SyncConvSeq(context.Background(), conversationID, message.Seq) + } + // 发送者自动已读：将发送者的 lastReadSeq 更新到消息 seq _ = s.repo.UpdateLastReadSeq(conversationID, senderID, message.Seq) if s.conversationCache != nil { @@ -1042,7 +1040,7 @@ func (s *chatServiceImpl) IsUserOnline(userID string) bool { // 适用于群聊等由调用方自行负责推送的场景 func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conversationID string, segments model.MessageSegments, replyToID *string) (*model.Message, error) { // 验证会话是否存在 - conv, err := s.getConversation(ctx, conversationID) + _, err := s.getConversation(ctx, conversationID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, errors.New("会话不存在，请重新创建会话") @@ -1073,19 +1071,17 @@ func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conv Status: model.MessageStatusNormal, } - // 从 Redis 获取下一个 seq（SaveMessage 方法） - if s.conversationCache != nil { - seq, err := s.conversationCache.GetNextSeqWithGroup(ctx, conversationID, conv.Type == model.ConversationTypeGroup) - if err != nil { - return nil, fmt.Errorf("failed to get next seq for conversation %s: %w", conversationID, err) - } - message.Seq = seq - } + // seq 由 CreateMessageWithSeq 在事务内原子分配 if err := s.repo.CreateMessageWithSeq(message); err != nil { return nil, fmt.Errorf("failed to save message: %w", err) } + // 写透 seq 到 Redis，保持缓存与 DB 一致 + if s.conversationCache != nil { + s.conversationCache.SyncConvSeq(context.Background(), conversationID, message.Seq) + } + // 新消息会改变分页结果，先失效分页缓存，避免读到旧列表 if s.conversationCache != nil { s.conversationCache.InvalidateMessagePages(conversationID) diff --git a/internal/service/group_service.go b/internal/service/group_service.go index 32c4237..a2d2aff 100644 --- a/internal/service/group_service.go +++ b/internal/service/group_service.go @@ -471,18 +471,7 @@ func (s *groupService) broadcastMemberJoinNotice(groupID string, targetUserID st Status: model.MessageStatusNormal, Category: model.CategoryNotification, } - // 从 Redis 获取下一个 seq - if s.conversationCache != nil { - seq, seqErr := s.conversationCache.GetNextSeqWithGroup(context.Background(), conv.ID, true) - if seqErr != nil { - zap.L().Warn("get next seq failed, skipping system message", - zap.String("convID", conv.ID), - zap.Error(seqErr), - ) - return - } - msg.Seq = seq - } + // seq 由 CreateMessageWithSeq 在事务内原子分配 if err := s.messageRepo.CreateMessageWithSeq(msg); err != nil { zap.L().Warn("保存入群提示消息失败", zap.String("component", "broadcastMemberJoinNotice"), @@ -492,6 +481,9 @@ func (s *groupService) broadcastMemberJoinNotice(groupID string, targetUserID st ) } else { savedMessage = msg + if s.conversationCache != nil { + s.conversationCache.SyncConvSeq(context.Background(), conv.ID, msg.Seq) + } s.invalidateConversationCachesAfterSystemMessage(conv.ID) } } else { @@ -1396,18 +1388,7 @@ func (s *groupService) MuteMember(userID string, groupID string, targetUserID st Category: model.CategoryNotification, } - // 从 Redis 获取下一个 seq - if s.conversationCache != nil { - seq, seqErr := s.conversationCache.GetNextSeqWithGroup(context.Background(), conv.ID, true) - if seqErr != nil { - zap.L().Warn("get next seq failed, skipping system message", - zap.String("convID", conv.ID), - zap.Error(seqErr), - ) - return nil - } - msg.Seq = seq - } + // seq 由 CreateMessageWithSeq 在事务内原子分配 if err := s.messageRepo.CreateMessageWithSeq(msg); err != nil { zap.L().Warn("保存禁言消息失败", @@ -1421,6 +1402,9 @@ func (s *groupService) MuteMember(userID string, groupID string, targetUserID st zap.String("msgID", msg.ID), zap.Int64("seq", msg.Seq), ) + if s.conversationCache != nil { + s.conversationCache.SyncConvSeq(context.Background(), conv.ID, msg.Seq) + } s.invalidateConversationCachesAfterSystemMessage(conv.ID) } } else { diff --git a/internal/service/message_service.go b/internal/service/message_service.go index d645666..1bc7d3c 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -2,7 +2,6 @@ package service import ( "context" - "fmt" "time" "with_you/internal/cache" @@ -88,21 +87,17 @@ func (s *MessageService) SendMessage(ctx context.Context, senderID, receiverID s Status: model.MessageStatusNormal, } - // 从 Redis 获取下一个 seq - if s.conversationCache != nil { - seq, seqErr := s.conversationCache.GetNextSeqWithGroup(context.Background(), conv.ID, false) - if seqErr != nil { - return nil, fmt.Errorf("failed to get next seq for conversation %s: %w", conv.ID, seqErr) - } - msg.Seq = seq - } - - // 使用事务创建消息并更新seq + // 事务内原子分配 seq + 创建消息（DB 是 seq 唯一来源） err = s.messageRepo.CreateMessageWithSeq(msg) if err != nil { return nil, err } + // 写透 seq 到 Redis，保持缓存与 DB 一致 + if s.conversationCache != nil { + s.conversationCache.SyncConvSeq(context.Background(), conv.ID, msg.Seq) + } + // 新消息会改变分页结果，先失效分页缓存，避免读到旧列表 if s.conversationCache != nil { s.conversationCache.InvalidateMessagePages(conv.ID)