refactor(core): distinguish network errors from auth failures and refactor real-time messaging pipeline

- Add isNetworkError() helper and FetchCurrentUserResult discriminated union to prevent logout on network failures - Refactor authService to return { kind: 'user' } | { kind: 'auth_failed' } | { kind: 'network_error' } - Update authStore to preserve login state on network errors, only logout on auth failures - Replace WSMessageHandler with RealtimeIngestionPipeline for improved real-time message handling - Remove MessageDeduplication service; add store-level idempotent addOrReplaceMessage for message dedup - Add atomic systemUnreadCount operations to prevent race conditions - Add SearchHeader component for consistent search UI across screens - Add 'home' TabBar variant with underline style matching HomeScreen - Add Jest test infrastructure and exclude tests from tsconfig - Fix ChatScreen history lock being stuck when scrolling to latest messages - Remove unused call_participant_joined/left WS event types
2026-06-21 17:17:15 +08:00
parent 94a202506b
commit 705b365536
47 changed files with 4357 additions and 1574 deletions
--- a/src/stores/auth/authStore.ts
+++ b/src/stores/auth/authStore.ts
@@ -331,9 +331,15 @@ export const useAuthStore = create<AuthState>()(
      }

      // 3. 纯 API 调用验证 Token 有效性（fetchCurrentUserFromAPI 完全不碰 DB）
-      const user = await authService.fetchCurrentUserFromAPI();
+      //    返回判别类型：
+      //    - 'user'         成功，继续登录流程
+      //    - 'auth_failed'  后端明确拒绝（401）→ 清除登录态
+      //    - 'network_error' 网络故障 → 保留旧登录态（持久化的 isAuthenticated），
+      //                       等待网络恢复后重试，避免断网即被登出
+      const result = await authService.fetchCurrentUserFromAPI();

