feat(message): implement incremental sync and atomic unread updates

Refactor the messaging system to improve synchronization efficiency and state consistency. - Implement incremental message synchronization using sequence numbers (seq) to reduce payload size. - Add `markAllAsRead` batch API support to optimize read receipt processing. - Introduce atomic state updates in `useMessageStore` to prevent inconsistent UI rendering during unread count changes. - Implement notification deduplication in `WSMessageHandler` to prevent duplicate unread increments during reconnection. - Optimize `MessageSyncService` to prioritize incremental sync over full snapshots when local data exists. - Refactor `ReadReceiptManager` to use optimistic updates with proper rollback mechanisms. - Fix minor UI issues in `SettingsScreen` and `PrivacySettingsScreen` related to theme picker z-index and style application.
2026-05-12 18:04:32 +08:00
parent 7c7aaf9108
commit 48f31e6617
11 changed files with 280 additions and 159 deletions
--- a/src/screens/profile/PrivacySettingsScreen.tsx
+++ b/src/screens/profile/PrivacySettingsScreen.tsx
@@ -333,7 +333,7 @@ export const PrivacySettingsScreen: React.FC = () => {
                              variant="body"
                              style={[
                                styles.dropdownOptionText,
-                                isActive && styles.dropdownOptionTextActive,
+                                ...(isActive ? [styles.dropdownOptionTextActive] : []),
                              ]}
                            >
                              {option.label}
--- a/src/screens/profile/SettingsScreen.tsx
+++ b/src/screens/profile/SettingsScreen.tsx
@@ -408,7 +408,7 @@ export const SettingsScreen: React.FC = () => {
    <View
      key={group.title}
      style={[
-        { overflow: 'visible', zIndex: group.items.some((i) => i.isThemePicker && openPicker === 'theme') ? 10 : groupIndex },
+        { overflow: 'visible', zIndex: group.items.some((i) => 'isThemePicker' in i && i.isThemePicker && openPicker === 'theme') ? 10 : groupIndex },
      ]}
    >
      <View style={styles.groupHeader}>
--- a/src/services/message/messageService.ts
+++ b/src/services/message/messageService.ts
@@ -568,6 +568,19 @@ class MessageService {
    });
  }

