fix(message): improve message synchronization and reliability

Refactor the messaging subsystem to enhance data consistency, prevent race conditions during WebSocket reconnection, and ensure accurate unread counts.

- **WebSocket Reliability**: Implement `client_msg_id` for precise message ACK matching, preventing incorrect pending message resolution in high-frequency scenarios.
- **Sync Logic**: Update `WSMessageHandler` and `MessageManager` to use a bootstrapping state, ensuring buffered SSE events are flushed only after the initial synchronization is complete.
- **State Consistency**:
    - Introduce atomic unread count increments to prevent lost updates.
    - Implement conditional rollback for optimistic read receipts, ensuring that failed API calls do not overwrite newer, valid read states.
- **Resource Management**: Add reference counting to `useMessages` hook to prevent premature clearing of loading states during rapid conversation switching or React StrictMode double-invocations.
- **Data Integrity**: Update `UserCacheService` to re-read the latest message list before applying sender info enrichment, ensuring new messages arriving during the async process are correctly processed.

src/services/core/wsService.ts

@@ -357,7 +357,7 @@ class WebSocketService {
         this.markActivity();
         this.startHeartbeat();
         this.connectionHandlers.forEach(h => h());
-        
+
         // 发送队列中的消息
         this.flushPendingMessages();
       };