-      if (user) {
+      if (result.kind === 'user') {
+        const user = result.user;
        const userId = String(user.id);

        // 4. 确保 DB 已初始化（如果预初始化失败，这里会重试）
@@ -364,27 +370,34 @@ export const useAuthStore = create<AuthState>()(
          isAuthenticated: true,
          currentUser: user,
          isLoading: false,
+          error: null,
        });

        // 8. 启动 SSE（不阻塞状态设置）
        startRealtime().catch((err) => {
          console.warn('[AuthStore] 冷启动实时服务失败:', err);
        });
+      } else if (result.kind === 'network_error') {
+        // 网络故障：不登出、不清 userId。
+        // 保留持久化的 isAuthenticated / currentUser（如有），让用户在网络恢复后仍可用。
+        set({
+          isLoading: false,
+          error: '网络连接失败，已使用本地登录态',
+        });
      } else {
-        // Token 已失效或不存在
+        // auth_failed：后端明确拒绝，Token 真失效 → 清除登录态
        await clearUserId();
        useSessionStore.getState().setUserId(null);
        set({
          isAuthenticated: false,
          currentUser: null,
          isLoading: false,
+          error: null,
        });
      }
    } catch (error) {
      console.error('[AuthStore] fetchCurrentUser 异常:', error);
      set({
-        isAuthenticated: false,
-        currentUser: null,
        isLoading: false,
        error: '获取用户信息失败',
      });
--- a/src/stores/call/callStore.ts
+++ b/src/stores/call/callStore.ts
@@ -680,24 +680,6 @@ export const callStore = create<CallState>((set, get) => ({
      })
    );

-    unsubs.push(
-      wsService.on('call_participant_joined', (msg) => {
-        console.log('[CallStore] Participant joined:', msg.user_id);
-        const { currentCall } = get();
-        if (!currentCall || currentCall.id !== msg.call_id) return;
-        get().addParticipant(msg.user_id, { name: msg.user_id });
-      })
-    );
-
-    unsubs.push(
-      wsService.on('call_participant_left', (msg) => {
-        console.log('[CallStore] Participant left:', msg.user_id, 'reason:', msg.reason);
-        const { currentCall } = get();
-        if (!currentCall || currentCall.id !== msg.call_id) return;
-        get().removeParticipant(msg.user_id);
-      })
-    );
-
    const cleanup = () => {
      cleanupResources();
      if (unsubInvited) {
--- a/src/stores/message/MessageManager.ts
+++ b/src/stores/message/MessageManager.ts
@@ -10,13 +10,12 @@ import { useAuthStore } from '../auth/authStore';
 import type { MessageManagerConversationListDeps } from './types';

 // 导入服务
-import { MessageDeduplication, messageDeduplication } from './services/MessageDeduplication';
-import { UserCacheService, userCacheService } from './services/UserCacheService';
+import { UserCacheService } from './services/UserCacheService';
 import { ReadReceiptManager } from './services/ReadReceiptManager';
 import { ConversationOperations } from './services/ConversationOperations';
 import { MessageSendService } from './services/MessageSendService';
 import { MessageSyncService } from './services/MessageSyncService';
-import { WSMessageHandler } from './services/WSMessageHandler';
+import { RealtimeIngestionPipeline } from './services/RealtimeIngestionPipeline';

 // 导入 store
 import { useMessageStore, normalizeConversationId, loadPersistedLastSystemMessageAt } from './store';
@@ -31,13 +30,12 @@ export type {

 class MessageManager {
  // 子模块
-  private deduplication: MessageDeduplication;
  private userCacheService: UserCacheService;
  private readReceiptManager: ReadReceiptManager;
  private conversationOps: ConversationOperations;
  private sendService: MessageSendService;
  private syncService: MessageSyncService;
-  private wsHandler: WSMessageHandler;
+  private pipeline: RealtimeIngestionPipeline;

  // 本地状态
  private currentUserId: string | null = null;
@@ -47,7 +45,6 @@ class MessageManager {

  constructor(deps?: MessageManagerConversationListDeps) {
    // 初始化子模块
-    this.deduplication = new MessageDeduplication();
    this.userCacheService = new UserCacheService();
    this.readReceiptManager = new ReadReceiptManager();
    this.conversationOps = new ConversationOperations();
@@ -58,18 +55,25 @@ class MessageManager {
      this.userCacheService,
      deps
    );
-    this.wsHandler = new WSMessageHandler(
-      this.deduplication,
-      this.userCacheService,
-      () => this.getCurrentUserId(),
-      {
-        markAsRead: (id, seq) => this.markAsRead(id, seq),
-        fetchConversations: (force, source) => this.fetchConversations(force, source),
-        fetchUnreadCount: () => this.fetchUnreadCount(),
-        fetchMessages: (id) => this.fetchMessages(id),
-        syncBySeq: () => this.syncService.syncBySeq(),
-      }
-    );
+    this.pipeline = new RealtimeIngestionPipeline({
+      userCacheService: this.userCacheService,
+      getCurrentUserId: () => this.getCurrentUserId(),
+      onActiveConvMessage: (id, seq) => this.markAsRead(id, seq),
+      triggerSync: async (source) => {
+        // 优先基于 seq 增量同步，失败则全量刷新会话 + 活动会话消息 + 未读
+        const ok = await this.syncService.syncBySeq();
+        if (!ok) {
+          await this.fetchConversations(true, source);
+          const activeConv = useMessageStore.getState().getActiveConversation();
+          if (activeConv) {
+            await this.fetchMessages(activeConv).catch(() => {});
+          }
+        }
+        await this.fetchUnreadCount().catch(() => {});
+      },
+      fetchConversations: (force, src) => this.fetchConversations(force, src),
+      fetchUnreadCount: () => this.fetchUnreadCount(),
+    });

    // 监听认证状态变化
    this.initAuthListener();
@@ -138,13 +142,13 @@ class MessageManager {
          return;
        }

-        this.wsHandler.setBootstrapping(true);
+        this.pipeline.setBootstrapping(true);

        // 从本地缓存恢复最后系统通知时间（避免首次渲染显示当前时间）
        await loadPersistedLastSystemMessageAt();

-        // 初始化SSE监听
-        this.wsHandler.connect();
+        // 注册实时事件监听（start 内部会先注销旧监听器再注册）
+        this.pipeline.start();

        // 加载会话列表
        await this.fetchConversations(false, 'initialize');
@@ -156,15 +160,15 @@ class MessageManager {

        // 先关闭 bootstrap（事件流转入正常处理路径），再做尾部 flush
        // 避免 flush 期间新到的事件被压入 buffer 后无人消费
-        this.wsHandler.setBootstrapping(false);
-        await this.wsHandler.flushBufferedSSEEvents();
+        this.pipeline.setBootstrapping(false);
+        await this.pipeline.flushBufferedEvents();

        if (useMessageStore.getState().currentConversationId) {
          await this.fetchMessages(useMessageStore.getState().currentConversationId!);
        }
      } catch (error) {
        console.error('[MessageManager] 初始化失败:', error);
-        this.wsHandler.setBootstrapping(false);
+        this.pipeline.setBootstrapping(false);
      }
    })();

@@ -176,11 +180,10 @@ class MessageManager {
  }

  destroy(): void {
-    this.wsHandler.disconnect();
+    this.pipeline.stop();
    useMessageStore.getState().reset();
    this.syncService.restartSources();
    this.readReceiptManager.reset();
-    this.deduplication.reset();
    this.userCacheService.reset();
    this.activatingConversationTasks.clear();
    this.initializePromise = null;
--- a/src/stores/message/tests/RealtimeIngestionPipeline.test.ts
+++ b/src/stores/message/tests/RealtimeIngestionPipeline.test.ts
@@ -0,0 +1,192 @@
+/**
+ * RealtimeIngestionPipeline 单测
+ *
+ * 锁定的不变量：
+ * 1. 重复 chat 事件只入库一次、副作用（未读递增）只触发一次。
+ * 2. ingest 中途副作用抛错时，消息仍已落库（修复「标记早于落库」竞态的核心保证）。
+ * 3. start() 重复调用不叠加监听器；stop() 彻底注销（杜绝监听器泄漏）。
+ */
+
+import { wsService } from '@/services/core';
+import { eventBus } from '@/core/events/EventBus';
+import { useMessageStore } from '../store';
+import { RealtimeIngestionPipeline } from '../services/RealtimeIngestionPipeline';
+import type { RealtimeIngestionPipelineDeps } from '../services/RealtimeIngestionPipeline';
+
+function makeDeps(overrides: Partial<RealtimeIngestionPipelineDeps> = {}): RealtimeIngestionPipelineDeps {
+  return {
+    userCacheService: { getSenderInfo: async () => null, enrichMessagesWithSenderInfo: () => {} },
+    getCurrentUserId: () => 'me',
+    onActiveConvMessage: async () => {},
+    triggerSync: async () => {},
+    fetchConversations: async () => {},
+    fetchUnreadCount: async () => {},
+    ...overrides,
+  };
+}
+
+function makeChatMessage(id: string, seq: number, senderId = 'other') {
+  return {
+    type: 'chat' as const,
+    conversation_id: 'c1',
+    id,
+    sender_id: senderId,
+    seq,
+    segments: [{ type: 'text', data: { text: 'hi' } }] as any[],
+    created_at: `2024-01-01T00:00:${String(seq).padStart(2, '0')}Z`,
+  };
+}
+
+describe('RealtimeIngestionPipeline', () => {
+  beforeEach(() => {
+    useMessageStore.getState().reset();
+    (wsService as any).reset();
+    (eventBus as any).reset();
+    // 确保 store 处于已初始化状态，否则事件会被 buffer
+    useMessageStore.getState().setInitialized(true);
+  });
+
+  describe('幂等 ingest', () => {
+    it('重复 chat 事件只入库一次', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      const msg = makeChatMessage('m1', 1);
+
+      await pipeline.ingestChatMessage(msg);
+      await pipeline.ingestChatMessage(msg); // 重复
+
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+    });
+
+    it('重复 chat 事件只递增未读一次', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      // 先确保会话存在，否则 incrementUnreadAtomic 会因找不到会话而跳过
+      useMessageStore.getState().updateConversation({
+        id: 'c1',
+        unread_count: 0,
+        last_seq: 0,
+      } as any);
+      const msg = makeChatMessage('m1', 1);
+
+      await pipeline.ingestChatMessage(msg);
+      await pipeline.ingestChatMessage(msg);
+
+      const conv = useMessageStore.getState().getConversation('c1');
+      expect(conv?.unread_count).toBe(1);
+    });
+
+    it('重复 chat 事件只发一次震动', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      useMessageStore.getState().updateConversation({
+        id: 'c1',
+        unread_count: 0,
+        last_seq: 0,
+      } as any);
+      const msg = makeChatMessage('m1', 1);
+
+      await pipeline.ingestChatMessage(msg);
+      await pipeline.ingestChatMessage(msg);
+
+      const vibrateEvents = (eventBus as any).emitted.filter((e: any) => e.type === 'VIBRATE');
+      expect(vibrateEvents).toHaveLength(1);
+    });
+
+    it('每次 ingest 都发 ACK（即使重复，服务器侧幂等）', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      const msg = makeChatMessage('m1', 1);
+
+      await pipeline.ingestChatMessage(msg);
+      await pipeline.ingestChatMessage(msg);
+
+      expect((wsService as any).ackCalls).toEqual(['m1', 'm1']);
+    });
+
+    it('不同消息独立入库', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      await pipeline.ingestChatMessage(makeChatMessage('m1', 1));
+      await pipeline.ingestChatMessage(makeChatMessage('m2', 2));
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs.map(m => m.id)).toEqual(['m1', 'm2']);
+    });
+  });
+
+  describe('落库先于副作用（不丢消息）', () => {
+    it('副作用回调抛错时消息仍已落库', async () => {
+      // triggerSync 抛错不应影响 ingest（虽然 ingest 本身不调 triggerSync，
+      // 这里验证 onActiveConvMessage 抛错时消息仍在 store）
+      const pipeline = new RealtimeIngestionPipeline(
+        makeDeps({
+          onActiveConvMessage: async () => {
+            throw new Error('markAsRead failed');
+          },
+        })
+      );
+      // 设为活动会话以触发 onActiveConvMessage
+      useMessageStore.getState().setCurrentConversation('c1');
+
+      await pipeline.ingestChatMessage(makeChatMessage('m1', 1));
+
+      // 关键不变量：消息已落库，副作用失败未回滚
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+    });
+  });
+
+  describe('监听器生命周期（杜绝泄漏）', () => {
+    it('start() 注册监听器，事件能被路由', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      pipeline.start();
+      // 触发 chat 事件
+      (wsService as any).emit('chat', makeChatMessage('m1', 1));
+      // ingestChatMessage 是 async，等一个微任务
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+      pipeline.stop();
+    });
+
+    it('stop() 彻底注销所有监听器', () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      pipeline.start();
+      const beforeChat = (wsService as any).listenerCount('chat');
+      const beforeConnect = (wsService as any).connectHandlerCount();
+      expect(beforeChat).toBeGreaterThan(0);
+      expect(beforeConnect).toBeGreaterThan(0);
+
+      pipeline.stop();
+
+      expect((wsService as any).listenerCount('chat')).toBe(0);
+      expect((wsService as any).listenerCount('group_message')).toBe(0);
+      expect((wsService as any).listenerCount('recall')).toBe(0);
+      expect((wsService as any).listenerCount('sync_required')).toBe(0);
+      expect((wsService as any).connectHandlerCount()).toBe(0);
+      expect((wsService as any).disconnectHandlerCount()).toBe(0);
+    });
+
+    it('重复 start() 不叠加监听器', () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      pipeline.start();
+      const firstCount = (wsService as any).listenerCount('chat');
+      expect(firstCount).toBeGreaterThan(0);
+
+      pipeline.start();
+      pipeline.start();
+
+      // 监听器数量应与首次注册后相同，而非 3 倍
+      expect((wsService as any).listenerCount('chat')).toBe(firstCount);
+      pipeline.stop();
+    });
+
+    it('stop() 后再 start() 可重新接收事件', async () => {
+      const pipeline = new RealtimeIngestionPipeline(makeDeps());
+      pipeline.start();
+      pipeline.stop();
+      pipeline.start();
+
+      (wsService as any).emit('chat', makeChatMessage('m1', 1));
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+      pipeline.stop();
+    });
+  });
+});
--- a/src/stores/message/tests/mocks/asyncStorageMock.ts
+++ b/src/stores/message/tests/mocks/asyncStorageMock.ts
@@ -0,0 +1,18 @@
+/**
+ * 测试用 AsyncStorage mock：内存 Map 实现。
+ */
+
+const store = new Map<string, string>();
+
+export default {
+  getItem: async (key: string): Promise<string | null> => store.get(key) ?? null,
+  setItem: async (key: string, value: string): Promise<void> => {
+    store.set(key, value);
+  },
+  removeItem: async (key: string): Promise<void> => {
+    store.delete(key);
+  },
+  __reset(): void {
+    store.clear();
+  },
+};
--- a/src/stores/message/tests/mocks/authStoreMock.ts
+++ b/src/stores/message/tests/mocks/authStoreMock.ts
@@ -0,0 +1,15 @@
+/**
+ * 测试用 authStore mock：提供 getState/subscribe 与 currentUser。
+ */
+
+import { create } from 'zustand';
+
+interface AuthState {
+  currentUser: { id: string } | null;
+  isAuthenticated: boolean;
+}
+
+export const useAuthStore = create<AuthState>(() => ({
+  currentUser: { id: 'me' },
+  isAuthenticated: true,
+}));
--- a/src/stores/message/tests/mocks/coreMock.ts
+++ b/src/stores/message/tests/mocks/coreMock.ts
@@ -0,0 +1,104 @@
+/**
+ * 测试用 core mock：提供 wsService、GroupNoticeType、ApiError 等最小可用实现。
+ * 仅覆盖 RealtimeIngestionPipeline 依赖的表面。
+ */
+
+export const GroupNoticeType = undefined as any;
+
+export class ApiError extends Error {
+  code?: number;
+  constructor(message: string, code?: number) {
+    super(message);
+    this.code = code;
+  }
+}
+
+type EventHandler = (msg: any) => void;
+type ConnHandler = () => void;
+
+/**
+ * 可控的 wsService mock：on/onConnect/onDisconnect 注册处理器，
+ * 测试通过 emit('chat', msg) / emitConnect() 等方法模拟 WS 事件。
+ */
+class WsServiceMock {
+  private handlers: Map<string, EventHandler[]> = new Map();
+  private connectHandlers: ConnHandler[] = [];
+  private disconnectHandlers: ConnHandler[] = [];
+  private syncRequiredHandlers: EventHandler[] = [];
+
+  ackCalls: string[] = [];
+
+  on(type: string, handler: EventHandler): () => void {
+    const list = this.handlers.get(type) || [];
+    list.push(handler);
+    this.handlers.set(type, list);
+    return () => {
+      const cur = this.handlers.get(type) || [];
+      const idx = cur.indexOf(handler);
+      if (idx >= 0) cur.splice(idx, 1);
+    };
+  }
+
+  onConnect(handler: ConnHandler): () => void {
+    this.connectHandlers.push(handler);
+    return () => {
+      const i = this.connectHandlers.indexOf(handler);
+      if (i >= 0) this.connectHandlers.splice(i, 1);
+    };
+  }
+
+  onDisconnect(handler: ConnHandler): () => void {
+    this.disconnectHandlers.push(handler);
+    return () => {
+      const i = this.disconnectHandlers.indexOf(handler);
+      if (i >= 0) this.disconnectHandlers.splice(i, 1);
+    };
+  }
+
+  sendAck(messageId: string): void {
+    this.ackCalls.push(messageId);
+  }
+
+  // ===== 测试驱动方法 =====
+  emit(type: string, msg: any): void {
+    const list = this.handlers.get(type) || [];
+    list.forEach(h => h(msg));
+  }
+
+  emitConnect(): void {
+    this.connectHandlers.forEach(h => h());
+  }
+
+  emitDisconnect(): void {
+    this.disconnectHandlers.forEach(h => h());
+  }
+
+  /** 当前已注册某类型监听器的数量（用于断言无泄漏） */
+  listenerCount(type: string): number {
+    return (this.handlers.get(type) || []).length;
+  }
+
+  connectHandlerCount(): number {
+    return this.connectHandlers.length;
+  }
+
+  disconnectHandlerCount(): number {
+    return this.disconnectHandlers.length;
+  }
+
+  reset(): void {
+    this.handlers.clear();
+    this.connectHandlers = [];
+    this.disconnectHandlers = [];
+    this.ackCalls = [];
+  }
+}
+
+export const wsService = new WsServiceMock();
+
+/**
+ * 供 WSServerMessage 等 wsService.ts 导出的类型在 mock 中可见。
+ * Pipeline 只用到 wsService 实例和 GroupNoticeType；类型 import 是 type-only，
+ * 在运行期被擦除，因此这里不需要真正导出接口类型。
+ */
+export {};
--- a/src/stores/message/tests/mocks/databaseMock.ts
+++ b/src/stores/message/tests/mocks/databaseMock.ts
@@ -0,0 +1,26 @@
+/**
+ * 测试用 database mock：messageRepository / conversationRepository / userCacheRepository。
+ * 仅提供方法存根，单测断言 store 层行为，不验证持久化。
+ */
+
+export const messageRepository = {
+  saveMessage: async () => {},
+  saveMessagesBatch: async () => {},
+  getByConversation: async () => [],
+  getBeforeSeq: async () => [],
+  updateStatus: async () => {},
+  markConversationAsRead: async () => {},
+  delete: async () => {},
+  clearConversation: async () => {},
+};
+
+export const conversationRepository = {
+  getListCache: async () => [],
+  saveWithRelated: async () => {},
+  delete: async () => {},
+};
+
+export const userCacheRepository = {
+  get: () => null,
+  set: () => {},
+};
--- a/src/stores/message/tests/mocks/eventBusMock.ts
+++ b/src/stores/message/tests/mocks/eventBusMock.ts
@@ -0,0 +1,23 @@
+/**
+ * 测试用 eventBus mock：记录 emit 的事件供断言。
+ */
+
+type AppEvent = { type: string; payload?: any };
+
+class EventBusMock {
+  emitted: AppEvent[] = [];
+
+  emit(event: AppEvent): void {
+    this.emitted.push(event);
+  }
+
+  subscribe(_fn: (event: AppEvent) => void): () => void {
+    return () => {};
+  }
+
+  reset(): void {
+    this.emitted = [];
+  }
+}
+
+export const eventBus = new EventBusMock();
--- a/src/stores/message/tests/mocks/messageServiceMock.ts
+++ b/src/stores/message/tests/mocks/messageServiceMock.ts
@@ -0,0 +1,18 @@
+/**
+ * 测试用 messageService mock：仅覆盖单测路径调用的方法。
+ */
+
+export const messageService = {
+  getMessages: async () => ({ messages: [], total: 0, page: 1, page_size: 20 }),
+  getConversationById: async () => null,
+  getUnreadCount: async () => ({ total_unread_count: 0 }),
+  getSystemUnreadCount: async () => ({ unread_count: 0 }),
+  getSyncData: async () => [],
+  markAsRead: async () => {},
+  markAllAsRead: async () => {},
+  markAllSystemMessagesRead: async () => {},
+  fetchRemoteConversationListPage: async () => ({ items: [], hasMore: false }),
+  getSystemMessagesCursor: async () => ({ list: [] }),
+};
+
+export type MessageListResponse = { messages: any[]; total: number; page: number; page_size: number };
--- a/src/stores/message/tests/store.idempotency.test.ts
+++ b/src/stores/message/tests/store.idempotency.test.ts
@@ -0,0 +1,112 @@
+/**
+ * Store 幂等性单测
+ *
+ * 锁定的不变量：
+ * 1. addOrReplaceMessage 以 message id 为幂等键，重复写入返回 false 且不产生重复消息。
+ * 2. 消息按 seq 升序排列。
+ * 3. 相邻消息不会因重复 ingest 而丢失（修复「消息被吞」的核心保证）。
+ * 4. 并发 merge（模拟多次同步写入）不会丢消息。
+ */
+
+import { useMessageStore } from '../store';
+import type { MessageResponse } from '../../../types/dto';
+
+function makeMessage(id: string, seq: number, extra: Partial<MessageResponse> = {}): MessageResponse {
+  return {
+    id,
+    conversation_id: 'c1',
+    sender_id: 'u1',
+    seq,
+    segments: [],
+    created_at: `2024-01-01T00:00:${String(seq).padStart(2, '0')}Z`,
+    status: 'normal',
+    ...extra,
+  };
+}
+
+describe('store 幂等性', () => {
+  beforeEach(() => {
+    useMessageStore.getState().reset();
+  });
+
+  describe('addOrReplaceMessage', () => {
+    it('首次写入返回 true，消息进入 store', () => {
+      const inserted = useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      expect(inserted).toBe(true);
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs).toHaveLength(1);
+      expect(msgs[0].id).toBe('m1');
+    });
+
+    it('同 id 重复写入返回 false 且不产生重复', () => {
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      const second = useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      expect(second).toBe(false);
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+    });
+
+    it('id 相同但 seq 不同（重排）仍视为幂等，不重复插入', () => {
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      const second = useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 999));
+      // 幂等键是 id，而非 seq
+      expect(second).toBe(false);
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+    });
+
+    it('相邻消息不会因中间消息重复 ingest 而丢失', () => {
+      // 模拟 WS 与 REST 增量同步交织：m1, m2 都先入，然后 m1 重复，最后 m3 入
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m2', 2));
+      // m1 重复（如发送回显 + WS echo）
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m3', 3));
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs.map(m => m.id)).toEqual(['m1', 'm2', 'm3']);
+      expect(msgs.map(m => m.seq)).toEqual([1, 2, 3]);
+    });
+
+    it('不同会话互不影响', () => {
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('m1', 1));
+      const inserted = useMessageStore.getState().addOrReplaceMessage('c2', makeMessage('m1', 1));
+      expect(inserted).toBe(true);
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+      expect(useMessageStore.getState().getMessages('c2')).toHaveLength(1);
+    });
+  });
+
+  describe('mergeMessages', () => {
+    it('批量合并同样按 id 去重', () => {
+      useMessageStore.getState().mergeMessages('c1', [makeMessage('m1', 1), makeMessage('m2', 2)]);
+      // 重复 m1 + 新 m3
+      useMessageStore.getState().mergeMessages('c1', [makeMessage('m1', 1), makeMessage('m3', 3)]);
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs.map(m => m.id)).toEqual(['m1', 'm2', 'm3']);
+    });
+
+    it('空 incoming 不改变现有消息', () => {
+      useMessageStore.getState().mergeMessages('c1', [makeMessage('m1', 1)]);
+      useMessageStore.getState().mergeMessages('c1', []);
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(1);
+    });
+
+    it('乱序 seq 入库后仍按 seq 升序排列', () => {
+      useMessageStore.getState().mergeMessages('c1', [
+        makeMessage('m3', 3),
+        makeMessage('m1', 1),
+        makeMessage('m2', 2),
+      ]);
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs.map(m => m.seq)).toEqual([1, 2, 3]);
+    });
+
+    it('并发连续 merge 不丢消息（模拟 WS + REST 增量同步竞态）', () => {
+      // 两条独立 merge，store 在 set 回调内原子合并
+      useMessageStore.getState().mergeMessages('c1', [makeMessage('m1', 1), makeMessage('m2', 2)]);
+      useMessageStore.getState().mergeMessages('c1', [makeMessage('m2', 2), makeMessage('m3', 3)]);
+      useMessageStore.getState().mergeMessages('c1', [makeMessage('m1', 1), makeMessage('m3', 3), makeMessage('m4', 4)]);
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs.map(m => m.id)).toEqual(['m1', 'm2', 'm3', 'm4']);
+    });
+  });
+});
--- a/src/stores/message/tests/tsconfig.json
+++ b/src/stores/message/tests/tsconfig.json
@@ -0,0 +1,11 @@
+{
+  "extends": "../../../../tsconfig.json",
+  "compilerOptions": {
+    "rootDir": "../../../../",
+    "isolatedModules": true,
+    "noEmit": true,
+    "types": ["jest", "node"]
+  },
+  "include": ["**/*.ts"],
+  "exclude": ["node_modules", "mocks/**"]
+}
--- a/src/stores/message/tests/unread.atomics.test.ts
+++ b/src/stores/message/tests/unread.atomics.test.ts
@@ -0,0 +1,112 @@
+/**
+ * 未读计数原子性单测
+ *
+ * 锁定的不变量：
+ * 1. store 原子 actions（incrementSystemUnreadAtomic / setSystemUnreadAtomic /
+ *    decrementSystemUnreadAtomic）保持 total = system + 会话未读的一致性。
+ * 2. ConversationOperations 的系统未读变更走原子路径，不丢更新（修复 RMW 竞态）。
+ * 3. 并发多次递增计数不丢失。
+ */
+
+import { useMessageStore } from '../store';
+import { ConversationOperations } from '../services/ConversationOperations';
+
+describe('未读计数原子性', () => {
+  let convOps: ConversationOperations;
+
+  beforeEach(() => {
+    useMessageStore.getState().reset();
+    convOps = new ConversationOperations();
+  });
+
+  describe('store 原子 actions', () => {
+    it('incrementSystemUnreadAtomic 同步推进 system 与 total', () => {
+      useMessageStore.getState().incrementSystemUnreadAtomic();
+      useMessageStore.getState().incrementSystemUnreadAtomic();
+      const { total, system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(2);
+      expect(total).toBe(2);
+    });
+
+    it('setSystemUnreadAtomic 保持 total 一致性（基于 delta）', () => {
+      useMessageStore.getState().incrementSystemUnreadAtomic(); // system=1,total=1
+      useMessageStore.getState().setSystemUnreadAtomic(5); // system=5, total 应 +4
+      const { total, system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(5);
+      expect(total).toBe(5);
+    });
+
+    it('setSystemUnreadAtomic 向下调整也保持一致', () => {
+      useMessageStore.getState().setSystemUnreadAtomic(10); // system=10,total=10
+      useMessageStore.getState().setSystemUnreadAtomic(3); // system=3,total=3
+      const { total, system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(3);
+      expect(total).toBe(3);
+    });
+
+    it('decrementSystemUnreadAtomic 不低于 0', () => {
+      useMessageStore.getState().setSystemUnreadAtomic(2);
+      useMessageStore.getState().decrementSystemUnreadAtomic(5); // 应被钳制到 0
+      const { total, system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(0);
+      expect(total).toBe(0);
+    });
+
+    it('decrementSystemUnreadAtomic 默认减 1', () => {
+      useMessageStore.getState().setSystemUnreadAtomic(3);
+      useMessageStore.getState().decrementSystemUnreadAtomic();
+      const { system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(2);
+    });
+
+    it('并发连续递增计数不丢失（每个 set 都是原子的）', () => {
+      // 模拟 WS 高并发推送 100 条系统通知
+      for (let i = 0; i < 100; i++) {
+        useMessageStore.getState().incrementSystemUnreadAtomic();
+      }
+      const { total, system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(100);
+      expect(total).toBe(100);
+    });
+  });
+
+  describe('ConversationOperations 走原子路径', () => {
+    it('incrementSystemUnreadCount 不丢更新', () => {
+      for (let i = 0; i < 50; i++) {
+        convOps.incrementSystemUnreadCount();
+      }
+      const { system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(50);
+    });
+
+    it('setSystemUnreadCount 后再并发递增不丢', () => {
+      convOps.setSystemUnreadCount(10);
+      for (let i = 0; i < 20; i++) {
+        convOps.incrementSystemUnreadCount();
+      }
+      const { system, total } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(30);
+      expect(total).toBe(30);
+    });
+
+    it('decrementSystemUnreadCount 与原子行为一致', () => {
+      convOps.setSystemUnreadCount(10);
+      convOps.decrementSystemUnreadCount(3);
+      expect(useMessageStore.getState().getUnreadCount().system).toBe(7);
+    });
+
+    it('会话未读 + 系统未读共同构成 total', () => {
+      // 先放一个有未读的会话
+      useMessageStore.getState().updateConversation({
+        id: 'c1',
+        unread_count: 5,
+        last_seq: 0,
+      } as any);
+      convOps.incrementSystemUnreadCount(); // system=1,total 应 = 5+1=6
+      const { total, system } = useMessageStore.getState().getUnreadCount();
+      expect(system).toBe(1);
+      // total 由会话未读累加与系统未读共同维护；这里仅断言 system 路径
+      expect(total).toBeGreaterThanOrEqual(system);
+    });
+  });
+});
--- a/src/stores/message/constants.ts
+++ b/src/stores/message/constants.ts
@@ -9,30 +9,12 @@
 */
 export const READ_STATE_PROTECTION_DELAY = 5000; // 5秒

-/**
- * 已处理消息ID过期时间（毫秒）
- * 用于消息去重机制，防止内存泄漏
- */
-export const PROCESSED_MESSAGE_ID_EXPIRY = 5 * 60 * 1000; // 5分钟
-
-/**
- * 已处理消息ID集合最大大小
- * 超过此大小时触发清理
- */
-export const PROCESSED_MESSAGE_ID_MAX_SIZE = 1000;
-
 /**
 * 缓冲SSE事件刷新最大迭代次数
 * 防止在线流持续不断导致长时间阻塞初始化
 */
 export const MAX_FLUSH_ITERATIONS = 3;

-/**
- * 会话列表刷新延迟检查间隔（毫秒）
- * 用于防抖处理
- */
-export const CONVERSATION_REFRESH_THROTTLE = 1000;
-
 /**
 * 重连同步最小间隔（毫秒）
 * 防止频繁重连导致的重复同步
--- a/src/stores/message/index.ts
+++ b/src/stores/message/index.ts
@@ -20,50 +20,41 @@ export type {
  ReadStateRecord,
  MessageManagerConversationListDeps,
  HandleNewMessageOptions,
-  IMessageDeduplication,
  IUserCacheService,
  IReadReceiptManager,
  IConversationOperations,
  IMessageSendService,
  IMessageSyncService,
-  IWSMessageHandler,
 } from './types';

 // ==================== 常量导出 ====================
 export {
  READ_STATE_PROTECTION_DELAY,
-  PROCESSED_MESSAGE_ID_EXPIRY,
-  PROCESSED_MESSAGE_ID_MAX_SIZE,
  MAX_FLUSH_ITERATIONS,
-  CONVERSATION_REFRESH_THROTTLE,
  MIN_RECONNECT_SYNC_INTERVAL,
 } from './constants';

 // ==================== 服务导出 ====================
 export {
-  // 消息去重服务
-  MessageDeduplication,
-  messageDeduplication,
-  
+  // 实时消息入口
+  RealtimeIngestionPipeline,
+
  // 用户缓存服务
  UserCacheService,
-  userCacheService,
-  
+
  // 已读回执管理器
  ReadReceiptManager,
-  
+
  // 会话操作服务
  ConversationOperations,
-  
+
  // 消息发送服务
  MessageSendService,
-  
+
  // 消息同步服务
  MessageSyncService,
-  
-  // WebSocket 消息处理器
-  WSMessageHandler,
 } from './services';
+export type { RealtimeIngestionPipelineDeps, IngestOptions } from './services';

 // ==================== Hooks 导出（便捷版，使用 messageManager 单例） ====================
 export {
--- a/src/stores/message/services/ConversationOperations.ts
+++ b/src/stores/message/services/ConversationOperations.ts
@@ -99,30 +99,23 @@ export class ConversationOperations implements IConversationOperations {
  }

  /**
-   * 设置系统消息未读数
+   * 设置系统消息未读数（原子：避免与 WS 推送的并发递增丢更新）
   */
  setSystemUnreadCount(count: number): void {
-    const store = useMessageStore.getState();
-    const currentUnread = store.getUnreadCount();
-    store.setUnreadCount(currentUnread.total, count);
+    useMessageStore.getState().setSystemUnreadAtomic(count);
  }

  /**
-   * 增加系统消息未读数
+   * 增加系统消息未读数（原子）
   */
  incrementSystemUnreadCount(): void {
-    const store = useMessageStore.getState();
-    const currentUnread = store.getUnreadCount();
-    store.setUnreadCount(currentUnread.total, currentUnread.system + 1);
+    useMessageStore.getState().incrementSystemUnreadAtomic();
  }

  /**
-   * 减少系统消息未读数
+   * 减少系统消息未读数（原子，不低于 0）
   */
  decrementSystemUnreadCount(count = 1): void {
-    const store = useMessageStore.getState();
-    const currentUnread = store.getUnreadCount();
-    const newSystemCount = Math.max(0, currentUnread.system - count);
-    store.setUnreadCount(currentUnread.total, newSystemCount);
+    useMessageStore.getState().decrementSystemUnreadAtomic(count);
  }
 }
--- a/src/stores/message/services/MessageDeduplication.ts
+++ b/src/stores/message/services/MessageDeduplication.ts
@@ -1,61 +0,0 @@
-/**
- * 消息去重服务
- * 处理消息去重逻辑，防止重复处理相同的消息
- */
-
-import type { IMessageDeduplication } from '../types';
-import {
-  PROCESSED_MESSAGE_ID_EXPIRY,
-  PROCESSED_MESSAGE_ID_MAX_SIZE,
-} from '../constants';
-
-export class MessageDeduplication implements IMessageDeduplication {
-  // 已处理消息ID集合（用于去重，防止ACK消息和正常消息重复处理）
-  private processedMessageIds: Set<string> = new Set();
-  private processedMessageIdsExpiry: Map<string, number> = new Map();
-
-  /**
-   * 检查消息是否已处理
-   */
-  isMessageProcessed(messageId: string): boolean {
-    return this.processedMessageIds.has(messageId);
-  }
-
-  /**
-   * 标记消息已处理
-   */
-  markMessageAsProcessed(messageId: string): void {
-    this.processedMessageIds.add(messageId);
-    this.processedMessageIdsExpiry.set(messageId, Date.now());
-    
-    // 定期清理
-    if (this.processedMessageIds.size > PROCESSED_MESSAGE_ID_MAX_SIZE) {
-      this.cleanupProcessedMessageIds();
-    }
-  }
-
-  /**
-   * 清理过期的消息ID（防止内存泄漏）
-   */
-  cleanupProcessedMessageIds(): void {
-    const now = Date.now();
-    for (const [id, timestamp] of this.processedMessageIdsExpiry.entries()) {
-      if (now - timestamp > PROCESSED_MESSAGE_ID_EXPIRY) {
-        this.processedMessageIds.delete(id);
-        this.processedMessageIdsExpiry.delete(id);
-      }
-    }
-  }
-
-  /**
-   * 重置所有已处理的消息ID
-   * 用于测试或需要清除缓存的场景
-   */
-  reset(): void {
-    this.processedMessageIds.clear();
-    this.processedMessageIdsExpiry.clear();
-  }
-}
-
-// 单例导出
-export const messageDeduplication = new MessageDeduplication();
--- a/src/stores/message/services/MessageSyncService.ts
+++ b/src/stores/message/services/MessageSyncService.ts
@@ -36,7 +36,6 @@ export class MessageSyncService implements IMessageSyncService {
  private inflightFetches: Map<string, Promise<void>> = new Map();
  private inflightFetchConversations: Promise<void> | null = null;
  private inflightSyncBySeq: Promise<boolean> | null = null;
-  private inflightSyncByVersion: Promise<boolean> | null = null;

  constructor(
    getCurrentUserId: () => string | null,
@@ -507,77 +506,6 @@ export class MessageSyncService implements IMessageSyncService {
    }
  }

-  /**
-   * 基于版本号的增量同步 — 去重入口
-   */
-  async syncByVersion(): Promise<boolean> {
-    if (this.inflightSyncByVersion) return this.inflightSyncByVersion;
-    this.inflightSyncByVersion = this._doSyncByVersion();
-    try {
-      return await this.inflightSyncByVersion;
-    } finally {
-      this.inflightSyncByVersion = null;
-    }
-  }
-
-  /**
-   * 基于版本号的增量同步 — 核心实现
-   */
-  private async _doSyncByVersion(): Promise<boolean> {
-    try {
-      const store = useMessageStore.getState();
-      const version = store.syncVersion ?? 0;
-
-      const result = await messageService.getSyncByVersion(version, 100);
-      if (!result) return false;
-
-      // full_sync 表示需要全量刷新
-      if (result.full_sync) {
-        return false;
-      }
-
-      if (result.changes.length === 0) {
-        // 保存当前版本号，下次增量同步时使用
-        useMessageStore.getState().setSyncVersion(result.current_version);
-        return true;
-      }
-
-      // 按变更类型处理
-      const staleIds: string[] = [];
-      for (const change of result.changes) {
-        const normId = normalizeConversationId(change.conversation_id);
-        if (change.change_type === 'hide') {
-          // 用户隐藏了会话，从本地移除
-          store.removeConversation(normId);
-        } else {
-          // new_msg, read, pin, unpin, mute, unmute, group_update
-          staleIds.push(normId);
-        }
-      }
-
-      // 刷新有变更的会话
-      if (staleIds.length > 0) {
-        const detailPromises = staleIds.map(id =>
-          this.fetchConversationDetail(id).catch(() => {})
-        );
-        await Promise.all(detailPromises);
-      }
-
-      // 保存版本号
-      useMessageStore.getState().setSyncVersion(result.current_version);
-
-      // has_more 时继续拉取
-      if (result.has_more) {
-        return this.syncByVersion();
-      }
-
-      return true;
-    } catch (error) {
-      console.error('[MessageSyncService] syncByVersion 失败:', error);
-      return false;
-    }
-  }
-
  /**
   * 检查是否可加载更多会话
   */
--- a/src/stores/message/services/ReadReceiptManager.ts
+++ b/src/stores/message/services/ReadReceiptManager.ts
@@ -180,69 +180,6 @@ export class ReadReceiptManager implements IReadReceiptManager {
    }
  }

-  /**
-   * 开始已读操作，返回版本号
-   */
-  beginReadOperation(conversationId: string, seq: number): number {
-    this.readStateVersion++;
-    const currentVersion = this.readStateVersion;
-    
-    this.pendingReadMap.set(conversationId, {
-      timestamp: Date.now(),
-      version: currentVersion,
-      lastReadSeq: seq,
-    });
-    
-    return currentVersion;
-  }
-
-  /**
-   * 结束已读操作
-   */
-  endReadOperation(conversationId: string, version: number, success: boolean): void {
-    if (!success) {
-      this.pendingReadMap.delete(conversationId);
-      return;
-    }
-
-    // API 成功后，设置延迟清除保护
-    const clearTimer = setTimeout(() => {
-      const record = this.pendingReadMap.get(conversationId);
-      if (record && record.version === version) {
-        this.pendingReadMap.delete(conversationId);
-      }
-    }, READ_STATE_PROTECTION_DELAY);
-
-    const existingRecord = this.pendingReadMap.get(conversationId);
-    if (existingRecord) {
-      this.pendingReadMap.set(conversationId, {
-        ...existingRecord,
-        clearTimer,
-      });
-    }
-  }
-
-  /**
-   * 检查是否有进行中的已读操作
-   */
-  isReadOperationPending(conversationId: string): boolean {
-    return this.pendingReadMap.has(conversationId);
-  }
-
-  /**
-   * 获取当前已读状态版本号
-   */
-  getReadStateVersion(): number {
-    return this.readStateVersion;
-  }
-
-  /**
-   * 获取待处理的已读映射
-   */
-  getPendingReadMap(): Map<string, ReadStateRecord> {
-    return this.pendingReadMap;
-  }
-
  /**
   * 根据已读保护状态智能合并会话
   */
--- a/src/stores/message/services/RealtimeIngestionPipeline.ts
+++ b/src/stores/message/services/RealtimeIngestionPipeline.ts
@@ -0,0 +1,605 @@
+/**
+ * RealtimeIngestionPipeline — 实时消息单一入口与幂等处理
+ *
+ * 设计目标（对标 Matrix/Element 的 idempotent event ingestion 与 Discord 的 gateway
+ * event deduplication）：
+ *
+ * 1. **幂等键下沉到 store**：不再在上层维护独立的「已处理消息 id」集合。
+ *    去重的唯一真理源是 `useMessageStore.addOrReplaceMessage`，它在 zustand set 的
+ *    原子边界内判断消息是否已存在。落库与去重原子绑定，**永不丢消息**
+ *    （最坏情况是重复 ingest，被 store 静默去重）。
+ *
+ * 2. **副作用与状态变更分离**：未读递增、震动、本地持久化、自动已读、会话更新都是
+ *    「副作用」，只有在消息**确认新入库后**（`addOrReplaceMessage` 返回 true）才执行。
+ *    任何副作用失败都不回滚已落库的消息——彻底消除旧实现「标记早于落库」导致的
+ *    永久丢消息竞态。
+ *
+ * 3. **显式监听器生命周期**：所有 `wsService.on/onConnect/onDisconnect` 返回的
+ *    unsubscribe 函数被收集到 `unsubscribers`，`start()` 先彻底注销旧监听器再注册，
+ *    `stop()` 一并释放——杜绝重复注册与重登录后的监听器泄漏。
+ *
+ * 该类取代旧的 `WSMessageHandler` + `MessageDeduplication`。
+ */
+
+import type {
+  WSChatMessage,
+  WSGroupChatMessage,
+  WSReadMessage,
+  WSGroupReadMessage,
+  WSRecallMessage,
+  WSGroupRecallMessage,
+  WSGroupTypingMessage,
+  WSGroupNoticeMessage,
+  WSNotificationMessage,
+} from '@/services/core';
+import { wsService, GroupNoticeType } from '@/services/core';
+import { eventBus } from '@/core/events/EventBus';
+import { messageRepository } from '@/database';
+import type { MessageResponse } from '../../../types/dto';
+import {
+  MAX_FLUSH_ITERATIONS,
+  MIN_RECONNECT_SYNC_INTERVAL,
+} from '../constants';
+import { useMessageStore, normalizeConversationId } from '../store';
+import type { IUserCacheService } from '../types';
+
+/** 实时事件统一缓冲类型（bootstrap 阶段暂存，初始化完成后回放） */
+type BufferedEvent =
+  | { kind: 'chat'; message: WSChatMessage | WSGroupChatMessage }
+  | { kind: 'read'; message: WSReadMessage | WSGroupReadMessage };
+
+/** ingest 时可抑制的副作用开关 */
+export interface IngestOptions {
+  suppressUnreadIncrement?: boolean;
+  suppressVibration?: boolean;
+  suppressConversationUpdates?: boolean;
+  suppressAutoMarkAsRead?: boolean;
+}
+
+/** Pipeline 对外依赖（由 MessageManager 注入） */
+export interface RealtimeIngestionPipelineDeps {
+  userCacheService: IUserCacheService;
+  getCurrentUserId: () => string | null;
+  /** 活动会话收到新消息时回调（用于自动标记已读） */
+  onActiveConvMessage: (conversationId: string, seq: number) => Promise<void>;
+  /** 重连 / sync_required 时触发的增量同步入口 */
+  triggerSync: (source: string) => Promise<void>;
+  /** 权威刷新会话列表入口（flush 阶段调用） */
+  fetchConversations: (forceRefresh: boolean, source: string) => Promise<void>;
+  /** 权威刷新未读数入口（flush 阶段调用） */
+  fetchUnreadCount: () => Promise<void>;
+}
+
+export class RealtimeIngestionPipeline {
+  private deps: RealtimeIngestionPipelineDeps;
+  /** 所有 wsService 监听器的注销函数；start() 先清空再注册，stop() 全部释放 */
+  private unsubscribers: Array<() => void> = [];
+  /** bootstrap 阶段是否缓冲事件（初始化完成前为 true） */
+  private isBootstrapping = false;
+  /** bootstrap 期间缓冲的实时事件 */
+  private bufferedEvents: BufferedEvent[] = [];
+
+  /** 同步触发器状态：防止 onConnect 和 sync_required 并发 */
+  private syncInProgress = false;
+  private lastSyncTriggerAt = 0;
+
+  /** 系统通知去重（防止重连时重复递增系统未读） */
+  private processedNotificationIds: Set<string> = new Set();
+
+  constructor(deps: RealtimeIngestionPipelineDeps) {
+    this.deps = deps;
+  }
+
+  // ==================== 生命周期 ====================
+
+  /**
+   * 注册所有 wsService 监听器。重复调用安全：先注销旧监听器再注册新的。
+   */
+  start(): void {
+    this.stop();
+
+    const push = <T>(unsub: () => void) => this.unsubscribers.push(unsub);
+
+    // 聊天消息：bootstrap 期间缓冲，否则直接 ingest
+    push(
+      wsService.on('chat', (message: WSChatMessage) => {
+        if (!this.isReady()) {
+          this.bufferedEvents.push({ kind: 'chat', message });
+          return;
+        }
+        this.ingestChatMessage(message).catch(error => {
+          console.error('[RealtimeIngestionPipeline] ingest chat 失败:', error);
+        });
+      })
+    );
+
+    push(
+      wsService.on('group_message', (message: WSGroupChatMessage) => {
+        if (!this.isReady()) {
+          this.bufferedEvents.push({ kind: 'chat', message });
+          return;
+        }
+        this.ingestChatMessage(message).catch(error => {
+          console.error('[RealtimeIngestionPipeline] ingest group_message 失败:', error);
+        });
+      })
+    );
+
+    // 已读回执：bootstrap 期间缓冲
+    push(
+      wsService.on('read', (message: WSReadMessage) => {
+        if (!this.isReady()) {
+          this.bufferedEvents.push({ kind: 'read', message });
+          return;
+        }
+        this.handleReadReceipt(message);
+      })
+    );
+
+    push(
+      wsService.on('group_read', (message: WSGroupReadMessage) => {
+        if (!this.isReady()) {
+          this.bufferedEvents.push({ kind: 'read', message });
+          return;
+        }
+        this.handleGroupReadReceipt(message);
+      })
+    );
+
+    // 撤回、输入状态、群通知：直接处理（这些操作幂等或无副作用丢失风险）
+    push(
+      wsService.on('recall', (message: WSRecallMessage) => this.handleRecallMessage(message))
+    );
+    push(
+      wsService.on('group_recall', (message: WSGroupRecallMessage) => this.handleGroupRecallMessage(message))
+    );
+    push(
+      wsService.on('group_typing', (message: WSGroupTypingMessage) => this.handleGroupTyping(message))
+    );
+    push(
+      wsService.on('group_notice', (message: WSGroupNoticeMessage) => this.handleGroupNotice(message))
+    );
+
+    // 系统通知：去重后递增系统未读
+    push(
+      wsService.on('notification', (message: WSNotificationMessage) => this.handleNotification(message))
+    );
+
+    // 连接状态
+    push(
+      wsService.onConnect(() => {
+        useMessageStore.getState().setSSEConnected(true);
+        this.triggerSync('sse-reconnect').catch(error => {
+          console.error('[RealtimeIngestionPipeline] 重连同步失败:', error);
+        });
+      })
+    );
+
+    push(
+      wsService.on('sync_required', () => {
+        this.triggerSync('sync_required').catch(error => {
+          console.error('[RealtimeIngestionPipeline] sync_required 同步失败:', error);
+        });
+      })
+    );
+
+    push(
+      wsService.onDisconnect(() => {
+        useMessageStore.getState().setSSEConnected(false);
+      })
+    );
+  }
+
+  /**
+   * 注销所有监听器并清空缓冲。重复调用安全。
+   */
+  stop(): void {
+    for (const unsub of this.unsubscribers) {
+      try {
+        unsub();
+      } catch (error) {
+        console.error('[RealtimeIngestionPipeline] 注销监听器失败:', error);
+      }
+    }
+    this.unsubscribers = [];
+  }
+
+  /** 设置 bootstrap 标志 */
+  setBootstrapping(value: boolean): void {
+    this.isBootstrapping = value;
+  }
+
+  /**
+   * 初始化完成后，回放缓冲的实时事件。
+   * 每轮取走当前 buffer 后立即清空（flush 期间新到事件走正常路径），直到稳定。
+   */
+  async flushBufferedEvents(): Promise<void> {
+    let lastLen = -1;
+    for (let i = 0; i < MAX_FLUSH_ITERATIONS; i++) {
+      const events = this.bufferedEvents;
+      if (events.length === lastLen) {
+        return;
+      }
+      lastLen = events.length;
+      this.bufferedEvents = [];
+
+      for (const event of events) {
+        if (event.kind === 'chat') {
+          await this.ingestChatMessage(event.message, {
+            suppressUnreadIncrement: true,
+            suppressVibration: true,
+            suppressConversationUpdates: true,
+            suppressAutoMarkAsRead: true,
+          });
+        } else if (event.message.type === 'read') {
+          this.handleReadReceipt(event.message as WSReadMessage);
+        } else {
+          this.handleGroupReadReceipt(event.message as WSGroupReadMessage);
+        }
+      }
+
+      // 使用服务器权威数据刷新会话与未读
+      await this.deps.fetchConversations(true, 'sse-buffer-flush');
+      await this.deps.fetchUnreadCount();
+    }
+  }
+
+  // ==================== 核心：聊天消息幂等 ingest ====================
+
+  /**
+   * 幂等 ingest 一条聊天消息（私聊 chat / 群聊 group_message）。
+   *
+   * 流程（关键：先落库，落库成功后才发副作用）：
+   * 1. 构造 MessageResponse
+   * 2. 通过 store.addOrReplaceMessage 原子写入并取得「是否新写入」标志
+   * 3. 始终发送 ACK（服务器侧幂等；ACK 失败不影响消息）
+   * 4. 仅当 inserted === true 时触发副作用：会话更新 / 未读递增 / 震动 /
+   *    自动已读 / 本地持久化 / 群聊 sender 补全
+   */
+  async ingestChatMessage(
+    message: (WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean },
+    options?: IngestOptions
+  ): Promise<void> {
+    const store = useMessageStore.getState();
+    const { conversation_id, id, sender_id, seq, segments, created_at, _isAck } = message;
+    const normalizedConversationId = normalizeConversationId(conversation_id);
+    const currentUserId = this.deps.getCurrentUserId();
+
+    const suppressUnreadIncrement = options?.suppressUnreadIncrement ?? false;
+    const suppressVibration = options?.suppressVibration ?? false;
+    const suppressConversationUpdates = options?.suppressConversationUpdates ?? false;
+    const suppressAutoMarkAsRead = options?.suppressAutoMarkAsRead ?? false;
+
+    // 1. 构造消息对象
+    const newMessage: MessageResponse = {
+      id,
+      conversation_id: normalizedConversationId,
+      sender_id,
+      seq: typeof seq === 'string' ? parseInt(seq, 10) : (seq ?? 0),
+      segments: segments || [],
+      created_at,
+      status: 'normal',
+    };
+
+    // 2. 原子幂等写入；inserted === true 才是真正的新消息
+    const inserted = useMessageStore.getState().addOrReplaceMessage(normalizedConversationId, newMessage);
+
+    // 3. 始终 ACK（即使重复，服务器侧幂等；ACK 失败不回滚消息）
+    wsService.sendAck(id);
+
+    // 4. 幂等：已存在的消息跳过所有副作用
+    if (!inserted) return;
+
+    // —— 以下副作用仅在消息新入库后执行，任一失败都不回滚消息 ——
+
+    if (!suppressConversationUpdates) {
+      this.applyConversationUpdate(normalizedConversationId, newMessage, created_at);
+    }
+
+    // 重新读取 store 以获得最新的活动会话状态（避免闭包陈旧）
+    const latestStore = useMessageStore.getState();
+    const isActiveConversation = normalizedConversationId === latestStore.getActiveConversation();
+    const isCurrentUserMessage = sender_id === currentUserId;
+
+    const shouldIncrementUnread =
+      !suppressUnreadIncrement &&
+      !isCurrentUserMessage &&
+      !isActiveConversation &&
+      !!currentUserId &&
+      !_isAck;
+
+    if (shouldIncrementUnread) {
+      const isNotificationMuted = latestStore.isNotificationMuted(normalizedConversationId);
+      if (!suppressVibration && !isNotificationMuted) {
+        const vibrationType = message.type === 'group_message' ? 'group_message' : 'message';
+        eventBus.emit({ type: 'VIBRATE', payload: { type: vibrationType } });
+      }
+      this.incrementUnreadCount(normalizedConversationId);
+    }
+
+    // 活动会话且非本人消息：自动标记已读
+    if (isActiveConversation && !isCurrentUserMessage && !suppressAutoMarkAsRead) {
+      this.deps.onActiveConvMessage(normalizedConversationId, seq).catch(error => {
+        console.error('[RealtimeIngestionPipeline] 自动标记已读失败:', error);
+      });
+    }
+
+    // 群聊消息：异步补 sender 信息（patchMessages 自身幂等，不影响去重）
+    if (
+      message.type === 'group_message' &&
+      sender_id &&
+      sender_id !== currentUserId &&
+      sender_id !== '10000'
+    ) {
+      this.deps.userCacheService
+        .getSenderInfo(sender_id)
+        .then(user => {
+          if (!user) return;
+          const patches = new Map<string, Partial<MessageResponse>>();
+          patches.set(String(id), { sender: user });
+          useMessageStore.getState().patchMessages(normalizedConversationId, patches);
+        })
+        .catch(error => {
+          console.error('[RealtimeIngestionPipeline] 获取发送者信息失败:', { userId: sender_id, error });
+        });
+    }
+
+    // 异步保存到本地数据库（fire-and-forget）
+    const textContent =
+      segments?.filter((s: any) => s.type === 'text').map((s: any) => s.data?.text || '').join('') || '';
+    messageRepository
+      .saveMessage({
+        id,
+        conversationId: normalizedConversationId,
+        senderId: sender_id || '',
+        content: textContent,
+        type: segments?.find((s: any) => s.type === 'image') ? 'image' : 'text',
+        isRead: isActiveConversation,
+        createdAt: (() => {
+          const d = new Date(created_at);
+          return Number.isNaN(d.getTime()) ? new Date().toISOString() : d.toISOString();
+        })(),
+        seq,
+        status: 'normal',
+        segments,
+      })
+      .catch(error => {
+        console.error('[RealtimeIngestionPipeline] 保存消息到本地失败:', error);
+      });
+  }
+
+  // ==================== 其他事件处理 ====================
+
+  /** 私聊已读回执（当前为占位，保留扩展点） */
+  handleReadReceipt(_message: WSReadMessage): void {
+    // 预留：可在此更新对方已读状态
+  }
+
+  /** 群聊已读回执（当前为占位，保留扩展点） */
+  handleGroupReadReceipt(_message: WSGroupReadMessage): void {
+    // 预留：可在此更新成员已读状态
+  }
+
+  /** 私聊消息撤回 */
+  handleRecallMessage(message: WSRecallMessage): void {
+    const { conversation_id, message_id } = message;
+    const normalizedConversationId = normalizeConversationId(conversation_id);
+
+    this.markMessageAsRecalled(normalizedConversationId, message_id);
+    this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id);
+
+    messageRepository.updateStatus(message_id, 'recalled', true).catch(error => {
+      console.error('[RealtimeIngestionPipeline] 更新本地撤回状态失败:', error);
+    });
+  }
+
+  /** 群聊消息撤回 */
+  handleGroupRecallMessage(message: WSGroupRecallMessage): void {
+    const { conversation_id, message_id } = message;
+    const normalizedConversationId = normalizeConversationId(conversation_id);
+
+    this.markMessageAsRecalled(normalizedConversationId, message_id);
+    this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id);
+
+    messageRepository.updateStatus(message_id, 'recalled', true).catch(error => {
+      console.error('[RealtimeIngestionPipeline] 更新本地撤回状态失败:', error);
+    });
+  }
+
+  /** 群聊输入状态 */
+  handleGroupTyping(message: WSGroupTypingMessage): void {
+    const store = useMessageStore.getState();
+    const { group_id, user_id, is_typing } = message;
+    const groupIdStr = String(group_id);
+
+    const currentTypingUsers = store.getTypingUsers(groupIdStr);
+    let updatedTypingUsers: string[];
+
+    if (is_typing) {
+      updatedTypingUsers = currentTypingUsers.includes(user_id)
+        ? currentTypingUsers
+        : [...currentTypingUsers, user_id];
+    } else {
+      updatedTypingUsers = currentTypingUsers.filter(id => id !== user_id);
+    }
+
+    if (updatedTypingUsers.length !== currentTypingUsers.length) {
+      store.setTypingUsers(groupIdStr, updatedTypingUsers);
+    }
+  }
+
+  /** 群通知（禁言/系统消息等） */
+  handleGroupNotice(message: WSGroupNoticeMessage): void {
+    const store = useMessageStore.getState();
+    const { notice_type, group_id, data, timestamp, message_id, seq } = message;
+    const groupIdStr = String(group_id);
+
+    // 处理禁言/解除禁言（仅本人相关）
+    if (notice_type === 'muted' || notice_type === 'unmuted') {
+      const mutedUserId = data?.user_id;
+      const currentUserId = this.deps.getCurrentUserId();
+      if (mutedUserId && mutedUserId === currentUserId) {
+        store.setMutedStatus(groupIdStr, notice_type === 'muted');
+      }
+    }
+
+    // 补一条系统消息到会话消息流（幂等：addOrReplaceMessage 内部去重）
+    if (message_id && typeof seq === 'number' && seq > 0) {
+      const conversationId = this.findConversationIdByGroupId(groupIdStr);
+      if (conversationId) {
+        const noticeText = this.buildGroupNoticeText(notice_type, data);
+        const systemNoticeMessage: MessageResponse = {
+          id: String(message_id),
+          conversation_id: conversationId,
+          sender_id: '10000',
+          seq,
+          segments: noticeText
+            ? [{ type: 'text', data: { text: noticeText } as any }]
+            : [],
+          status: 'normal',
+          category: 'notification',
+          created_at: (() => {
+            const d = new Date(timestamp || Date.now());
+            return Number.isNaN(d.getTime()) ? new Date().toISOString() : d.toISOString();
+          })(),
+        };
+        useMessageStore.getState().addOrReplaceMessage(conversationId, systemNoticeMessage);
+      }
+    }
+  }
+
+  /** 系统通知：去重后递增系统未读 */
+  handleNotification(message: WSNotificationMessage): void {
+    if (message.is_read) return;
+
+    // 按通知 ID 去重，防止重连时重复递增
+    const notificationId = (message as any).id || (message as any).message_id;
+    if (notificationId) {
+      const idStr = String(notificationId);
+      if (this.processedNotificationIds.has(idStr)) return;
+      this.processedNotificationIds.add(idStr);
+      // 防止内存泄漏：超过 500 条淘汰最旧 100 条
+      if (this.processedNotificationIds.size > 500) {
+        const firstIds = Array.from(this.processedNotificationIds).slice(0, 100);
+        firstIds.forEach(id => this.processedNotificationIds.delete(id));
+      }
+    }
+
+    useMessageStore.getState().incrementSystemUnreadAtomic();
+
+    if (message.created_at) {
+      const store = useMessageStore.getState();
+      const currentLast = store.lastSystemMessageAt;
+      const msgDate = new Date(message.created_at);
+      if (!Number.isNaN(msgDate.getTime()) && (!currentLast || msgDate > new Date(currentLast))) {
+        store.setLastSystemMessageAt(message.created_at);
+      }
+    }
+  }
+
+  // ==================== 同步触发器 ====================
+
+  /**
+   * 统一同步入口：节流 + 去重，onConnect 与 sync_required 共用。
+   * bootstrap 期间跳过（由 flushBufferedEvents 统一回放）。
+   */
+  private async triggerSync(source: string): Promise<void> {
+    if (this.isBootstrapping) return;
+
+    const now = Date.now();
+    if (now - this.lastSyncTriggerAt < MIN_RECONNECT_SYNC_INTERVAL) return;
+    if (this.syncInProgress) return;
+
+    this.syncInProgress = true;
+    this.lastSyncTriggerAt = now;
+
+    try {
+      await this.deps.triggerSync(source);
+    } catch (error) {
+      console.error('[RealtimeIngestionPipeline] 同步失败:', error);
+    } finally {
+      this.syncInProgress = false;
+    }
+  }
+
+  // ==================== 私有工具方法 ====================
+
+  /** store 已初始化且非 bootstrap 阶段时才直接处理事件 */
+  private isReady(): boolean {
+    return useMessageStore.getState().isInitialized && !this.isBootstrapping;
+  }
+
+  private applyConversationUpdate(
+    conversationId: string,
+    message: MessageResponse,
+    createdAt: string
+  ): void {
+    const store = useMessageStore.getState();
+    const conversation = store.getConversation(conversationId);
+    if (!conversation) return;
+
+    const updatedConv = {
+      ...conversation,
+      last_seq: message.seq,
+      last_message: message,
+      last_message_at: createdAt,
+      updated_at: createdAt,
+    };
+    store.updateConversation(updatedConv);
+  }
+
+  private incrementUnreadCount(conversationId: string): void {
+    useMessageStore.getState().incrementUnreadAtomic(conversationId);
+  }
+
+  private markMessageAsRecalled(conversationId: string, messageId: string): void {
+    const patches = new Map<string, Partial<MessageResponse>>();
+    patches.set(String(messageId), {
+      status: 'recalled' as MessageResponse['status'],
+      segments: [],
+    });
+    useMessageStore.getState().patchMessages(conversationId, patches);
+  }
+
+  private syncConversationLastMessageOnRecall(conversationId: string, messageId: string): void {
+    const store = useMessageStore.getState();
+    const conversation = store.getConversation(conversationId);
+    if (!conversation?.last_message) return;
+    if (String(conversation.last_message.id) !== String(messageId)) return;
+    if (conversation.last_message.status === 'recalled') return;
+
+    const updatedConversation = {
+      ...conversation,
+      last_message: { ...conversation.last_message, status: 'recalled' as const },
+    };
+    store.updateConversation(updatedConversation);
+  }
+
+  private findConversationIdByGroupId(groupId: string): string | null {
+    const conversations = useMessageStore.getState().getConversations();
+    for (const conv of conversations) {
+      if (String(conv.group?.id || '') === groupId) {
+        return conv.id;
+      }
+    }
+    return null;
+  }
+
+  private buildGroupNoticeText(noticeType: GroupNoticeType, data: any): string {
+    const username = data?.username || '用户';
+    switch (noticeType) {
+      case 'member_join':
+        return `"${username}" 加入了群聊`;
+      case 'member_leave':
+        return `"${username}" 退出了群聊`;
+      case 'member_removed':
+        return `"${username}" 被移出群聊`;
+      case 'muted':
+        return `"${username}" 已被管理员禁言`;
+      case 'unmuted':
+        return `"${username}" 已被管理员解除禁言`;
+      default:
+        return '';
+    }
+  }
+}
--- a/src/stores/message/services/UserCacheService.ts
+++ b/src/stores/message/services/UserCacheService.ts
@@ -120,7 +120,4 @@ export class UserCacheService implements IUserCacheService {
  reset(): void {
    this.pendingUserRequests.clear();
  }
-}
-
-// 单例导出
-export const userCacheService = new UserCacheService();
+}
--- a/src/stores/message/services/WSMessageHandler.ts
+++ b/src/stores/message/services/WSMessageHandler.ts
@@ -1,597 +0,0 @@
-/**
- * WebSocket 消息处理器
- * 处理所有来自 WebSocket 的消息事件
- */
-
-import type {
-  WSChatMessage,
-  WSGroupChatMessage,
-  WSReadMessage,
-  WSGroupReadMessage,
-  WSRecallMessage,
-  WSGroupRecallMessage,
-  WSGroupTypingMessage,
-  WSGroupNoticeMessage,
-  WSNotificationMessage,
-} from '@/services/core';
-import { wsService, GroupNoticeType } from '@/services/core';
-import { eventBus } from '@/core/events/EventBus';
-import { messageRepository } from '@/database';
-import type { MessageResponse, ConversationResponse } from '../../../types/dto';
-import type {
-  IWSMessageHandler,
-  IMessageDeduplication,
-  IUserCacheService,
-  HandleNewMessageOptions,
-} from '../types';
-import { MAX_FLUSH_ITERATIONS, MIN_RECONNECT_SYNC_INTERVAL } from '../constants';
-import { useMessageStore, normalizeConversationId } from '../store';
-
-export class WSMessageHandler implements IWSMessageHandler {
-  private deduplication: IMessageDeduplication;
-  private userCacheService: IUserCacheService;
-  private getCurrentUserId: () => string | null;
-  private markAsReadCallback: (conversationId: string, seq: number) => Promise<void>;
-  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;
-
-  // 同步触发器状态：防止 onConnect 和 sync_required 并发
-  private syncInProgress: boolean = false;
-  private lastSyncTriggerAt: number = 0;
-
-  // 系统通知去重（防止重连时重复递增系统未读）
-  private processedNotificationIds: Set<string> = new Set();
-
-  // 初始化阶段缓冲来自 SSE 的消息事件
-  private bufferedChatMessages: Array<(WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean }> = [];
-  private bufferedReadReceipts: Array<WSReadMessage | WSGroupReadMessage> = [];
-
-  constructor(
-    deduplication: IMessageDeduplication,
-    userCacheService: IUserCacheService,
-    getCurrentUserId: () => string | null,
-    callbacks: {
-      markAsRead: (conversationId: string, seq: number) => Promise<void>;
-      fetchConversations: (forceRefresh: boolean, source: string) => Promise<void>;
-      fetchUnreadCount: () => Promise<void>;
-      fetchMessages: (conversationId: string) => Promise<void>;
-      syncBySeq: () => Promise<boolean>;
-    }
-  ) {
-    this.deduplication = deduplication;
-    this.userCacheService = userCacheService;
-    this.getCurrentUserId = getCurrentUserId;
-    this.markAsReadCallback = callbacks.markAsRead;
-    this.fetchConversationsCallback = callbacks.fetchConversations;
-    this.fetchUnreadCountCallback = callbacks.fetchUnreadCount;
-    this.fetchMessagesCallback = callbacks.fetchMessages;
-    this.syncBySeqCallback = callbacks.syncBySeq;
-  }
-
-  /**
-   * 初始化 SSE 监听器
-   */
-  connect(): void {
-    if (this.sseUnsubscribe) {
-      this.sseUnsubscribe();
-    }
-
-    wsService.on('chat', (message: WSChatMessage) => {
-      if (!useMessageStore.getState().isInitialized || this.isBootstrapping) {
-        this.bufferedChatMessages.push(message);
-        return;
-      }
-      this.handleNewMessage(message);
-    });
-
-    wsService.on('group_message', (message: WSGroupChatMessage) => {
-      if (!useMessageStore.getState().isInitialized || this.isBootstrapping) {
-        this.bufferedChatMessages.push(message);
-        return;
-      }
-      this.handleNewMessage(message);
-    });
-
-    wsService.on('read', (message: WSReadMessage) => {
-      if (!useMessageStore.getState().isInitialized || this.isBootstrapping) {
-        this.bufferedReadReceipts.push(message);
-        return;
-      }
-      this.handleReadReceipt(message);
-    });
-
-    // 监听群聊已读回执
-    wsService.on('group_read', (message: WSGroupReadMessage) => {
-      if (!useMessageStore.getState().isInitialized || this.isBootstrapping) {
-        this.bufferedReadReceipts.push(message);
-        return;
-      }
-      this.handleGroupReadReceipt(message);
-    });
-
-    // 监听私聊消息撤回
-    wsService.on('recall', (message: WSRecallMessage) => {
-      this.handleRecallMessage(message);
-    });
-
-    // 监听群聊消息撤回
-    wsService.on('group_recall', (message: WSGroupRecallMessage) => {
-      this.handleGroupRecallMessage(message);
-    });
-
-    // 监听群聊输入状态
-    wsService.on('group_typing', (message: WSGroupTypingMessage) => {
-      this.handleGroupTyping(message);
-    });
-
-    // 监听群通知
-    wsService.on('group_notice', (message: WSGroupNoticeMessage) => {
-      this.handleGroupNotice(message);
-    });
-
-    // 监听系统通知，实时更新系统未读数
-    wsService.on('notification', (message: WSNotificationMessage) => {
-      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();
-        const currentLast = store.lastSystemMessageAt;
-        const msgDate = new Date(message.created_at);
-        if (Number.isNaN(msgDate.getTime())) return;
-        if (!currentLast || msgDate > new Date(currentLast)) {
-          store.setLastSystemMessageAt(message.created_at);
-        }
-      }
-    });
-
-    // 监听连接状态
-    wsService.onConnect(() => {
-      useMessageStore.getState().setSSEConnected(true);
-      this.triggerSync('sse-reconnect');
-    });
-
-    // 监听 sync_required 事件，触发增量同步
-    wsService.on('sync_required', () => {
-      console.log('[WSMessageHandler] 收到 sync_required，开始增量同步');
-      this.triggerSync('sync_required');
-    });
-
-    wsService.onDisconnect(() => {
-      useMessageStore.getState().setSSEConnected(false);
-    });
-  }
-
-  /**
-   * 断开 SSE 连接
-   */
-  disconnect(): void {
-    if (this.sseUnsubscribe) {
-      this.sseUnsubscribe();
-      this.sseUnsubscribe = null;
-    }
-    this.processedNotificationIds.clear();
-  }
-
-  /**
-   * 检查是否已连接
-   */
-  isConnected(): boolean {
-    return useMessageStore.getState().isConnected();
-  }
-
-  /**
-   * 设置启动状态
-   */
-  setBootstrapping(value: boolean): void {
-    this.isBootstrapping = value;
-  }
-
-  /**
-   * 统一同步入口 — 节流 + 去重
-   * 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;
-
-    this.syncInProgress = true;
-    this.lastSyncTriggerAt = now;
-
-    try {
-      const ok = await this.syncBySeqCallback();
-      if (!ok) {
-        await this.fetchConversationsCallback(true, source);
-        const activeConv = useMessageStore.getState().getActiveConversation();
-        if (activeConv) {
-          await this.fetchMessagesCallback(activeConv).catch(() => {});
-        }
-      }
-      await this.fetchUnreadCountCallback().catch(() => {});
-    } catch {
-      await this.fetchConversationsCallback(true, source).catch(() => {});
-      const activeConv = useMessageStore.getState().getActiveConversation();
-      if (activeConv) {
-        await this.fetchMessagesCallback(activeConv).catch(() => {});
-      }
-      await this.fetchUnreadCountCallback().catch(() => {});
-    } finally {
-      this.syncInProgress = false;
-    }
-  }
-
-  /**
-   * 初始化完成后，处理缓冲的 SSE 事件
-   * 每次循环：取走当前 buffer 后立即清空（避免在 fetch 期间新事件再被覆盖）
-   * 退出条件：连续一轮没有新事件，确保 flush 期间到达的事件也被消费
-   */
-  async flushBufferedSSEEvents(): Promise<void> {
-    let lastChatLen = -1;
-    let lastReadLen = -1;
-    for (let i = 0; i < MAX_FLUSH_ITERATIONS; i++) {
-      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 = [];
-
-      // 处理消息
-      for (const m of chatEvents) {
-        await this.handleNewMessage(m, {
-          suppressUnreadIncrement: true,
-          suppressVibration: true,
-          suppressConversationUpdates: true,
-          suppressAutoMarkAsRead: true,
-        });
-      }
-
-      // 处理已读回执
-      for (const r of readEvents) {
-        if (r.type === 'read') {
-          this.handleReadReceipt(r);
-        } else {
-          this.handleGroupReadReceipt(r as WSGroupReadMessage);
-        }
-      }
-
-      // 使用服务器权威刷新
-      await this.fetchConversationsCallback(true, 'sse-buffer-flush');
-      await this.fetchUnreadCountCallback();
-    }
-  }
-
-  /**
-   * 处理新消息（SSE推送）
-   */
-  async handleNewMessage(
-    message: (WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean },
-    options?: HandleNewMessageOptions
-  ): Promise<void> {
-    const store = useMessageStore.getState();
-    const { conversation_id, id, sender_id, seq, segments, created_at, _isAck } = message;
-    const normalizedConversationId = normalizeConversationId(conversation_id);
-    const currentUserId = this.getCurrentUserId();
-    const suppressUnreadIncrement = options?.suppressUnreadIncrement ?? false;
-    const suppressVibration = options?.suppressVibration ?? false;
-    const suppressConversationUpdates = options?.suppressConversationUpdates ?? false;
-    const suppressAutoMarkAsRead = options?.suppressAutoMarkAsRead ?? false;
-
-    // 消息去重检查
-    if (this.deduplication.isMessageProcessed(id)) {
-      return;
-    }
-    this.deduplication.markMessageAsProcessed(id);
-
-    // 发送 ACK 确认
-    wsService.sendAck(id);
-
-    // 对于群聊消息，获取发送者信息
-    if (message.type === 'group_message' && sender_id && sender_id !== currentUserId && sender_id !== '10000') {
-      this.userCacheService.getSenderInfo(sender_id).then(user => {
-        if (user) {
-          const patches = new Map<string, Partial<MessageResponse>>();
-          patches.set(String(id), { sender: user });
-          useMessageStore.getState().patchMessages(normalizedConversationId, patches);
-        }
-      }).catch(error => {
-        console.error('[WSMessageHandler] 获取发送者信息失败:', { userId: sender_id, error });
-      });
-    }
-
-    // 构造消息对象
-    const newMessage: MessageResponse = {
-      id,
-      conversation_id: normalizedConversationId,
-      sender_id,
-      seq: typeof seq === 'string' ? parseInt(seq, 10) : (seq ?? 0),
-      segments: segments || [],
-      created_at,
-      status: 'normal',
-    };
-
-    // 立即更新内存中的消息列表 — 使用原子性 mergeMessages 防止并发写入丢失
-    const messageExists = useMessageStore.getState().getMessages(normalizedConversationId).some(m => String(m.id) === String(id));
-
-    if (!messageExists) {
-      useMessageStore.getState().mergeMessages(normalizedConversationId, [newMessage]);
-    }
-
-    // 更新会话信息
-    if (!suppressConversationUpdates) {
-      this.updateConversationWithNewMessage(normalizedConversationId, newMessage, created_at);
-    }
-
-    // 更新未读数
-    const isCurrentUserMessage = sender_id === currentUserId;
-    const isActiveConversation = normalizedConversationId === store.getActiveConversation();
-
-    const shouldIncrementUnread =
-      !suppressUnreadIncrement &&
-      !isCurrentUserMessage &&
-      !isActiveConversation &&
-      !!currentUserId &&
-      !_isAck;
-
-    if (shouldIncrementUnread) {
-      const isNotificationMuted = store.isNotificationMuted(normalizedConversationId);
-      if (!suppressVibration && !isNotificationMuted) {
-        const vibrationType = message.type === 'group_message' ? 'group_message' : 'message';
-        eventBus.emit({ type: 'VIBRATE', payload: { type: vibrationType } });
-      }
-      this.incrementUnreadCount(normalizedConversationId);
-    }
-
-    // 如果是当前活动会话，自动标记已读
-    if (isActiveConversation && !isCurrentUserMessage && !suppressAutoMarkAsRead) {
-      this.markAsReadCallback(normalizedConversationId, seq).catch(() => {});
-    }
-
-    // 异步保存到本地数据库
-    const textContent = segments?.filter((s: any) => s.type === 'text').map((s: any) => s.data?.text || '').join('') || '';
-    messageRepository.saveMessage({
-      id,
-      conversationId: normalizedConversationId,
-      senderId: sender_id || '',
-      content: textContent,
-      type: segments?.find((s: any) => s.type === 'image') ? 'image' : 'text',
-      isRead: isActiveConversation,
-      createdAt: (() => { const d = new Date(created_at); return Number.isNaN(d.getTime()) ? new Date().toISOString() : d.toISOString(); })(),
-      seq,
-      status: 'normal',
-      segments,
-    }).catch(error => {
-      console.error('[WSMessageHandler] 保存消息到本地失败:', error);
-    });
-  }
-
-  /**
-   * 处理私聊已读回执
-   */
-  handleReadReceipt(message: WSReadMessage): void {
-    // 可以在这里处理对方已读的状态更新
-  }
-
-  /**
-   * 处理群聊已读回执
-   */
-  handleGroupReadReceipt(message: WSGroupReadMessage): void {
-    // 群聊已读回执处理
-  }
-
-  /**
-   * 处理私聊消息撤回
-   */
-  handleRecallMessage(message: WSRecallMessage): void {
-    const { conversation_id, message_id } = message;
-    const normalizedConversationId = normalizeConversationId(conversation_id);
-
-    this.markMessageAsRecalled(normalizedConversationId, message_id);
-    this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id);
-
-    messageRepository.updateStatus(message_id, 'recalled', true).catch(error => {
-      console.error('[WSMessageHandler] 更新本地消息撤回状态失败:', error);
-    });
-  }
-
-  /**
-   * 处理群聊消息撤回
-   */
-  handleGroupRecallMessage(message: WSGroupRecallMessage): void {
-    const { conversation_id, message_id } = message;
-    const normalizedConversationId = normalizeConversationId(conversation_id);
-
-    this.markMessageAsRecalled(normalizedConversationId, message_id);
-    this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id);
-
-    messageRepository.updateStatus(message_id, 'recalled', true).catch(error => {
-      console.error('[WSMessageHandler] 更新本地消息撤回状态失败:', error);
-    });
-  }
-
-  /**
-   * 处理群聊输入状态
-   */
-  handleGroupTyping(message: WSGroupTypingMessage): void {
-    const store = useMessageStore.getState();
-    const { group_id, user_id, is_typing } = message;
-    const groupIdStr = String(group_id);
-
-    const currentTypingUsers = store.getTypingUsers(groupIdStr);
-    let updatedTypingUsers: string[];
-
-    if (is_typing) {
-      if (!currentTypingUsers.includes(user_id)) {
-        updatedTypingUsers = [...currentTypingUsers, user_id];
-      } else {
-        updatedTypingUsers = currentTypingUsers;
-      }
-    } else {
-      updatedTypingUsers = currentTypingUsers.filter(id => id !== user_id);
-    }
-
-    if (updatedTypingUsers.length !== currentTypingUsers.length) {
-      store.setTypingUsers(groupIdStr, updatedTypingUsers);
-    }
-  }
-
-  /**
-   * 处理群通知
-   */
-  handleGroupNotice(message: WSGroupNoticeMessage): void {
-    const store = useMessageStore.getState();
-    const { notice_type, group_id, data, timestamp, message_id, seq } = message;
-    const groupIdStr = String(group_id);
-
-    // 处理禁言/解除禁言通知
-    if (notice_type === 'muted' || notice_type === 'unmuted') {
-      const mutedUserId = data?.user_id;
-      const currentUserId = this.getCurrentUserId();
-
-      if (mutedUserId && mutedUserId === currentUserId) {
-        const isMuted = notice_type === 'muted';
-        store.setMutedStatus(groupIdStr, isMuted);
-      }
-    }
-
-    // 补一条系统消息到当前会话消息流（原子性 merge）
-    if (message_id && typeof seq === 'number' && seq > 0) {
-      const conversationId = this.findConversationIdByGroupId(groupIdStr);
-      if (conversationId) {
-        const messageExists = store.getMessages(conversationId).some(m => String(m.id) === String(message_id));
-        if (!messageExists) {
-          const noticeText = this.buildGroupNoticeText(notice_type, data);
-          const systemNoticeMessage: MessageResponse = {
-            id: String(message_id),
-            conversation_id: conversationId,
-            sender_id: '10000',
-            seq,
-            segments: noticeText
-              ? [{ type: 'text', data: { text: noticeText } as any }]
-              : [],
-            status: 'normal',
-            category: 'notification',
-            created_at: (() => { const d = new Date(timestamp || Date.now()); return Number.isNaN(d.getTime()) ? new Date().toISOString() : d.toISOString(); })(),
-          };
-
-          useMessageStore.getState().mergeMessages(conversationId, [systemNoticeMessage]);
-        }
-      }
-    }
-  }
-
-  // ==================== 私有工具方法 ====================
-
-  private updateConversationWithNewMessage(
-    conversationId: string,
-    message: MessageResponse,
-    createdAt: string
-  ): void {
-    const store = useMessageStore.getState();
-    const conversation = store.getConversation(conversationId);
-
-    if (conversation) {
-      const updatedConv: ConversationResponse = {
-        ...conversation,
-        last_seq: message.seq,
-        last_message: message,
-        last_message_at: createdAt,
-        updated_at: createdAt,
-      };
-      store.updateConversation(updatedConv);
-    }
-  }
-
-  private incrementUnreadCount(conversationId: string): void {
-    useMessageStore.getState().incrementUnreadAtomic(conversationId);
-  }
-
-  private incrementSystemUnread(): void {
-    // 原子递增：避免外部 read-modify-write 造成丢计数
-    useMessageStore.getState().incrementSystemUnreadAtomic();
-  }
-
-  private markMessageAsRecalled(conversationId: string, messageId: string): void {
-    const patches = new Map<string, Partial<MessageResponse>>();
-    patches.set(String(messageId), { status: 'recalled' as MessageResponse['status'], segments: [] });
-    useMessageStore.getState().patchMessages(conversationId, patches);
-  }
-
-  private syncConversationLastMessageOnRecall(conversationId: string, messageId: string): void {
-    const store = useMessageStore.getState();
-    const conversation = store.getConversation(conversationId);
-    if (!conversation?.last_message) return;
-    if (String(conversation.last_message.id) !== String(messageId)) return;
-    if (conversation.last_message.status === 'recalled') return;
-
-    const updatedConversation: ConversationResponse = {
-      ...conversation,
-      last_message: {
-        ...conversation.last_message,
-        status: 'recalled',
-      },
-    };
-    store.updateConversation(updatedConversation);
-  }
-
-  private findConversationIdByGroupId(groupId: string): string | null {
-    const store = useMessageStore.getState();
-    const conversations = store.getConversations();
-    for (const conv of conversations) {
-      if (String(conv.group?.id || '') === groupId) {
-        return conv.id;
-      }
-    }
-    return null;
-  }
-
-  private buildGroupNoticeText(noticeType: GroupNoticeType, data: any): string {
-    const username = data?.username || '用户';
-    switch (noticeType) {
-      case 'member_join':
-        return `"${username}" 加入了群聊`;
-      case 'member_leave':
-        return `"${username}" 退出了群聊`;
-      case 'member_removed':
-        return `"${username}" 被移出群聊`;
-      case 'muted':
-        return `"${username}" 已被管理员禁言`;
-      case 'unmuted':
-        return `"${username}" 已被管理员解除禁言`;
-      default:
-        return '';
-    }
-  }
-}
--- a/src/stores/message/services/index.ts
+++ b/src/stores/message/services/index.ts
@@ -2,11 +2,15 @@
 * 消息服务层统一导出
 */