+  /**
+   * 批量标记所有会话已读
+   * POST /api/v1/conversations/read-all
+   */
+  async markAllAsRead(conversations: Array<{ id: string; lastSeq: number }>): Promise<void> {
+    await api.post('/conversations/read-all', {
+      conversations: conversations.map(c => ({
+        conversation_id: c.id,
+        last_read_seq: Number(c.lastSeq),
+      })),
+    });
+  }
+
  /**
   * 上报输入状态
   * 优先使用WebSocket，失败时降级到HTTP
@@ -597,6 +610,15 @@ class MessageService {
    }
  }

+  /**
+   * 获取同步元数据（轻量级 seq + 时间，用于增量同步判断）
+   * GET /api/v1/conversations/sync-data
+   */
+  async getSyncData(): Promise<Array<{ id: string; max_seq: number; last_message_at: string }>> {
+    const response = await api.get<{ conversations: Array<{ id: string; max_seq: number; last_message_at: string }> }>('/conversations/sync-data');
+    return response.data?.conversations || [];
+  }
+
  /**
   * 获取单个会话未读数
   * GET /api/v1/conversations/unread/count?conversation_id=xxx
--- a/src/stores/message/MessageManager.ts
+++ b/src/stores/message/MessageManager.ts
@@ -29,7 +29,6 @@ import { useMessageStore, normalizeConversationId, loadPersistedLastSystemMessag

 // 重新导出类型，保持兼容性
 export type {
-  MessageManagerState,
  ReadStateRecord,
  MessageManagerConversationListDeps,
 } from './types';
@@ -74,6 +73,7 @@ class MessageManager {
        fetchConversations: (force, source) => this.fetchConversations(force, source),
        fetchUnreadCount: () => this.fetchUnreadCount(),
        fetchMessages: (id) => this.fetchMessages(id),
+        syncBySeq: () => this.syncService.syncBySeq(),
      }
    );

@@ -380,7 +380,7 @@ class MessageManager {
  private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean {
    if (!this.getActiveConversation()) return false;
    if (!forceRefresh) return false;
-    return source === 'sse-reconnect' || source === 'prefetch' || source === 'hooks-initial-refresh';
+    return source === 'sse-reconnect' || source === 'prefetch';
  }
 }

--- a/src/stores/message/index.ts
+++ b/src/stores/message/index.ts
@@ -24,7 +24,6 @@ export type { MessageState, MessageActions, MessageStore } from './store';
 // ==================== 类型导出 ====================
 // 从 types.ts 导出所有类型
 export type {
-  MessageManagerState,
  ReadStateRecord,
  MessageManagerConversationListDeps,
  HandleNewMessageOptions,
--- a/src/stores/message/services/ConversationOperations.ts
+++ b/src/stores/message/services/ConversationOperations.ts
@@ -102,32 +102,6 @@ export class ConversationOperations implements IConversationOperations {
    );
  }

-  /**
-   * 增加会话未读数
-   */
-  incrementUnreadCount(conversationId: string): void {
-    const store = useMessageStore.getState();
-    const conversation = store.getConversation(conversationId);
-    if (!conversation) return;
-
-    const prevUnreadCount = conversation.unread_count || 0;
-    const newUnreadCount = prevUnreadCount + 1;
-
-    // 创建新对象而不是直接修改，确保状态一致性
-    const updatedConv: ConversationResponse = {
-      ...conversation,
-      unread_count: newUnreadCount,
-    };
-    
-    store.updateConversation(updatedConv);
-
-    const currentUnread = store.getUnreadCount();
-    store.setUnreadCount(
-      currentUnread.total + 1,
-      currentUnread.system
-    );
-  }
-
  /**
   * 设置系统消息未读数
   */