@@ -776,10 +776,19 @@ class WebSocketService {
   }
 
   private handleMessageSent(payload: any): void {
-    // 找到对应的发送请求并resolve
+    // 使用后端回带的 client_msg_id 精确匹配单条消息
-    const pendingMsg = this.pendingMessages.find(p => 
+    // 避免同会话连续发送时，第一条 ACK 误清掉第二条的 pending 记录
-      p.type === 'chat' && p.payload.conversation_id === payload.conversation_id
+    const clientId = payload?.client_msg_id;
-    );
+    let pendingMsg: PendingMessage | undefined;
+    if (clientId) {
+      pendingMsg = this.pendingMessages.find(p => p.id === clientId);
+    }
+    // 兜底：旧版后端不回带 client_msg_id 时，按 conversation_id + 类型匹配最旧一条
+    if (!pendingMsg) {
+      pendingMsg = this.pendingMessages.find(p =>
+        p.type === 'chat' && p.payload.conversation_id === payload.conversation_id
+      );
+    }
     if (pendingMsg) {
       pendingMsg.resolve(payload);
       this.removePendingMessage(pendingMsg.id);
@@ -787,9 +796,16 @@ class WebSocketService {
   }
 
   private handleMessageRecalled(payload: any): void {
-    const pendingMsg = this.pendingMessages.find(p => 
+    const clientId = payload?.client_msg_id;
-      p.type === 'recall' && p.payload.message_id === payload.message_id
+    let pendingMsg: PendingMessage | undefined;
-    );
+    if (clientId) {
+      pendingMsg = this.pendingMessages.find(p => p.id === clientId);
+    }
+    if (!pendingMsg) {
+      pendingMsg = this.pendingMessages.find(p =>
+        p.type === 'recall' && p.payload.message_id === payload.message_id
+      );
+    }
     if (pendingMsg) {
       pendingMsg.resolve(payload);
       this.removePendingMessage(pendingMsg.id);
@@ -880,16 +896,18 @@ class WebSocketService {
       }
 
       const messageId = `msg_${++this.messageIdCounter}_${Date.now()}`;
+      // 携带 client_msg_id：用于服务器回 ACK / message_sent 时精确匹配单条消息
+      const payloadWithId = { ...payload, client_msg_id: messageId };
       const message = {
         type,
-        payload,
+        payload: payloadWithId,
       };
 
       // 添加到待处理队列
       const pendingMsg: PendingMessage = {
         id: messageId,
         type,
-        payload,
+        payload: payloadWithId,
         resolve,
         reject,
         timestamp: Date.now(),
@@ -929,6 +947,11 @@ class WebSocketService {
   }
 
   // 刷新待发送消息队列
+  // 过期判定：
+  //   - 单条 sendViaWebSocket 自带 10s 超时，正常情况下 pending 消息不会超过这么久
+  //   - 重连后链路刚恢复时，距离"刚入队"的消息一律视为有效（避免把刚刚挂起的消息误清）
+  //   - 兜底：超过 STALE_PENDING_MS（60s）一定视为过期丢弃
+  private static readonly STALE_PENDING_MS = 60000;
   private flushPendingMessages(): void {
     if (this.pendingMessages.length === 0) return;
 
@@ -938,7 +961,9 @@ class WebSocketService {
 
     // 分离过期和有效的消息
     for (const msg of this.pendingMessages) {
-      if (now - msg.timestamp > 30000) {
+      const ageMs = now - msg.timestamp;
+      const isStale = ageMs > WebSocketService.STALE_PENDING_MS;
+      if (isStale) {
         expiredMessages.push(msg);
       } else {
         validMessages.push(msg);
@@ -950,13 +975,32 @@ class WebSocketService {
       msg.reject(new Error('Message expired'));
     });
 
-    // 重新发送有效消息
+    // 重新发送有效消息：直接复用原 client_msg_id，否则重连后服务端 ACK 无法回带到原 pending 记录
-    this.pendingMessages = [];
+    // 注意：不能调 sendViaWebSocket（会重新生成 id 并入队），必须直接 ws.send
-    validMessages.forEach(msg => {
+    this.pendingMessages = validMessages;
-      this.sendViaWebSocket(msg.type, msg.payload)
+    for (let i = 0; i < validMessages.length; i++) {
-        .then(msg.resolve)
+      const msg = validMessages[i];
-        .catch(msg.reject);
+      if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
-    });
+        // 链路断开，剩余消息留在队列等下次重连
+        break;
+      }
+      try {
+        this.ws.send(JSON.stringify({ type: msg.type, payload: msg.payload }));
+      } catch {
+        // ws.send 抛异常时 reject 并移出队列，避免无限残留
+        msg.reject(new Error('WebSocket send failed on flush'));
+        this.removePendingMessage(msg.id);
+        continue;
+      }
+      // 重置 10s 超时
+      setTimeout(() => {
+        const index = this.pendingMessages.findIndex(p => p.id === msg.id);
+        if (index >= 0) {
+          this.pendingMessages[index].reject(new Error('Message timeout'));
+          this.removePendingMessage(msg.id);
+        }
+      }, 10000);
+    }
   }
 
   private removePendingMessage(id: string): void {

src/stores/message/MessageManager.ts

@@ -154,10 +154,10 @@ class MessageManager {
 
         useMessageStore.getState().setInitialized(true);
 
-        // 处理缓冲的 SSE 事件
+        // 先关闭 bootstrap（事件流转入正常处理路径），再做尾部 flush
-        await this.wsHandler.flushBufferedSSEEvents();
+        // 避免 flush 期间新到的事件被压入 buffer 后无人消费
-
         this.wsHandler.setBootstrapping(false);
+        await this.wsHandler.flushBufferedSSEEvents();
 
         if (useMessageStore.getState().currentConversationId) {
           await this.fetchMessages(useMessageStore.getState().currentConversationId!);

src/stores/message/hooks.ts

@@ -100,9 +100,14 @@ interface UseMessagesReturn {
  *
  * 核心修复：
  * 1. useFocusEffect：屏幕重新获得焦点时强制同步消息，解决"需要退出重进才能看到新消息"的问题
- * 2. 清理 isLoadingMessages：组件卸载时清除残留的 loading 锁，防止下次进入被阻断
+ * 2. 引用计数清理 isLoadingMessages：仅在最后一个订阅者卸载时清除 loading 锁，
+ *    避免 StrictMode 双调用或快速切换会话时，第一次 fetch 还没完成就被清锁
  * 3. 重入时 forceSync：相同 conversationId 重复进入时强制拉取最新消息
  */
+
+// 记录每个 conversationId 上的活跃订阅数，仅在归零时清锁
+const loadingLockRefCount: Map<string, number> = new Map();
+
 export function useMessages(conversationId: string | null): UseMessagesReturn {
   const normalizedConversationId = conversationId ? String(conversationId) : null;
 
@@ -128,6 +133,10 @@ export function useMessages(conversationId: string | null): UseMessagesReturn {
       return;
     }
 
+    // 引用计数 +1
+    const prevCount = loadingLockRefCount.get(normalizedConversationId) ?? 0;
+    loadingLockRefCount.set(normalizedConversationId, prevCount + 1);
+
     lastActivatedAtRef.current = Date.now();
 
     const isReentry = lastActivatedIdRef.current === normalizedConversationId;
@@ -141,11 +150,17 @@ export function useMessages(conversationId: string | null): UseMessagesReturn {
     lastActivatedIdRef.current = normalizedConversationId;
 
     return () => {
-      // 清理活动会话 + 清除残留的 loading 锁
+      // 引用计数 -1，仅在最后一个订阅者卸载时清活动会话 + 清 loading 锁
-      if (messageManager.getActiveConversation() === normalizedConversationId) {
+      const nextCount = (loadingLockRefCount.get(normalizedConversationId) ?? 1) - 1;
-        messageManager.setActiveConversation(null);
+      if (nextCount <= 0) {
+        loadingLockRefCount.delete(normalizedConversationId);
+        if (messageManager.getActiveConversation() === normalizedConversationId) {
+          messageManager.setActiveConversation(null);
+        }
+        useMessageStore.getState().setLoadingMessages(normalizedConversationId, false);
+      } else {
+        loadingLockRefCount.set(normalizedConversationId, nextCount);
       }
-      useMessageStore.getState().setLoadingMessages(normalizedConversationId, false);
     };
   }, [normalizedConversationId]);
 

src/stores/message/services/MessageSyncService.ts

@@ -197,9 +197,10 @@ export class MessageSyncService implements IMessageSyncService {
 
   /**
    * 获取会话消息（增量同步）— 去重入口
+   * key 包含 force 维度，强制刷新会跳过非强制的 in-flight 合并
    */
   async fetchMessages(conversationId: string, afterSeq?: number, force?: boolean): Promise<void> {
-    const key = `${conversationId}:${afterSeq ?? 'all'}`;
+    const key = `${conversationId}:${afterSeq ?? 'all'}:${force ? 'force' : 'normal'}`;
     const existing = this.inflightFetches.get(key);
     if (existing) return existing;
 

src/stores/message/services/ReadReceiptManager.ts

@@ -30,7 +30,7 @@ export class ReadReceiptManager implements IReadReceiptManager {
     const normalizedId = normalizeConversationId(conversationId);
     const store = useMessageStore.getState();
     const conversation = store.getConversation(normalizedId);
-    
+
     if (!conversation) {
       console.warn('[ReadReceiptManager] 会话不存在:', normalizedId);
       return;
@@ -53,6 +53,9 @@ export class ReadReceiptManager implements IReadReceiptManager {
     this.readStateVersion++;
     const currentVersion = this.readStateVersion;
 
+    // 记录本次写入的 my_last_read_seq，用于失败时条件回滚
+    const writtenReadSeq = Math.max(Number(conversation.my_last_read_seq || 0), seq);
+
     // 1. 标记此会话有进行中的已读请求
     this.pendingReadMap.set(normalizedId, {
       timestamp: Date.now(),
@@ -64,7 +67,7 @@ export class ReadReceiptManager implements IReadReceiptManager {
     const updatedConv: ConversationResponse = {
       ...conversation,
       unread_count: 0,
-      my_last_read_seq: Math.max(Number(conversation.my_last_read_seq || 0), seq),
+      my_last_read_seq: writtenReadSeq,
     };
 
     const currentUnread = store.getUnreadCount();
@@ -79,11 +82,11 @@ export class ReadReceiptManager implements IReadReceiptManager {
       await messageService.markAsRead(normalizedId, seq);
     } catch (error) {
       console.error('[ReadReceiptManager] 标记已读API失败:', error);
-      
+
-      // 失败时回滚状态
+      // 条件回滚（含 totalUnreadCount）：仅在 my_last_read_seq 仍是本次写入值时回滚，
-      store.updateConversation(conversation);
+      // 避免覆盖并发 markAsRead 已写入的更大读游标或更小未读
-      store.setUnreadCount(currentUnread.total, currentUnread.system);
+      store.rollbackReadIfUnchanged(normalizedId, writtenReadSeq, prevUnreadCount);
-      
+
       // API 失败时立即清除保护
       this.pendingReadMap.delete(normalizedId);
       return;
@@ -137,6 +140,9 @@ export class ReadReceiptManager implements IReadReceiptManager {
     });
     store.setConversationsWithUnread(newConversations, 0, currentSystem);
 
+    // 记录本次乐观写入的"目标快照"，用于失败时条件回滚
+    const optimisticSnapshot = new Map(newConversations);
+
     try {
       // 标记系统消息已读
       await messageService.markAllSystemMessagesRead();
@@ -155,8 +161,22 @@ export class ReadReceiptManager implements IReadReceiptManager {
       }
     } catch (error) {
       console.error('[ReadReceiptManager] 标记所有已读失败:', error);
-      // 回滚：恢复原始会话和未读数
+      // 条件回滚：仅在 conversations 的关键字段未被其他操作修改时还原，
-      store.setConversationsWithUnread(new Map(conversationsMap), prevTotalUnread, currentSystem);
+      // 避免覆盖期间到达的 markAsRead / WS 推送造成的新未读
+      const current = useMessageStore.getState().conversations;
+      let stillMatches = current.size === optimisticSnapshot.size;
+      if (stillMatches) {
+        for (const [id, conv] of optimisticSnapshot.entries()) {
+          const cur = current.get(id);
+          if (!cur || cur.unread_count !== conv.unread_count || cur.my_last_read_seq !== conv.my_last_read_seq) {
+            stillMatches = false;
+            break;
+          }
+        }
+      }
+      if (stillMatches) {
+        store.setConversationsWithUnread(new Map(conversationsMap), prevTotalUnread, currentSystem);
+      }
     }
   }
 

src/stores/message/services/UserCacheService.ts

@@ -6,6 +6,7 @@
 import type { MessageResponse, UserDTO } from '../../../types/dto';
 import { api } from '@/services/core';
 import { userCacheRepository } from '@/database';
+import { useMessageStore } from '../store';
 import type { IUserCacheService } from '../types';
 
 export class UserCacheService implements IUserCacheService {
@@ -61,16 +62,20 @@ export class UserCacheService implements IUserCacheService {
 
   /**
    * 批量异步填充消息的 sender 信息
-   * 填充完成后调用 notifyUpdate 更新内存并通知订阅者
+   * 注意：fetch 期间新消息可能已到达，回调内必须重读最新 messages 列表，
+   * 否则新消息的 sender 字段不会被填充。
    */
   enrichMessagesWithSenderInfo(
     conversationId: string,
-    messages: MessageResponse[],
+    _messages: MessageResponse[],
     notifyUpdate: (conversationId: string, messages: MessageResponse[]) => void
   ): void {
+    // 取一次最新列表作为本次填充目标
+    const latestMessages = useMessageStore.getState().getMessages(conversationId);
+
     // 收集所有需要查询的唯一 sender_id（排除系统用户）
     const senderIds = [...new Set(
-      messages
+      latestMessages
         .filter(m => m.sender_id && m.sender_id !== '10000' && !m.sender)
         .map(m => m.sender_id)
     )];
@@ -87,8 +92,10 @@ export class UserCacheService implements IUserCacheService {
 
         if (senderMap.size === 0) return;
 
+        // 重新读取最新消息列表（期间可能又有新消息到达）
+        const currentMessages = useMessageStore.getState().getMessages(conversationId);
         let changed = false;
-        const updated = messages.map(m => {
+        const updated = currentMessages.map(m => {
           if (!m.sender && m.sender_id && senderMap.has(m.sender_id)) {
             changed = true;
             return { ...m, sender: senderMap.get(m.sender_id) };

src/stores/message/services/WSMessageHandler.ts

@@ -212,10 +212,13 @@ export class WSMessageHandler implements IWSMessageHandler {
   /**
    * 统一同步入口 — 节流 + 去重
    * onConnect 和 sync_required 共用此入口，防止并发触发重复同步
+   * 启动阶段（isBootstrapping=true）跳过，由 MessageManager.initialize 完成后统一触发
    */
   private async triggerSync(source: string): Promise<void> {
+    if (this.isBootstrapping) return;
+
     const now = Date.now();
-    // 最小间隔节流
+    // 最小间隔节流：避免连续 sync_required 事件短时间内重复触发全量同步
     if (now - this.lastSyncTriggerAt < MIN_RECONNECT_SYNC_INTERVAL) return;
     if (this.syncInProgress) return;
 
@@ -246,14 +249,22 @@ export class WSMessageHandler implements IWSMessageHandler {
 
   /**
    * 初始化完成后，处理缓冲的 SSE 事件
+   * 每次循环：取走当前 buffer 后立即清空（避免在 fetch 期间新事件再被覆盖）
+   * 退出条件：连续一轮没有新事件，确保 flush 期间到达的事件也被消费
    */
   async flushBufferedSSEEvents(): Promise<void> {
+    let lastChatLen = -1;
+    let lastReadLen = -1;
     for (let i = 0; i < MAX_FLUSH_ITERATIONS; i++) {
-      const hasBuffered = this.bufferedChatMessages.length > 0 || this.bufferedReadReceipts.length > 0;
-      if (!hasBuffered) return;
-
       const chatEvents = this.bufferedChatMessages;
       const readEvents = this.bufferedReadReceipts;
+      // 上一轮长度未变且都已清空，说明已稳定
+      if (chatEvents.length === lastChatLen && readEvents.length === lastReadLen) {
+        return;
+      }
+      lastChatLen = chatEvents.length;
+      lastReadLen = readEvents.length;
+      // 立即清空 buffer；flush 期间新到的事件将走正常处理路径（isBootstrapping 已是 false）
       this.bufferedChatMessages = [];
       this.bufferedReadReceipts = [];
 
@@ -528,9 +539,8 @@ export class WSMessageHandler implements IWSMessageHandler {
   }
 
   private incrementSystemUnread(): void {
-    const store = useMessageStore.getState();
+    // 原子递增：避免外部 read-modify-write 造成丢计数
-    const currentUnread = store.getUnreadCount();
+    useMessageStore.getState().incrementSystemUnreadAtomic();
-    store.setUnreadCount(currentUnread.total + 1, currentUnread.system + 1);
   }
 
   private markMessageAsRecalled(conversationId: string, messageId: string): void {

src/stores/message/store.ts

@@ -88,6 +88,10 @@ export interface MessageActions {
   removeMessage: (conversationId: string, messageId: string) => void;
   patchMessages: (conversationId: string, patches: Map<string, Partial<MessageResponse>>) => void;
   setUnreadCount: (total: number, system: number) => void;
+  /**
+   * 原子递增系统未读数：避免外部 RMW 造成丢失
+   */
+  incrementSystemUnreadAtomic: () => void;
   setLastSystemMessageAt: (time: string | null) => void;
   setSSEConnected: (connected: boolean) => void;
   setCurrentConversation: (conversationId: string | null) => void;
@@ -107,6 +111,16 @@ export interface MessageActions {
   ) => void;
   incrementUnreadAtomic: (conversationId: string) => void;
 
+  /**
+   * 原子回滚乐观已读：仅在当前会话的 my_last_read_seq 仍等于 expectedReadSeq 时才回滚
+   * 避免覆盖后续并发 markAsRead 已更新的更大值
+   */
+  rollbackReadIfUnchanged: (
+    conversationId: string,
+    expectedReadSeq: number,
+    prevUnreadCount: number,
+  ) => void;
+
   // 重置
   reset: () => void;
 }
@@ -393,6 +407,13 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
     set({ totalUnreadCount: total, systemUnreadCount: system });
   },
 
+  incrementSystemUnreadAtomic: () => {
+    set(state => ({
+      totalUnreadCount: state.totalUnreadCount + 1,
+      systemUnreadCount: state.systemUnreadCount + 1,
+    }));
+  },
+
   setLastSystemMessageAt: (time: string | null) => {
     set({ lastSystemMessageAt: time });
     if (time) {
@@ -506,6 +527,36 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
     });
   },
 
+  rollbackReadIfUnchanged: (
+    conversationId: string,
+    expectedReadSeq: number,
+    prevUnreadCount: number,
+  ) => {
+    const normalizedId = normalizeConversationId(conversationId);
+    set(state => {
+      const conversation = state.conversations.get(normalizedId);
+      if (!conversation) return state;
+      // 仅当本地乐观已读游标仍是本次写入的值时，才回滚未读
+      // 否则说明已被更新的 markAsRead 覆盖，不应回滚
+      if (Number(conversation.my_last_read_seq || 0) !== expectedReadSeq) {
+        return state;
+      }
+      const newConversations = new Map(state.conversations);
+      newConversations.set(normalizedId, {
+        ...conversation,
+        unread_count: prevUnreadCount,
+      });
+      const conversationList = sortConversationList(newConversations);
+      // 同步回滚 totalUnreadCount：加回本次扣减的未读数
+      const unreadDelta = prevUnreadCount - (conversation.unread_count || 0);
+      return {
+        conversations: newConversations,
+        conversationList,
+        totalUnreadCount: state.totalUnreadCount + unreadDelta,
+      };
+    });
+  },
+
   // ==================== 重置 ====================
 
   reset: () => {