-// 消息去重服务
-export { MessageDeduplication, messageDeduplication } from './MessageDeduplication';
+// 实时消息入口（幂等 ingest + 监听器生命周期）
+export { RealtimeIngestionPipeline } from './RealtimeIngestionPipeline';
+export type {
+  RealtimeIngestionPipelineDeps,
+  IngestOptions,
+} from './RealtimeIngestionPipeline';

 // 用户缓存服务
-export { UserCacheService, userCacheService } from './UserCacheService';
+export { UserCacheService } from './UserCacheService';

 // 已读回执管理器
 export { ReadReceiptManager } from './ReadReceiptManager';
@@ -19,6 +23,3 @@ export { MessageSendService } from './MessageSendService';

 // 消息同步服务
 export { MessageSyncService } from './MessageSyncService';
-
-// WebSocket 消息处理器
-export { WSMessageHandler } from './WSMessageHandler';
--- a/src/stores/message/store.ts
+++ b/src/stores/message/store.ts
@@ -85,6 +85,16 @@ export interface MessageActions {
  removeConversation: (conversationId: string) => void;
  setMessages: (conversationId: string, messages: MessageResponse[]) => void;
  mergeMessages: (conversationId: string, incoming: MessageResponse[]) => void;
+  /**
+   * 幂等写入单条消息：以 message id 为幂等键，在 zustand set 原子边界内去重。
+   * 返回 true 表示这条消息是新写入的（store 之前没有同 id 的消息），
+   * 返回 false 表示已存在（被静默去重，副作用调用方应据此跳过未读/通知等副作用）。
+   *
+   * 这是消息幂等性的唯一真理源：任何来源（WS 推送、REST 拉取、发送回显、
+   * loadMore）都应通过 addOrReplaceMessage/mergeMessages 走 store，避免上层
+   * 维护独立去重表带来的「标记早于落库」竞态。
+   */
+  addOrReplaceMessage: (conversationId: string, message: MessageResponse) => boolean;
  removeMessage: (conversationId: string, messageId: string) => void;
  patchMessages: (conversationId: string, patches: Map<string, Partial<MessageResponse>>) => void;
  setUnreadCount: (total: number, system: number) => void;
@@ -92,6 +102,15 @@ export interface MessageActions {
   * 原子递增系统未读数：避免外部 RMW 造成丢失
   */
  incrementSystemUnreadAtomic: () => void;
+  /**
+   * 原子设置系统未读数：保持 totalUnreadCount 一致性（total = sum(各会话未读) + system），
+   * 在 zustand set 回调内一次性完成读-改-写，避免外部 RMW 丢更新。
+   */
+  setSystemUnreadAtomic: (count: number) => void;
+  /**
+   * 原子递减系统未读数（不低于 0）：避免外部 RMW 造成丢失。
+   */
+  decrementSystemUnreadAtomic: (count?: number) => void;
  setLastSystemMessageAt: (time: string | null) => void;
  setSSEConnected: (connected: boolean) => void;
  setCurrentConversation: (conversationId: string | null) => void;
@@ -364,6 +383,31 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
    });
  },