--- a/src/stores/message/services/MessageSyncService.ts
+++ b/src/stores/message/services/MessageSyncService.ts
@@ -33,8 +33,6 @@ export class MessageSyncService implements IMessageSyncService {

  /** 正在加载会话列表下一页 */
  private loadingMoreConversations = false;
-  /** 聊天页活跃期间延迟的会话列表刷新 */
-  private deferredConversationRefresh = false;
    /** fetchUnreadCount 去重：复用 in-flight promise */
  private fetchUnreadCountPromise: Promise<void> | null = null;

@@ -70,7 +68,6 @@ export class MessageSyncService implements IMessageSyncService {
    }

    if (this.shouldDeferConversationRefresh(source, forceRefresh)) {
-      this.deferredConversationRefresh = true;
      if (__DEV__) {
        console.log('[MessageSyncService] defer fetchConversations', {
          source,
@@ -208,8 +205,6 @@ export class MessageSyncService implements IMessageSyncService {
        if (!hasInMemoryMessages) {
          try {
            const localMessages = await messageRepository.getByConversation(conversationId, 20);
-            const localMaxSeq = await messageRepository.getMaxSeq(conversationId);
-            baselineMaxSeq = localMaxSeq;

            if (localMessages.length > 0) {
              const formattedMessages: MessageResponse[] = localMessages.map(m => ({
@@ -224,7 +219,6 @@ export class MessageSyncService implements IMessageSyncService {

              store.setMessages(conversationId, formattedMessages);
            } else {
-              // 冷启动兜底：先设置空列表
              store.setMessages(conversationId, []);
            }
          } catch (error) {
@@ -232,32 +226,20 @@ export class MessageSyncService implements IMessageSyncService {
          }
        }

-        // 服务端快照 + 增量同步
+        // 更新 baseline：合并本地DB后重新计算
+        const currentMessages = store.getMessages(conversationId);
+        baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
+
+        // 策略：有本地数据时仅增量同步，无数据时拉快照
+        if (baselineMaxSeq > 0) {
+          // 增量同步：只拉 baselineMaxSeq 之后的消息
          try {
-          const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
-          const snapshotMessages = snapshotResp?.messages || [];
-          
-          if (snapshotMessages.length > 0) {
-            const existingMessages = store.getMessages(conversationId);
-            const mergedSnapshot = mergeMessagesById(existingMessages, snapshotMessages);
-            store.setMessages(conversationId, mergedSnapshot);
-
-            messageRepository.saveMessagesBatch(
-              MessageMapper.toCachedMessages(snapshotMessages, conversationId)
-            ).catch(error => {
-              console.error('[MessageSyncService] 保存快照消息到本地失败:', error);
-            });
-          }
-
-          // 增量补齐
-          const snapshotMaxSeq = snapshotMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
-          if (snapshotMaxSeq > baselineMaxSeq) {
            const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
            const newMessages = incrementalResp?.messages || [];

            if (newMessages.length > 0) {
-              const existingMessages = store.getMessages(conversationId);
-              const mergedMessages = mergeMessagesById(existingMessages, newMessages);
+              const existingMsgs = store.getMessages(conversationId);
+              const mergedMessages = mergeMessagesById(existingMsgs, newMessages);
              store.setMessages(conversationId, mergedMessages);

              messageRepository.saveMessagesBatch(
@@ -266,9 +248,27 @@ export class MessageSyncService implements IMessageSyncService {
                console.error('[MessageSyncService] 保存增量消息到本地失败:', error);
              });
            }
+          } catch (error) {
+            console.error('[MessageSyncService] 增量同步失败:', error);
+          }
+        } else {
+          // 冷启动：本地无数据，拉快照
+          try {
+            const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
+            const snapshotMessages = snapshotResp?.messages || [];
+
+            if (snapshotMessages.length > 0) {
+              store.setMessages(conversationId, snapshotMessages);
+
+              messageRepository.saveMessagesBatch(
+                MessageMapper.toCachedMessages(snapshotMessages, conversationId)
+              ).catch(error => {
+                console.error('[MessageSyncService] 保存快照消息到本地失败:', error);
+              });
            }
          } catch (error) {
-          console.error('[MessageSyncService] 快照/增量同步失败:', error);
+            console.error('[MessageSyncService] 快照同步失败:', error);
+          }
        }
      } else {
        // 指定了 afterSeq
@@ -399,35 +399,77 @@ export class MessageSyncService implements IMessageSyncService {
        }
      }

-      // 服务端汇总未读为 0 时，仅在本地也一致时才清零（避免缓存竞态导致的抖动）
-      if (totalUnread === 0) {
-        const localSum = Array.from(store.getState().conversations.values())
-          .reduce((sum, conv) => sum + (conv.unread_count || 0), 0);
-        if (localSum > 0) {
-          // 本地有未读但服务端返回 0 → 可能是后端缓存竞态，保留本地状态
-          // 下一次 fetchConversations 会提供权威的 per-conversation 数据
-        } else {
-          // 本地也无未读，确认清零
-          const currentConversations = store.getState().conversations;
-          let anyCleared = false;
-          const newConversations = new Map(currentConversations);
-          for (const [cid, conv] of newConversations) {
-            if ((conv.unread_count || 0) > 0) {
-              newConversations.set(cid, { ...conv, unread_count: 0 });
-              anyCleared = true;
-            }
-          }
-          if (anyCleared) {
-            store.setConversations(newConversations);
-            this.persistConversationListCache();
-          }
-        }
-      }
+      // 服务端汇总未读为 0：不主动清零本地未读，等 fetchConversations 提供权威数据
+      // 避免后端缓存竞态导致的未读数抖动
    } catch (error) {
      console.error('[MessageSyncService] 获取未读数失败:', error);
    }
  }

+  /**
+   * 基于 seq 的轻量增量同步（用于重连场景，替代全量刷新）
+   * 返回 true 表示同步成功，false 表示需要退化为全量刷新
+   */
+  async syncBySeq(): Promise<boolean> {
+    try {
+      const serverItems = await messageService.getSyncData();
+      if (!serverItems || serverItems.length === 0) return false;
+
+      const store = useMessageStore.getState();
+      const localConversations = store.conversations;
+
+      // 没有本地数据，无法增量同步
+      if (localConversations.size === 0) return false;
+
+      // 构建 serverMap
+      const serverMap = new Map<string, { max_seq: number; last_message_at: string }>();
+      for (const item of serverItems) {
+        serverMap.set(normalizeConversationId(item.id), item);
+      }
+
+      // 对比差异
+      const staleIds: string[] = [];
+      for (const [id, conv] of localConversations) {
+        const serverItem = serverMap.get(id);
+        if (!serverItem) continue;
+        const localMaxSeq = Number(conv.last_seq || 0);
+        const serverMaxSeq = Number(serverItem.max_seq || 0);
+        if (serverMaxSeq > localMaxSeq) {
+          staleIds.push(id);
+        }
+      }
+
+      // 差异过大（>50%），退化为全量刷新
+      if (staleIds.length > localConversations.size * 0.5) {
+        return false;
+      }
+
+      // 对有差异的会话更新数据
+      if (staleIds.length > 0) {
+        const detailPromises = staleIds.map(id =>
+          this.fetchConversationDetail(id).catch(() => {})
+        );
+        await Promise.all(detailPromises);
+
+        // 重新读取 store，避免 stale 引用
+        const activeId = useMessageStore.getState().getActiveConversation();
+        if (activeId && staleIds.includes(activeId)) {
+          const localMaxSeq = useMessageStore.getState().getMessages(activeId).reduce(
+            (max, m) => Math.max(max, m.seq || 0), 0
+          );
+          if (localMaxSeq > 0) {
+            await this.fetchMessages(activeId, localMaxSeq);
+          }
+        }
+      }
+
+      return true;
+    } catch (error) {
+      console.error('[MessageSyncService] syncBySeq 失败:', error);
+      return false;
+    }
+  }
+
  /**
   * 检查是否可加载更多会话
   */
--- a/src/stores/message/services/ReadReceiptManager.ts
+++ b/src/stores/message/services/ReadReceiptManager.ts
@@ -64,18 +64,16 @@ export class ReadReceiptManager implements IReadReceiptManager {
      lastReadSeq: seq,
    });

-    // 2. 乐观更新本地状态
+    // 2. 乐观更新本地状态（原子更新，避免中间渲染状态）
    const updatedConv: ConversationResponse = {
      ...conversation,
      unread_count: 0,
      my_last_read_seq: Math.max(Number(conversation.my_last_read_seq || 0), seq),
    };

-    store.updateConversation(updatedConv);
-    
    const currentUnread = store.getUnreadCount();
    const newTotalUnread = Math.max(0, currentUnread.total - prevUnreadCount);
-    store.setUnreadCount(newTotalUnread, currentUnread.system);
+    store.updateConversationWithUnread(updatedConv, newTotalUnread, currentUnread.system);

    // 3. 更新本地数据库
    messageRepository.markConversationAsRead(normalizedId).catch(console.error);
@@ -117,34 +115,52 @@ export class ReadReceiptManager implements IReadReceiptManager {
   */
  async markAllAsRead(): Promise<void> {
    const store = useMessageStore.getState();
-    const conversations = store.conversations;
+    const conversationsMap = store.conversations;
    const prevTotalUnread = store.getUnreadCount().total;
    const currentSystem = store.getUnreadCount().system;

-    // 乐观更新
-    conversations.forEach(conv => {
-      conv.unread_count = 0;
+    // 先收集所有有未读数的会话（不变原 Map）
+    const unreadConversations: Array<{ id: string; lastSeq: number }> = [];
+    conversationsMap.forEach(conv => {
+      if ((conv.unread_count || 0) > 0) {
+        unreadConversations.push({
+          id: conv.id,
+          lastSeq: Number(conv.last_seq || 0),
+        });
+      }
    });

-    store.setUnreadCount(0, currentSystem);
+    if (unreadConversations.length === 0 && prevTotalUnread === 0) return;
+
+    // 乐观更新：创建新 Map，将所有有未读的会话置零（不变原对象）
+    const newConversations = new Map(conversationsMap);
+    conversationsMap.forEach((conv, id) => {
+      if ((conv.unread_count || 0) > 0) {
+        newConversations.set(id, { ...conv, unread_count: 0 });
+      }
+    });
+    store.setConversationsWithUnread(newConversations, 0, currentSystem);

    try {
      // 标记系统消息已读
      await messageService.markAllSystemMessagesRead();

-      // 标记所有会话已读
-      const promises: Promise<any>[] = [];
-      conversations.forEach(conv => {
-        if ((conv.unread_count || 0) > 0) {
-          promises.push(messageService.markAsRead(conv.id, conv.last_seq));
-        }
-      });
-
+      // 批量标记会话已读
+      if (unreadConversations.length > 0) {
+        try {
+          await messageService.markAllAsRead(unreadConversations);
+        } catch {
+          // 降级：逐个调用
+          const promises = unreadConversations.map(conv =>
+            messageService.markAsRead(conv.id, conv.lastSeq)
+          );
          await Promise.all(promises);
+        }
+      }
    } catch (error) {
      console.error('[ReadReceiptManager] 标记所有已读失败:', error);
-      // 回滚
-      store.setUnreadCount(prevTotalUnread, currentSystem);
+      // 回滚：恢复原始会话和未读数
+      store.setConversationsWithUnread(new Map(conversationsMap), prevTotalUnread, currentSystem);
    }
  }

--- a/src/stores/message/services/WSMessageHandler.ts
+++ b/src/stores/message/services/WSMessageHandler.ts
@@ -39,11 +39,15 @@ export class WSMessageHandler implements IWSMessageHandler {
  private fetchConversationsCallback: (forceRefresh: boolean, source: string) => Promise<void>;
  private fetchUnreadCountCallback: () => Promise<void>;
  private fetchMessagesCallback: (conversationId: string) => Promise<void>;
+  private syncBySeqCallback: () => Promise<boolean>;

  private sseUnsubscribe: (() => void) | null = null;
  private isBootstrapping: boolean = false;
  private lastReconnectSyncAt: number = 0;

+  // 系统通知去重（防止重连时重复递增系统未读）
+  private processedNotificationIds: Set<string> = new Set();
+
  // 初始化阶段缓冲来自 SSE 的消息事件
  private bufferedChatMessages: Array<(WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean }> = [];
  private bufferedReadReceipts: Array<WSReadMessage | WSGroupReadMessage> = [];
@@ -57,6 +61,7 @@ export class WSMessageHandler implements IWSMessageHandler {
      fetchConversations: (forceRefresh: boolean, source: string) => Promise<void>;
      fetchUnreadCount: () => Promise<void>;
      fetchMessages: (conversationId: string) => Promise<void>;
+      syncBySeq: () => Promise<boolean>;
    }
  ) {
    this.deduplication = deduplication;
@@ -66,6 +71,7 @@ export class WSMessageHandler implements IWSMessageHandler {
    this.fetchConversationsCallback = callbacks.fetchConversations;
    this.fetchUnreadCountCallback = callbacks.fetchUnreadCount;
    this.fetchMessagesCallback = callbacks.fetchMessages;
+    this.syncBySeqCallback = callbacks.syncBySeq;
  }

  /**
@@ -134,6 +140,21 @@ export class WSMessageHandler implements IWSMessageHandler {
      if (message.is_read) {
        return;
      }
+
+      // 按通知 ID 去重，防止重连时重复递增
+      const notificationId = (message as any).id || (message as any).message_id;
+      if (notificationId) {
+        if (this.processedNotificationIds.has(String(notificationId))) {
+          return;
+        }
+        this.processedNotificationIds.add(String(notificationId));
+        // 防止内存泄漏：超过 500 条时淘汰最旧的 100 条
+        if (this.processedNotificationIds.size > 500) {
+          const firstIds = Array.from(this.processedNotificationIds).slice(0, 100);
+          firstIds.forEach(id => this.processedNotificationIds.delete(id));
+        }
+      }
+
      this.incrementSystemUnread();
      if (message.created_at) {
        const store = useMessageStore.getState();
@@ -154,43 +175,62 @@ export class WSMessageHandler implements IWSMessageHandler {
      const now = Date.now();
      if (now - this.lastReconnectSyncAt > 1500) {
        this.lastReconnectSyncAt = now;
-        const activeConversation = useMessageStore.getState().getActiveConversation();

-        if (!activeConversation) {
+        // 优先尝试 seq 驱动的轻量同步（内部会为 stale 的活动会话拉取消息）
+        this.syncBySeqCallback().then(ok => {
+          if (!ok) {
+            // 增量同步失败或差异过大，退化为全量刷新
            this.fetchConversationsCallback(true, 'sse-reconnect').catch(error => {
              console.error('[WSMessageHandler] 连接后同步会话失败:', error);
            });
-        } else {
+            // 退化场景下仍需同步活动会话消息
+            const activeConv = useMessageStore.getState().getActiveConversation();
+            if (activeConv) {
+              this.fetchMessagesCallback(activeConv).catch(() => {});
+            }
+          }
          this.fetchUnreadCountCallback().catch(error => {
            console.error('[WSMessageHandler] 连接后同步未读数失败:', error);
          });
+        }).catch(() => {
+          // syncBySeq 异常，退化为全量
+          this.fetchConversationsCallback(true, 'sse-reconnect').catch(() => {});
+          const activeConv = useMessageStore.getState().getActiveConversation();
+          if (activeConv) {
+            this.fetchMessagesCallback(activeConv).catch(() => {});
          }
-
-        if (activeConversation) {
-          this.fetchMessagesCallback(activeConversation).catch(error => {
-            console.error('[WSMessageHandler] 连接后同步当前会话消息失败:', error);
+          this.fetchUnreadCountCallback().catch(() => {});
        });
      }
-      }
    });

    // 监听 sync_required 事件，触发增量同步
    wsService.on('sync_required', () => {
      console.log('[WSMessageHandler] 收到 sync_required，开始增量同步');
-      const activeConversation = useMessageStore.getState().getActiveConversation();

+      // 先尝试 seq 驱动的轻量同步（内部会为 stale 的活动会话拉取消息）
+      this.syncBySeqCallback().then(ok => {
+        if (!ok) {
          this.fetchConversationsCallback(true, 'sync_required').catch(error => {
            console.error('[WSMessageHandler] sync_required 同步会话失败:', error);
          });
+          // 退化场景下仍需同步活动会话消息
+          const activeConv = useMessageStore.getState().getActiveConversation();
+          if (activeConv) {
+            this.fetchMessagesCallback(activeConv).catch(() => {});
+          }
+        }
        this.fetchUnreadCountCallback().catch(error => {
          console.error('[WSMessageHandler] sync_required 同步未读数失败:', error);
        });
-
-      if (activeConversation) {
-        this.fetchMessagesCallback(activeConversation).catch(error => {
-          console.error('[WSMessageHandler] sync_required 同步消息失败:', error);
-        });
+      }).catch(() => {
+        this.fetchConversationsCallback(true, 'sync_required').catch(() => {});
+        const activeConv = useMessageStore.getState().getActiveConversation();
+        if (activeConv) {
+          this.fetchMessagesCallback(activeConv).catch(() => {});
        }
+        this.fetchUnreadCountCallback().catch(() => {});
+      });
    });

    wsService.onDisconnect(() => {
@@ -206,6 +246,7 @@ export class WSMessageHandler implements IWSMessageHandler {
      this.sseUnsubscribe();
      this.sseUnsubscribe = null;
    }
+    this.processedNotificationIds.clear();
  }

  /**
@@ -510,19 +551,7 @@ export class WSMessageHandler implements IWSMessageHandler {
  }

  private incrementUnreadCount(conversationId: string): void {
-    const store = useMessageStore.getState();
-    const conversation = store.getConversation(conversationId);
-    if (conversation) {
-      const prevUnreadCount = conversation.unread_count || 0;
-      const updatedConv: ConversationResponse = {
-        ...conversation,
-        unread_count: prevUnreadCount + 1,
-      };
-      store.updateConversation(updatedConv);
-
-      const currentUnread = store.getUnreadCount();
-      store.setUnreadCount(currentUnread.total + 1, currentUnread.system);
-    }
+    useMessageStore.getState().incrementUnreadAtomic(conversationId);
  }

  private incrementSystemUnread(): void {
--- a/src/stores/message/store.ts
+++ b/src/stores/message/store.ts
@@ -100,6 +100,14 @@ export interface MessageActions {
  setNotificationMuted: (conversationId: string, notificationMuted: boolean) => void;
  setInitialized: (initialized: boolean) => void;

+  // 原子性更新（避免中间渲染状态）
+  updateConversationWithUnread: (
+    conversation: ConversationResponse,
+    totalUnread: number,
+    systemUnread: number,
+  ) => void;
+  incrementUnreadAtomic: (conversationId: string) => void;
+
  // 重置
  reset: () => void;
 }
@@ -439,6 +447,53 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
    set({ isInitialized: initialized });
  },

+  updateConversationWithUnread: (
+    conversation: ConversationResponse,
+    totalUnread: number,
+    systemUnread: number,
+  ) => {
+    const id = normalizeConversationId(conversation.id);
+    set(state => {
+      const newConversations = new Map(state.conversations);
+      newConversations.set(id, { ...conversation, id });
+      const conversationList = sortConversationList(newConversations);
+      const newNotificationMutedMap = new Map(state.notificationMutedMap);
+      if (conversation.notification_muted) {
+        newNotificationMutedMap.set(id, true);
+      } else {
+        newNotificationMutedMap.delete(id);
+      }
+      return {
+        conversations: newConversations,
+        conversationList,
+        notificationMutedMap: newNotificationMutedMap,
+        totalUnreadCount: totalUnread,
+        systemUnreadCount: systemUnread,
+      };
+    });
+  },
+
+  incrementUnreadAtomic: (conversationId: string) => {
+    const normalizedId = normalizeConversationId(conversationId);
+    set(state => {
+      const conversation = state.conversations.get(normalizedId);
+      if (!conversation) return state;
+
+      const newConversations = new Map(state.conversations);
+      newConversations.set(normalizedId, {
+        ...conversation,
+        unread_count: (conversation.unread_count || 0) + 1,
+      });
+      const conversationList = sortConversationList(newConversations);
+
+      return {
+        conversations: newConversations,
+        conversationList,
+        totalUnreadCount: state.totalUnreadCount + 1,
+      };
+    });
+  },
+
  // ==================== 重置 ====================

  reset: () => {
--- a/src/stores/message/types.ts
+++ b/src/stores/message/types.ts
@@ -1,21 +1,6 @@
 import type { ConversationResponse, MessageResponse, UserDTO } from '../../types/dto';
 import type { IConversationListPagedSource } from './sources';

-export interface MessageManagerState {
-  conversations: Map<string, ConversationResponse>;
-  conversationList: ConversationResponse[];
-  messagesMap: Map<string, MessageResponse[]>;
-  totalUnreadCount: number;
-  systemUnreadCount: number;
-  isWSConnected: boolean;
-  currentConversationId: string | null;
-  isLoadingConversations: boolean;
-  loadingMessagesSet: Set<string>;
-  isInitialized: boolean;
-  typingUsersMap: Map<string, string[]>;
-  mutedStatusMap: Map<string, boolean>;
-}
-
 export interface ReadStateRecord {
  timestamp: number;
  version: number;
@@ -68,7 +53,6 @@ export interface IConversationOperations {
  removeConversation(conversationId: string): void;
  clearConversations(): void;
  updateUnreadCount(conversationId: string, count: number): void;
-  incrementUnreadCount(conversationId: string): void;
  setSystemUnreadCount(count: number): void;
  incrementSystemUnreadCount(): void;
  decrementSystemUnreadCount(count?: number): void;