From 6bf87fec46bb9ef5763a9a73d5aa7d480d6ec570 Mon Sep 17 00:00:00 2001 From: lan Date: Sun, 17 May 2026 23:38:04 +0800 Subject: [PATCH] feat(chat): implement sequence pre-allocation, versioned sync, and push worker Introduce several performance and synchronization enhancements: - Implement `SeqBufferManager` to allow sequence number pre-allocation via Redis Lua scripts, reducing atomic increment overhead. - Add `PushWorker` to handle asynchronous message pushing using Redis Streams. - Implement incremental conversation synchronization via `ConversationVersionLog` to allow clients to fetch only recent changes. - Add support for Gzip compression in WebSocket communications to reduce bandwidth usage. - Update dependency injection and configuration to support these new components. --- cmd/server/app.go | 14 + cmd/server/wire_gen.go | 7 +- configs/config.yaml | 21 + internal/cache/conversation_cache.go | 38 +- internal/cache/keys.go | 6 + internal/cache/seq_buffer.go | 303 ++++++++++++++ internal/config/config.go | 22 + internal/config/push_worker.go | 13 + internal/config/seq_buffer.go | 10 + internal/config/version_log.go | 8 + internal/dto/sync_dto.go | 23 ++ internal/handler/message_handler.go | 31 ++ internal/handler/ws_handler.go | 103 +++-- internal/model/conversation_version_log.go | 13 + internal/pkg/ws/compress.go | 71 ++++ internal/pkg/ws/compress_test.go | 94 +++++ internal/pkg/ws/hub.go | 11 +- .../conversation_version_log_repo.go | 58 +++ internal/router/router.go | 1 + internal/service/chat_service.go | 229 +++++++++-- internal/service/group_service.go | 1 + internal/service/message_service.go | 1 + internal/service/push_worker.go | 381 ++++++++++++++++++ internal/wire/infrastructure.go | 51 +++ internal/wire/repository.go | 3 + internal/wire/service.go | 19 +- 26 files changed, 1450 insertions(+), 82 deletions(-) create mode 100644 internal/cache/seq_buffer.go create mode 100644 internal/config/push_worker.go create mode 100644 internal/config/seq_buffer.go create mode 100644 internal/config/version_log.go create mode 100644 internal/dto/sync_dto.go create mode 100644 internal/model/conversation_version_log.go create mode 100644 internal/pkg/ws/compress.go create mode 100644 internal/pkg/ws/compress_test.go create mode 100644 internal/repository/conversation_version_log_repo.go create mode 100644 internal/service/push_worker.go diff --git a/cmd/server/app.go b/cmd/server/app.go index 37cd8e3..ce0825a 100644 --- a/cmd/server/app.go +++ b/cmd/server/app.go @@ -28,6 +28,7 @@ type App struct { Publisher ws.MessagePublisher TaskDispatcher runner.TaskDispatcher Logger *zap.Logger + PushWorker *service.PushWorker } // NewApp 创建应用程序 @@ -41,6 +42,7 @@ func NewApp( publisher ws.MessagePublisher, taskDispatcher runner.TaskDispatcher, logger *zap.Logger, + pushWorker *service.PushWorker, ) *App { return &App{ Config: cfg, @@ -52,6 +54,7 @@ func NewApp( Publisher: publisher, TaskDispatcher: taskDispatcher, Logger: logger, + PushWorker: pushWorker, } } @@ -73,6 +76,11 @@ func (a *App) Start(ctx context.Context) error { a.HotRankWorker.Start(ctx) } + // 2.2 启动 PushWorker（Redis Stream 异步推送） + if a.PushWorker != nil { + a.PushWorker.Start(ctx) + } + // 3. 创建 HTTP 服务器 addr := fmt.Sprintf("%s:%d", a.Config.Server.Host, a.Config.Server.Port) a.Logger.Info("Starting HTTP server", @@ -148,6 +156,12 @@ func (a *App) Stop(ctx context.Context) error { a.HotRankWorker.Stop() } + // 5.1 停止 PushWorker + if a.PushWorker != nil { + a.Logger.Info("Stopping push worker (stream)") + a.PushWorker.Stop() + } + // 6. 关闭数据库连接 a.Logger.Info("Closing database connection") sqlDB, err := a.DB.DB() diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index ac31d25..d061d0c 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -78,7 +78,10 @@ func InitializeApp() (*App, error) { return nil, err } uploadService := wire.ProvideUploadService(s3Client, userService) - chatService := wire.ProvideChatService(messageRepository, userRepository, messagePublisher, cache, uploadService, pushService) + seqBufferManager := wire.ProvideSeqBufferManager(config, client, cache) + pushWorker := wire.ProvidePushWorker(config, client, messagePublisher, pushService, messageRepository, userRepository) + conversationVersionLogRepository := repository.NewConversationVersionLogRepository(db) + chatService := wire.ProvideChatService(messageRepository, userRepository, messagePublisher, cache, uploadService, pushService, seqBufferManager, pushWorker, client, conversationVersionLogRepository, config) messageService := wire.ProvideMessageService(messageRepository, cache, uploadService) groupRepository := repository.NewGroupRepository(db) groupJoinRequestRepository := repository.NewGroupJoinRequestRepository(db) @@ -158,7 +161,7 @@ func InitializeApp() (*App, error) { server := wire.ProvideGRPCServer(config, logger, runnerHub) runnerRegistry := wire.ProvideRunnerRegistry(config, client) taskDispatcher := wire.ProvideTaskDispatcher(runnerHub, runnerRegistry, config, client, logger) - app := NewApp(config, db, router, pushService, hotRankWorker, server, messagePublisher, taskDispatcher, logger) + app := NewApp(config, db, router, pushService, hotRankWorker, server, messagePublisher, taskDispatcher, logger, pushWorker) return app, nil } diff --git a/configs/config.yaml b/configs/config.yaml index a49c9aa..e3f9571 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -312,3 +312,24 @@ websocket: msg_channel: "ws:msg" # Redis Pub/Sub 频道 online_ttl: 60 # 在线状态 TTL（秒） heartbeat_interval: 20 # 心跳间隔（秒） + +# Seq 预分配配置 +# 环境变量: +# APP_SEQ_BUFFER_ENABLED, APP_SEQ_BUFFER_PRIVATE_BUFFER_SIZE +# APP_SEQ_BUFFER_GROUP_BUFFER_SIZE, APP_SEQ_BUFFER_LOCK_TIMEOUT_MS +# APP_SEQ_BUFFER_MAX_RETRIES +seq_buffer: + enabled: false # 启用后使用 Lua 预分配代替 INCR，关闭则回退旧逻辑 + private_buffer_size: 50 # 私聊每次向 Redis 预分配步长 + group_buffer_size: 200 # 群聊每次向 Redis 预分配步长（群聊消息更频繁） + lock_timeout_ms: 5000 # 分布式锁超时（毫秒） + max_retries: 3 # 冷启动获取锁失败最大重试次数 + +# 版本日志增量同步配置 +# 环境变量: +# APP_VERSION_LOG_ENABLED, APP_VERSION_LOG_SYNC_LIMIT +# APP_VERSION_LOG_MAX_SYNC_GAP +version_log: + enabled: false # 启用后客户端可使用 /conversations/sync 增量同步 + sync_limit: 100 # 单次同步最大返回变更条数 + max_sync_gap: 1000 # 版本差距超过此值建议全量同步 diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index 82f3675..6cee2fc 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -230,22 +230,24 @@ type MessageRepository interface { // ConversationCache 会话缓存管理器 type ConversationCache struct { - cache Cache // 底层缓存 - settings *ConversationCacheSettings // 配置 - repo ConversationRepository // 数据仓库接口（用于 cache-aside 回源） - msgRepo MessageRepository // 消息数据仓库接口（用于消息缓存回源） + cache Cache // 底层缓存 + settings *ConversationCacheSettings // 配置 + repo ConversationRepository // 数据仓库接口（用于 cache-aside 回源） + msgRepo MessageRepository // 消息数据仓库接口（用于消息缓存回源） + seqBufferMgr *SeqBufferManager // seq 预分配管理器（nil 则使用旧 INCR 逻辑） } // NewConversationCache 创建会话缓存管理器 -func NewConversationCache(cache Cache, repo ConversationRepository, msgRepo MessageRepository, settings *ConversationCacheSettings) *ConversationCache { +func NewConversationCache(cache Cache, repo ConversationRepository, msgRepo MessageRepository, settings *ConversationCacheSettings, seqBufferMgr *SeqBufferManager) *ConversationCache { if settings == nil { settings = DefaultConversationCacheSettings() } return &ConversationCache{ - cache: cache, - settings: settings, - repo: repo, - msgRepo: msgRepo, + cache: cache, + settings: settings, + repo: repo, + msgRepo: msgRepo, + seqBufferMgr: seqBufferMgr, } } @@ -694,8 +696,16 @@ func (c *ConversationCache) ComputeAllUnreadCount(ctx context.Context, userID st } // GetNextSeq 获取会话的下一个 seq 值（原子递增） -// 使用 Redis INCR 实现，首次调用时从 DB 初始化并补齐差值 +// 启用 seq 预分配时使用本地缓冲区 + Redis Lua 分配；否则使用旧 INCR 逻辑 func (c *ConversationCache) GetNextSeq(ctx context.Context, convID string) (int64, error) { + // 优先使用 seq 预分配 + if c.seqBufferMgr != nil && c.seqBufferMgr.IsEnabled() { + // 判断是否群聊（通过会话类型判断，这里先按私聊处理，由上层传入 isGroup） + // 由于接口只传 convID，这里默认私聊逻辑，群聊的 isGroup 由调用方决定 + return c.seqBufferMgr.GetNextSeq(ctx, convID, false) + } + + // 降级到旧 INCR 逻辑 seqKey := MessageSeqKey(convID) newSeq, err := c.cache.Incr(ctx, seqKey) @@ -732,6 +742,14 @@ func (c *ConversationCache) GetNextSeq(ctx context.Context, convID string) (int6 return newSeq, nil } +// GetNextSeqWithGroup 获取会话的下一个 seq 值（支持群聊标识，用于 seq 预分配步长区分） +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) + } + return c.GetNextSeq(ctx, convID) +} + // ============================================================ // 消息缓存方法 // ============================================================ diff --git a/internal/cache/keys.go b/internal/cache/keys.go index a8b856b..2f9912a 100644 --- a/internal/cache/keys.go +++ b/internal/cache/keys.go @@ -44,6 +44,7 @@ const ( keyPrefixMsgSeq = "msg_seq" // Seq 计数器 keyPrefixMsgPage = "msg_page" // 分页缓存 keyPrefixMsgIdempotent = "msg_idem" // 消息幂等键 + keyPrefixSeqBuffer = "seq_buf" // Seq 预分配缓冲区 Hash ) // PostListKey 生成帖子列表缓存键 @@ -197,6 +198,11 @@ func MessageIdempotentKey(senderID, clientMsgID string) string { return fmt.Sprintf("%s:%s:%s", keyPrefixMsgIdempotent, senderID, clientMsgID) } +// SeqBufferKey seq 预分配缓冲区 Hash key +func SeqBufferKey(convID string) string { + return fmt.Sprintf("%s:%s", keyPrefixSeqBuffer, convID) +} + // UserReadSeqKey 用户已读位置缓存键（OpenIM 风格 SEQ_USER_READ:{convID}:{userID}） func UserReadSeqKey(convID, userID string) string { diff --git a/internal/cache/seq_buffer.go b/internal/cache/seq_buffer.go new file mode 100644 index 0000000..d2e6ce4 --- /dev/null +++ b/internal/cache/seq_buffer.go @@ -0,0 +1,303 @@ +package cache + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + + redisPkg "with_you/internal/pkg/redis" +) + +// 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"` // 冷启动重试次数 +} + +// localSeqBuffer 本地 seq 缓冲区 +type localSeqBuffer struct { + mu sync.Mutex + curr int64 // 当前已分配到的 seq + last int64 // 本地缓冲区的上界 +} + +// SeqBufferManager seq 预分配管理器 +type SeqBufferManager struct { + rdb *redis.Client // 原始 Redis 客户端（用于执行 Lua 脚本） + config SeqBufferConfig + repo ConversationRepository + msgRepo MessageRepository + buffers sync.Map // convID -> *localSeqBuffer +} + +// 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) +-- 同步更新 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 预分配管理器 +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, + repo: repo, + msgRepo: msgRepo, + } +} + +// 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. 更新本地缓冲区 + newBuf := &localSeqBuffer{curr: startSeq, last: endSeq} + m.buffers.Store(convID, newBuf) + + // startSeq 已经被 Redis 分配，返回它 + 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) + 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 + } + } + + // 所有重试失败，降级到 DB + zap.L().Warn("seq malloc retries exhausted, falling back to DB", zap.String("convID", convID)) + if m.msgRepo != nil { + if dbSeq, dbErr := m.msgRepo.GetNextSeq(convID); dbErr == nil { + return dbSeq, dbSeq, nil + } + } + return 0, 0, fmt.Errorf("malloc retries exhausted") +} + +// initFromDB 冷启动时从 DB 加载 LastSeq 并初始化 Redis seq_buf Hash +func (m *SeqBufferManager) initFromDB(ctx context.Context, convID string, bufferSize int) (startSeq, endSeq int64, err error) { + if m.repo == nil { + return 0, 0, fmt.Errorf("conversation repository not configured") + } + + conv, err := m.repo.GetConversationByID(convID) + if err != nil { + return 0, 0, fmt.Errorf("load conversation: %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() + if err != nil { + return 0, 0, fmt.Errorf("init lua script error: %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 +} + +// InvalidateBuffer 清除本地缓冲区（会话 seq 被外部重置时调用） +func (m *SeqBufferManager) InvalidateBuffer(convID string) { + m.buffers.Delete(convID) +} + +// 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) +} + +// NewSeqBufferManagerFromRedisPkg 从 pkg/redis.Client 创建 SeqBufferManager +func NewSeqBufferManagerFromRedisPkg(redisPkgClient *redisPkg.Client, cfg SeqBufferConfig, repo ConversationRepository, msgRepo MessageRepository) *SeqBufferManager { + if redisPkgClient == nil { + zap.L().Warn("SeqBuffer: redis client is nil, seq pre-allocation disabled") + return &SeqBufferManager{config: cfg, repo: repo, msgRepo: msgRepo} + } + rdb := redisPkgClient.GetClient() + return NewSeqBufferManager(rdb, cfg, repo, msgRepo) +} \ No newline at end of file diff --git a/internal/config/config.go b/internal/config/config.go index d0088a0..cb217be 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -34,6 +34,9 @@ type Config struct { SetupSecret string `mapstructure:"setup_secret"` WebSocket WSConfig `mapstructure:"websocket"` Runner RunnerConfig `mapstructure:"runner"` + SeqBuffer SeqBufferConfig `mapstructure:"seq_buffer"` + VersionLog VersionLogConfig `mapstructure:"version_log"` + PushWorker PushWorkerConfig `mapstructure:"push_worker"` } // Load 加载配置文件 @@ -191,6 +194,25 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("runner.cluster.cap_prefix", "runner:cap:") viper.SetDefault("runner.cluster.registry_ttl", 90) viper.SetDefault("runner.cluster.heartbeat_interval", 30) + // SeqBuffer 默认值 + viper.SetDefault("seq_buffer.enabled", false) + viper.SetDefault("seq_buffer.private_buffer_size", 50) + viper.SetDefault("seq_buffer.group_buffer_size", 200) + viper.SetDefault("seq_buffer.lock_timeout_ms", 5000) + viper.SetDefault("seq_buffer.max_retries", 3) + // VersionLog 默认值 + viper.SetDefault("version_log.enabled", false) + viper.SetDefault("version_log.sync_limit", 100) + viper.SetDefault("version_log.max_sync_gap", 1000) + // PushWorker 默认值 + viper.SetDefault("push_worker.enabled", false) + viper.SetDefault("push_worker.stream", "msg_push") + viper.SetDefault("push_worker.group", "push_worker") + viper.SetDefault("push_worker.batch_size", 50) + viper.SetDefault("push_worker.poll_timeout_ms", 2000) + viper.SetDefault("push_worker.max_retries", 3) + viper.SetDefault("push_worker.max_stream_len", 100000) + viper.SetDefault("push_worker.idle_timeout_ms", 30000) if err := viper.ReadInConfig(); err != nil { return nil, fmt.Errorf("failed to read config: %w", err) diff --git a/internal/config/push_worker.go b/internal/config/push_worker.go new file mode 100644 index 0000000..f47a0ad --- /dev/null +++ b/internal/config/push_worker.go @@ -0,0 +1,13 @@ +package config + +// PushWorkerConfig 推送 Worker 配置（Redis Stream 异步推送） +type PushWorkerConfig struct { + Enabled bool `mapstructure:"enabled"` + Stream string `mapstructure:"stream"` // Redis Stream 名称，默认 "msg_push" + Group string `mapstructure:"group"` // Consumer Group 名称，默认 "push_worker" + BatchSize int `mapstructure:"batch_size"` // XREADGROUP 每次读取条数，默认 50 + PollTimeoutMs int `mapstructure:"poll_timeout_ms"` // XREADGROUP BLOCK 超时 ms，默认 2000 + MaxRetries int `mapstructure:"max_retries"` // 消息最大重试次数，默认 3 + MaxStreamLen int64 `mapstructure:"max_stream_len"` // Stream MAXLEN，默认 100000 + IdleTimeoutMs int64 `mapstructure:"idle_timeout_ms"` // XPENDING 空闲超时 ms，默认 30000 +} \ No newline at end of file diff --git a/internal/config/seq_buffer.go b/internal/config/seq_buffer.go new file mode 100644 index 0000000..4a13793 --- /dev/null +++ b/internal/config/seq_buffer.go @@ -0,0 +1,10 @@ +package config + +// SeqBufferConfig seq 预分配配置 +type SeqBufferConfig struct { + Enabled bool `mapstructure:"enabled"` + PrivateBufferSize int `mapstructure:"private_buffer_size"` // 私聊预分配步长（默认 50） + GroupBufferSize int `mapstructure:"group_buffer_size"` // 群聊预分配步长（默认 200） + LockTimeoutMs int `mapstructure:"lock_timeout_ms"` // 分布式锁超时毫秒（默认 5000） + MaxRetries int `mapstructure:"max_retries"` // 冷启动重试次数（默认 3） +} diff --git a/internal/config/version_log.go b/internal/config/version_log.go new file mode 100644 index 0000000..c9e7478 --- /dev/null +++ b/internal/config/version_log.go @@ -0,0 +1,8 @@ +package config + +// VersionLogConfig 版本日志增量同步配置 +type VersionLogConfig struct { + Enabled bool `mapstructure:"enabled"` + SyncLimit int `mapstructure:"sync_limit"` // 单次同步最大返回条数 + MaxSyncGap int64 `mapstructure:"max_sync_gap"` // 超过此版本差距建议全量同步 +} diff --git a/internal/dto/sync_dto.go b/internal/dto/sync_dto.go new file mode 100644 index 0000000..6ad2bfe --- /dev/null +++ b/internal/dto/sync_dto.go @@ -0,0 +1,23 @@ +package dto + +// SyncVersionRequest 增量同步请求 +type SyncVersionRequest struct { + Version int64 `form:"version" binding:"min=0"` // 客户端上次同步版本号，0 表示全量同步 +} + +// SyncVersionResponse 增量同步响应 +type SyncVersionResponse struct { + CurrentVersion int64 `json:"current_version"` // 服务端当前最大版本号 + Changes []SyncChangeItem `json:"changes"` // 变更列表 + FullSync bool `json:"full_sync"` // 是否需要全量同步（版本差距过大时） + HasMore bool `json:"has_more"` // 是否还有更多变更 +} + +// SyncChangeItem 单条变更项 +type SyncChangeItem struct { + ConversationID string `json:"conversation_id"` + ChangeType string `json:"change_type"` // new_msg, read, pin, unpin, mute, unmute, hide, group_update + MaxSeq int64 `json:"max_seq,omitempty"` + LastMessageAt int64 `json:"last_message_at,omitempty"` + Version int64 `json:"version"` +} \ No newline at end of file diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 843ad97..11d9a1a 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -585,6 +585,37 @@ func (h *MessageHandler) HandleGetSyncData(c *gin.Context) { }) } +// HandleGetSyncByVersion 增量同步：按版本号获取会话变更 +// GET /api/v1/conversations/sync?version=0&limit=100 +func (h *MessageHandler) HandleGetSyncByVersion(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + versionStr := c.DefaultQuery("version", "0") + sinceVersion, err := strconv.ParseInt(versionStr, 10, 64) + if err != nil || sinceVersion < 0 { + response.BadRequest(c, "invalid version parameter") + return + } + + limitStr := c.DefaultQuery("limit", "100") + limit, err := strconv.Atoi(limitStr) + if err != nil || limit < 1 { + limit = 100 + } + + result, err := h.chatService.GetSyncByVersion(c.Request.Context(), userID, sinceVersion, limit) + if err != nil { + response.InternalServerError(c, err.Error()) + return + } + + response.Success(c, result) +} + // GetConversationUnreadCount 获取单个会话的未读数 // GET /api/conversations/:id/unread/count func (h *MessageHandler) GetConversationUnreadCount(c *gin.Context) { diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 32e1a11..8078628 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -151,12 +151,14 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { // 3. 创建客户端 clientID := atomic.AddUint64(&h.clientSeq, 1) + compressEnabled := c.Query("compress") == "1" client := &ws.Client{ - ID: clientID, - UserID: userID, - Send: make(chan []byte, defaultUserBufferSize), - Quit: make(chan struct{}), - PendingAcks: make(map[string]time.Time), + ID: clientID, + UserID: userID, + Send: make(chan []byte, defaultUserBufferSize), + Quit: make(chan struct{}), + PendingAcks: make(map[string]time.Time), + CompressEnabled: compressEnabled, } // 4. 注册客户端 @@ -243,9 +245,21 @@ func (h *WSHandler) readPump(conn *websocket.Conn, client *ws.Client) { break } - // 解析消息 + // 解析消息（支持 gzip 解压） + var rawData []byte + if client.CompressEnabled && ws.IsCompressed(message) { + decompressed, dErr := ws.DefaultCompressor.Decompress(message) + if dErr != nil { + h.publisher.SendError(client, "decompress_error", "failed to decompress gzip message") + continue + } + rawData = decompressed + } else { + rawData = message + } + var msg ws.Message - if err := json.Unmarshal(message, &msg); err != nil { + if err := json.Unmarshal(rawData, &msg); err != nil { h.publisher.SendError(client, "parse_error", "invalid message format") continue } @@ -280,31 +294,60 @@ func (h *WSHandler) writePump(conn *websocket.Conn, client *ws.Client) { return } - w, err := conn.NextWriter(websocket.TextMessage) - if err != nil { - zap.L().Warn("WebSocket write error (NextWriter)", - zap.String("user_id", client.UserID), - zap.Uint64("client_id", client.ID), - zap.Error(err), - ) - return - } - w.Write(message) + if client.CompressEnabled { + buf := make([]byte, 0, len(message)+64) + buf = append(buf, message...) + n := len(client.Send) + for i := 0; i < n; i++ { + buf = append(buf, '\n') + buf = append(buf, <-client.Send...) + } + compressed, cErr := ws.DefaultCompressor.Compress(buf) + if cErr != nil { + zap.L().Warn("WebSocket compress error, sending raw", + zap.String("user_id", client.UserID), + zap.Error(cErr), + ) + w, err := conn.NextWriter(websocket.TextMessage) + if err != nil { + return + } + w.Write(buf) + w.Close() + } else { + w, err := conn.NextWriter(websocket.BinaryMessage) + if err != nil { + return + } + w.Write(compressed) + w.Close() + } + } else { + w, err := conn.NextWriter(websocket.TextMessage) + if err != nil { + zap.L().Warn("WebSocket write error (NextWriter)", + zap.String("user_id", client.UserID), + zap.Uint64("client_id", client.ID), + zap.Error(err), + ) + return + } + w.Write(message) - // 批量发送缓冲区中的消息 - n := len(client.Send) - for i := 0; i < n; i++ { - w.Write([]byte{'\n'}) - w.Write(<-client.Send) - } + n := len(client.Send) + for i := 0; i < n; i++ { + w.Write([]byte{'\n'}) + w.Write(<-client.Send) + } - if err := w.Close(); err != nil { - zap.L().Warn("WebSocket write error (Close)", - zap.String("user_id", client.UserID), - zap.Uint64("client_id", client.ID), - zap.Error(err), - ) - return + if err := w.Close(); err != nil { + zap.L().Warn("WebSocket write error (Close)", + zap.String("user_id", client.UserID), + zap.Uint64("client_id", client.ID), + zap.Error(err), + ) + return + } } case <-ticker.C: conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) diff --git a/internal/model/conversation_version_log.go b/internal/model/conversation_version_log.go new file mode 100644 index 0000000..9b98b14 --- /dev/null +++ b/internal/model/conversation_version_log.go @@ -0,0 +1,13 @@ +package model + +import "time" + +// ConversationVersionLog 会话版本日志（用于增量同步） +type ConversationVersionLog struct { + ID uint `gorm:"primaryKey;autoIncrement"` + UserID string `gorm:"not null;size:36;index:idx_vlog_user_version,priority:1"` + ConversationID string `gorm:"not null;size:20;index:idx_vlog_conv"` + Version int64 `gorm:"not null;index:idx_vlog_user_version,priority:2"` + ChangeType string `gorm:"not null;type:varchar(20)"` // new_msg, read, pin, unpin, mute, unmute, hide, group_update + CreatedAt time.Time `gorm:"autoCreateTime"` +} \ No newline at end of file diff --git a/internal/pkg/ws/compress.go b/internal/pkg/ws/compress.go new file mode 100644 index 0000000..0394ddc --- /dev/null +++ b/internal/pkg/ws/compress.go @@ -0,0 +1,71 @@ +package ws + +import ( + "bytes" + "compress/gzip" + "io" + "sync" +) + +const gzipMagicByte = 0x1F + +// Compressor 管理 gzip Writer/Reader 的 sync.Pool 复用 +type Compressor struct { + writerPool sync.Pool + readerPool sync.Pool + level int +} + +// NewCompressor 创建 Compressor（level: gzip.DefaultCompression/BestSpeed/BestCompression 等） +func NewCompressor(level int) *Compressor { + return &Compressor{ + level: level, + writerPool: sync.Pool{ + New: func() any { + w, _ := gzip.NewWriterLevel(io.Discard, level) + return w + }, + }, + readerPool: sync.Pool{ + New: func() any { return new(gzip.Reader) }, + }, + } +} + +// Compress 压缩数据，返回 gzip 字节 +func (c *Compressor) Compress(data []byte) ([]byte, error) { + buf := &bytes.Buffer{} + w := c.writerPool.Get().(*gzip.Writer) + w.Reset(buf) + if _, err := w.Write(data); err != nil { + c.writerPool.Put(w) + return nil, err + } + if err := w.Close(); err != nil { + c.writerPool.Put(w) + return nil, err + } + c.writerPool.Put(w) + return buf.Bytes(), nil +} + +// Decompress 解压 gzip 数据 +func (c *Compressor) Decompress(data []byte) ([]byte, error) { + r := c.readerPool.Get().(*gzip.Reader) + if err := r.Reset(bytes.NewReader(data)); err != nil { + c.readerPool.Put(r) + return nil, err + } + out, err := io.ReadAll(r) + r.Close() + c.readerPool.Put(r) + return out, err +} + +// IsCompressed 检查数据是否以 gzip magic byte 开头 +func IsCompressed(data []byte) bool { + return len(data) > 0 && data[0] == gzipMagicByte +} + +// DefaultCompressor 全局默认压缩器（gzip.DefaultCompression = level 6） +var DefaultCompressor = NewCompressor(gzip.DefaultCompression) \ No newline at end of file diff --git a/internal/pkg/ws/compress_test.go b/internal/pkg/ws/compress_test.go new file mode 100644 index 0000000..c893647 --- /dev/null +++ b/internal/pkg/ws/compress_test.go @@ -0,0 +1,94 @@ +package ws + +import ( + "bytes" + "testing" +) + +func TestCompressDecompressRoundTrip(t *testing.T) { + c := NewCompressor(6) + + original := []byte(`{"type":"chat_message","payload":{"conversation_id":"conv1","message":"hello world"}}`) + + compressed, err := c.Compress(original) + if err != nil { + t.Fatalf("Compress failed: %v", err) + } + + if len(compressed) >= len(original) { + t.Logf("warning: compressed size (%d) >= original size (%d), but this can happen for small payloads", len(compressed), len(original)) + } + + if !IsCompressed(compressed) { + t.Fatal("IsCompressed should return true for gzip data") + } + + decompressed, err := c.Decompress(compressed) + if err != nil { + t.Fatalf("Decompress failed: %v", err) + } + + if !bytes.Equal(decompressed, original) { + t.Fatalf("round-trip mismatch: got %q, want %q", decompressed, original) + } +} + +func TestIsCompressed(t *testing.T) { + if IsCompressed([]byte{}) { + t.Fatal("empty data should not be considered compressed") + } + if IsCompressed([]byte{0x00}) { + t.Fatal("0x00 should not be considered gzip magic byte") + } + if !IsCompressed([]byte{0x1F, 0x8B}) { + t.Fatal("gzip magic bytes should be detected") + } +} + +func TestDecompressNonGzipData(t *testing.T) { + c := NewCompressor(6) + _, err := c.Decompress([]byte("not gzip data")) + if err == nil { + t.Fatal("expected error when decompressing non-gzip data") + } +} + +func TestDefaultCompressor(t *testing.T) { + data := []byte("test data for default compressor") + compressed, err := DefaultCompressor.Compress(data) + if err != nil { + t.Fatalf("DefaultCompressor.Compress failed: %v", err) + } + decompressed, err := DefaultCompressor.Decompress(compressed) + if err != nil { + t.Fatalf("DefaultCompressor.Decompress failed: %v", err) + } + if !bytes.Equal(decompressed, data) { + t.Fatalf("DefaultCompressor round-trip mismatch") + } +} + +func BenchmarkCompress(b *testing.B) { + c := NewCompressor(6) + data := bytes.Repeat([]byte(`{"type":"chat_message","payload":{"message":"hello"}}`), 10) + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := c.Compress(data) + if err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkDecompress(b *testing.B) { + c := NewCompressor(6) + data := bytes.Repeat([]byte(`{"type":"chat_message","payload":{"message":"hello"}}`), 10) + compressed, _ := c.Compress(data) + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := c.Decompress(compressed) + if err != nil { + b.Fatal(err) + } + } +} \ No newline at end of file diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go index 5f0a87f..4e84a5e 100644 --- a/internal/pkg/ws/hub.go +++ b/internal/pkg/ws/hub.go @@ -31,11 +31,12 @@ type Event struct { // Client 表示一个WebSocket客户端连接 type Client struct { - ID uint64 - UserID string - Send chan []byte - Quit chan struct{} - PendingAcks map[string]time.Time + ID uint64 + UserID string + Send chan []byte + Quit chan struct{} + PendingAcks map[string]time.Time + CompressEnabled bool // 客户端是否启用 gzip 压缩 } // ErrConnectionLimit 连接数限制错误 diff --git a/internal/repository/conversation_version_log_repo.go b/internal/repository/conversation_version_log_repo.go new file mode 100644 index 0000000..87bc111 --- /dev/null +++ b/internal/repository/conversation_version_log_repo.go @@ -0,0 +1,58 @@ +package repository + +import ( + "context" + + "with_you/internal/model" + + "gorm.io/gorm" +) + +// ConversationVersionLogRepository 会话版本日志仓储接口 +type ConversationVersionLogRepository interface { + Create(ctx context.Context, log *model.ConversationVersionLog) error + BatchCreate(ctx context.Context, logs []*model.ConversationVersionLog) error + GetByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) ([]*model.ConversationVersionLog, error) + GetMaxVersion(ctx context.Context, userID string) (int64, error) +} + +// conversationVersionLogRepository GORM 实现 +type conversationVersionLogRepository struct { + db *gorm.DB +} + +// NewConversationVersionLogRepository 创建会话版本日志仓储 +func NewConversationVersionLogRepository(db *gorm.DB) ConversationVersionLogRepository { + return &conversationVersionLogRepository{db: db} +} + +func (r *conversationVersionLogRepository) Create(ctx context.Context, log *model.ConversationVersionLog) error { + return r.db.WithContext(ctx).Create(log).Error +} + +func (r *conversationVersionLogRepository) BatchCreate(ctx context.Context, logs []*model.ConversationVersionLog) error { + if len(logs) == 0 { + return nil + } + return r.db.WithContext(ctx).CreateInBatches(logs, 100).Error +} + +func (r *conversationVersionLogRepository) GetByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) ([]*model.ConversationVersionLog, error) { + var logs []*model.ConversationVersionLog + err := r.db.WithContext(ctx). + Where("user_id = ? AND version > ?", userID, sinceVersion). + Order("version ASC"). + Limit(limit). + Find(&logs).Error + return logs, err +} + +func (r *conversationVersionLogRepository) GetMaxVersion(ctx context.Context, userID string) (int64, error) { + var maxVersion int64 + err := r.db.WithContext(ctx). + Model(&model.ConversationVersionLog{}). + Where("user_id = ?", userID). + Select("COALESCE(MAX(version), 0)"). + Scan(&maxVersion).Error + return maxVersion, err +} \ No newline at end of file diff --git a/internal/router/router.go b/internal/router/router.go index dad91c3..c4f57ac 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -431,6 +431,7 @@ func (r *Router) setupRoutes() { conversations.PUT("/:id/notification_muted", r.messageHandler.HandleSetConversationNotificationMuted) // 免打扰设置 conversations.GET("/unread/count", r.messageHandler.GetUnreadCount) // 未读数 conversations.GET("/sync-data", r.messageHandler.HandleGetSyncData) // 同步元数据 + conversations.GET("/sync", r.messageHandler.HandleGetSyncByVersion) // 增量同步(版本日志) 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 de1e278..047ba7f 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -13,6 +13,7 @@ import ( "with_you/internal/pkg/ws" "with_you/internal/repository" + "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" ) @@ -48,6 +49,9 @@ type ChatService interface { // 消息同步 GetSyncData(ctx context.Context, userID string) ([]dto.SyncDataItem, error) + // 增量同步（版本日志） + GetSyncByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) (*dto.SyncVersionResponse, error) + // 消息扩展功能 RecallMessage(ctx context.Context, messageID string, userID string) error DeleteMessage(ctx context.Context, messageID string, userID string) error @@ -80,6 +84,11 @@ type chatServiceImpl struct { conversationCache *cache.ConversationCache uploadService *UploadService + versionLogRepo repository.ConversationVersionLogRepository + versionLogEnabled bool + versionLogMaxGap int64 + pushWorker *PushWorker + redisClient *redis.Client } // NewChatService 创建聊天服务 @@ -91,6 +100,10 @@ func NewChatService( cacheBackend cache.Cache, uploadService *UploadService, pushSvc PushService, + seqBufferMgr *cache.SeqBufferManager, + pushWorker *PushWorker, + redisClient *redis.Client, + versionLogRepo ...repository.ConversationVersionLogRepository, ) ChatService { // 创建适配器 convRepoAdapter := cache.NewConversationRepositoryAdapter(repo) @@ -102,9 +115,10 @@ func NewChatService( convRepoAdapter, msgRepoAdapter, cache.DefaultConversationCacheSettings(), + seqBufferMgr, ) - return &chatServiceImpl{ + impl := &chatServiceImpl{ repo: repo, userRepo: userRepo, sensitive: sensitive, @@ -113,7 +127,17 @@ func NewChatService( cache: cacheBackend, conversationCache: conversationCache, uploadService: uploadService, + pushWorker: pushWorker, + redisClient: redisClient, } + + if len(versionLogRepo) > 0 && versionLogRepo[0] != nil { + impl.versionLogRepo = versionLogRepo[0] + impl.versionLogEnabled = true + impl.versionLogMaxGap = 1000 + } + + return impl } func (s *chatServiceImpl) publishToUsers(userIDs []string, event string, payload any) { @@ -123,6 +147,115 @@ func (s *chatServiceImpl) publishToUsers(userIDs []string, event string, payload s.wsHub.PublishToUsers(userIDs, event, payload) } +// writeVersionLog 写入版本日志（异步，失败仅 warn） +func (s *chatServiceImpl) writeVersionLog(ctx context.Context, userID, conversationID, changeType string) { + if !s.versionLogEnabled || s.versionLogRepo == nil { + return + } + maxV, err := s.versionLogRepo.GetMaxVersion(ctx, userID) + if err != nil { + zap.L().Warn("writeVersionLog: get max version failed", zap.String("userID", userID), zap.Error(err)) + return + } + vlog := &model.ConversationVersionLog{ + UserID: userID, + ConversationID: conversationID, + Version: maxV + 1, + ChangeType: changeType, + } + if err := s.versionLogRepo.Create(ctx, vlog); err != nil { + zap.L().Warn("writeVersionLog: create failed", zap.String("userID", userID), zap.Error(err)) + } +} + +// writeVersionLogBatch 批量写入版本日志（异步，失败仅 warn） +func (s *chatServiceImpl) writeVersionLogBatch(ctx context.Context, userIDs []string, conversationID, changeType string) { + if !s.versionLogEnabled || s.versionLogRepo == nil { + return + } + logs := make([]*model.ConversationVersionLog, 0, len(userIDs)) + for _, uid := range userIDs { + maxV, err := s.versionLogRepo.GetMaxVersion(ctx, uid) + if err != nil { + zap.L().Warn("writeVersionLogBatch: get max version failed", zap.String("userID", uid), zap.Error(err)) + continue + } + logs = append(logs, &model.ConversationVersionLog{ + UserID: uid, + ConversationID: conversationID, + Version: maxV + 1, + ChangeType: changeType, + }) + } + if len(logs) > 0 { + if err := s.versionLogRepo.BatchCreate(ctx, logs); err != nil { + zap.L().Warn("writeVersionLogBatch: batch create failed", zap.Error(err)) + } + } +} + +// GetSyncByVersion 增量同步：按版本号获取会话变更 +func (s *chatServiceImpl) GetSyncByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) (*dto.SyncVersionResponse, error) { + if !s.versionLogEnabled || s.versionLogRepo == nil { + return nil, errors.New("version log sync is not enabled") + } + + if limit <= 0 { + limit = 100 + } + + currentVersion, err := s.versionLogRepo.GetMaxVersion(ctx, userID) + if err != nil { + return nil, fmt.Errorf("failed to get current version: %w", err) + } + + resp := &dto.SyncVersionResponse{ + CurrentVersion: currentVersion, + } + + if currentVersion == 0 { + resp.FullSync = true + return resp, nil + } + + gap := currentVersion - sinceVersion + if gap > s.versionLogMaxGap { + resp.FullSync = true + return resp, nil + } + + vlogs, err := s.versionLogRepo.GetByVersion(ctx, userID, sinceVersion, limit+1) + if err != nil { + return nil, fmt.Errorf("failed to get version logs: %w", err) + } + + hasMore := len(vlogs) > limit + if hasMore { + vlogs = vlogs[:limit] + } + + changes := make([]dto.SyncChangeItem, 0, len(vlogs)) + for _, vl := range vlogs { + item := dto.SyncChangeItem{ + ConversationID: vl.ConversationID, + ChangeType: vl.ChangeType, + Version: vl.Version, + } + conv, cErr := s.repo.GetConversation(vl.ConversationID) + if cErr == nil && conv != nil { + item.LastMessageAt = conv.UpdatedAt.Unix() + if maxSeq, sErr := s.conversationCache.GetConvMaxSeq(ctx, vl.ConversationID); sErr == nil { + item.MaxSeq = maxSeq + } + } + changes = append(changes, item) + } + + resp.Changes = changes + resp.HasMore = hasMore + return resp, nil +} + // GetOrCreateConversation 获取或创建私聊会话 func (s *chatServiceImpl) GetOrCreateConversation(ctx context.Context, user1ID, user2ID string) (*model.Conversation, error) { conv, err := s.repo.GetOrCreatePrivateConversation(user1ID, user2ID) @@ -205,6 +338,8 @@ func (s *chatServiceImpl) DeleteConversationForSelf(ctx context.Context, convers s.conversationCache.InvalidateConversationList(userID) } + go s.writeVersionLog(context.Background(), userID, conversationID, "hide") + return nil } @@ -387,6 +522,13 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv // 获取会话中的参与者并发送消息 participants, err := s.getParticipants(ctx, conversationID) if err == nil { + // 版本日志：为所有参与者写入 new_msg + pids := make([]string, 0, len(participants)) + for _, p := range participants { + pids = append(pids, p.UserID) + } + go s.writeVersionLogBatch(context.Background(), pids, conversationID, "new_msg") + // 先更新未读计数器，再推送 WS 事件，确保事件中的 total_unread 已包含新消息 // 注意：已移除 IncrementUnread，未读数完全由 maxSeq - readSeq 算术计算 if s.conversationCache != nil { @@ -405,38 +547,63 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv } } - targetIDs := make([]string, 0, len(participants)) - for _, p := range participants { - if conv.Type == model.ConversationTypePrivate && p.UserID == senderID { - continue + // PushWorker 异步推送 or inline fallback + if s.pushWorker != nil && s.pushWorker.IsEnabled() { + streamMsg := &PushStreamMsg{ + ConversationID: conversationID, + SenderID: senderID, + MessageID: message.ID, + ConvType: string(conv.Type), + Timestamp: time.Now().UnixMilli(), + } + if pubErr := PublishToPush(context.Background(), s.redisClient, "msg_push", streamMsg); pubErr != nil { + zap.L().Warn("push stream publish failed, falling back to inline push", + zap.String("conversationID", conversationID), + zap.Error(pubErr), + ) + s.inlinePush(ctx, conv, participants, senderID, conversationID, message) } - targetIDs = append(targetIDs, p.UserID) - } - detailType := "private" - if conv.Type == model.ConversationTypeGroup { - detailType = "group" - } - if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup { - s.wsHub.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{ - "detail_type": detailType, - "message": dto.ConvertMessageToResponse(message), - }) } else { - s.publishToUsers(targetIDs, "chat_message", map[string]any{ - "detail_type": detailType, - "message": dto.ConvertMessageToResponse(message), - }) + s.inlinePush(ctx, conv, participants, senderID, conversationID, message) } - for _, p := range participants { - if p.UserID == senderID { - continue - } - 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, - }) - } + } + + return message, nil +} + +// inlinePush 内联推送（原有逻辑，PushWorker 失败时的 fallback） +func (s *chatServiceImpl) inlinePush(ctx context.Context, conv *model.Conversation, participants []*model.ConversationParticipant, senderID, conversationID string, message *model.Message) { + targetIDs := make([]string, 0, len(participants)) + for _, p := range participants { + if conv.Type == model.ConversationTypePrivate && p.UserID == senderID { + continue + } + targetIDs = append(targetIDs, p.UserID) + } + detailType := "private" + if conv.Type == model.ConversationTypeGroup { + detailType = "group" + } + if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup { + s.wsHub.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{ + "detail_type": detailType, + "message": dto.ConvertMessageToResponse(message), + }) + } else { + s.publishToUsers(targetIDs, "chat_message", map[string]any{ + "detail_type": detailType, + "message": dto.ConvertMessageToResponse(message), + }) + } + for _, p := range participants { + if p.UserID == senderID { + continue + } + 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, + }) } } @@ -482,8 +649,6 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv }(offlineTargetIDs, sender, groupID, groupAvatar) } } - - return message, nil } func (s *chatServiceImpl) getConversation(ctx context.Context, conversationID string) (*model.Conversation, error) { diff --git a/internal/service/group_service.go b/internal/service/group_service.go index ae98157..5140832 100644 --- a/internal/service/group_service.go +++ b/internal/service/group_service.go @@ -126,6 +126,7 @@ func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.U convRepoAdapter, msgRepoAdapter, cache.DefaultConversationCacheSettings(), + nil, ) return &groupService{ diff --git a/internal/service/message_service.go b/internal/service/message_service.go index 71201de..963654c 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -49,6 +49,7 @@ func NewMessageService(messageRepo repository.MessageRepository, cacheBackend ca convRepoAdapter, msgRepoAdapter, cache.DefaultConversationCacheSettings(), + nil, ) return &MessageService{ diff --git a/internal/service/push_worker.go b/internal/service/push_worker.go new file mode 100644 index 0000000..a1c0fa2 --- /dev/null +++ b/internal/service/push_worker.go @@ -0,0 +1,381 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "with_you/internal/dto" + "with_you/internal/model" + "with_you/internal/pkg/ws" + "with_you/internal/repository" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// PushWorkerConfig 推送 Worker 配置 +type PushWorkerConfig struct { + Enabled bool + Stream string // Redis Stream 名称，默认 "msg_push" + Group string // Consumer Group 名称，默认 "push_worker" + BatchSize int // XREADGROUP 每次读取条数，默认 50 + PollTimeoutMs int // XREADGROUP BLOCK 超时 ms，默认 2000 + MaxRetries int // 消息最大重试次数，默认 3 + MaxStreamLen int64 // Stream MAXLEN，默认 100000 + IdleTimeoutMs int64 // XPENDING 空闲超时 ms，默认 30000 +} + +// PushStreamMsg 推送到 Redis Stream 的消息结构 +type PushStreamMsg struct { + ConversationID string `json:"conversation_id"` + SenderID string `json:"sender_id"` + MessageID string `json:"message_id"` + ConvType string `json:"conv_type"` + Timestamp int64 `json:"timestamp"` +} + +// PushWorker 从 Redis Stream 消费消息并异步推送 +type PushWorker struct { + rdb *redis.Client + publisher ws.MessagePublisher + pushSvc PushService + msgRepo repository.MessageRepository + userRepo repository.UserRepository + config PushWorkerConfig + stopCh chan struct{} + doneCh chan struct{} +} + +// NewPushWorker 创建推送 Worker +func NewPushWorker( + rdb *redis.Client, + publisher ws.MessagePublisher, + pushSvc PushService, + msgRepo repository.MessageRepository, + userRepo repository.UserRepository, + config PushWorkerConfig, +) *PushWorker { + if config.Stream == "" { + config.Stream = "msg_push" + } + if config.Group == "" { + config.Group = "push_worker" + } + if config.BatchSize <= 0 { + config.BatchSize = 50 + } + if config.PollTimeoutMs <= 0 { + config.PollTimeoutMs = 2000 + } + if config.MaxRetries <= 0 { + config.MaxRetries = 3 + } + if config.MaxStreamLen <= 0 { + config.MaxStreamLen = 100000 + } + if config.IdleTimeoutMs <= 0 { + config.IdleTimeoutMs = 30000 + } + return &PushWorker{ + rdb: rdb, + publisher: publisher, + pushSvc: pushSvc, + msgRepo: msgRepo, + userRepo: userRepo, + config: config, + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } +} + +// IsEnabled 是否启用 +func (w *PushWorker) IsEnabled() bool { + return w != nil && w.config.Enabled +} + +// Start 启动 Worker +func (w *PushWorker) Start(ctx context.Context) { + if !w.config.Enabled { + return + } + + // 创建 Consumer Group（忽略 BUSYGROUP 错误） + err := w.rdb.XGroupCreateMkStream(ctx, w.config.Stream, w.config.Group, "0").Err() + if err != nil { + // BUSYGROUP 错误可以忽略 + zap.L().Debug("XGroupCreateMkStream", zap.Error(err)) + } + + go w.consumeLoop() + go w.claimPendingLoop() + + zap.L().Info("PushWorker started", + zap.String("stream", w.config.Stream), + zap.String("group", w.config.Group), + ) +} + +// Stop 停止 Worker +func (w *PushWorker) Stop() { + if !w.config.Enabled { + return + } + close(w.stopCh) + // 等待两个循环退出 + <-w.doneCh + <-w.doneCh + zap.L().Info("PushWorker stopped") +} + +// consumeLoop 主消费循环 +func (w *PushWorker) consumeLoop() { + defer func() { w.doneCh <- struct{}{} }() + + consumerName := fmt.Sprintf("pw-%d", time.Now().UnixNano()) + args := &redis.XReadGroupArgs{ + Group: w.config.Group, + Consumer: consumerName, + Streams: []string{w.config.Stream, ">"}, + Count: int64(w.config.BatchSize), + Block: time.Duration(w.config.PollTimeoutMs) * time.Millisecond, + } + + for { + select { + case <-w.stopCh: + return + default: + } + + results, err := w.rdb.XReadGroup(context.Background(), args).Result() + if err != nil { + if err == redis.Nil { + continue + } + zap.L().Warn("PushWorker XReadGroup error", zap.Error(err)) + time.Sleep(1 * time.Second) + continue + } + + for _, stream := range results { + for _, msg := range stream.Messages { + if err := w.handleMsg(msg); err != nil { + zap.L().Warn("PushWorker handle message error", + zap.String("stream_id", msg.ID), + zap.Error(err), + ) + } + // ACK 无论成功失败（失败消息由 claimPendingLoop 重试） + w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID) + } + } + } +} + +// claimPendingLoop 定期认领超时未 ACK 的消息 +func (w *PushWorker) claimPendingLoop() { + defer func() { w.doneCh <- struct{}{} }() + + consumerName := fmt.Sprintf("pw-claim-%d", time.Now().UnixNano()) + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { + select { + case <-w.stopCh: + return + case <-ticker.C: + } + + // 查询 pending 消息 + pendingResult, err := w.rdb.XPendingExt(context.Background(), &redis.XPendingExtArgs{ + Stream: w.config.Stream, + Group: w.config.Group, + Start: "-", + End: "+", + Count: int64(w.config.BatchSize), + Idle: time.Duration(w.config.IdleTimeoutMs) * time.Millisecond, + }).Result() + if err != nil { + if err == redis.Nil { + continue + } + zap.L().Warn("PushWorker XPENDING error", zap.Error(err)) + continue + } + + for _, pending := range pendingResult { + // 认领消息 + claimed, err := w.rdb.XClaim(context.Background(), &redis.XClaimArgs{ + Stream: w.config.Stream, + Group: w.config.Group, + Consumer: consumerName, + MinIdle: time.Duration(w.config.IdleTimeoutMs) * time.Millisecond, + Messages: []string{pending.ID}, + }).Result() + if err != nil || len(claimed) == 0 { + continue + } + + retryCount := int(pending.RetryCount) + for _, msg := range claimed { + if retryCount >= w.config.MaxRetries { + zap.L().Warn("PushWorker message exceeded max retries, deleting", + zap.String("stream_id", msg.ID), + zap.Int("retry_count", retryCount), + ) + w.rdb.XDel(context.Background(), w.config.Stream, msg.ID) + continue + } + + if err := w.handleMsg(msg); err != nil { + zap.L().Warn("PushWorker claim retry handle error", + zap.String("stream_id", msg.ID), + zap.Int("retry_count", retryCount), + zap.Error(err), + ) + } + w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID) + } + } + } +} + +// handleMsg 处理一条 Stream 消息 +func (w *PushWorker) handleMsg(streamMsg redis.XMessage) error { + data, ok := streamMsg.Values["data"] + if !ok { + return fmt.Errorf("missing data field in stream message %s", streamMsg.ID) + } + + var pushMsg PushStreamMsg + dataStr, ok := data.(string) + if !ok { + return fmt.Errorf("data field is not string in stream message %s", streamMsg.ID) + } + if err := json.Unmarshal([]byte(dataStr), &pushMsg); err != nil { + return fmt.Errorf("unmarshal push message: %w", err) + } + + ctx := context.Background() + + // 加载消息 + message, err := w.msgRepo.GetMessageByID(pushMsg.MessageID) + if err != nil { + return fmt.Errorf("load message %s: %w", pushMsg.MessageID, err) + } + + // 加载会话 + conv, err := w.msgRepo.GetConversation(pushMsg.ConversationID) + if err != nil { + return fmt.Errorf("load conversation %s: %w", pushMsg.ConversationID, err) + } + + // 加载参与者 + participants, err := w.msgRepo.GetConversationParticipants(pushMsg.ConversationID) + if err != nil { + return fmt.Errorf("load participants for %s: %w", pushMsg.ConversationID, err) + } + + // ---- WS Push ---- + targetIDs := make([]string, 0, len(participants)) + for _, p := range participants { + if conv.Type == model.ConversationTypePrivate && p.UserID == pushMsg.SenderID { + continue + } + targetIDs = append(targetIDs, p.UserID) + } + + detailType := "private" + if conv.Type == model.ConversationTypeGroup { + detailType = "group" + } + + if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup { + w.publisher.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{ + "detail_type": detailType, + "message": dto.ConvertMessageToResponse(message), + }) + } else { + w.publisher.PublishToUsers(targetIDs, "chat_message", map[string]any{ + "detail_type": detailType, + "message": dto.ConvertMessageToResponse(message), + }) + } + + // ---- WS unread count push ---- + for _, p := range participants { + if p.UserID == pushMsg.SenderID { + continue + } + // PushWorker doesn't have conversationCache, skip unread count push here + // The inline push in chat_service still handles this when push worker is disabled + } + + // ---- JPush for offline users ---- + if w.pushSvc == nil { + return nil + } + + sender := ChatMessageSender{ID: pushMsg.SenderID} + if senderUser, sErr := w.userRepo.GetByID(pushMsg.SenderID); sErr == nil { + sender.Name = senderUser.Nickname + sender.Avatar = senderUser.Avatar + } + + convName := "" + groupID := "" + groupAvatar := "" + if conv.Type == model.ConversationTypeGroup && conv.Group != nil { + convName = conv.Group.Name + groupID = conv.Group.ID + groupAvatar = conv.Group.Avatar + } + + allTargetIDs := make([]string, 0, len(participants)) + for _, p := range participants { + if p.UserID == pushMsg.SenderID { + continue + } + allTargetIDs = append(allTargetIDs, p.UserID) + } + + var offlineTargetIDs []string + if w.publisher != nil && len(allTargetIDs) > 0 { + _, offlineTargetIDs = w.publisher.FilterOnline(allTargetIDs) + } else { + offlineTargetIDs = allTargetIDs + } + + for _, uid := range offlineTargetIDs { + if pushErr := w.pushSvc.PushChatMessage(ctx, uid, pushMsg.ConversationID, &sender, conv.Type, convName, message, groupID, groupAvatar); pushErr != nil { + zap.L().Debug("push worker: JPush chat message failed", + zap.String("userID", uid), + zap.String("conversationID", pushMsg.ConversationID), + zap.Error(pushErr), + ) + } + } + + return nil +} + +// PublishToPush 将消息发布到推送 Stream +func PublishToPush(ctx context.Context, rdb *redis.Client, stream string, msg *PushStreamMsg) error { + data, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshal push message: %w", err) + } + + return rdb.XAdd(ctx, &redis.XAddArgs{ + Stream: stream, + MaxLen: 100000, + Approx: true, + ID: "*", + Values: map[string]any{ + "data": string(data), + }, + }).Err() +} \ No newline at end of file diff --git a/internal/wire/infrastructure.go b/internal/wire/infrastructure.go index 862ac4f..38fbe84 100644 --- a/internal/wire/infrastructure.go +++ b/internal/wire/infrastructure.go @@ -18,6 +18,7 @@ import ( "with_you/internal/repository" "with_you/internal/service" + redisPkg "github.com/redis/go-redis/v9" "github.com/google/wire" "go.uber.org/zap" "gorm.io/gorm" @@ -49,6 +50,12 @@ var InfrastructureSet = wire.NewSet( // WebSocket MessagePublisher ProvideWSMessagePublisher, + // Seq 预分配管理器 + ProvideSeqBufferManager, + + // 推送 Worker + ProvidePushWorker, + // 外部服务客户端 ProvideOpenAIClient, ProvideTencentClient, @@ -275,3 +282,47 @@ func ProvideLogger() *zap.Logger { zap.ReplaceGlobals(logger) return logger } + +// ProvideSeqBufferManager 提供 Seq 预分配管理器 +func ProvideSeqBufferManager(cfg *config.Config, redisClient *redis.Client, cacheBackend cache.Cache) *cache.SeqBufferManager { + seqCfg := cache.SeqBufferConfig{ + Enabled: cfg.SeqBuffer.Enabled, + PrivateBufferSize: cfg.SeqBuffer.PrivateBufferSize, + GroupBufferSize: cfg.SeqBuffer.GroupBufferSize, + LockTimeoutMs: cfg.SeqBuffer.LockTimeoutMs, + MaxRetries: cfg.SeqBuffer.MaxRetries, + } + if !seqCfg.Enabled { + return cache.NewSeqBufferManager(nil, seqCfg, nil, nil) + } + return cache.NewSeqBufferManagerFromRedisPkg(redisClient, seqCfg, nil, nil) +} + +// ProvidePushWorker 提供推送 Worker（Redis Stream 异步推送） +func ProvidePushWorker( + cfg *config.Config, + redisClient *redis.Client, + publisher ws.MessagePublisher, + pushSvc service.PushService, + msgRepo repository.MessageRepository, + userRepo repository.UserRepository, +) *service.PushWorker { + pushCfg := service.PushWorkerConfig{ + Enabled: cfg.PushWorker.Enabled, + Stream: cfg.PushWorker.Stream, + Group: cfg.PushWorker.Group, + BatchSize: cfg.PushWorker.BatchSize, + PollTimeoutMs: cfg.PushWorker.PollTimeoutMs, + MaxRetries: cfg.PushWorker.MaxRetries, + MaxStreamLen: cfg.PushWorker.MaxStreamLen, + IdleTimeoutMs: cfg.PushWorker.IdleTimeoutMs, + } + if !pushCfg.Enabled { + return service.NewPushWorker(nil, nil, nil, nil, nil, pushCfg) + } + var rdb *redisPkg.Client + if redisClient != nil { + rdb = redisClient.GetClient() + } + return service.NewPushWorker(rdb, publisher, pushSvc, msgRepo, userRepo, pushCfg) +} diff --git a/internal/wire/repository.go b/internal/wire/repository.go index 0bcdfbe..7e47c60 100644 --- a/internal/wire/repository.go +++ b/internal/wire/repository.go @@ -51,6 +51,9 @@ var RepositorySet = wire.NewSet( // 二手交易相关 repository.NewTradeRepository, + + // 版本日志同步 + repository.NewConversationVersionLogRepository, ) // ProvideUserActivityRepository 提供用户活跃数据仓储 diff --git a/internal/wire/service.go b/internal/wire/service.go index eecc4f2..2c2fd6c 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -18,7 +18,7 @@ import ( "with_you/internal/service" "github.com/google/wire" - redislib "github.com/redis/go-redis/v9" + redisPkg "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" ) @@ -212,8 +212,21 @@ func ProvideChatService( cacheBackend cache.Cache, uploadService *service.UploadService, pushService service.PushService, + seqBufferMgr *cache.SeqBufferManager, + pushWorker *service.PushWorker, + redisClient *redis.Client, + versionLogRepo repository.ConversationVersionLogRepository, + cfg *config.Config, ) service.ChatService { - return service.NewChatService(messageRepo, userRepo, nil, publisher, cacheBackend, uploadService, pushService) + var vlogOpt []repository.ConversationVersionLogRepository + if cfg.VersionLog.Enabled { + vlogOpt = []repository.ConversationVersionLogRepository{versionLogRepo} + } + var rdb *redisPkg.Client + if redisClient != nil { + rdb = redisClient.GetClient() + } + return service.NewChatService(messageRepo, userRepo, nil, publisher, cacheBackend, uploadService, pushService, seqBufferMgr, pushWorker, rdb, vlogOpt...) } // ProvideScheduleService 提供日程服务 @@ -398,7 +411,7 @@ func ProvideLogService( // ProvideQRCodeLoginService 提供二维码登录服务 func ProvideHotRankWorker(cfg *config.Config, postRepo repository.PostRepository, channelRepo repository.ChannelRepository, cacheBackend cache.Cache, redisClient *redis.Client) *service.HotRankWorker { - var rdb *redislib.Client + var rdb *redisPkg.Client if redisClient != nil { rdb = redisClient.GetClient() }