+  /**
+   * 幂等写入单条消息，返回是否为新写入。
+   * 去重发生在 zustand set 的原子边界内，调用方据此决定是否触发未读/通知等副作用。
+   */
+  addOrReplaceMessage: (conversationId: string, message: MessageResponse): boolean => {
+    const normalizedId = normalizeConversationId(conversationId);
+    let inserted = false;
+    set(state => {
+      const existing = state.messagesMap.get(normalizedId);
+      const targetId = String(message.id);
+      // 已存在同 id：静默去重（幂等），不视为新增
+      if (existing && existing.some(m => String(m.id) === targetId)) {
+        inserted = false;
+        return state;
+      }
+      inserted = true;
+      const base = existing || [];
+      const merged = mergeMessagesById(base, [message]);
+      const newMessagesMap = new Map(state.messagesMap);
+      newMessagesMap.set(normalizedId, merged);
+      return { messagesMap: newMessagesMap };
+    });
+    return inserted;
+  },
+
  removeMessage: (conversationId: string, messageId: string) => {
    const normalizedId = normalizeConversationId(conversationId);
    set(state => {
@@ -414,6 +458,27 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
    }));
  },

+  setSystemUnreadAtomic: (count: number) => {
+    set(state => {
+      const delta = count - state.systemUnreadCount;
+      return {
+        systemUnreadCount: count,
+        totalUnreadCount: Math.max(0, state.totalUnreadCount + delta),
+      };
+    });
+  },
+
+  decrementSystemUnreadAtomic: (count = 1) => {
+    set(state => {
+      const newSystemCount = Math.max(0, state.systemUnreadCount - count);
+      const actualDecrement = state.systemUnreadCount - newSystemCount;
+      return {
+        systemUnreadCount: newSystemCount,
+        totalUnreadCount: Math.max(0, state.totalUnreadCount - actualDecrement),
+      };
+    });
+  },
+
  setLastSystemMessageAt: (time: string | null) => {
    set({ lastSystemMessageAt: time });
    if (time) {
--- a/src/stores/message/types.ts
+++ b/src/stores/message/types.ts
@@ -22,12 +22,6 @@ export interface HandleNewMessageOptions {
  suppressAutoMarkAsRead?: boolean;
 }

-export interface IMessageDeduplication {
-  isMessageProcessed(messageId: string): boolean;
-  markMessageAsProcessed(messageId: string): void;
-  cleanupProcessedMessageIds(): void;
-}
-
 export interface IUserCacheService {
  getSenderInfo(userId: string): Promise<UserDTO | null>;
  enrichMessagesWithSenderInfo(
@@ -40,11 +34,6 @@ export interface IUserCacheService {
 export interface IReadReceiptManager {
  markAsRead(conversationId: string, seq: number): Promise<void>;
  markAllAsRead(): Promise<void>;
-  beginReadOperation(conversationId: string, seq: number): number;
-  endReadOperation(conversationId: string, version: number, success: boolean): void;
-  isReadOperationPending(conversationId: string): boolean;
-  getReadStateVersion(): number;
-  getPendingReadMap(): Map<string, ReadStateRecord>;
 }

 export interface IConversationOperations {
@@ -77,9 +66,3 @@ export interface IMessageSyncService {
  canLoadMoreConversations(): boolean;
  restartSources(): void;
 }
-
-export interface IWSMessageHandler {
-  connect(): void;
-  disconnect(): void;
-  isConnected(): boolean;
-}
--- a/src/stores/user/UserManager.ts
+++ b/src/stores/user/UserManager.ts
@@ -14,6 +14,14 @@ import { createDedupe } from '../utils/requestDedupe';

 const dedupe = createDedupe(() => useUserManagerStore);

+// 将 fetchCurrentUserFromAPI 的判别结果转为 User | null。
+// 网络故障/认证失败时返回 null，由上层缓存兜底（避免空数据覆盖缓存）。
+// 注意：UserManager 只负责缓存，不触发登出，登出由 authStore 统一处理。
+async function fetchCurrentUserOrNull(): Promise<UserDTO | null> {
+  const result = await authService.fetchCurrentUserFromAPI();
+  return result.kind === 'user' ? result.user : null;
+}
+
 // ==================== UserManager 类 ====================

 class UserManager {
@@ -41,11 +49,11 @@ class UserManager {
    }

    return dedupe('users:current', async () => {
-      const user = await authService.fetchCurrentUserFromAPI();
-      const newEntry = createCacheEntry(user, DEFAULT_TTL.USER);
-      store.setCurrentUserEntry(newEntry);
-
+      const user = await fetchCurrentUserOrNull();
+      // 仅在拿到真实用户时刷新缓存；网络/认证失败时保留旧缓存，避免误清空
      if (user) {
+        const newEntry = createCacheEntry(user, DEFAULT_TTL.USER);
+        store.setCurrentUserEntry(newEntry);
        userCacheRepository.saveCurrent(user).catch(() => {});
        userCacheRepository.save(user).catch(() => {});
      }
@@ -56,11 +64,10 @@ class UserManager {
  private refreshCurrentUserInBackground(): void {
    dedupe('users:current:bg', async () => {
      const store = useUserManagerStore.getState();
-      const user = await authService.fetchCurrentUserFromAPI();
-      const newEntry = createCacheEntry(user, DEFAULT_TTL.USER);
-      store.setCurrentUserEntry(newEntry);
-
+      const user = await fetchCurrentUserOrNull();
      if (user) {
+        const newEntry = createCacheEntry(user, DEFAULT_TTL.USER);
+        store.setCurrentUserEntry(newEntry);
        userCacheRepository.saveCurrent(user).catch(() => {});
        userCacheRepository.save(user).catch(() => {});
      }