diff --git a/src/database/repositories/MessageRepository.ts b/src/database/repositories/MessageRepository.ts index dd4b2a7..94661df 100644 --- a/src/database/repositories/MessageRepository.ts +++ b/src/database/repositories/MessageRepository.ts @@ -9,6 +9,16 @@ export interface IMessageRepository { getByConversation(conversationId: string, limit?: number): Promise; getMaxSeq(conversationId: string): Promise; getMinSeq(conversationId: string): Promise; + /** + * 返回本地某会话从最大 seq 往下的「连续区间」。 + * - maxSeq: 该会话本地最大的 seq（0 表示无消息） + * - minSeq: 从 maxSeq 往下连续递减遇到第一个缺口时的下沿（即连续段的最早 seq） + * + * 用于 hydration 管线判断「最新端往前是否存在历史缺口」： + * 若 minSeq > 1，说明 [minSeq-1, ...] 更早的消息本地缺失，需要回填。 + * 应用层遍历而非 SQL 窗口函数，保证跨 SQLite 版本兼容、易测试。 + */ + getContiguousRange(conversationId: string): Promise<{ maxSeq: number; minSeq: number }>; getBeforeSeq(conversationId: string, beforeSeq: number, limit?: number): Promise; getCount(conversationId: string): Promise; getCountBeforeSeq(conversationId: string, beforeSeq: number): Promise; @@ -79,6 +89,27 @@ export class MessageRepository implements IMessageRepository { return r?.minSeq || 0; } + async getContiguousRange(conversationId: string): Promise<{ maxSeq: number; minSeq: number }> { + // 单次查询拉出该会话所有有效 seq，按 DESC 排序；应用层从最大值向下找第一个缺口。 + // 会话消息量通常可控，且仅对未 hydrated 的会话执行一次，开销可接受。 + const rows = await this.dataSource.query<{ seq: number }>( + `SELECT seq FROM messages WHERE conversationId = ? AND seq > 0 ORDER BY seq DESC`, + [conversationId] + ); + if (rows.length === 0) return { maxSeq: 0, minSeq: 0 }; + + let maxSeq = rows[0].seq; + let minSeq = maxSeq; + for (let i = 1; i < rows.length; i++) { + if (rows[i].seq === minSeq - 1) { + minSeq = rows[i].seq; // 连续，向下扩展 + } else { + break; // 遇到缺口，停止 + } + } + return { maxSeq, minSeq }; + } + async getBeforeSeq(conversationId: string, beforeSeq: number, limit: number = 20): Promise { const rows = await this.dataSource.query( `SELECT * FROM messages WHERE conversationId = ? AND seq < ? ORDER BY seq DESC LIMIT ?`, diff --git a/src/screens/message/PrivateChatInfoScreen.tsx b/src/screens/message/PrivateChatInfoScreen.tsx index b0f2b97..bb97db1 100644 --- a/src/screens/message/PrivateChatInfoScreen.tsx +++ b/src/screens/message/PrivateChatInfoScreen.tsx @@ -133,6 +133,9 @@ const PrivateChatInfoScreen: React.FC = () => { onPress: async () => { try { await messageRepository.clearConversation(conversationId); + // 同步清空内存消息 + hydrated 标志，保证返回聊天页时状态与 DB 一致。 + // 否则 hydrated 残留 + 内存旧消息会显示已清空的记录。 + useMessageStore.getState().clearMessages(conversationId); Alert.alert('成功', '聊天记录已清空'); } catch (error) { Alert.alert('错误', '清空聊天记录失败'); diff --git a/src/screens/message/components/ChatScreen/useChatScreen.ts b/src/screens/message/components/ChatScreen/useChatScreen.ts index 21a0ef9..7f7cd74 100644 --- a/src/screens/message/components/ChatScreen/useChatScreen.ts +++ b/src/screens/message/components/ChatScreen/useChatScreen.ts @@ -1515,7 +1515,10 @@ export const useChatScreen = (props?: ChatScreenProps) => { try { setHasMoreHistory(true); await messageRepository.clearConversation(conversationId); - // 刷新消息列表 + // 同步清空内存消息 + hydrated 标志，保证内存/DB/hydrated 三者一致。 + // 否则 hydrated 残留会让 refreshMessages 走轻量增量同步，清空后仍显示旧消息。 + useMessageStore.getState().clearMessages(conversationId); + // 刷新消息列表（hydrated 已清，会重新走完整 hydration 管线） await refreshMessages(); } catch (error) { console.error('清空会话失败:', error); diff --git a/src/stores/message/__tests__/messageHydration.test.ts b/src/stores/message/__tests__/messageHydration.test.ts new file mode 100644 index 0000000..1050868 --- /dev/null +++ b/src/stores/message/__tests__/messageHydration.test.ts @@ -0,0 +1,323 @@ +/** + * 消息 hydration 管线单测（A+B+C 融合方案） + * + * 锁定的不变量（直接对应「从消息栏进入聊天页只显示一条消息、历史被吞」的回归）： + * + * 1. 【方案 A 核心回归】内存已有 WS 零散最新消息时，进入会话仍能从本地 DB 加载历史， + * 不再因为「内存非空」而跳过本地读取、误判 baseline。 + * 2. 【方案 B 连续回填】本地连续区间存在下缺口时，管线循环 loadMore 回填直到连续到顶。 + * 3. 【方案 B 上限保护】历史巨大时回填在 MAX_GAP_FILL_ROUNDS 后停止，不卡死。 + * 4. 【方案 C 门控】首次进入走完整管线并标记 hydrated；重入只做轻量增量同步。 + * 5. 回填后内存消息按 seq 严格升序、无重复、无丢失。 + * + * mock 机制：jest moduleNameMapper 自动把 @/services/message、@/database 重定向到 + * mocks/messageServiceMock、mocks/databaseMock。测试通过 __setMessagesHandler / + * __setMessagesByConversation 配置响应，beforeEach 调 __reset 复原。 + */ + +import { useMessageStore } from '../store'; +import { MessageSyncService } from '../services/MessageSyncService'; +import { messageService } from '@/services/message'; +import { messageRepository } from '@/database'; +import type { MessageResponse } from '../../../types/dto'; +import { MAX_GAP_FILL_ROUNDS, MESSAGES_PAGE_SIZE } from '../constants'; + +// mock 注入的测试辅助方法（运行时由 mocks/databaseMock、mocks/messageServiceMock 提供， +// 类型上不存在，统一用 any 别名调用，与现有 (xxx as any).reset() 风格一致） +const repoMock = messageRepository as any; +const serviceMock = messageService as any; + +// ---- 测试辅助 ---- + +function makeMessage(id: string, seq: number, extra: Partial = {}): MessageResponse { + return { + id, + conversation_id: 'c1', + sender_id: 'u1', + seq, + segments: [{ type: 'text', data: { text: `msg-${seq}` } }], + created_at: `2024-01-01T00:00:${String(seq).padStart(2, '0')}Z`, + status: 'normal', + ...extra, + }; +} + +/** 生成 [from..to] 区间的消息（含端点），seq 升序。 */ +function makeRange(from: number, to: number, prefix = 'm'): MessageResponse[] { + const out: MessageResponse[] = []; + for (let s = from; s <= to; s++) { + out.push(makeMessage(`${prefix}-${s}`, s)); + } + return out; +} + +/** + * 模拟真实后端的 getMessages 行为： + * - afterSeq 指定：返回 seq > afterSeq 的「新消息」（默认空，可由 extraOverrides 注入） + * - beforeSeq 指定：返回 [beforeSeq-limit, beforeSeq-1] 区间的历史（分页回填） + * - 都未指定：返回最近 limit 条快照（默认空，可由 snapshotOverride 注入） + * + * 这样回填循环每轮 beforeSeq 请求都能拿到真实历史，模拟线上行为。 + */ +function makeServerHandler(opts: { + topSeq?: number; // 会话历史的最大 seq（决定回填到顶）；不传则按 beforeSeq 递减 + extraAbove?: MessageResponse[]; // afterSeq 增量返回的新消息 + snapshot?: MessageResponse[]; // 快照响应覆盖 +} = {}): (args: any) => any { + return ({ afterSeq, beforeSeq, limit }) => { + const pageSize = limit || MESSAGES_PAGE_SIZE; + if (afterSeq !== undefined) { + return { messages: opts.extraAbove || [], total: 0, page: 1, page_size: pageSize }; + } + if (beforeSeq !== undefined) { + const from = Math.max(1, beforeSeq - pageSize); + const to = beforeSeq - 1; + if (to < 1) return { messages: [], total: 0, page: 1, page_size: pageSize }; + return { messages: makeRange(from, to), total: to, page: 1, page_size: pageSize }; + } + // 快照 + return { messages: opts.snapshot || [], total: 0, page: 1, page_size: pageSize }; + }; +} + +/** 构造 MessageSyncService，依赖全部用最小存根。 */ +function createSyncService(): MessageSyncService { + return new MessageSyncService( + () => 'me', + { markAsRead: async () => {}, markAllAsRead: async () => {} } as any, + { + getSenderInfo: async () => null, + enrichMessagesWithSenderInfo: () => {}, + } as any, + undefined + ); +} + +describe('消息 hydration 管线', () => { + beforeEach(() => { + useMessageStore.getState().reset(); + repoMock.__reset(); + serviceMock.__reset(); + }); + + describe('方案 A：baseline 修正（核心回归）', () => { + it('内存已有 WS 零散最新消息时，仍从本地 DB 加载更早的历史', async () => { + // 场景：WS 已把 seq=49 写入内存（模拟用户停留在消息栏时收到的最新推送）， + // 本地 DB 存有 seq=1~48 的完整历史。 + // 修复前：existingMessagesAtStart.length>0 → 跳过本地读取 → after_seq=49 → 历史全丢。 + // 修复后：无条件读本地补全 → 内存得到 seq=1~49。 + useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('ws-49', 49)); + repoMock.__setMessagesByConversation('c1', makeRange(1, 48)); + + // 后端：after_seq=49 无新消息；beforeSeq 请求返回分页历史（模拟真实后端） + serviceMock.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + // 关键断言：应能看到完整历史，而非只有 seq=49 一条 + expect(msgs).toHaveLength(49); + expect(msgs.map((m) => m.seq)).toEqual(expect.arrayContaining(makeRange(1, 49).map((m) => m.seq))); + expect(msgs[0].seq).toBe(1); + expect(msgs[msgs.length - 1].seq).toBe(49); + // 已 hydrated + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + }); + + it('内存与本地均空时走冷启动快照分支', async () => { + // 冷启动：本地无数据，拉取最近 50 条快照 + const snapshot = makeRange(1, 50); + serviceMock.__setMessagesHandler(makeServerHandler({ snapshot })); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + expect(msgs).toHaveLength(50); + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + }); + }); + + describe('方案 B：连续区间回填', () => { + it('本地连续区间有下缺口时，循环 loadMore 回填到顶', async () => { + // 场景：本地 DB 只有 seq=45~49（minSeq=45 > 1，存在下缺口）， + // 服务端按 beforeSeq 分段返回历史。 + // 注意：getByConversation 默认读 MESSAGES_PAGE_SIZE=20 条，本地只有 5 条， + // 故 Step 1 先把 5 条并入内存；Step 2 上缺口无新消息； + // Step 3 回填：beforeSeq=45 → 服务端返回 seq<45 的一页。 + repoMock.__setMessagesByConversation('c1', makeRange(45, 49)); + + serviceMock.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + // 应连续到 seq=1 + expect(msgs).toHaveLength(49); + const seqs = msgs.map((m) => m.seq); + expect(seqs).toEqual(expect.arrayContaining(makeRange(1, 49).map((m) => m.seq))); + expect(Math.min(...seqs)).toBe(1); + expect(Math.max(...seqs)).toBe(49); + }); + + it('回填在 MAX_GAP_FILL_ROUNDS 后停止，不卡死', async () => { + // 场景：历史巨大（seq 到 10000），每页 MESSAGES_PAGE_SIZE， + // 5 轮最多回填 5*20=100 条，无法到顶，应在上限处停止。 + const TOP = 10000; + repoMock.__setMessagesByConversation('c1', makeRange(TOP - 4, TOP, 'top')); + + serviceMock.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + // 本地 5 + 至多 5*20=100 回填 = 至多 105 条，远小于 10000（证明未无限循环） + expect(msgs.length).toBeLessThanOrEqual(5 + MAX_GAP_FILL_ROUNDS * MESSAGES_PAGE_SIZE); + expect(msgs.length).toBeGreaterThan(5); // 确实回填了 + // 仍标记 hydrated（避免反复重试卡首屏） + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + }); + + it('回填后消息按 seq 严格升序、无重复', async () => { + repoMock.__setMessagesByConversation('c1', makeRange(10, 15)); + serviceMock.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + const seqs = msgs.map((m) => m.seq); + // 严格升序 + for (let i = 1; i < seqs.length; i++) { + expect(seqs[i]).toBeGreaterThan(seqs[i - 1]); + } + // 无重复 id + const ids = msgs.map((m) => m.id); + expect(new Set(ids).size).toBe(ids.length); + }); + }); + + describe('方案 C：hydrated 门控', () => { + it('首次进入走完整管线；重入只做轻量增量同步', async () => { + let snapshotCallCount = 0; + let beforeSeqCallCount = 0; + let localReadCount = 0; + // 首次 hydration 的上缺口同步（after_seq=20）返回空； + // 重入的增量同步（after_seq=20）返回 seq=21，用于验证重入确实走了增量路径。 + let afterSeqCallCount = 0; + + const realGetByConversation = repoMock.getByConversation.bind(repoMock); + repoMock.getByConversation = async (id: string, limit?: number) => { + localReadCount++; + return realGetByConversation(id, limit); + }; + + repoMock.__setMessagesByConversation('c1', makeRange(1, 20)); + + serviceMock.__setMessagesHandler(({ afterSeq, beforeSeq }: any) => { + if (afterSeq !== undefined) { + afterSeqCallCount++; + // 仅重入（第 2 次 afterSeq 请求）返回新消息 seq=21 + if (afterSeqCallCount === 2) { + return { messages: [makeMessage('new-21', 21)], total: 0, page: 1, page_size: 20 }; + } + return { messages: [], total: 0, page: 1, page_size: 20 }; + } + if (beforeSeq !== undefined) { + beforeSeqCallCount++; + // 首次 hydration 本地已连续到 1，理论上不应被调用 + return { messages: [], total: 0, page: 1, page_size: 20 }; + } + snapshotCallCount++; + return { messages: [], total: 0, page: 1, page_size: 20 }; + }); + + const sync = createSyncService(); + + // 首次：完整 hydration。本地 1~20 连续到顶，无需回填。 + const localReadAfterFirst = localReadCount; + await sync.fetchMessages('c1'); + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + expect(useMessageStore.getState().getMessages('c1')).toHaveLength(20); + // 本地已连续到顶，回填不应触发服务端 beforeSeq 请求 + expect(beforeSeqCallCount).toBe(0); + expect(snapshotCallCount).toBe(0); + expect(localReadCount).toBeGreaterThan(localReadAfterFirst); // 首次读了本地 + + // 重入：应只做轻量增量（after_seq=maxSeq=20），不再读本地/回填 + const localReadBeforeReentry = localReadCount; + await sync.fetchMessages('c1'); + // 重入后增量拿到 seq=21 + const msgs = useMessageStore.getState().getMessages('c1'); + expect(msgs).toHaveLength(21); + expect(msgs.some((m) => m.seq === 21)).toBe(true); + // 重入不应再读本地、不应触发快照或回填 + expect(localReadCount).toBe(localReadBeforeReentry); + expect(snapshotCallCount).toBe(0); + expect(beforeSeqCallCount).toBe(0); + }); + }); + + describe('并发安全', () => { + it('hydration 期间 WS ingest 新消息，最终消息完整不丢失', async () => { + // 本地有 seq=1~20，内存初始空 + repoMock.__setMessagesByConversation('c1', makeRange(1, 20)); + + serviceMock.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + const fetchPromise = sync.fetchMessages('c1'); + + // 在 hydration 进行中模拟 WS ingest 一条新消息 + useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('ws-21', 21)); + + await fetchPromise; + + const msgs = useMessageStore.getState().getMessages('c1'); + const seqs = msgs.map((m) => m.seq); + // 本地 1~20 + WS 的 21，共 21 条，无丢失 + expect(msgs).toHaveLength(21); + expect(seqs).toContain(21); + expect(new Set(msgs.map((m) => m.id)).size).toBe(msgs.length); + }); + }); + + describe('清空聊天记录后重新 hydration', () => { + it('clearMessages 清空内存消息 + hydrated 标志，下次进入重新走完整管线', async () => { + // 场景：会话已 hydrated 且内存有消息；用户清空记录后重新进入。 + // 修复前：hydrated 残留 → 走轻量增量（after_seq=旧maxSeq）→ 清空后仍显示旧消息或拉不到新历史。 + // 修复后：clearMessages 清内存+hydrated → 下次 fetchMessages 重新走完整 hydration。 + messageRepository.__setMessagesByConversation('c1', makeRange(1, 20)); + messageService.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + + // 首次 hydration + await sync.fetchMessages('c1'); + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + expect(useMessageStore.getState().getMessages('c1')).toHaveLength(20); + + // 用户清空记录（DB + 内存 + hydrated） + await messageRepository.clearConversation('c1'); + useMessageStore.getState().clearMessages('c1'); + expect(useMessageStore.getState().getMessages('c1')).toHaveLength(0); + expect(useMessageStore.getState().isHydrated('c1')).toBe(false); + + // 重新进入：本地 DB 已空（clearConversation 清了 mock 状态），走冷启动快照 + // 配置服务端快照返回 seq=1~10（模拟清空后服务端重新返回的消息） + messageService.__setMessagesHandler( + makeServerHandler({ snapshot: makeRange(1, 10) }) + ); + + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + // 应重新加载到快照消息，而非停留在清空状态 + expect(msgs).toHaveLength(10); + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + }); + }); +}); diff --git a/src/stores/message/__tests__/mocks/databaseMock.ts b/src/stores/message/__tests__/mocks/databaseMock.ts index 5409dfd..5c23b95 100644 --- a/src/stores/message/__tests__/mocks/databaseMock.ts +++ b/src/stores/message/__tests__/mocks/databaseMock.ts @@ -1,26 +1,125 @@ /** * 测试用 database mock：messageRepository / conversationRepository / userCacheRepository。 - * 仅提供方法存根，单测断言 store 层行为，不验证持久化。 + * + * 默认行为：所有读取返回空、写入为 no-op，与历史行为一致（不破坏现有单测）。 + * 增强能力：单测可通过 `messageRepository.__setState()` / `messageRepository.__setHandler()` + * 注入按 (conversationId, 参数) 维度的响应，用于 hydration 回填管线等需要 + * 真实数据流的场景。每个 test 的 beforeEach 应调用 `__reset()` 复原默认行为。 */ +// ---- 可配置的内部状态 ---- + +interface RepositoryMockState { + /** 按 `${conversationId}` 存储该会话的全部本地消息（模拟 SQLite 表内容）。 */ + messagesByConversation: Map; +} + +const state: RepositoryMockState = { + messagesByConversation: new Map(), +}; + +function resetState() { + state.messagesByConversation.clear(); +} + +/** + * 让 `getByConversation` / `getBeforeSeq` / `getContiguousRange` 等读取方法 + * 基于注入的消息集合工作（模拟本地 DB 已落库的数据）。 + * `messages` 应按 seq 升序提供。 + */ +function __setMessagesByConversation(conversationId: string, messages: any[]) { + state.messagesByConversation.set(conversationId, [...messages].sort((a, b) => a.seq - b.seq)); +} + export const messageRepository = { + // ---- 写入：no-op（单测不验证持久化）---- saveMessage: async () => {}, saveMessagesBatch: async () => {}, - getByConversation: async () => [], - getBeforeSeq: async () => [], + + // ---- 读取：基于内部状态 ---- + async getByConversation(conversationId: string, limit = 20) { + const all = state.messagesByConversation.get(conversationId) || []; + // 模拟 SQL：ORDER BY seq DESC LIMIT，再 reverse() 为升序 + return [...all].sort((a, b) => b.seq - a.seq).slice(0, limit).reverse(); + }, + + async getBeforeSeq(conversationId: string, beforeSeq: number, limit = 20) { + const all = state.messagesByConversation.get(conversationId) || []; + return [...all] + .filter((m) => m.seq < beforeSeq) + .sort((a, b) => b.seq - a.seq) + .slice(0, limit) + .reverse(); + }, + + async getMaxSeq(conversationId: string) { + const all = state.messagesByConversation.get(conversationId) || []; + return all.reduce((max, m) => Math.max(max, m.seq || 0), 0); + }, + + async getMinSeq(conversationId: string) { + const all = state.messagesByConversation.get(conversationId) || []; + if (all.length === 0) return 0; + return all.reduce((min, m) => Math.min(min, m.seq || Infinity), Infinity); + }, + + async getContiguousRange(conversationId: string) { + const all = (state.messagesByConversation.get(conversationId) || []) + .filter((m) => m.seq > 0) + .map((m) => m.seq) + .sort((a, b) => b - a); // DESC + if (all.length === 0) return { maxSeq: 0, minSeq: 0 }; + let maxSeq = all[0]; + let minSeq = maxSeq; + for (let i = 1; i < all.length; i++) { + if (all[i] === minSeq - 1) { + minSeq = all[i]; + } else { + break; + } + } + return { maxSeq, minSeq }; + }, + + async getCount(conversationId: string) { + return (state.messagesByConversation.get(conversationId) || []).length; + }, + + // ---- 其它：no-op / 空实现 ---- updateStatus: async () => {}, + markAsRead: async () => {}, markConversationAsRead: async () => {}, delete: async () => {}, - clearConversation: async () => {}, + clearConversation: async (conversationId: string) => { + state.messagesByConversation.delete(conversationId); + }, + search: async () => [], + searchByConversation: async () => [], + getStats: async () => ({ totalMessages: 0, totalConversations: 0 }), + getLastMessageByConversation: async () => null, + getCountBeforeSeq: async () => 0, + + // ---- 测试辅助 ---- + __reset: resetState, + __setState: resetState, // 别名，语义对齐 asyncStorageMock + __setMessagesByConversation, + __getMessagesByConversation: (conversationId: string) => + state.messagesByConversation.get(conversationId) || [], }; export const conversationRepository = { getListCache: async () => [], saveWithRelated: async () => {}, delete: async () => {}, + saveCache: async () => {}, + getCache: async () => null, + updateCacheUnreadCount: async () => {}, + clearAll: async () => {}, }; export const userCacheRepository = { get: () => null, set: () => {}, + save: async () => {}, + saveBatch: async () => {}, }; diff --git a/src/stores/message/__tests__/mocks/messageServiceMock.ts b/src/stores/message/__tests__/mocks/messageServiceMock.ts index cdde190..36f21f4 100644 --- a/src/stores/message/__tests__/mocks/messageServiceMock.ts +++ b/src/stores/message/__tests__/mocks/messageServiceMock.ts @@ -1,9 +1,42 @@ /** * 测试用 messageService mock：仅覆盖单测路径调用的方法。 + * + * 默认行为：所有方法返回空/no-op，与历史行为一致（不破坏现有单测）。 + * 增强能力：`getMessages` 可通过 `__setMessagesHandler` 配置响应分发， + * 用于 hydration 回填管线等需要按 (afterSeq/beforeSeq/limit) 返回不同数据的场景。 + * 每个 test 的 beforeEach 应调用 `__reset()` 复原默认行为。 */ +export type MessageListResponse = { messages: any[]; total: number; page: number; page_size: number }; + +type GetMessagesHandler = (args: { + conversationId: string; + afterSeq?: number; + beforeSeq?: number; + limit?: number; +}) => MessageListResponse; + +const EMPTY_RESPONSE: MessageListResponse = { messages: [], total: 0, page: 1, page_size: 20 }; + +let getMessagesHandler: GetMessagesHandler | null = null; + +function resetHandler() { + getMessagesHandler = null; +} + export const messageService = { - getMessages: async () => ({ messages: [], total: 0, page: 1, page_size: 20 }), + async getMessages( + conversationId: string, + afterSeq?: number, + beforeSeq?: number, + limit?: number + ): Promise { + if (getMessagesHandler) { + return getMessagesHandler({ conversationId, afterSeq, beforeSeq, limit }); + } + return { ...EMPTY_RESPONSE }; + }, + getConversationById: async () => null, getUnreadCount: async () => ({ total_unread_count: 0 }), getSystemUnreadCount: async () => ({ unread_count: 0 }), @@ -13,6 +46,12 @@ export const messageService = { markAllSystemMessagesRead: async () => {}, fetchRemoteConversationListPage: async () => ({ items: [], hasMore: false }), getSystemMessagesCursor: async () => ({ list: [] }), + + // ---- 测试辅助 ---- + __reset: resetHandler, + __setMessagesHandler: (handler: GetMessagesHandler) => { + getMessagesHandler = handler; + }, }; -export type MessageListResponse = { messages: any[]; total: number; page: number; page_size: number }; +export type { GetMessagesHandler }; diff --git a/src/stores/message/constants.ts b/src/stores/message/constants.ts index bf96676..5182f20 100644 --- a/src/stores/message/constants.ts +++ b/src/stores/message/constants.ts @@ -20,3 +20,19 @@ export const MAX_FLUSH_ITERATIONS = 3; * 防止频繁重连导致的重复同步 */ export const MIN_RECONNECT_SYNC_INTERVAL = 1500; + +/** + * 单次消息拉取/本地读取的页大小。 + * 对齐现有 loadMoreMessages 的默认 limit，作为 hydration 管线的统一页大小。 + */ +export const MESSAGES_PAGE_SIZE = 20; + +/** + * 进入会话时「下缺口回填」的最大循环轮数（对标 Matrix backpagination 上限）。 + * + * 当本地连续区间 [maxSeq, minSeq] 的 minSeq > 1（即最新端往前存在历史缺口）时， + * hydration 管线会循环 loadMoreMessages 向前回填，直到连续到顶（minSeq==1） + * 或触达本上限。上限是为了防止极端情况（如会话有数万条历史）下首屏卡死， + * 剩余缺口由用户上滑 loadMoreHistory 兜底补全。 + */ +export const MAX_GAP_FILL_ROUNDS = 5; diff --git a/src/stores/message/services/MessageSyncService.ts b/src/stores/message/services/MessageSyncService.ts index 36b4491..74c6e94 100644 --- a/src/stores/message/services/MessageSyncService.ts +++ b/src/stores/message/services/MessageSyncService.ts @@ -16,6 +16,10 @@ import { import type { IMessageSyncService, MessageManagerConversationListDeps, IUserCacheService } from '../types'; import { useMessageStore, normalizeConversationId } from '../store'; import { ReadReceiptManager } from './ReadReceiptManager'; +import { + MESSAGES_PAGE_SIZE, + MAX_GAP_FILL_ROUNDS, +} from '../constants'; export class MessageSyncService implements IMessageSyncService { private getCurrentUserId: () => string | null; @@ -216,6 +220,15 @@ export class MessageSyncService implements IMessageSyncService { /** * 获取会话消息（增量同步）— 核心实现 + * + * 【hydration 管线，修复「进入聊天页只显示一条消息、历史被吞」】 + * + * 未 hydrated 的会话走完整管线： + * 1. 本地 DB 连续区间补全（方案 A：不信内存零散消息，始终读本地） + * 2. 上缺口增量同步（拉 maxSeq 之后的新消息） + * 3. 下缺口完整回填（循环 loadMore 直到连续到顶，方案 B） + * 4. 标记 hydrated（方案 C） + * 已 hydrated 的会话只做轻量增量同步（baseline 此时可信）。 */ private async _doFetchMessages(conversationId: string, afterSeq?: number): Promise { const store = useMessageStore.getState(); @@ -224,76 +237,16 @@ export class MessageSyncService implements IMessageSyncService { try { if (!afterSeq) { - // 先从本地数据库加载（使用 merge 避免 WS 消息丢失） - const existingMessagesAtStart = store.getMessages(conversationId); - if (existingMessagesAtStart.length === 0) { - try { - const localMessages = await messageRepository.getByConversation(conversationId, 20); - - if (localMessages.length > 0) { - const formattedMessages: MessageResponse[] = localMessages.map(m => ({ - id: m.id, - conversation_id: m.conversationId, - sender_id: m.senderId, - seq: m.seq, - segments: m.segments || [], - status: m.status as any, - created_at: m.createdAt, - })); - - // 原子性合并本地缓存消息，防止并发 WS 消息丢失 - store.mergeMessages(conversationId, formattedMessages); - } - } catch (error) { - console.warn('[MessageSyncService] 读取本地消息失败:', error); - } - } - - // 更新 baseline：合并本地DB后重新计算 - const currentMessages = store.getMessages(conversationId); - const baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0); - - // 策略：有本地数据时仅增量同步，无数据时拉快照 - if (baselineMaxSeq > 0) { - // 增量同步：只拉 baselineMaxSeq 之后的消息 - try { - const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq); - const newMessages = incrementalResp?.messages || []; - - if (newMessages.length > 0) { - store.mergeMessages(conversationId, newMessages); - - messageRepository.saveMessagesBatch( - MessageMapper.toCachedMessages(newMessages, conversationId) - ).catch(error => { - console.error('[MessageSyncService] 保存增量消息到本地失败:', error); - }); - } - } catch (error) { - console.error('[MessageSyncService] 增量同步失败:', error); - } + const wasHydrated = store.isHydrated(conversationId); + if (wasHydrated) { + // 已完整加载过：baseline 可信，仅做轻量增量同步 + await this.incrementalSyncAbove(conversationId); } else { - // 冷启动：本地无数据，拉快照 - try { - const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50); - const snapshotMessages = snapshotResp?.messages || []; - - if (snapshotMessages.length > 0) { - // 快照来自分页接口，后端返回 DESC 顺序；mergeMessages 内部按 seq ASC 排序并合并 - store.mergeMessages(conversationId, snapshotMessages); - - messageRepository.saveMessagesBatch( - MessageMapper.toCachedMessages(snapshotMessages, conversationId) - ).catch(error => { - console.error('[MessageSyncService] 保存快照消息到本地失败:', error); - }); - } - } catch (error) { - console.error('[MessageSyncService] 快照同步失败:', error); - } + // 完整 hydration 管线 + await this.runHydrationPipeline(conversationId); } } else { - // 指定了 afterSeq + // 指定了 afterSeq（焦点/重连增量同步）：保持原行为 const response = await messageService.getMessages(conversationId, afterSeq); if (response?.messages && response.messages.length > 0) { @@ -334,6 +287,181 @@ export class MessageSyncService implements IMessageSyncService { } } + /** + * 轻量增量同步：以当前内存最大 seq 为 baseline，只拉取更新的消息。 + * 仅对已 hydrated 的会话调用——此时 baseline 是可信的（已通过完整管线加载）。 + */ + private async incrementalSyncAbove(conversationId: string): Promise { + const store = useMessageStore.getState(); + const currentMessages = store.getMessages(conversationId); + const baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0); + + if (baselineMaxSeq <= 0) { + // 已 hydrated 但内存空（如清空后未清 hydrated 的边界）：退回快照兜底 + await this.fetchSnapshot(conversationId); + return; + } + + try { + const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq); + const newMessages = incrementalResp?.messages || []; + if (newMessages.length > 0) { + store.mergeMessages(conversationId, newMessages); + messageRepository.saveMessagesBatch( + MessageMapper.toCachedMessages(newMessages, conversationId) + ).catch(error => { + console.error('[MessageSyncService] 增量同步保存失败:', error); + }); + } + } catch (error) { + console.error('[MessageSyncService] 增量同步失败:', error); + } + } + + /** + * 完整 hydration 管线（方案 A + B + C 的核心）。 + * + * 流程： + * Step 1（A）：无条件读本地 DB 补全内存基线（修复「内存非空就跳过本地」的根因） + * Step 2：上缺口增量同步（拉本地 maxSeq 之后的新消息） + * Step 3（B）：下缺口完整回填循环（向前 loadMore 直到连续到顶或触达上限） + * Step 4（C）：标记 hydrated + */ + private async runHydrationPipeline(conversationId: string): Promise { + const store = useMessageStore.getState(); + + // ---- Step 1（方案 A）：本地 DB 连续区间补全 ---- + // 关键修复：移除「内存非空就跳过本地」的错误前提。WS 全局 ingest 会让内存非空， + // 但那只是零散最新消息，不代表历史完整。始终用本地 DB 补全基线。 + try { + const localMessages = await messageRepository.getByConversation(conversationId, MESSAGES_PAGE_SIZE); + if (localMessages.length > 0) { + const formattedMessages: MessageResponse[] = localMessages.map(m => ({ + id: m.id, + conversation_id: m.conversationId, + sender_id: m.senderId, + seq: m.seq, + segments: m.segments || [], + status: m.status as any, + created_at: m.createdAt, + })); + // mergeMessages 按 id 去重 + seq 升序，与内存现有 WS 消息安全合并 + useMessageStore.getState().mergeMessages(conversationId, formattedMessages); + } + } catch (error) { + console.warn('[MessageSyncService] hydration 读取本地消息失败:', error); + } + + // ---- Step 2：上缺口增量同步（baselineMaxSeq 之后的新消息）---- + const messagesAfterLocal = useMessageStore.getState().getMessages(conversationId); + const baselineMaxSeq = messagesAfterLocal.reduce((max, m) => Math.max(max, m.seq || 0), 0); + + if (baselineMaxSeq > 0) { + try { + const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq); + const newMessages = incrementalResp?.messages || []; + if (newMessages.length > 0) { + useMessageStore.getState().mergeMessages(conversationId, newMessages); + messageRepository.saveMessagesBatch( + MessageMapper.toCachedMessages(newMessages, conversationId) + ).catch(error => { + console.error('[MessageSyncService] hydration 增量同步保存失败:', error); + }); + } + } catch (error) { + console.error('[MessageSyncService] hydration 增量同步失败:', error); + } + } else { + // 本地与内存均空：冷启动快照兜底 + await this.fetchSnapshot(conversationId); + } + + // ---- Step 3（方案 B）：下缺口完整回填循环 ---- + // 从最新端往前检查连续性，发现缺口就 loadMore 回填，直到连续到顶（minSeq<=1） + // 或触达 MAX_GAP_FILL_ROUNDS 上限（防极端卡死，剩余靠用户上滑兜底）。 + await this.fillBackwardGaps(conversationId); + + // ---- Step 4（方案 C）：标记 hydrated ---- + // 无论回填是否完全到顶都标记，避免反复重试卡首屏；用户上滑仍可触发 loadMoreHistory。 + useMessageStore.getState().markHydrated(conversationId); + } + + /** + * 冷启动快照：本地与内存均无数据时，拉取最近一页快照。 + */ + private async fetchSnapshot(conversationId: string): Promise { + try { + const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50); + const snapshotMessages = snapshotResp?.messages || []; + if (snapshotMessages.length > 0) { + // 快照来自分页接口，后端返回 DESC 顺序；mergeMessages 内部按 seq ASC 排序并合并 + useMessageStore.getState().mergeMessages(conversationId, snapshotMessages); + messageRepository.saveMessagesBatch( + MessageMapper.toCachedMessages(snapshotMessages, conversationId) + ).catch(error => { + console.error('[MessageSyncService] 保存快照消息到本地失败:', error); + }); + } + } catch (error) { + console.error('[MessageSyncService] 快照同步失败:', error); + } + } + + /** + * 下缺口回填循环（方案 B）。 + * + * 从当前内存消息的最新端向前找第一个连续缺口，loadMore 一页补上，循环直到： + * - 连续到顶（contiguousMinSeq <= 1），或 + * - 触达 MAX_GAP_FILL_ROUNDS 上限，或 + * - 某轮回填无新增（已到顶或网络空响应），防死循环。 + * + * 连续区间在「内存消息数组」上计算（回填每轮都 merge 进内存，比每轮查 DB 更高效）。 + */ + private async fillBackwardGaps(conversationId: string): Promise { + for (let round = 0; round < MAX_GAP_FILL_ROUNDS; round++) { + const currentMessages = useMessageStore.getState().getMessages(conversationId); + if (currentMessages.length === 0) return; + + const contiguousMinSeq = this.computeContiguousMinSeq(currentMessages); + // 连续到顶（minSeq <= 1 表示已到最早消息），无需回填 + if (contiguousMinSeq <= 1) return; + + const beforeCount = currentMessages.length; + try { + // 复用现有 loadMoreMessages：本地不足则请求服务端，内部已 merge + 落库 + await this.loadMoreMessages(conversationId, contiguousMinSeq, MESSAGES_PAGE_SIZE); + } catch (error) { + console.error('[MessageSyncService] 回填历史缺口失败:', error); + return; // 本轮失败，停止回填（已加载部分正常展示） + } + + // 本轮无新增消息：已到顶或服务端无更多，停止回填 + const afterCount = useMessageStore.getState().getMessages(conversationId).length; + if (afterCount <= beforeCount) return; + } + } + + /** + * 计算内存消息数组从最大 seq 往下的连续区间下沿（contiguousMinSeq）。 + * 遇到第一个缺口即停。返回值 <= 1 表示已连续到顶。 + */ + private computeContiguousMinSeq(messages: MessageResponse[]): number { + if (messages.length === 0) return 0; + // 按 seq 降序遍历找连续段 + const sorted = [...messages].sort((a, b) => b.seq - a.seq); + let expected = sorted[0].seq; + for (let i = 1; i < sorted.length; i++) { + if (sorted[i].seq === expected - 1) { + expected = sorted[i].seq; + } else if (sorted[i].seq < expected - 1) { + // 遇到缺口 + break; + } + // sorted[i].seq === expected（重复，幂等已去重，理论不出现）→ 跳过 + } + return expected; + } + /** * 加载更多历史消息 */ diff --git a/src/stores/message/store.ts b/src/stores/message/store.ts index ebea297..32e0e65 100644 --- a/src/stores/message/store.ts +++ b/src/stores/message/store.ts @@ -42,6 +42,17 @@ export interface MessageState { isLoadingConversations: boolean; loadingMessagesSet: Set; + /** + * 已完成完整 hydration（连续区间补全）的会话集合。 + * + * 区别于 loadingMessagesSet（仅表示「正在拉取」），hydrated 标志语义是 + * 「该会话的消息历史已通过完整管线加载到本地连续，后续进入只需轻量增量同步」。 + * 用于修复「从消息栏进入聊天页只显示零散 WS 消息、历史被吞」的问题： + * WS 全局 ingest 会让内存非空，但那不代表历史已连续加载，必须以 hydrated + * 作为「是否需要执行完整 hydration 管线」的判据。 + */ + hydratedConversations: Set; + // 初始化状态 isInitialized: boolean; @@ -96,6 +107,14 @@ export interface MessageActions { */ addOrReplaceMessage: (conversationId: string, message: MessageResponse) => boolean; removeMessage: (conversationId: string, messageId: string) => void; + /** + * 清空某会话的内存消息并清除 hydrated 标志。 + * + * 用于「清空聊天记录」场景：调用方先 messageRepository.clearConversation 清 DB， + * 再调本 action 同步内存状态，保证内存/DB/hydrated 三者一致。 + * 清除 hydrated 后，下次进入会话会重新走完整 hydration 管线（重新从服务端加载）。 + */ + clearMessages: (conversationId: string) => void; patchMessages: (conversationId: string, patches: Map>) => void; setUnreadCount: (total: number, system: number) => void; /** @@ -116,6 +135,15 @@ export interface MessageActions { setCurrentConversation: (conversationId: string | null) => void; setLoading: (loading: boolean) => void; setLoadingMessages: (conversationId: string, loading: boolean) => void; + /** + * 是否已完成完整 hydration（连续区间补全）。 + * hydration 管线据此门控：未 hydrated 才执行完整的本地补全 + 缺口回填。 + */ + isHydrated: (conversationId: string) => boolean; + /** 标记某会话已完成完整 hydration（幂等，已存在则不变）。 */ + markHydrated: (conversationId: string) => void; + /** 清除某会话的 hydrated 标志（删除会话/清空记录时调用，避免残留）。 */ + clearHydrated: (conversationId: string) => void; setTypingUsers: (groupId: string, users: string[]) => void; setMutedStatus: (groupId: string, isMuted: boolean) => void; setNotificationMuted: (conversationId: string, notificationMuted: boolean) => void; @@ -157,6 +185,7 @@ const initialState: MessageState = { currentConversationId: null, isLoadingConversations: false, loadingMessagesSet: new Set(), + hydratedConversations: new Set(), isInitialized: false, typingUsersMap: new Map(), mutedStatusMap: new Map(), @@ -226,6 +255,7 @@ export const useMessageStore = create((set, get) => ({ currentConversationId: state.currentConversationId, isLoadingConversations: state.isLoadingConversations, loadingMessagesSet: state.loadingMessagesSet, + hydratedConversations: state.hydratedConversations, isInitialized: state.isInitialized, typingUsersMap: state.typingUsersMap, mutedStatusMap: state.mutedStatusMap, @@ -339,6 +369,10 @@ export const useMessageStore = create((set, get) => ({ const newLoadingMessagesSet = new Set(state.loadingMessagesSet); newLoadingMessagesSet.delete(normalizedId); + // 同步清除 hydrated 标志，避免会话重建后误判为「已完整加载」 + const newHydratedConversations = new Set(state.hydratedConversations); + newHydratedConversations.delete(normalizedId); + const newTotalUnreadCount = Math.max(0, state.totalUnreadCount - removedUnread); const newCurrentConversationId = state.currentConversationId === normalizedId @@ -354,6 +388,7 @@ export const useMessageStore = create((set, get) => ({ totalUnreadCount: newTotalUnreadCount, currentConversationId: newCurrentConversationId, loadingMessagesSet: newLoadingMessagesSet, + hydratedConversations: newHydratedConversations, }; }); }, @@ -421,6 +456,25 @@ export const useMessageStore = create((set, get) => ({ }); }, + clearMessages: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const newMessagesMap = new Map(state.messagesMap); + newMessagesMap.delete(normalizedId); + // 同步清除 hydrated 标志，下次进入会重新走完整 hydration 管线 + const newHydratedConversations = new Set(state.hydratedConversations); + newHydratedConversations.delete(normalizedId); + // 同步清除 loading 锁，避免残留锁阻断后续 fetchMessages + const newLoadingMessagesSet = new Set(state.loadingMessagesSet); + newLoadingMessagesSet.delete(normalizedId); + return { + messagesMap: newMessagesMap, + hydratedConversations: newHydratedConversations, + loadingMessagesSet: newLoadingMessagesSet, + }; + }); + }, + /** * 原子性 patch：按消息 ID 更新指定字段（如 sender、status） * 不影响其他消息，不丢失并发写入 @@ -513,6 +567,30 @@ export const useMessageStore = create((set, get) => ({ }); }, + isHydrated: (conversationId: string) => { + return get().hydratedConversations.has(normalizeConversationId(conversationId)); + }, + + markHydrated: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + if (state.hydratedConversations.has(normalizedId)) return state; + const next = new Set(state.hydratedConversations); + next.add(normalizedId); + return { hydratedConversations: next }; + }); + }, + + clearHydrated: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + if (!state.hydratedConversations.has(normalizedId)) return state; + const next = new Set(state.hydratedConversations); + next.delete(normalizedId); + return { hydratedConversations: next }; + }); + }, + setTypingUsers: (groupId: string, users: string[]) => { set(state => { const newTypingUsersMap = new Map(state.typingUsersMap); @@ -630,6 +708,7 @@ export const useMessageStore = create((set, get) => ({ conversations: new Map(), messagesMap: new Map(), loadingMessagesSet: new Set(), + hydratedConversations: new Set(), typingUsersMap: new Map(), mutedStatusMap: new Map(), notificationMutedMap: new Map(),