diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 2f2f892..17bd49f 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -48,6 +48,8 @@ type Cache interface { HGetAll(ctx context.Context, key string) (map[string]string, error) // HDel 删除 Hash 字段 HDel(ctx context.Context, key string, fields ...string) error + // HIncrBy 原子递增 Hash 字段的值 + HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error) // ==================== Sorted Set 操作 ==================== // ZAdd 添加 Sorted Set 成员 @@ -68,6 +70,8 @@ type Cache interface { // ==================== 计数器操作 ==================== // Incr 原子递增（返回新值） Incr(ctx context.Context, key string) (int64, error) + // IncrBySeq 原子递增指定值（用于 seq 对齐） + IncrBySeq(ctx context.Context, key string, delta int64) (int64, error) // Expire 设置过期时间 Expire(ctx context.Context, key string, ttl time.Duration) error } @@ -95,7 +99,7 @@ var settings = Settings{ NullTTL: 5 * time.Second, JitterRatio: 0.1, PostListTTL: 30 * time.Second, - ConversationTTL: 60 * time.Second, + ConversationTTL: 120 * time.Second, UnreadCountTTL: 30 * time.Second, GroupMembersTTL: 120 * time.Second, DisableFlushDB: true, diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index ceabeea..94ad026 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -74,19 +74,23 @@ type ConversationCacheSettings struct { MessageListTTL time.Duration // 消息分页列表缓存 (5min) MessageIndexTTL time.Duration // 消息索引缓存 (30min) MessageCountTTL time.Duration // 消息计数缓存 (30min) + + // Seq 缓存配置 + SeqTTL time.Duration // Seq 计数器缓存 (24h) } // DefaultConversationCacheSettings 返回默认配置 func DefaultConversationCacheSettings() *ConversationCacheSettings { return &ConversationCacheSettings{ - DetailTTL: 5 * time.Minute, - ListTTL: 60 * time.Second, - ParticipantTTL: 5 * time.Minute, + DetailTTL: 15 * time.Minute, + ListTTL: 120 * time.Second, + ParticipantTTL: 15 * time.Minute, UnreadTTL: 30 * time.Second, MessageDetailTTL: 30 * time.Minute, MessageListTTL: 5 * time.Minute, MessageIndexTTL: 30 * time.Minute, MessageCountTTL: 30 * time.Minute, + SeqTTL: 24 * time.Hour, } } @@ -480,6 +484,142 @@ func (c *ConversationCache) InvalidateUnreadCount(userID, convID string) { c.cache.Delete(UnreadDetailKey(userID, convID)) } +// IncrementUnread 递增用户在某会话的未读数（Redis Hash + 总数计数器） +func (c *ConversationCache) IncrementUnread(ctx context.Context, userID, convID string) error { + hashKey := UnreadHashKey(userID) + totalKey := UnreadTotalKey(userID) + + _, err := c.cache.HIncrBy(ctx, hashKey, convID, 1) + if err != nil { + return fmt.Errorf("hincrby unread failed: %w", err) + } + + _, err = c.cache.Incr(ctx, totalKey) + if err != nil { + return fmt.Errorf("incr unread total failed: %w", err) + } + + _ = c.cache.Expire(ctx, hashKey, c.settings.UnreadTTL*10) + _ = c.cache.Expire(ctx, totalKey, c.settings.UnreadTTL*10) + return nil +} + +// ClearUnread 清零用户在某会话的未读数 +func (c *ConversationCache) ClearUnread(ctx context.Context, userID, convID string) error { + hashKey := UnreadHashKey(userID) + totalKey := UnreadTotalKey(userID) + + // 读取当前未读数 + currentStr, err := c.cache.HGet(ctx, hashKey, convID) + if err != nil { + currentStr = "0" + } + current := int64(0) + if v, parseErr := fmt.Sscanf(currentStr, "%d", ¤t); parseErr != nil || v != 1 { + current = 0 + } + + // 清零该会话未读数 + if err := c.cache.HSet(ctx, hashKey, convID, "0"); err != nil { + return fmt.Errorf("hset clear unread failed: %w", err) + } + + // 减少总未读数 + if current > 0 { + c.cache.IncrBySeq(ctx, totalKey, -current) + } + + return nil +} + +// GetUnreadCountFromHash 从 Redis Hash 获取单个会话未读数 +func (c *ConversationCache) GetUnreadCountFromHash(ctx context.Context, userID, convID string) (int64, error) { + hashKey := UnreadHashKey(userID) + val, err := c.cache.HGet(ctx, hashKey, convID) + if err != nil { + return 0, err + } + var count int64 + fmt.Sscanf(val, "%d", &count) + return count, nil +} + +// GetAllUnreadCounts 从 Redis Hash 获取用户所有会话的未读数 +func (c *ConversationCache) GetAllUnreadCounts(ctx context.Context, userID string) (map[string]int64, error) { + hashKey := UnreadHashKey(userID) + all, err := c.cache.HGetAll(ctx, hashKey) + if err != nil { + return nil, err + } + result := make(map[string]int64, len(all)) + for k, v := range all { + var count int64 + fmt.Sscanf(v, "%d", &count) + result[k] = count + } + return result, nil +} + +// GetTotalUnread 从 Redis 获取用户总未读数 +func (c *ConversationCache) GetTotalUnread(ctx context.Context, userID string) (int64, error) { + totalKey := UnreadTotalKey(userID) + + redisCache, ok := c.cache.(*RedisCache) + if ok { + val, err := redisCache.GetRedisClient().GetClient().Get(ctx, normalizeKey(totalKey)).Int64() + if err != nil { + return 0, nil + } + return val, nil + } + + raw, ok := c.cache.Get(totalKey) + if !ok { + return 0, nil + } + switch v := raw.(type) { + case int64: + return v, nil + case string: + var n int64 + fmt.Sscanf(v, "%d", &n) + return n, nil + } + return 0, nil +} + +// GetNextSeq 获取会话的下一个 seq 值（原子递增） +// 使用 Redis INCR 实现，首次调用时从 DB 初始化并补齐差值 +func (c *ConversationCache) GetNextSeq(ctx context.Context, convID string) (int64, error) { + seqKey := MessageSeqKey(convID) + + newSeq, err := c.cache.Incr(ctx, seqKey) + if err != nil { + return 0, fmt.Errorf("redis incr seq failed: %w", err) + } + + // 首次调用时 Redis key 不存在，INCR 从 0 返回 1 + 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 { + // 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) + } + + return newSeq, nil +} + // ============================================================ // 消息缓存方法 // ============================================================ diff --git a/internal/cache/keys.go b/internal/cache/keys.go index 6ae5598..8ee7d0b 100644 --- a/internal/cache/keys.go +++ b/internal/cache/keys.go @@ -26,6 +26,8 @@ const ( PrefixUnreadSystem = "unread:system" PrefixUnreadConversation = "unread:conversation" PrefixUnreadDetail = "unread:detail" + PrefixUnreadHash = "unread:hash" + PrefixUnreadTotal = "unread:total" // 用户相关 PrefixUserInfo = "users:info" @@ -118,6 +120,16 @@ func UnreadDetailKey(userID, conversationID string) string { return fmt.Sprintf("%s:%s:%s", PrefixUnreadDetail, userID, conversationID) } +// UnreadHashKey 生成用户未读数 Hash 缓存键 +func UnreadHashKey(userID string) string { + return fmt.Sprintf("%s:%s", PrefixUnreadHash, userID) +} + +// UnreadTotalKey 生成用户总未读数计数器键 +func UnreadTotalKey(userID string) string { + return fmt.Sprintf("%s:%s", PrefixUnreadTotal, userID) +} + // UserInfoKey 生成用户信息缓存键 func UserInfoKey(userID string) string { return fmt.Sprintf("%s:%s", PrefixUserInfo, userID) diff --git a/internal/cache/layered_cache.go b/internal/cache/layered_cache.go index 4455727..6f776a0 100644 --- a/internal/cache/layered_cache.go +++ b/internal/cache/layered_cache.go @@ -294,6 +294,14 @@ func (c *LayeredCache) HDel(ctx context.Context, key string, fields ...string) e return c.redis.HDel(ctx, key, fields...) } +func (c *LayeredCache) HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error) { + if !c.enabled || c.redis == nil { + return 0, ErrKeyNotFound + } + c.local.Delete(key) + return c.redis.HIncrBy(ctx, key, field, incr) +} + func (c *LayeredCache) ZAdd(ctx context.Context, key string, score float64, member string) error { if !c.enabled || c.redis == nil { return nil @@ -354,6 +362,14 @@ func (c *LayeredCache) Incr(ctx context.Context, key string) (int64, error) { return c.redis.Incr(ctx, key) } +func (c *LayeredCache) IncrBySeq(ctx context.Context, key string, delta int64) (int64, error) { + if !c.enabled || c.redis == nil { + return 0, ErrKeyNotFound + } + c.local.Delete(key) + return c.redis.IncrBySeq(ctx, key, delta) +} + func (c *LayeredCache) Expire(ctx context.Context, key string, ttl time.Duration) error { if !c.enabled || c.redis == nil { return nil diff --git a/internal/cache/redis_cache.go b/internal/cache/redis_cache.go index 05814ec..906ac80 100644 --- a/internal/cache/redis_cache.go +++ b/internal/cache/redis_cache.go @@ -222,6 +222,12 @@ func (c *RedisCache) HDel(ctx context.Context, key string, fields ...string) err return c.client.HDel(ctx, key, fields...) } +// HIncrBy 原子递增 Hash 字段的值 +func (c *RedisCache) HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error) { + key = normalizeKey(key) + return c.client.HIncrBy(ctx, key, field, incr) +} + // ==================== RedisCache Sorted Set 操作 ==================== // ZAdd 添加 Sorted Set 成员 @@ -281,6 +287,13 @@ func (c *RedisCache) Incr(ctx context.Context, key string) (int64, error) { return c.client.Incr(ctx, key) } +// IncrBySeq 原子递增指定值（用于 seq 对齐） +func (c *RedisCache) IncrBySeq(ctx context.Context, key string, delta int64) (int64, error) { + key = normalizeKey(key) + rdb := c.client.GetClient() + return rdb.IncrBy(ctx, key, delta).Result() +} + // Expire 设置过期时间 func (c *RedisCache) Expire(ctx context.Context, key string, ttl time.Duration) error { key = normalizeKey(key) diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 033e871..4654e13 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -14,6 +14,76 @@ import ( "with_you/internal/service" ) +// enrichConversations 批量填充会话列表响应数据（解决 N+1 问题） +func (h *MessageHandler) enrichConversations(ctx context.Context, convs []*model.Conversation, userID string) []*dto.ConversationResponse { + if len(convs) == 0 { + return nil + } + + convIDs := make([]string, len(convs)) + for i, c := range convs { + convIDs[i] = c.ID + } + + // 批量查询：4-5 次查询替代 N*5 次 + unreadCounts, _ := h.chatService.GetUnreadCountBatch(ctx, userID, convIDs) + lastMessages, _ := h.chatService.GetLastMessagesBatch(ctx, convIDs) + myParticipants, _ := h.messageService.GetMyParticipantsBatch(ctx, convIDs, userID) + allParticipants, _ := h.messageService.GetParticipantsBatch(ctx, convIDs) + + // 收集需要额外查询的 ID + var groupIDs []string + var userIDs []string + for _, conv := range convs { + if conv.Type == model.ConversationTypeGroup && conv.GroupID != nil && *conv.GroupID != "" { + groupIDs = append(groupIDs, *conv.GroupID) + } + } + for _, participants := range allParticipants { + for _, p := range participants { + if p.UserID != userID { + userIDs = append(userIDs, p.UserID) + } + } + } + + memberCounts, _ := h.groupService.GetMemberCountBatch(ctx, groupIDs) + usersMap, _ := h.userService.GetUsersByIDs(ctx, userIDs) + + // 组装响应 + result := make([]*dto.ConversationResponse, len(convs)) + for i, conv := range convs { + unreadCount := unreadCounts[conv.ID] + lastMessage := lastMessages[conv.ID] + + myP := myParticipants[conv.ID] + isPinned := myP != nil && myP.IsPinned + notificationMuted := myP != nil && myP.NotificationMuted + + var resp *dto.ConversationResponse + if conv.Type == model.ConversationTypeGroup && conv.GroupID != nil && *conv.GroupID != "" { + resp = dto.ConvertConversationToResponse(conv, nil, int(unreadCount), lastMessage, isPinned, notificationMuted) + if mc, ok := memberCounts[*conv.GroupID]; ok { + resp.MemberCount = int(mc) + } + } else { + participants := allParticipants[conv.ID] + var users []*model.User + for _, p := range participants { + if p.UserID == userID { + continue + } + if u, ok := usersMap[p.UserID]; ok { + users = append(users, u) + } + } + resp = dto.ConvertConversationToResponse(conv, users, int(unreadCount), lastMessage, isPinned, notificationMuted) + } + result[i] = resp + } + return result +} + // MessageHandler 消息处理器 type MessageHandler struct { chatService service.ChatService @@ -80,37 +150,8 @@ func (h *MessageHandler) GetConversations(c *gin.Context) { } } - // 转换为响应格式 - result := make([]*dto.ConversationResponse, len(filteredConvs)) - for i, conv := range filteredConvs { - // 获取未读数 - unreadCount, _ := h.chatService.GetUnreadCount(c.Request.Context(), conv.ID, userID) - - // 获取最后一条消息 - var lastMessage *model.Message - messages, _, _ := h.chatService.GetMessages(c.Request.Context(), conv.ID, userID, 1, 1) - if len(messages) > 0 { - lastMessage = messages[0] - } - - // 群聊时返回member_count，私聊时返回participants - var resp *dto.ConversationResponse - myParticipant, _ := h.getMyConversationParticipant(conv.ID, userID) - isPinned := myParticipant != nil && myParticipant.IsPinned - notificationMuted := myParticipant != nil && myParticipant.NotificationMuted - if conv.Type == model.ConversationTypeGroup && conv.GroupID != nil && *conv.GroupID != "" { - // 群聊：实时计算群成员数量 - memberCount, _ := h.groupService.GetMemberCount(*conv.GroupID) - // 创建响应并设置member_count - resp = dto.ConvertConversationToResponse(conv, nil, int(unreadCount), lastMessage, isPinned, notificationMuted) - resp.MemberCount = memberCount - } else { - // 私聊：获取参与者信息 - participants, _ := h.getConversationParticipants(c.Request.Context(), conv.ID, userID) - resp = dto.ConvertConversationToResponse(conv, participants, int(unreadCount), lastMessage, isPinned, notificationMuted) - } - result[i] = resp - } + // 批量填充会话数据（解决 N+1 问题） + result := h.enrichConversations(c.Request.Context(), filteredConvs, userID) // 更新 total 为过滤后的数量 response.Paginated(c, result, int64(len(filteredConvs)), page, pageSize) @@ -437,37 +478,8 @@ func (h *MessageHandler) HandleGetConversationList(c *gin.Context) { } } - // 转换为响应格式 - result := make([]*dto.ConversationResponse, len(filteredConvs)) - for i, conv := range filteredConvs { - // 获取未读数 - unreadCount, _ := h.chatService.GetUnreadCount(c.Request.Context(), conv.ID, userID) - - // 获取最后一条消息 - var lastMessage *model.Message - messages, _, _ := h.chatService.GetMessages(c.Request.Context(), conv.ID, userID, 1, 1) - if len(messages) > 0 { - lastMessage = messages[0] - } - - // 群聊时返回member_count，私聊时返回participants - var resp *dto.ConversationResponse - myParticipant, _ := h.getMyConversationParticipant(conv.ID, userID) - isPinned := myParticipant != nil && myParticipant.IsPinned - notificationMuted := myParticipant != nil && myParticipant.NotificationMuted - if conv.Type == model.ConversationTypeGroup && conv.GroupID != nil && *conv.GroupID != "" { - // 群聊：实时计算群成员数量 - memberCount, _ := h.groupService.GetMemberCount(*conv.GroupID) - // 创建响应并设置member_count - resp = dto.ConvertConversationToResponse(conv, nil, int(unreadCount), lastMessage, isPinned, notificationMuted) - resp.MemberCount = memberCount - } else { - // 私聊：获取参与者信息 - participants, _ := h.getConversationParticipants(c.Request.Context(), conv.ID, userID) - resp = dto.ConvertConversationToResponse(conv, participants, int(unreadCount), lastMessage, isPinned, notificationMuted) - } - result[i] = resp - } + // 批量填充会话数据（解决 N+1 问题） + result := h.enrichConversations(c.Request.Context(), filteredConvs, userID) response.Paginated(c, result, int64(len(filteredConvs)), page, pageSize) } @@ -1017,37 +1029,8 @@ func (h *MessageHandler) GetConversationsByCursor(c *gin.Context) { } } - // 转换为响应格式 - items := make([]*dto.ConversationResponse, len(filteredItems)) - for i, conv := range filteredItems { - // 获取未读数 - unreadCount, _ := h.chatService.GetUnreadCount(c.Request.Context(), conv.ID, userID) - - // 获取最后一条消息 - var lastMessage *model.Message - messages, _, _ := h.chatService.GetMessages(c.Request.Context(), conv.ID, userID, 1, 1) - if len(messages) > 0 { - lastMessage = messages[0] - } - - // 群聊时返回member_count，私聊时返回participants - var resp *dto.ConversationResponse - myParticipant, _ := h.getMyConversationParticipant(conv.ID, userID) - isPinned := myParticipant != nil && myParticipant.IsPinned - notificationMuted := myParticipant != nil && myParticipant.NotificationMuted - if conv.Type == model.ConversationTypeGroup && conv.GroupID != nil && *conv.GroupID != "" { - // 群聊：实时计算群成员数量 - memberCount, _ := h.groupService.GetMemberCount(*conv.GroupID) - resp = dto.ConvertConversationToResponse(conv, nil, int(unreadCount), lastMessage, isPinned, notificationMuted) - resp.MemberCount = memberCount - } else { - // 私聊：获取参与者信息 - participants, _ := h.getConversationParticipants(c.Request.Context(), conv.ID, userID) - resp = dto.ConvertConversationToResponse(conv, participants, int(unreadCount), lastMessage, isPinned, notificationMuted) - } - items[i] = resp - } - + // 批量填充会话数据（解决 N+1 问题） + items := h.enrichConversations(c.Request.Context(), filteredItems, userID) response.Success(c, &dto.ConversationCursorPageResponse{ Items: items, NextCursor: result.NextCursor, diff --git a/internal/handler/qrcode_handler.go b/internal/handler/qrcode_handler.go index 4825f26..dae209d 100644 --- a/internal/handler/qrcode_handler.go +++ b/internal/handler/qrcode_handler.go @@ -53,7 +53,7 @@ func (h *QRCodeHandler) WSEvents(c *gin.Context) { return } - ch, cancel, replay := h.qrcodeService.GetWSHub().Subscribe(sessionID, 0) + ch, cancel := h.qrcodeService.GetWSHub().Subscribe(sessionID, 0) defer cancel() w := c.Writer @@ -82,13 +82,6 @@ func (h *QRCodeHandler) WSEvents(c *gin.Context) { return true } - // 发送历史事件 - for _, ev := range replay { - if !writeEvent(ev) { - return - } - } - // 心跳 heartbeat := time.NewTicker(25 * time.Second) defer heartbeat.Stop() diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 006192a..0abdab9 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -140,14 +140,15 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { // 3. 创建客户端 clientID := atomic.AddUint64(&h.clientSeq, 1) client := &ws.Client{ - ID: clientID, - UserID: userID, - Send: make(chan []byte, defaultUserBufferSize), - Quit: make(chan struct{}), + ID: clientID, + UserID: userID, + Send: make(chan []byte, defaultUserBufferSize), + Quit: make(chan struct{}), + PendingAcks: make(map[string]time.Time), } // 4. 注册客户端 - replayEvents, regErr := h.wsHub.Register(client) + regErr := h.wsHub.Register(client) if regErr != nil { zap.L().Warn("WebSocket registration rejected", zap.String("user_id", userID), @@ -165,23 +166,18 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { zap.Uint64("client_id", clientID), ) - // 5. 发送历史回放消息 - go func() { - for _, ev := range replayEvents { - msg := ws.ResponseMessage{ - EventID: ev.ID, - Type: ev.Type, - TS: ev.TS, - Payload: ev.Payload, - } - data, _ := json.Marshal(msg) - select { - case <-client.Quit: - return - case client.Send <- data: - } + // 5. 提示客户端进行 seq 同步 + syncMsg := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "sync_required", + TS: time.Now().UnixMilli(), + } + if syncData, err := json.Marshal(syncMsg); err == nil { + select { + case <-client.Quit: + case client.Send <- syncData: } - }() + } // 6. 启动读写goroutine go h.writePump(conn, client) @@ -322,6 +318,8 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) { h.handleRead(ctx, client, msg.Payload) case "recall": h.handleRecall(ctx, client, msg.Payload) + case "ack": + h.handleAck(client, msg.Payload) // 通话信令 case "call_invite": h.handleCallInvite(ctx, client, msg.Payload) @@ -511,6 +509,17 @@ func (h *WSHandler) handleRecall(ctx context.Context, client *ws.Client, payload const defaultUserBufferSize = 128 +// handleAck 处理客户端ACK确认 +func (h *WSHandler) handleAck(client *ws.Client, payload json.RawMessage) { + var req struct { + MessageID string `json:"message_id"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.MessageID == "" { + return + } + h.wsHub.AckMessage(client.UserID, req.MessageID) +} + // isVerified 检查用户是否已通过身份认证 func (h *WSHandler) isVerified(ctx context.Context, client *ws.Client) bool { user, err := h.userRepo.GetByID(client.UserID) diff --git a/internal/pkg/redis/redis.go b/internal/pkg/redis/redis.go index 22b4ad4..c850b4c 100644 --- a/internal/pkg/redis/redis.go +++ b/internal/pkg/redis/redis.go @@ -150,6 +150,11 @@ func (c *Client) HDel(ctx context.Context, key string, fields ...string) error { return c.rdb.HDel(ctx, key, fields...).Err() } +// HIncrBy 原子递增 Hash 字段的值 +func (c *Client) HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error) { + return c.rdb.HIncrBy(ctx, key, field, incr).Result() +} + // HExists 检查 Hash 字段是否存在 func (c *Client) HExists(ctx context.Context, key string, field string) (bool, error) { return c.rdb.HExists(ctx, key, field).Result() diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go index deb2bc5..5f0a87f 100644 --- a/internal/pkg/ws/hub.go +++ b/internal/pkg/ws/hub.go @@ -12,11 +12,12 @@ import ( const ( defaultUserBufferSize = 128 - maxReplayEvents = 200 groupPushConcurrency = 64 groupPushQueueSize = 512 maxTotalConnections = 100000 maxConnectionsPerUser = 10 + ackTimeout = 30 * time.Second + ackCheckInterval = 10 * time.Second ) // Event 表示一个WebSocket事件 @@ -30,10 +31,11 @@ type Event struct { // Client 表示一个WebSocket客户端连接 type Client struct { - ID uint64 - UserID string - Send chan []byte - Quit chan struct{} + ID uint64 + UserID string + Send chan []byte + Quit chan struct{} + PendingAcks map[string]time.Time } // ErrConnectionLimit 连接数限制错误 @@ -86,7 +88,6 @@ type Hub struct { mu sync.RWMutex clients map[string]map[uint64]*Client - history map[string][]Event subscribers map[string]map[uint64]*subscriber disconnectHandlers []DisconnectHandler connectHandlers []ConnectHandler @@ -96,7 +97,6 @@ type Hub struct { func NewHub() *Hub { return &Hub{ clients: make(map[string]map[uint64]*Client), - history: make(map[string][]Event), subscribers: make(map[string]map[uint64]*subscriber), disconnectHandlers: make([]DisconnectHandler, 0), connectHandlers: make([]ConnectHandler, 0), @@ -108,17 +108,17 @@ func (h *Hub) NextID() uint64 { return atomic.AddUint64(&h.seq, 1) } -// Register 注册客户端连接，返回回放事件和可能的错误（连接数超限） -func (h *Hub) Register(client *Client) ([]Event, error) { +// Register 注册客户端连接，返回可能的错误（连接数超限） +func (h *Hub) Register(client *Client) error { if h.connCount.Load() >= int64(maxTotalConnections) { - return nil, ErrConnectionLimit + return ErrConnectionLimit } h.mu.Lock() if len(h.clients[client.UserID]) >= maxConnectionsPerUser { h.mu.Unlock() - return nil, ErrConnectionLimit + return ErrConnectionLimit } _, existed := h.clients[client.UserID] @@ -130,11 +130,6 @@ func (h *Hub) Register(client *Client) ([]Event, error) { h.clients[client.UserID][client.ID] = client h.connCount.Add(1) - replay := make([]Event, 0) - for _, e := range h.history[client.UserID] { - replay = append(replay, e) - } - zap.L().Debug("WebSocket client registered", zap.String("user_id", client.UserID), zap.Uint64("client_id", client.ID), @@ -155,7 +150,7 @@ func (h *Hub) Register(client *Client) ([]Event, error) { }() } - return replay, nil + return nil } // Unregister 注销客户端连接 @@ -290,18 +285,12 @@ func (h *Hub) PublishToUsers(userIDs []string, eventType string, payload any) { // publishPreMarshaled 使用预序列化的数据发送事件 func (h *Hub) publishPreMarshaled(userID string, ev Event, data []byte) { - h.mu.Lock() - history := append(h.history[userID], ev) - if len(history) > maxReplayEvents { - history = history[len(history)-maxReplayEvents:] - } - h.history[userID] = history - + h.mu.RLock() targets := make([]*Client, 0, len(h.clients[userID])) for _, c := range h.clients[userID] { targets = append(targets, c) } - h.mu.Unlock() + h.mu.RUnlock() for _, c := range targets { select { @@ -392,20 +381,12 @@ func (h *Hub) PublishToUserOnlineReliable(userID string, eventType string, paylo // publish 内部发布方法 func (h *Hub) publish(userID string, ev Event) { - h.mu.Lock() - // 存储历史事件（最多保留200条） - history := append(h.history[userID], ev) - if len(history) > maxReplayEvents { - history = history[len(history)-maxReplayEvents:] - } - h.history[userID] = history - - // 复制客户端列表避免锁竞争 + h.mu.RLock() targets := make([]*Client, 0, len(h.clients[userID])) for _, c := range h.clients[userID] { targets = append(targets, c) } - h.mu.Unlock() + h.mu.RUnlock() // 构建消息 msg := ResponseMessage{ @@ -532,17 +513,6 @@ func (h *Hub) PublishToUsersConcurrent(userIDs []string, eventType string, paylo } wg.Wait() - - // Store history for each user for reconnection replay - h.mu.Lock() - for _, uid := range userIDs { - history := append(h.history[uid], ev) - if len(history) > maxReplayEvents { - history = history[len(history)-maxReplayEvents:] - } - h.history[uid] = history - } - h.mu.Unlock() } // GetOnlineUsers 获取在线用户列表 @@ -588,7 +558,7 @@ func (h *Hub) CloseAllConnections() { } // Subscribe 订阅事件 -func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Event) { +func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func()) { subID := h.NextID() sub := &subscriber{ id: subID, @@ -601,13 +571,6 @@ func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Ev h.subscribers[userID] = make(map[uint64]*subscriber) } h.subscribers[userID][subID] = sub - - replay := make([]Event, 0) - for _, e := range h.history[userID] { - if e.ID > afterID { - replay = append(replay, e) - } - } h.mu.Unlock() cancel := func() { @@ -625,7 +588,59 @@ func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Ev } } - return sub.ch, cancel, replay + return sub.ch, cancel +} + +// AckMessage 确认消息已送达，从 pendingAcks 中移除 +func (h *Hub) AckMessage(userID, messageID string) bool { + h.mu.RLock() + defer h.mu.RUnlock() + + for _, client := range h.clients[userID] { + if _, ok := client.PendingAcks[messageID]; ok { + delete(client.PendingAcks, messageID) + return true + } + } + return false +} + +// startAckChecker 启动超时 ACK 检查器 +func (h *Hub) startAckChecker(onTimeout func(userID, messageID string)) { + ticker := time.NewTicker(ackCheckInterval) + defer ticker.Stop() + + for range ticker.C { + h.mu.RLock() + // 收集超时的 ack + type timeoutEntry struct { + userID string + messageID string + } + var timeouts []timeoutEntry + + now := time.Now() + for uid, userClients := range h.clients { + for _, client := range userClients { + for msgID, sentAt := range client.PendingAcks { + if now.Sub(sentAt) > ackTimeout { + timeouts = append(timeouts, timeoutEntry{uid, msgID}) + delete(client.PendingAcks, msgID) + } + } + } + } + h.mu.RUnlock() + + for _, entry := range timeouts { + onTimeout(entry.userID, entry.messageID) + } + } +} + +// StartAckChecker 启动 ACK 检查器（公开方法） +func (h *Hub) StartAckChecker(onTimeout func(userID, messageID string)) { + go h.startAckChecker(onTimeout) } // EncodeData 编码事件数据 diff --git a/internal/repository/group_repo.go b/internal/repository/group_repo.go index b4a1cf8..0a3e623 100644 --- a/internal/repository/group_repo.go +++ b/internal/repository/group_repo.go @@ -27,6 +27,7 @@ type GroupRepository interface { UpdateMember(member *model.GroupMember) error RemoveMember(groupID string, userID string) error GetMemberCount(groupID string) (int64, error) + GetMemberCountBatch(ctx context.Context, groupIDs []string) (map[string]int64, error) IsMember(groupID string, userID string) (bool, error) GetUserGroups(userID string, page, pageSize int) ([]model.Group, int64, error) @@ -225,6 +226,34 @@ func (r *groupRepository) GetMemberCount(groupID string) (int64, error) { return count, err } +// GetMemberCountBatch 批量获取多个群的成员数量 +func (r *groupRepository) GetMemberCountBatch(ctx context.Context, groupIDs []string) (map[string]int64, error) { + result := make(map[string]int64, len(groupIDs)) + if len(groupIDs) == 0 { + return result, nil + } + + type row struct { + GroupID string + Count int64 + } + var rows []row + err := r.db.WithContext(ctx). + Model(&model.GroupMember{}). + Select("group_id, COUNT(*) as count"). + Where("group_id IN ?", groupIDs). + Group("group_id"). + Scan(&rows).Error + if err != nil { + return nil, err + } + + for _, r := range rows { + result[r.GroupID] = r.Count + } + return result, nil +} + // IsMember 检查是否是群成员 func (r *groupRepository) IsMember(groupID string, userID string) (bool, error) { var count int64 diff --git a/internal/repository/message_repo.go b/internal/repository/message_repo.go index 7dfb597..7b78130 100644 --- a/internal/repository/message_repo.go +++ b/internal/repository/message_repo.go @@ -54,6 +54,12 @@ type MessageRepository interface { UpdateConversationLastSeqWithTx(tx *gorm.DB, convID string, lastSeq int64, lastMsgTime time.Time) error GetMessagesByCursor(ctx context.Context, conversationID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Message], error) GetConversationsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Conversation], error) + + // 批量查询方法（解决 N+1 问题） + GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) + GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) + GetParticipantsBatch(ctx context.Context, convIDs []string) (map[string][]*model.ConversationParticipant, error) + GetMyParticipantsBatch(ctx context.Context, convIDs []string, userID string) (map[string]*model.ConversationParticipant, error) } // messageRepository 消息仓储实现 @@ -405,17 +411,10 @@ func (r *messageRepository) GetNextSeq(conversationID string) (int64, error) { } // CreateMessageWithSeq 创建消息并更新seq（事务操作） +// msg.Seq 应由调用方通过 Redis INCR 预分配 func (r *messageRepository) CreateMessageWithSeq(msg *model.Message) error { return r.db.Transaction(func(tx *gorm.DB) error { - // 获取当前seq并+1 - var conv model.Conversation - if err := tx.Select("last_seq").Where("id = ?", msg.ConversationID).First(&conv).Error; err != nil { - return err - } - - msg.Seq = conv.LastSeq + 1 - - // 创建消息 + // 创建消息（seq 已由调用方预分配） if err := tx.Create(msg).Error; err != nil { return err } @@ -430,7 +429,7 @@ func (r *messageRepository) CreateMessageWithSeq(msg *model.Message) error { return err } - // 新消息到达后，自动恢复被“仅自己删除”的会话 + // 新消息到达后，自动恢复被"仅自己删除"的会话 if err := tx.Model(&model.ConversationParticipant{}). Where("conversation_id = ?", msg.ConversationID). Update("hidden_at", nil).Error; err != nil { @@ -883,3 +882,101 @@ func (r *messageRepository) GetConversationsByCursor(ctx context.Context, userID return cursor.NewCursorPageResult(conversations, nextCursor, prevCursor, hasMore), nil } + +// GetUnreadCountBatch 批量获取用户在多个会话中的未读数 +func (r *messageRepository) GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) { + result := make(map[string]int64, len(convIDs)) + if len(convIDs) == 0 { + return result, nil + } + + type row struct { + ConversationID string + Count int64 + } + var rows []row + err := r.db.WithContext(ctx). + Table("conversation_participants cp"). + Select("cp.conversation_id, COALESCE(cnt.unread, 0) as count"). + Joins("LEFT JOIN (SELECT m.conversation_id, COUNT(m.id) as unread FROM messages m WHERE m.sender_id <> ? AND m.deleted_at IS NULL AND m.conversation_id IN (?) GROUP BY m.conversation_id) cnt ON cnt.conversation_id = cp.conversation_id AND cnt.unread > cp.last_read_seq", userID, convIDs). + Where("cp.user_id = ? AND cp.conversation_id IN (?)", userID, convIDs). + Scan(&rows).Error + if err != nil { + return nil, err + } + + for _, id := range convIDs { + result[id] = 0 + } + for _, r := range rows { + result[r.ConversationID] = r.Count + } + return result, nil +} + +// GetLastMessagesBatch 批量获取每个会话的最后一条消息 +func (r *messageRepository) GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) { + result := make(map[string]*model.Message, len(convIDs)) + if len(convIDs) == 0 { + return result, nil + } + + var messages []*model.Message + err := r.db.WithContext(ctx). + Where("id IN (?)", r.db. + Select("MAX(id)"). + Table("messages"). + Where("conversation_id IN (?) AND deleted_at IS NULL", convIDs). + Group("conversation_id")). + Find(&messages).Error + if err != nil { + return nil, err + } + + for _, m := range messages { + result[m.ConversationID] = m + } + return result, nil +} + +// GetParticipantsBatch 批量获取多个会话的参与者列表 +func (r *messageRepository) GetParticipantsBatch(ctx context.Context, convIDs []string) (map[string][]*model.ConversationParticipant, error) { + result := make(map[string][]*model.ConversationParticipant, len(convIDs)) + if len(convIDs) == 0 { + return result, nil + } + + var participants []*model.ConversationParticipant + err := r.db.WithContext(ctx). + Where("conversation_id IN (?)", convIDs). + Find(&participants).Error + if err != nil { + return nil, err + } + + for _, p := range participants { + result[p.ConversationID] = append(result[p.ConversationID], p) + } + return result, nil +} + +// GetMyParticipantsBatch 批量获取用户在多个会话中的参与者信息 +func (r *messageRepository) GetMyParticipantsBatch(ctx context.Context, convIDs []string, userID string) (map[string]*model.ConversationParticipant, error) { + result := make(map[string]*model.ConversationParticipant, len(convIDs)) + if len(convIDs) == 0 { + return result, nil + } + + var participants []*model.ConversationParticipant + err := r.db.WithContext(ctx). + Where("conversation_id IN (?) AND user_id = ?", convIDs, userID). + Find(&participants).Error + if err != nil { + return nil, err + } + + for _, p := range participants { + result[p.ConversationID] = p + } + return result, nil +} diff --git a/internal/repository/user_repo.go b/internal/repository/user_repo.go index d6d4fdf..f88b98f 100644 --- a/internal/repository/user_repo.go +++ b/internal/repository/user_repo.go @@ -1,6 +1,7 @@ package repository import ( + "context" "strings" "time" "with_you/internal/model" @@ -13,6 +14,7 @@ import ( type UserRepository interface { Create(user *model.User) error GetByID(id string) (*model.User, error) + GetUsersByIDs(ctx context.Context, ids []string) (map[string]*model.User, error) GetByUsername(username string) (*model.User, error) GetByEmail(email string) (*model.User, error) GetByPhone(phone string) (*model.User, error) @@ -607,3 +609,22 @@ func (r *userRepository) GetPendingDeletionUsers(beforeTime time.Time) ([]*model func (r *userRepository) HardDeleteUser(userID string) error { return r.db.Unscoped().Delete(&model.User{}, "id = ?", userID).Error } + +// GetUsersByIDs 批量获取用户信息 +func (r *userRepository) GetUsersByIDs(ctx context.Context, ids []string) (map[string]*model.User, error) { + result := make(map[string]*model.User, len(ids)) + if len(ids) == 0 { + return result, nil + } + + var users []*model.User + err := r.db.WithContext(ctx).Where("id IN ?", ids).Find(&users).Error + if err != nil { + return nil, err + } + + for _, u := range users { + result[u.ID] = u + } + return result, nil +} diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index fbf577d..dede620 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -56,6 +56,10 @@ type ChatService interface { // 仅保存消息到数据库，不发送实时推送（供群聊等自行推送的场景使用） SaveMessage(ctx context.Context, senderID string, conversationID string, segments model.MessageSegments, replyToID *string) (*model.Message, error) + + // 批量查询（解决 N+1 问题） + GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) + GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) } // chatServiceImpl 聊天服务实现 @@ -333,6 +337,19 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv Status: model.MessageStatusNormal, } + // 从 Redis 获取下一个 seq + 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 + } + } + // 使用事务创建消息并更新seq if err := s.repo.CreateMessageWithSeq(message); err != nil { return nil, fmt.Errorf("failed to save message: %w", err) @@ -397,6 +414,20 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv } if s.conversationCache != nil { + // 使用 Redis Hash 预计算未读数 + for _, p := range participants { + if p.UserID == senderID { + continue + } + if incrErr := s.conversationCache.IncrementUnread(context.Background(), p.UserID, conversationID); incrErr != nil { + zap.L().Warn("increment unread from redis hash failed", + zap.String("userID", p.UserID), + zap.String("convID", conversationID), + zap.Error(incrErr), + ) + } + } + keys := make([]string, 0, len(participants)*2) for _, p := range participants { keys = append(keys, @@ -559,11 +590,18 @@ func (s *chatServiceImpl) MarkAsRead(ctx context.Context, conversationID string, return fmt.Errorf("failed to update last read seq: %w", err) } - // 2. DB 写入成功后，失效缓存（Cache-Aside 模式） + // 2. DB 写入成功后，清除 Redis 未读数 if s.conversationCache != nil { + if clearErr := s.conversationCache.ClearUnread(ctx, userID, conversationID); clearErr != nil { + zap.L().Warn("clear unread from redis hash failed", + zap.String("userID", userID), + zap.String("convID", conversationID), + zap.Error(clearErr), + ) + } // 失效参与者缓存，下次读取时会从 DB 加载最新数据 s.conversationCache.InvalidateParticipant(conversationID, userID) - // 失效未读数缓存 + // 失效未读数缓存（旧 cache-aside 键） s.conversationCache.InvalidateUnreadCount(userID, conversationID) // 失效会话列表缓存 s.conversationCache.InvalidateConversationList(userID) @@ -612,8 +650,13 @@ func (s *chatServiceImpl) GetUnreadCount(ctx context.Context, conversationID str return 0, fmt.Errorf("failed to get participant: %w", err) } - // 优先使用缓存 + // 优先从 Redis Hash 读取 if s.conversationCache != nil { + count, err := s.conversationCache.GetUnreadCountFromHash(ctx, userID, conversationID) + if err == nil { + return count, nil + } + // 降级到旧缓存路径 return s.conversationCache.GetUnreadCount(ctx, userID, conversationID) } @@ -622,6 +665,12 @@ func (s *chatServiceImpl) GetUnreadCount(ctx context.Context, conversationID str // GetAllUnreadCount 获取所有会话的未读消息总数 func (s *chatServiceImpl) GetAllUnreadCount(ctx context.Context, userID string) (int64, error) { + // 优先从 Redis Hash 读取 + if s.conversationCache != nil { + if total, err := s.conversationCache.GetTotalUnread(ctx, userID); err == nil { + return total, nil + } + } return s.repo.GetAllUnreadCount(userID) } @@ -798,6 +847,19 @@ func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conv Status: model.MessageStatusNormal, } + // 从 Redis 获取下一个 seq（SaveMessage 方法） + 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 + } + } + if err := s.repo.CreateMessageWithSeq(message); err != nil { return nil, fmt.Errorf("failed to save message: %w", err) } @@ -821,3 +883,13 @@ func (s *chatServiceImpl) SaveMessage(ctx context.Context, senderID string, conv return message, nil } + +// GetUnreadCountBatch 批量获取用户在多个会话中的未读数 +func (s *chatServiceImpl) GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) { + return s.repo.GetUnreadCountBatch(ctx, userID, convIDs) +} + +// GetLastMessagesBatch 批量获取每个会话的最后一条消息 +func (s *chatServiceImpl) GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) { + return s.repo.GetLastMessagesBatch(ctx, convIDs) +} diff --git a/internal/service/group_service.go b/internal/service/group_service.go index d2fee95..9113989 100644 --- a/internal/service/group_service.go +++ b/internal/service/group_service.go @@ -61,6 +61,7 @@ type GroupService interface { TransferOwner(userID string, groupID string, newOwnerID string) error GetUserGroups(userID string, page, pageSize int) ([]model.Group, int64, error) GetMemberCount(groupID string) (int, error) + GetMemberCountBatch(ctx context.Context, groupIDs []string) (map[string]int64, error) // 成员管理 InviteMembers(userID string, groupID string, memberIDs []string) error @@ -106,25 +107,36 @@ type GroupMembersResult struct { // groupService 群组服务实现 type groupService struct { - groupRepo repository.GroupRepository - userRepo repository.UserRepository - messageRepo repository.MessageRepository - requestRepo repository.GroupJoinRequestRepository - notifyRepo repository.SystemNotificationRepository - wsHub *ws.Hub - cache cache.Cache + groupRepo repository.GroupRepository + userRepo repository.UserRepository + messageRepo repository.MessageRepository + requestRepo repository.GroupJoinRequestRepository + notifyRepo repository.SystemNotificationRepository + wsHub *ws.Hub + cache cache.Cache + conversationCache *cache.ConversationCache } // NewGroupService 创建群组服务 func NewGroupService(groupRepo repository.GroupRepository, userRepo repository.UserRepository, messageRepo repository.MessageRepository, requestRepo repository.GroupJoinRequestRepository, notifyRepo repository.SystemNotificationRepository, wsHub *ws.Hub, cacheBackend cache.Cache) GroupService { + convRepoAdapter := cache.NewConversationRepositoryAdapter(messageRepo) + msgRepoAdapter := cache.NewMessageRepositoryAdapter(messageRepo) + conversationCache := cache.NewConversationCache( + cacheBackend, + convRepoAdapter, + msgRepoAdapter, + cache.DefaultConversationCacheSettings(), + ) + return &groupService{ - groupRepo: groupRepo, - userRepo: userRepo, - messageRepo: messageRepo, - requestRepo: requestRepo, - notifyRepo: notifyRepo, - wsHub: wsHub, - cache: cacheBackend, + groupRepo: groupRepo, + userRepo: userRepo, + messageRepo: messageRepo, + requestRepo: requestRepo, + notifyRepo: notifyRepo, + wsHub: wsHub, + cache: cacheBackend, + conversationCache: conversationCache, } } @@ -299,6 +311,11 @@ func (s *groupService) GetMemberCount(groupID string) (int, error) { return int(count), nil } +// GetMemberCountBatch 批量获取多个群的成员数量 +func (s *groupService) GetMemberCountBatch(ctx context.Context, groupIDs []string) (map[string]int64, error) { + return s.groupRepo.GetMemberCountBatch(ctx, groupIDs) +} + // UpdateGroup 更新群组信息 func (s *groupService) UpdateGroup(userID string, groupID string, updates map[string]any) error { // 检查群组是否存在 @@ -453,6 +470,18 @@ func (s *groupService) broadcastMemberJoinNotice(groupID string, targetUserID st Status: model.MessageStatusNormal, Category: model.CategoryNotification, } + // 从 Redis 获取下一个 seq + 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 + } + } if err := s.messageRepo.CreateMessageWithSeq(msg); err != nil { zap.L().Warn("保存入群提示消息失败", zap.String("component", "broadcastMemberJoinNotice"), @@ -1366,7 +1395,19 @@ func (s *groupService) MuteMember(userID string, groupID string, targetUserID st Category: model.CategoryNotification, } - // 保存消息并获取 seq + // 从 Redis 获取下一个 seq + 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 + } + } + if err := s.messageRepo.CreateMessageWithSeq(msg); err != nil { zap.L().Warn("保存禁言消息失败", zap.String("component", "MuteMember"), diff --git a/internal/service/message_service.go b/internal/service/message_service.go index 642b878..6da0fd1 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -86,6 +86,19 @@ func (s *MessageService) SendMessage(ctx context.Context, senderID, receiverID s Status: model.MessageStatusNormal, } + // 从 Redis 获取下一个 seq + 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 + } + } + // 使用事务创建消息并更新seq err = s.messageRepo.CreateMessageWithSeq(msg) if err != nil { @@ -284,6 +297,16 @@ func (s *MessageService) GetConversationParticipants(conversationID string) ([]* return s.messageRepo.GetConversationParticipants(conversationID) } +// GetParticipantsBatch 批量获取多个会话的参与者列表 +func (s *MessageService) GetParticipantsBatch(ctx context.Context, convIDs []string) (map[string][]*model.ConversationParticipant, error) { + return s.messageRepo.GetParticipantsBatch(ctx, convIDs) +} + +// GetMyParticipantsBatch 批量获取用户在多个会话中的参与者信息 +func (s *MessageService) GetMyParticipantsBatch(ctx context.Context, convIDs []string, userID string) (map[string]*model.ConversationParticipant, error) { + return s.messageRepo.GetMyParticipantsBatch(ctx, convIDs, userID) +} + // ParseConversationID 辅助函数：直接返回字符串ID（已经是string类型） func ParseConversationID(idStr string) (string, error) { return idStr, nil diff --git a/internal/service/push_service.go b/internal/service/push_service.go index 9ff9b86..7ef1246 100644 --- a/internal/service/push_service.go +++ b/internal/service/push_service.go @@ -94,6 +94,7 @@ type pushTask struct { userID string message *model.Message priority int + recordID int64 // DB record ID for crash recovery } // NewPushService 创建推送服务 @@ -152,21 +153,31 @@ func (s *pushServiceImpl) PushToUser(ctx context.Context, userID string, message } } - // 极光推送也失败，加入推送队列等待重试 + // 极光推送也失败，先持久化推送记录再入内存队列 + expiredAt := time.Now().Add(DefaultExpiredTime) + record := &model.PushRecord{ + UserID: userID, + MessageID: message.ID, + PushChannel: model.PushChannelJPush, + PushStatus: model.PushStatusPending, + MaxRetry: MaxRetryCount, + ExpiredAt: &expiredAt, + } + if err := s.pushRepo.Create(record); err != nil { + return fmt.Errorf("failed to persist push record: %w", err) + } + select { case s.pushQueue <- &pushTask{ userID: userID, message: message, priority: priority, + recordID: record.ID, }: return nil default: - // 队列已满，直接创建待推送记录 - _, err := s.CreatePushRecord(ctx, userID, message.ID, model.PushChannelJPush) - if err != nil { - return fmt.Errorf("failed to create pending push record: %w", err) - } - return errors.New("push queue is full, message queued for later delivery") + // 队列已满，记录已持久化，等待重试协程处理 + return errors.New("push queue is full, message persisted for later delivery") } } @@ -435,10 +446,50 @@ func (s *pushServiceImpl) GetUserDevices(ctx context.Context, userID string) ([] // StartPushWorker 启动推送工作协程 func (s *pushServiceImpl) StartPushWorker(ctx context.Context) { + s.recoverPendingPushes() go s.processPushQueue() go s.retryFailedPushes() } +// recoverPendingPushes 从数据库恢复未处理的推送记录 +func (s *pushServiceImpl) recoverPendingPushes() { + records, err := s.pushRepo.GetPendingPushes(100) + if err != nil { + zap.L().Error("failed to recover pending push records", zap.Error(err)) + return + } + + recovered := 0 + for _, record := range records { + if record.IsExpired() { + s.pushRepo.UpdateStatus(record.ID, model.PushStatusExpired) + continue + } + + message, err := s.messageRepo.GetMessageByID(record.MessageID) + if err != nil { + s.pushRepo.MarkAsFailed(record.ID, "message not found during recovery") + continue + } + + select { + case s.pushQueue <- &pushTask{ + userID: record.UserID, + message: message, + priority: int(PriorityNormal), + recordID: record.ID, + }: + recovered++ + default: + // 队列已满，保留 pending 状态等待下次恢复 + } + } + + if recovered > 0 { + zap.L().Info("recovered pending push records", zap.Int("count", recovered)) + } +} + // StopPushWorker 停止推送工作协程 func (s *pushServiceImpl) StopPushWorker() { close(s.stopChan) @@ -464,8 +515,8 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) { // 获取用户活跃设备 devices, err := s.deviceRepo.GetActiveByUserID(task.userID) if err != nil || len(devices) == 0 { - // 没有可用设备，创建待推送记录 - s.CreatePushRecord(ctx, task.userID, task.message.ID, model.PushChannelJPush) + // 没有可用设备，保留记录为 pending 等待重试 + s.pushRepo.UpdateStatus(task.recordID, model.PushStatusFailed) return } @@ -480,6 +531,8 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) { err := s.pushViaJPushBatch(mobileDevices, payload.Title, payload.Content, payload.Extras) if err == nil { s.batchCreatePushedRecords(ctx, task.userID, task.message.ID, mobileDevices) + // 原始 pending 记录标记为已推送（被批量记录替代） + s.pushRepo.UpdateStatus(task.recordID, model.PushStatusPushed) return } } @@ -508,6 +561,8 @@ func (s *pushServiceImpl) processPushTask(task *pushTask) { } s.pushRepo.Update(record) } + // 原始 pending 记录标记为已处理 + s.pushRepo.UpdateStatus(task.recordID, model.PushStatusPushed) } } diff --git a/internal/service/user_service.go b/internal/service/user_service.go index 97a5390..060a23c 100644 --- a/internal/service/user_service.go +++ b/internal/service/user_service.go @@ -28,6 +28,7 @@ type UserService interface { // 用户查询 GetUserByID(ctx context.Context, id string) (*model.User, error) + GetUsersByIDs(ctx context.Context, ids []string) (map[string]*model.User, error) GetUserPostCount(ctx context.Context, userID string) (int64, error) GetUserPostCountBatch(ctx context.Context, userIDs []string) (map[string]int64, error) GetUserByIDWithFollowingStatus(ctx context.Context, userID, currentUserID string) (*model.User, bool, error) @@ -386,6 +387,11 @@ func (s *userServiceImpl) GetUserByID(ctx context.Context, id string) (*model.Us return s.userRepo.GetByID(id) } +// GetUsersByIDs 批量获取用户信息 +func (s *userServiceImpl) GetUsersByIDs(ctx context.Context, ids []string) (map[string]*model.User, error) { + return s.userRepo.GetUsersByIDs(ctx, ids) +} + // GetUserPostCount 获取用户帖子数（实时计算） func (s *userServiceImpl) GetUserPostCount(ctx context.Context, userID string) (int64, error) { return s.userRepo.GetPostsCount(userID)