diff --git a/src/core/entities/Message.ts b/src/core/entities/Message.ts index 1fd66e6..0349937 100644 --- a/src/core/entities/Message.ts +++ b/src/core/entities/Message.ts @@ -15,7 +15,7 @@ export interface Message { seq: number; segments: MessageSegment[]; createdAt: string; - status: 'normal' | 'recalled' | 'deleted'; + status: 'normal' | 'pending' | 'failed' | 'recalled' | 'deleted'; category?: string; sender?: { id: string; diff --git a/src/database/repositories/MessageRepository.ts b/src/database/repositories/MessageRepository.ts index 94661df..32ee937 100644 --- a/src/database/repositories/MessageRepository.ts +++ b/src/database/repositories/MessageRepository.ts @@ -27,6 +27,12 @@ export interface IMessageRepository { delete(messageId: string): Promise; updateStatus(messageId: string, status: string, clearContent?: boolean): Promise; clearConversation(conversationId: string): Promise; + /** + * 查询所有待发送(pending)或发送失败(failed)的消息(Outbox 恢复用)。 + * 按 createdAt 升序,保证恢复重试时按发送顺序处理。 + * 用于 App 启动时扫描本地未完成发送的消息,读入 store 并触发重试。 + */ + getPendingOrFailed(): Promise; search(keyword: string): Promise; searchByConversation(conversationId: string, keyword: string, limit?: number, offset?: number): Promise; getStats(): Promise<{ totalMessages: number; totalConversations: number }>; @@ -169,6 +175,18 @@ export class MessageRepository implements IMessageRepository { }); } + async getPendingOrFailed(): Promise { + const rows = await this.dataSource.query( + `SELECT * FROM messages WHERE status IN ('pending', 'failed') ORDER BY createdAt ASC`, + [] + ); + return rows.map(r => ({ + ...r, + isRead: r.isRead === 1, + segments: r.segments ? JSON.parse(r.segments) : undefined, + })); + } + async search(keyword: string): Promise { const rows = await this.dataSource.query( `SELECT * FROM messages WHERE content LIKE ? ORDER BY createdAt DESC LIMIT 50`, [`%${keyword}%`] diff --git a/src/database/schema/MigrationManager.ts b/src/database/schema/MigrationManager.ts index 6a71a5a..7993fbd 100644 --- a/src/database/schema/MigrationManager.ts +++ b/src/database/schema/MigrationManager.ts @@ -23,6 +23,42 @@ import { CREATE_POSTS_CACHE_TABLE, } from './PostSchema'; +/** + * 迁移版本号管理(对标 expo-sqlite 官方推荐用 PRAGMA user_version)。 + * + * 设计: + * - version 0:基线(建表 + 建索引),所有老库都已执行过(靠 CREATE IF NOT EXISTS 幂等) + * - version 1:列级迁移(补 seq/status/segments/lastSeq 列),即原 MIGRATE_* 函数 + * + * 向后兼容:老库 user_version 默认为 0,会重新跑 version 1 的列迁移, + * 但 MIGRATE_* 靠 PRAGMA table_info 列检查幂等,重复执行无副作用。 + * + * 扩展方式(未来破坏性迁移):在 MIGRATIONS 注册表新增 version 2/3/..., + * 每个 step 靠 transaction 包裹保证原子性(失败回滚,不污染 user_version)。 + */ +export const CURRENT_SCHEMA_VERSION = 1; + +/** 读取当前 user_version */ +async function getUserVersion(db: SQLiteDatabase): Promise { + const result = await db.getFirstAsync<{ user_version: number }>('PRAGMA user_version'); + return result?.user_version ?? 0; +} + +/** 设置 user_version */ +async function setUserVersion(db: SQLiteDatabase, version: number): Promise { + await execWithRetry(db, `PRAGMA user_version = ${version};`); +} + +/** 版本化迁移注册表:key = 目标版本,value = 该版本要执行的迁移逻辑 */ +const MIGRATIONS: Map Promise> = new Map([ + // version 1:列级迁移(补 seq/status/segments/lastSeq 列) + // 历史背景:最初表结构不含这些列,靠 MIGRATE_* 增量 ALTER。现在纳入版本管理。 + [1, async (db: SQLiteDatabase) => { + await MIGRATE_MESSAGES_TABLE(db); + await MIGRATE_CONVERSATIONS_TABLE(db); + }], +]); + async function execWithRetry(db: SQLiteDatabase, sql: string, maxRetries = 5): Promise { for (let i = 0; i < maxRetries; i++) { try { @@ -41,9 +77,19 @@ async function execWithRetry(db: SQLiteDatabase, sql: string, maxRetries = 5): P } } +/** + * 执行数据库迁移。 + * + * 流程: + * 1. 设置 busy_timeout + * 2. 基线建表 + 建索引(version 0,CREATE IF NOT EXISTS 幂等,每次都执行以保证新表存在) + * 3. 按 user_version 顺序执行版本化迁移,每步成功后递增 user_version + */ export async function runMigrations(db: SQLiteDatabase): Promise { await execWithRetry(db, 'PRAGMA busy_timeout = 10000;'); - + + // ---- 基线:version 0(建表 + 建索引)---- + // 保留每次执行,确保新库/损坏恢复时表结构就位;对老库靠 IF NOT EXISTS 幂等。 await execWithRetry(db, CREATE_MESSAGES_TABLE); await execWithRetry(db, CREATE_CONVERSATIONS_TABLE); await execWithRetry(db, CREATE_CONVERSATION_CACHE_TABLE); @@ -61,6 +107,32 @@ export async function runMigrations(db: SQLiteDatabase): Promise { await execWithRetry(db, indexSql); } - await MIGRATE_MESSAGES_TABLE(db); - await MIGRATE_CONVERSATIONS_TABLE(db); -} \ No newline at end of file + // ---- 版本化迁移:version 1 ~ CURRENT_SCHEMA_VERSION ---- + const currentVersion = await getUserVersion(db); + + // 当前版本已是最新(含未来更高版本),跳过 + if (currentVersion >= CURRENT_SCHEMA_VERSION) { + return; + } + + // 从 currentVersion+1 开始,逐版本执行迁移 + for (let target = currentVersion + 1; target <= CURRENT_SCHEMA_VERSION; target++) { + const migration = MIGRATIONS.get(target); + if (!migration) { + console.warn(`[MigrationManager] 未找到 version ${target} 的迁移逻辑,跳过`); + continue; + } + + try { + await migration(db); + // 每个版本迁移成功后立即持久化 user_version, + // 即使后续版本失败,已完成的版本不会重复执行 + await setUserVersion(db, target); + } catch (error) { + console.error(`[MigrationManager] 迁移到 version ${target} 失败:`, error); + // 不抛出致命错误以避免阻塞 App 启动;降级运行(表可能缺列,MIGRATE_* 内部会兜底) + // 但记录版本未推进,下次启动会重试该版本 + break; + } + } +} diff --git a/src/services/core/errorHandler.ts b/src/services/core/errorHandler.ts index d622ab0..426be56 100644 --- a/src/services/core/errorHandler.ts +++ b/src/services/core/errorHandler.ts @@ -11,6 +11,7 @@ */ import { showPrompt } from '../ui/promptService'; +import { ApiError, isNetworkError } from './api'; export enum AppErrorCode { NETWORK_ERROR = 'NETWORK_ERROR', @@ -45,6 +46,19 @@ const ERROR_MESSAGES: Record = { function classifyError(error: unknown): AppErrorCode { if (!error) return AppErrorCode.UNKNOWN; + // 优先识别项目自有 ApiError(含 errorCode 字段,比 message 关键词匹配更可靠) + // 修复衔接缺口:api.ts 用 status=503 + errorCode='NETWORK_ERROR' 表示网络错误, + // 但原逻辑只看 status≥500 会误判为 SERVER_ERROR。这里用 isNetworkError 精确识别。 + if (isNetworkError(error)) return AppErrorCode.NETWORK_ERROR; + if (error instanceof ApiError) { + if (error.errorCode === 'AUTH_ERROR' || error.code === 401) return AppErrorCode.AUTH_ERROR; + if (error.errorCode === 'FORBIDDEN' || error.code === 403) return AppErrorCode.FORBIDDEN; + if (error.errorCode === 'NOT_FOUND' || error.code === 404) return AppErrorCode.NOT_FOUND; + if (error.errorCode === 'VERIFICATION_REQUIRED') return AppErrorCode.VALIDATION_ERROR; + if (error.code === 400 || error.code === 422) return AppErrorCode.VALIDATION_ERROR; + if (typeof error.code === 'number' && error.code >= 500) return AppErrorCode.SERVER_ERROR; + } + if (error instanceof Error) { const message = error.message.toLowerCase(); diff --git a/src/services/core/wsService.ts b/src/services/core/wsService.ts index 7aca32e..2494df0 100644 --- a/src/services/core/wsService.ts +++ b/src/services/core/wsService.ts @@ -304,6 +304,11 @@ class WebSocketService { preventDisconnectOnBackground = false; private getWSUrl(token: string | null): string { + // 当前仅携带 token + compress。 + // 待后端支持 event 级 resume 后,可在此拼接 last_event_id 实现断点续传: + // const eventIdParam = this.lastEventId ? `&last_event_id=${encodeURIComponent(this.lastEventId)}` : ''; + // return `${WS_URL}?token=${encodeURIComponent(token || '')}&compress=1${eventIdParam}`; + // 届时重连后服务端从 lastEventId 续传断线期间的事件,无需全量 re-sync。 return `${WS_URL}?token=${encodeURIComponent(token || '')}&compress=1`; } diff --git a/src/stores/message/MessageManager.ts b/src/stores/message/MessageManager.ts index 5b60574..7498b74 100644 --- a/src/stores/message/MessageManager.ts +++ b/src/stores/message/MessageManager.ts @@ -16,9 +16,13 @@ import { ConversationOperations } from './services/ConversationOperations'; import { MessageSendService } from './services/MessageSendService'; import { MessageSyncService } from './services/MessageSyncService'; import { RealtimeIngestionPipeline } from './services/RealtimeIngestionPipeline'; +import { messageRepository } from '@/database'; // 导入 store -import { useMessageStore, normalizeConversationId, loadPersistedLastSystemMessageAt } from './store'; +import { useMessageStore, normalizeConversationId, loadPersistedLastSystemMessageAt, loadPersistedSyncVersion } from './store'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('MessageManager'); // 重新导出类型,保持兼容性 export type { @@ -60,8 +64,18 @@ class MessageManager { getCurrentUserId: () => this.getCurrentUserId(), onActiveConvMessage: (id, seq) => this.markAsRead(id, seq), triggerSync: async (source) => { - // 优先基于 seq 增量同步,失败则全量刷新会话 + 活动会话消息 + 未读 - const ok = await this.syncService.syncBySeq(); + // 三级恢复策略(精确度从高到低): + // 1. syncByVersion:基于持久化游标的精确增量(对标 Matrix resume),最省流量 + // 2. syncBySeq:基于 seq 比对的增量(原有路径) + // 3. 全量刷新:会话列表 + 活动会话消息 + 未读 + const persistedVersion = useMessageStore.getState().syncVersion; + let ok = false; + if (persistedVersion && persistedVersion > 0) { + ok = await this.syncService.syncByVersion(persistedVersion); + } + if (!ok) { + ok = await this.syncService.syncBySeq(); + } if (!ok) { await this.fetchConversations(true, source); const activeConv = useMessageStore.getState().getActiveConversation(); @@ -102,7 +116,7 @@ class MessageManager { if (nextUserId && useMessageStore.getState().currentConversationId) { this.activateConversation(useMessageStore.getState().currentConversationId!, { forceSync: true }).catch(error => { - console.error('[MessageManager] 登录态就绪后重激活会话失败:', error); + errorHandler(error); }); } @@ -120,6 +134,74 @@ class MessageManager { return this.currentUserId; } + /** + * Outbox 恢复:扫描本地未完成发送的消息(pending/failed),读入 store 并自动重试。 + * + * 对标 Matrix Send Queue:发送是离线可靠操作,pending/failed 消息必须持久化, + * App 重启后能恢复未完成的发送。 + * + * 策略: + * - pending:重启后视为「发送可能已到服务端但未收到 ACK」,直接重试一次(幂等由服务端保证, + * 或 WS 回执去重)。重试成功则落库 normal + 删 tempId;仍失败则保持 failed 等用户手动。 + * - failed:重启后保持 failed 显示(用户可见失败气泡),不自动重试(避免无限重试刷屏), + * 等待用户点击重试。 + */ + private async restoreOutboxMessages(): Promise { + try { + const pendingOrFailed = await messageRepository.getPendingOrFailed(); + if (pendingOrFailed.length === 0) return; + + const store = useMessageStore.getState(); + // 1. 全部读入 store(按 conversationId 分组合并,幂等) + const byConv = new Map(); + for (const cached of pendingOrFailed) { + const msg: MessageResponse = { + id: cached.id, + conversation_id: cached.conversationId, + sender_id: cached.senderId, + seq: cached.seq, + segments: cached.segments || [], + created_at: cached.createdAt, + status: cached.status as any, + }; + const arr = byConv.get(cached.conversationId) || []; + arr.push(msg); + byConv.set(cached.conversationId, arr); + } + for (const [convId, msgs] of byConv) { + store.mergeMessages(convId, msgs); + } + + // 2. 自动重试 pending 消息(failed 不自动重试,等用户手动) + const pending = pendingOrFailed.filter(m => m.status === 'pending'); + if (pending.length > 0) { + console.log(`[MessageManager] 恢复 ${pending.length} 条 pending 消息,自动重试`); + // 串行重试,避免并发发送打乱顺序 + for (const cached of pending) { + try { + // pending 重启后无法确认是否已到服务端,统一按 failed 处理后重试: + // retrySendMessage 会删旧记录、用新 tempId 重新走乐观更新流程 + await messageRepository.updateStatus(cached.id, 'failed'); + const failedMsg: MessageResponse = { + id: cached.id, + conversation_id: cached.conversationId, + sender_id: cached.senderId, + seq: cached.seq, + segments: cached.segments || [], + created_at: cached.createdAt, + status: 'failed', + }; + await this.sendService.retrySendMessage(cached.conversationId, failedMsg); + } catch (error) { + errorHandler(error); + } + } + } + } catch (error) { + errorHandler(error); + } + } + // ==================== 初始化与销毁 ==================== async initialize(): Promise { @@ -147,6 +229,9 @@ class MessageManager { // 从本地缓存恢复最后系统通知时间(避免首次渲染显示当前时间) await loadPersistedLastSystemMessageAt(); + // 恢复持久化的同步游标(供 triggerSync 的 syncByVersion 精确增量恢复) + await loadPersistedSyncVersion(); + // 注册实时事件监听(start 内部会先注销旧监听器再注册) this.pipeline.start(); @@ -166,8 +251,12 @@ class MessageManager { if (useMessageStore.getState().currentConversationId) { await this.fetchMessages(useMessageStore.getState().currentConversationId!); } + + // Outbox 恢复:扫描本地未完成发送的消息(pending/failed),读入 store 并自动重试。 + // 修复前:pending/failed 消息只存在内存,App 重启后完全丢失。 + await this.restoreOutboxMessages(); } catch (error) { - console.error('[MessageManager] 初始化失败:', error); + errorHandler(error); this.pipeline.setBootstrapping(false); } })(); diff --git a/src/stores/message/__tests__/messageHydration.test.ts b/src/stores/message/__tests__/messageHydration.test.ts index 1050868..596a39c 100644 --- a/src/stores/message/__tests__/messageHydration.test.ts +++ b/src/stores/message/__tests__/messageHydration.test.ts @@ -320,4 +320,26 @@ describe('消息 hydration 管线', () => { expect(useMessageStore.getState().isHydrated('c1')).toBe(true); }); }); + + describe('边界:临时消息(seq<=0)不干扰连续性判断', () => { + it('内存混入 seq=0 的发送中消息时,回填仍按真实历史序号进行', async () => { + // 场景:本地有 seq=5~10,内存还混入一条 seq=0 的「发送中」临时消息。 + // 修复后:computeContiguousMinSeq 过滤 seq<=0,按 5 判定下缺口,回填 1~4。 + // 修复前(理论上):seq=0 恰好等于 expected-1 当 expected=1,不会误判;但过滤后语义更稳。 + messageRepository.__setMessagesByConversation('c1', makeRange(5, 10)); + // 预置一条 seq=0 的临时消息到内存 + useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('temp-0', 0)); + + messageService.__setMessagesHandler(makeServerHandler()); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + const msgs = useMessageStore.getState().getMessages('c1'); + const realSeqs = msgs.filter((m) => m.seq > 0).map((m) => m.seq); + // 真实历史应回填到 seq=1 + expect(Math.min(...realSeqs)).toBe(1); + expect(Math.max(...realSeqs)).toBe(10); + }); + }); }); diff --git a/src/stores/message/__tests__/messageOutbox.test.ts b/src/stores/message/__tests__/messageOutbox.test.ts new file mode 100644 index 0000000..23e3031 --- /dev/null +++ b/src/stores/message/__tests__/messageOutbox.test.ts @@ -0,0 +1,150 @@ +/** + * 发送 Outbox 单测(批次二②) + * + * 锁定 Outbox 持久化的核心不变量: + * 1. 发送时临时消息(pending)落库(status='pending'),重启后可恢复 + * 2. 发送成功后正式消息落库(status='normal'),且 tempId 的 DB 记录被删除 + * 3. 发送失败后落库(status='failed'),重启后可见 + * 4. getPendingOrFailed 正确返回 pending/failed 消息 + * + * 对标 Matrix Send Queue:发送是离线可靠操作,pending/failed 必须持久化。 + */ + +import { useMessageStore } from '../store'; +import { MessageSendService } from '../services/MessageSendService'; +import { messageService } from '@/services/message'; +import { messageRepository } from '@/database'; +import type { MessageSegment } from '../../../types/dto'; + +const TEXT_SEGMENTS: MessageSegment[] = [{ type: 'text', data: { text: 'hello' } }]; + +function createSendService(): MessageSendService { + return new MessageSendService(() => 'me'); +} + +describe('发送 Outbox 持久化', () => { + beforeEach(() => { + useMessageStore.getState().reset(); + (messageRepository as any).__reset(); + (messageService as any).__reset(); + }); + + it('发送成功:临时消息落库(pending) → 正式消息落库(normal) → tempId 删除', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + // 服务端发送成功,返回 id/seq + serviceMock.__setSendMessageHandler(() => ({ id: 'srv-1', seq: 42 })); + + const send = createSendService(); + await send.sendMessage('c1', TEXT_SEGMENTS, { detailType: 'private' }); + + const saved = repoMock.__getSavedMessages(); + // 至少落库两条:pending 临时消息 + normal 正式消息 + const statuses = saved.map((m: any) => m.status); + expect(statuses).toContain('pending'); + expect(statuses).toContain('normal'); + + // tempId 的 DB 记录应被删除 + const deleted = repoMock.__getDeletedIds(); + expect(deleted.length).toBeGreaterThan(0); + // 被删除的 id 应是 temp_ 前缀 + expect(deleted.some((id: string) => id.startsWith('temp_'))).toBe(true); + + // store 里最终只有正式消息 + const msgs = useMessageStore.getState().getMessages('c1'); + expect(msgs).toHaveLength(1); + expect(msgs[0].id).toBe('srv-1'); + expect(msgs[0].status).toBe('normal'); + expect(msgs[0].seq).toBe(42); + }); + + it('发送失败:临时消息落库(failed),重启后可见', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + // 服务端发送失败(返回 null) + serviceMock.__setSendMessageHandler(() => null); + + const send = createSendService(); + await send.sendMessage('c1', TEXT_SEGMENTS, { detailType: 'private' }); + + const saved = repoMock.__getSavedMessages(); + // pending 落库 + failed 覆盖落库 + const statuses = saved.map((m: any) => m.status); + expect(statuses).toContain('pending'); + expect(statuses).toContain('failed'); + + // store 里是 failed 消息 + const msgs = useMessageStore.getState().getMessages('c1'); + expect(msgs).toHaveLength(1); + expect(msgs[0].status).toBe('failed'); + + // getPendingOrFailed 应返回这条 failed 消息(模拟重启恢复) + const pending = await messageRepository.getPendingOrFailed(); + expect(pending.length).toBeGreaterThanOrEqual(1); + expect(pending.some((m: any) => m.status === 'failed')).toBe(true); + }); + + it('发送异常(throw):临时消息落库(failed)', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + // 服务端抛异常 + serviceMock.__setSendMessageHandler(() => { + throw new Error('网络断开'); + }); + + const send = createSendService(); + await send.sendMessage('c1', TEXT_SEGMENTS, { detailType: 'private' }); + + const saved = repoMock.__getSavedMessages(); + const statuses = saved.map((m: any) => m.status); + expect(statuses).toContain('failed'); + + const msgs = useMessageStore.getState().getMessages('c1'); + expect(msgs[0].status).toBe('failed'); + }); + + it('重试失败消息:旧 failed 记录被删除,新消息重新落库', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + + // 第一次发送失败 + serviceMock.__setSendMessageHandler(() => null); + const send = createSendService(); + await send.sendMessage('c1', TEXT_SEGMENTS, { detailType: 'private' }); + + const failedMsg = useMessageStore.getState().getMessages('c1')[0]; + expect(failedMsg.status).toBe('failed'); + + // 重试:这次成功 + serviceMock.__setSendMessageHandler(() => ({ id: 'srv-2', seq: 50 })); + await send.retrySendMessage('c1', failedMsg); + + // 旧 failedId 的 DB 记录应被删除 + const deleted = repoMock.__getDeletedIds(); + expect(deleted).toContain(failedMsg.id); + + // store 里是新的正式消息 + const msgs = useMessageStore.getState().getMessages('c1'); + expect(msgs.some((m) => m.id === 'srv-2')).toBe(true); + }); + + it('getPendingOrFailed 只返回 pending/failed,不含 normal', async () => { + const repoMock = messageRepository as any; + // 注入混合状态的消息到 mock DB + repoMock.__setMessagesByConversation('c1', [ + { id: 'm1', conversationId: 'c1', senderId: 'u1', seq: 1, status: 'normal', createdAt: '2024-01-01', segments: [] }, + { id: 'm2', conversationId: 'c1', senderId: 'u1', seq: 0, status: 'pending', createdAt: '2024-01-02', segments: [] }, + { id: 'm3', conversationId: 'c1', senderId: 'u1', seq: 0, status: 'failed', createdAt: '2024-01-03', segments: [] }, + { id: 'm4', conversationId: 'c1', senderId: 'u1', seq: 2, status: 'recalled', createdAt: '2024-01-04', segments: [] }, + ]); + + const result = await messageRepository.getPendingOrFailed(); + const ids = result.map((m: any) => m.id); + expect(ids).toContain('m2'); + expect(ids).toContain('m3'); + expect(ids).not.toContain('m1'); + expect(ids).not.toContain('m4'); + // 按 createdAt 升序 + expect(result[0].id).toBe('m2'); + }); +}); diff --git a/src/stores/message/__tests__/mocks/coreMock.ts b/src/stores/message/__tests__/mocks/coreMock.ts index 7a51c50..9d741fe 100644 --- a/src/stores/message/__tests__/mocks/coreMock.ts +++ b/src/stores/message/__tests__/mocks/coreMock.ts @@ -96,6 +96,64 @@ class WsServiceMock { export const wsService = new WsServiceMock(); +/** + * errorHandler 相关导出:测试环境用简化实现(不依赖 RN UI)。 + * + * 真实 errorHandler 的副作用是 console.error(默认开启)+ 可选 showPrompt(依赖 RN)。 + * 我们的错误接入都用默认选项(showErrorPrompt=false),不触发 showPrompt, + * 所以测试环境用简化实现即可,安全无副作用。 + * + * 注意:测试环境不透传真实 errorHandler,因为 jest moduleNameMapper 只把 + * `^@/services/core$` 映射到本文件,`@/services/core/errorHandler` 不匹配, + * 会落到真实实现,而真实实现 import 了 api.ts 等可能有副作用的模块。 + */ +export function isNetworkError(error: unknown): boolean { + if (error instanceof ApiError) { + return (error as any).errorCode === 'NETWORK_ERROR'; + } + if (error instanceof Error) { + const msg = error.message.toLowerCase(); + return msg.includes('network request failed') || msg.includes('failed to fetch'); + } + return false; +} + +// errorHandler 的核心函数(简化版:分类 + console.error,不调 showPrompt) +export const handleError = (error: unknown, options?: any) => { + const context = options?.context; + const prefix = context ? `[${context}]` : '[error]'; + if (options?.logToConsole !== false) { + console.error(`${prefix} ${error instanceof Error ? error.message : String(error)}`, error); + } + return { code: 'UNKNOWN', message: String(error), originalError: error, context, silent: options?.silent }; +}; + +export const createErrorHandler = (defaultContext: string) => { + return (error: unknown, options: any = {}) => + handleError(error, { ...options, context: defaultContext }); +}; + +export const safeAsync = async (fn: () => Promise, options?: any): Promise<{ data: T | null; error: any | null }> => { + try { + const data = await fn(); + return { data, error: null }; + } catch (error) { + const appError = handleError(error, options); + return { data: null, error: appError }; + } +}; + +export const AppErrorCode = { + NETWORK_ERROR: 'NETWORK_ERROR', + AUTH_ERROR: 'AUTH_ERROR', + FORBIDDEN: 'FORBIDDEN', + NOT_FOUND: 'NOT_FOUND', + VALIDATION_ERROR: 'VALIDATION_ERROR', + SERVER_ERROR: 'SERVER_ERROR', + TIMEOUT: 'TIMEOUT', + UNKNOWN: 'UNKNOWN', +} as const; + /** * 供 WSServerMessage 等 wsService.ts 导出的类型在 mock 中可见。 * Pipeline 只用到 wsService 实例和 GroupNoticeType;类型 import 是 type-only, diff --git a/src/stores/message/__tests__/mocks/databaseMock.ts b/src/stores/message/__tests__/mocks/databaseMock.ts index 5c23b95..b43d9a3 100644 --- a/src/stores/message/__tests__/mocks/databaseMock.ts +++ b/src/stores/message/__tests__/mocks/databaseMock.ts @@ -12,14 +12,26 @@ interface RepositoryMockState { /** 按 `${conversationId}` 存储该会话的全部本地消息(模拟 SQLite 表内容)。 */ messagesByConversation: Map; + /** saveMessage 实际写入的调用记录(验证 Outbox 落库用)。 */ + savedMessages: any[]; + /** delete 调用记录的 messageId(验证 Outbox tempId 清理用)。 */ + deletedIds: string[]; + /** updateStatus 调用记录(验证 Outbox 状态流转用)。 */ + statusUpdates: { messageId: string; status: string }[]; } const state: RepositoryMockState = { messagesByConversation: new Map(), + savedMessages: [], + deletedIds: [], + statusUpdates: [], }; function resetState() { state.messagesByConversation.clear(); + state.savedMessages = []; + state.deletedIds = []; + state.statusUpdates = []; } /** @@ -32,9 +44,23 @@ function __setMessagesByConversation(conversationId: string, messages: any[]) { } export const messageRepository = { - // ---- 写入:no-op(单测不验证持久化)---- - saveMessage: async () => {}, - saveMessagesBatch: async () => {}, + // ---- 写入:记录调用(验证 Outbox 落库),同时维护内部状态 ---- + async saveMessage(message: any) { + state.savedMessages.push(message); + // 同步更新 messagesByConversation,让 getPendingOrFailed 等读取能反映写入 + const convId = message.conversationId; + const arr = state.messagesByConversation.get(convId) || []; + const idx = arr.findIndex((m) => m.id === message.id); + if (idx >= 0) arr[idx] = message; + else arr.push(message); + state.messagesByConversation.set(convId, arr); + }, + + async saveMessagesBatch(messages: any[]) { + for (const m of messages) { + await this.saveMessage(m); + } + }, // ---- 读取:基于内部状态 ---- async getByConversation(conversationId: string, limit = 20) { @@ -85,14 +111,47 @@ export const messageRepository = { return (state.messagesByConversation.get(conversationId) || []).length; }, - // ---- 其它:no-op / 空实现 ---- - updateStatus: async () => {}, + // ---- 其它:状态维护版本(验证 Outbox 状态流转)---- + async updateStatus(messageId: string, status: string, clearContent = false) { + state.statusUpdates.push({ messageId, status }); + // 同步更新内部状态 + for (const [, arr] of state.messagesByConversation) { + const m = arr.find((x) => x.id === messageId); + if (m) { + m.status = status; + if (clearContent) { + m.content = ''; + m.segments = []; + } + } + } + }, markAsRead: async () => {}, markConversationAsRead: async () => {}, - delete: async () => {}, + async delete(messageId: string) { + state.deletedIds.push(messageId); + for (const [, arr] of state.messagesByConversation) { + const idx = arr.findIndex((x) => x.id === messageId); + if (idx >= 0) arr.splice(idx, 1); + } + }, clearConversation: async (conversationId: string) => { state.messagesByConversation.delete(conversationId); }, + async getPendingOrFailed() { + // 扫描所有会话,收集 status 为 pending/failed 的消息,按 createdAt 升序 + const out: any[] = []; + for (const [, arr] of state.messagesByConversation) { + for (const m of arr) { + if (m.status === 'pending' || m.status === 'failed') { + out.push(m); + } + } + } + return out.sort((a, b) => + String(a.createdAt).localeCompare(String(b.createdAt)) + ); + }, search: async () => [], searchByConversation: async () => [], getStats: async () => ({ totalMessages: 0, totalConversations: 0 }), @@ -105,6 +164,12 @@ export const messageRepository = { __setMessagesByConversation, __getMessagesByConversation: (conversationId: string) => state.messagesByConversation.get(conversationId) || [], + /** 已 saveMessage 落库的消息记录(断言 Outbox 持久化) */ + __getSavedMessages: () => state.savedMessages, + /** 已 delete 调用的 messageId 列表(断言 Outbox tempId 清理) */ + __getDeletedIds: () => state.deletedIds, + /** 已 updateStatus 调用的记录(断言 Outbox 状态流转) */ + __getStatusUpdates: () => state.statusUpdates, }; export const conversationRepository = { diff --git a/src/stores/message/__tests__/mocks/messageServiceMock.ts b/src/stores/message/__tests__/mocks/messageServiceMock.ts index 36f21f4..2f46aeb 100644 --- a/src/stores/message/__tests__/mocks/messageServiceMock.ts +++ b/src/stores/message/__tests__/mocks/messageServiceMock.ts @@ -16,12 +16,32 @@ type GetMessagesHandler = (args: { limit?: number; }) => MessageListResponse; +/** 发送消息的模拟响应 */ +type SendMessageResponse = { id: string; seq: number }; +type SendMessageHandler = (args: { + conversationId: string; + segments: any[]; + replyToId?: string; +}) => SendMessageResponse | null; + +/** syncByVersion 的模拟响应 */ +type SyncByVersionHandler = (version: number) => { + current_version: number; + changes: any[]; + full_sync: boolean; + has_more: boolean; +}; + const EMPTY_RESPONSE: MessageListResponse = { messages: [], total: 0, page: 1, page_size: 20 }; let getMessagesHandler: GetMessagesHandler | null = null; +let sendMessageHandler: SendMessageHandler | null = null; +let syncByVersionHandler: SyncByVersionHandler | null = null; function resetHandler() { getMessagesHandler = null; + sendMessageHandler = null; + syncByVersionHandler = null; } export const messageService = { @@ -37,6 +57,33 @@ export const messageService = { return { ...EMPTY_RESPONSE }; }, + async sendMessage( + conversationId: string, + payload: { segments: any[]; reply_to_id?: string } + ): Promise { + if (sendMessageHandler) { + return sendMessageHandler({ + conversationId, + segments: payload.segments, + replyToId: payload.reply_to_id, + }); + } + // 默认:返回一个递增的 id/seq,模拟发送成功 + return { id: `srv_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`, seq: 0 }; + }, + + async getSyncByVersion(version: number): Promise<{ + current_version: number; + changes: any[]; + full_sync: boolean; + has_more: boolean; + }> { + if (syncByVersionHandler) { + return syncByVersionHandler(version); + } + return { current_version: version, changes: [], full_sync: false, has_more: false }; + }, + getConversationById: async () => null, getUnreadCount: async () => ({ total_unread_count: 0 }), getSystemUnreadCount: async () => ({ unread_count: 0 }), @@ -52,6 +99,12 @@ export const messageService = { __setMessagesHandler: (handler: GetMessagesHandler) => { getMessagesHandler = handler; }, + __setSendMessageHandler: (handler: SendMessageHandler) => { + sendMessageHandler = handler; + }, + __setSyncByVersionHandler: (handler: SyncByVersionHandler) => { + syncByVersionHandler = handler; + }, }; -export type { GetMessagesHandler }; +export type { GetMessagesHandler, SendMessageHandler, SyncByVersionHandler }; diff --git a/src/stores/message/__tests__/syncByVersion.test.ts b/src/stores/message/__tests__/syncByVersion.test.ts new file mode 100644 index 0000000..cff0985 --- /dev/null +++ b/src/stores/message/__tests__/syncByVersion.test.ts @@ -0,0 +1,117 @@ +/** + * syncByVersion 游标同步单测(批次二③) + * + * 锁定基于版本号的精确增量恢复不变量: + * 1. 成功同步并推进持久化游标 + * 2. full_sync=true 时退化(返回 false) + * 3. 异常时退化(返回 false) + * 4. 无 version(<=0)时直接退化 + * + * 对标 Matrix resume 游标:断线恢复从全量轮询升级为精确增量。 + */ + +import { useMessageStore } from '../store'; +import { MessageSyncService } from '../services/MessageSyncService'; +import { messageService } from '@/services/message'; +import { messageRepository } from '@/database'; + +function createSyncService(): MessageSyncService { + return new MessageSyncService( + () => 'me', + { markAsRead: async () => {}, markAllAsRead: async () => {} } as any, + { getSenderInfo: async () => null, enrichMessagesWithSenderInfo: () => {} } as any, + undefined + ); +} + +describe('syncByVersion 游标同步', () => { + beforeEach(() => { + useMessageStore.getState().reset(); + (messageRepository as any).__reset(); + (messageService as any).__reset(); + }); + + it('无变更时返回成功且推进游标', async () => { + const serviceMock = messageService as any; + serviceMock.__setSyncByVersionHandler((version: number) => ({ + current_version: version + 10, + changes: [], + full_sync: false, + has_more: false, + })); + + const sync = createSyncService(); + const ok = await sync.syncByVersion(100); + + expect(ok).toBe(true); + // 游标推进到 current_version + expect(useMessageStore.getState().syncVersion).toBe(110); + }); + + it('full_sync=true 时退化为 false', async () => { + const serviceMock = messageService as any; + serviceMock.__setSyncByVersionHandler(() => ({ + current_version: 0, + changes: [], + full_sync: true, + has_more: false, + })); + + const sync = createSyncService(); + const ok = await sync.syncByVersion(100); + + expect(ok).toBe(false); + }); + + it('API 抛异常时退化为 false', async () => { + const serviceMock = messageService as any; + serviceMock.__setSyncByVersionHandler(() => { + throw new Error('网络故障'); + }); + + const sync = createSyncService(); + const ok = await sync.syncByVersion(100); + + expect(ok).toBe(false); + }); + + it('version <= 0 时直接退化', async () => { + const sync = createSyncService(); + const ok1 = await sync.syncByVersion(0); + const ok2 = await sync.syncByVersion(-1); + expect(ok1).toBe(false); + expect(ok2).toBe(false); + }); + + it('有会话变更时按需同步会话详情', async () => { + const serviceMock = messageService as any; + // 先在 store 放一个会话,让 fetchConversationDetail 有上下文 + useMessageStore.getState().setConversationsWithUnread( + new Map([['c1', { id: 'c1', last_seq: 5, unread_count: 0 } as any]]), + 0, 0 + ); + + let detailCalled = false; + serviceMock.__setSyncByVersionHandler((version: number) => ({ + current_version: version + 5, + changes: [ + { conversation_id: 'c1', change_type: 'new_message', max_seq: 10, version: version + 1 }, + ], + full_sync: false, + has_more: false, + })); + // fetchConversationDetail 内部调 getConversationById + serviceMock.getConversationById = async () => { + detailCalled = true; + return { id: 'c1', last_seq: 10 } as any; + }; + serviceMock.getMessages = async () => ({ messages: [], total: 0, page: 1, page_size: 20 }); + + const sync = createSyncService(); + const ok = await sync.syncByVersion(100); + + expect(ok).toBe(true); + expect(detailCalled).toBe(true); + expect(useMessageStore.getState().syncVersion).toBe(105); + }); +}); diff --git a/src/stores/message/__tests__/syncState.test.ts b/src/stores/message/__tests__/syncState.test.ts new file mode 100644 index 0000000..0539518 --- /dev/null +++ b/src/stores/message/__tests__/syncState.test.ts @@ -0,0 +1,129 @@ +/** + * 同步状态机单测(批次一④) + * + * 锁定 conversationSyncStateMap 的状态流转不变量: + * - 初始 idle + * - hydration 进行中 → hydrating + * - hydration 成功 → synced + * - hydration 失败 → error + * - clearMessages/removeConversation → 回到 idle(隐式删除) + */ + +import { useMessageStore } from '../store'; +import { MessageSyncService } from '../services/MessageSyncService'; +import { messageService } from '@/services/message'; +import { messageRepository } from '@/database'; +import type { MessageResponse } from '../../../types/dto'; + +function makeRange(from: number, to: number): MessageResponse[] { + const out: MessageResponse[] = []; + for (let s = from; s <= to; s++) { + out.push({ + id: `m-${s}`, + conversation_id: 'c1', + sender_id: 'u1', + seq: s, + segments: [], + created_at: `2024-01-01T00:00:${String(s).padStart(2, '0')}Z`, + status: 'normal', + }); + } + return out; +} + +function createSyncService(): MessageSyncService { + return new MessageSyncService( + () => 'me', + { markAsRead: async () => {}, markAllAsRead: async () => {} } as any, + { getSenderInfo: async () => null, enrichMessagesWithSenderInfo: () => {} } as any, + undefined + ); +} + +describe('同步状态机 conversationSyncState', () => { + beforeEach(() => { + useMessageStore.getState().reset(); + (messageRepository as any).__reset(); + (messageService as any).__reset(); + }); + + it('初始状态为 idle', () => { + expect(useMessageStore.getState().getConversationSyncState('c1')).toBe('idle'); + }); + + it('hydration 成功后状态转为 synced', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + repoMock.__setMessagesByConversation('c1', makeRange(1, 20)); + // 后端:本地已连续到顶,无需回填 + serviceMock.__setMessagesHandler(() => ({ messages: [], total: 0, page: 1, page_size: 20 })); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + + expect(useMessageStore.getState().isHydrated('c1')).toBe(true); + expect(useMessageStore.getState().getConversationSyncState('c1')).toBe('synced'); + }); + + it('hydration 失败后状态转为 error(store 层验证)', () => { + // 直接验证状态机 action:外部(如 catch 块)设置 error 态 + const store = useMessageStore.getState(); + store.setConversationSyncState('c1', 'hydrating'); + expect(store.getConversationSyncState('c1')).toBe('hydrating'); + + store.setConversationSyncState('c1', 'error'); + expect(store.getConversationSyncState('c1')).toBe('error'); + + // error 态下 isHydrated 仍为 false(未成功完成) + expect(store.isHydrated('c1')).toBe(false); + }); + + it('error 态可重新进入 hydrating(用户重试场景)', async () => { + const store = useMessageStore.getState(); + // 模拟上次失败留下 error 态 + store.setConversationSyncState('c1', 'error'); + // 手动清除 hydrated(确保重新 hydration) + store.clearHydrated('c1'); + + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + repoMock.__setMessagesByConversation('c1', makeRange(1, 5)); + serviceMock.__setMessagesHandler(() => ({ messages: [], total: 0, page: 1, page_size: 20 })); + + // 重新 fetch:error → hydrating → synced + const sync = createSyncService(); + await sync.fetchMessages('c1'); + expect(store.getConversationSyncState('c1')).toBe('synced'); + }); + + it('clearMessages 后状态回到 idle', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + repoMock.__setMessagesByConversation('c1', makeRange(1, 5)); + serviceMock.__setMessagesHandler(() => ({ messages: [], total: 0, page: 1, page_size: 20 })); + + const sync = createSyncService(); + await sync.fetchMessages('c1'); + expect(useMessageStore.getState().getConversationSyncState('c1')).toBe('synced'); + + useMessageStore.getState().clearMessages('c1'); + expect(useMessageStore.getState().getConversationSyncState('c1')).toBe('idle'); + }); + + it('重入已 synced 会话不再进入 hydrating', async () => { + const repoMock = messageRepository as any; + const serviceMock = messageService as any; + repoMock.__setMessagesByConversation('c1', makeRange(1, 20)); + serviceMock.__setMessagesHandler(() => ({ messages: [], total: 0, page: 1, page_size: 20 })); + + const sync = createSyncService(); + // 首次:idle → hydrating → synced + await sync.fetchMessages('c1'); + expect(useMessageStore.getState().getConversationSyncState('c1')).toBe('synced'); + + // 重入:已 synced,走轻量增量,不重置为 hydrating + await sync.fetchMessages('c1'); + // 仍为 synced(不会回到 hydrating,因为 wasHydratedAtStart=true 时不设 hydrating) + expect(useMessageStore.getState().getConversationSyncState('c1')).toBe('synced'); + }); +}); diff --git a/src/stores/message/hooks.ts b/src/stores/message/hooks.ts index c3090b9..c07b45f 100644 --- a/src/stores/message/hooks.ts +++ b/src/stores/message/hooks.ts @@ -17,7 +17,11 @@ import { useShallow } from 'zustand/react/shallow'; import { ConversationResponse, MessageResponse, MessageSegment } from '../../types/dto'; import { messageManager } from './MessageManager'; import { useMessageStore } from './store'; +import type { ConversationSyncState } from './store'; import { useUnreadCountQuery } from '../../hooks/useUnreadCount'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('message-hooks'); const EMPTY_MESSAGES: MessageResponse[] = []; @@ -44,7 +48,7 @@ export function useConversations(): UseConversationsReturn { useEffect(() => { // 初始化时确保 MessageManager 已初始化 messageManager.initialize().catch(error => { - console.error('[useConversations] 初始化失败:', error); + errorHandler(error); }); // 仅在首次初始化时做一次冷启动兜底刷新,避免每次挂载都触发 @@ -52,7 +56,7 @@ export function useConversations(): UseConversationsReturn { if (!store.isInitialized) { const coldStartSyncTimer = setTimeout(() => { messageManager.refreshConversations(true, 'hooks-initial-refresh').catch(error => { - console.error('[useConversations] 冷启动强制刷新失败:', error); + errorHandler(error); }); }, 1200); @@ -89,6 +93,7 @@ export function useConversations(): UseConversationsReturn { interface UseMessagesReturn { messages: MessageResponse[]; isLoading: boolean; + syncState: ConversationSyncState; hasMore: boolean; loadMore: () => Promise; refresh: () => Promise; @@ -121,6 +126,12 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { const isLoadingMessages = useMessageStore(state => normalizedConversationId ? state.loadingMessagesSet.has(normalizedConversationId) : false ); + // 会话同步状态机(hydration 管线维护),供 UI 展示错误态/重试 + const syncState = useMessageStore(state => + normalizedConversationId + ? (state.conversationSyncStateMap.get(normalizedConversationId) || 'idle') + : 'idle' + ); const [hasMore, setHasMore] = useState(true); const loadMoreInFlightRef = useRef(false); // 追踪上次激活的 conversationId,避免同一会话重复激活时不必要的 forceSync @@ -144,7 +155,7 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { // 架构入口:激活会话并完成初始化/同步 // 重入同一会话时 forceSync,确保拿到最新消息 messageManager.activateConversation(normalizedConversationId, { forceSync: isReentry }).catch(error => { - console.error('[useMessages] 激活会话失败:', error); + errorHandler(error); }); lastActivatedIdRef.current = normalizedConversationId; @@ -180,11 +191,11 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { const maxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0); if (maxSeq > 0) { messageManager.fetchMessages(normalizedConversationId, maxSeq).catch(error => { - console.error('[useMessages] 焦点同步消息失败:', error); + errorHandler(error); }); } else { messageManager.fetchMessages(normalizedConversationId).catch(error => { - console.error('[useMessages] 焦点同步消息(冷启动)失败:', error); + errorHandler(error); }); } }, [normalizedConversationId]) @@ -219,6 +230,7 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { return { messages, isLoading: isLoadingMessages, + syncState, hasMore, loadMore, refresh, @@ -413,7 +425,7 @@ export function useMessageManager(): UseMessageManagerReturn { useEffect(() => { // 初始化 messageManager.initialize().catch(error => { - console.error('[useMessageManager] 初始化失败:', error); + errorHandler(error); }); }, []); @@ -528,7 +540,7 @@ export function useMessageListRefresh(): UseMessageListRefreshReturn { try { await messageManager.refreshConversations(true, 'hooks-load-more-fallback'); } catch (error) { - console.error('[useMessageListRefresh] 刷新失败:', error); + errorHandler(error); } finally { setIsRefreshing(false); } diff --git a/src/stores/message/services/ConversationOperations.ts b/src/stores/message/services/ConversationOperations.ts index fd918dc..e558e85 100644 --- a/src/stores/message/services/ConversationOperations.ts +++ b/src/stores/message/services/ConversationOperations.ts @@ -8,6 +8,9 @@ import { messageService } from '@/services/message'; import { conversationRepository } from '@/database'; import type { IConversationOperations } from '../types'; import { useMessageStore, normalizeConversationId } from '../store'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('ConversationOperations'); export class ConversationOperations implements IConversationOperations { /** @@ -24,7 +27,7 @@ export class ConversationOperations implements IConversationOperations { return conversation; } catch (error) { - console.error('[ConversationOperations] 创建会话失败:', error); + errorHandler(error); return null; } } @@ -57,7 +60,7 @@ export class ConversationOperations implements IConversationOperations { // 删除本地数据库中的会话 conversationRepository.delete(conversationId).catch(error => { - console.error('[ConversationOperations] 删除本地会话失败:', error); + errorHandler(error); }); } diff --git a/src/stores/message/services/MessageSendService.ts b/src/stores/message/services/MessageSendService.ts index 539c9f4..165aa84 100644 --- a/src/stores/message/services/MessageSendService.ts +++ b/src/stores/message/services/MessageSendService.ts @@ -9,6 +9,9 @@ import { messageService } from '@/services/message'; import { messageRepository } from '@/database'; import type { IMessageSendService } from '../types'; import { useMessageStore } from '../store'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('MessageSendService'); // 临时ID前缀,用于乐观更新 const TEMP_ID_PREFIX = 'temp_'; @@ -65,6 +68,8 @@ export class MessageSendService implements IMessageSendService { // 2. 立即添加到store,用户可以马上看到 store.addOrReplaceMessage(conversationId, tempMessage); + // Outbox:临时消息落库(status='pending'),保证 App 重启后能恢复未完成的发送 + this.persistMessage(tempMessage); // 3. 更新会话最后消息(乐观更新) this.updateConversationWithNewMessage(conversationId, tempMessage); @@ -96,24 +101,10 @@ export class MessageSendService implements IMessageSendService { // 更新会话最后消息 this.updateConversationWithNewMessage(conversationId, successMessage); - // 保存到本地数据库 - const textContent = segments - ?.filter((s: any) => s.type === 'text') - .map((s: any) => s.data?.text || '') - .join('') || ''; - - messageRepository.saveMessage({ - id: response.id, - conversationId, - senderId: currentUserId || '', - content: textContent, - type: segments?.find((s: any) => s.type === 'image') ? 'image' : 'text', - isRead: true, - createdAt: successMessage.created_at, - seq: response.seq, - status: 'normal', - segments, - }).catch(console.error); + // Outbox:成功后持久化正式消息(INSERT OR REPLACE 按 id 覆盖), + // 并删除 tempId 的 DB 记录,避免重启后重复恢复 + this.persistMessage(successMessage); + messageRepository.delete(tempId).catch((e: unknown) => errorHandler(e)); return successMessage; } @@ -123,17 +114,27 @@ export class MessageSendService implements IMessageSendService { ...tempMessage, status: 'failed', }; - store.addOrReplaceMessage(conversationId, failedMessage); + // 注意:addOrReplaceMessage 按 id 幂等(已存在则不更新),改用 patchMessages 更新 status + const patches = new Map>(); + patches.set(String(failedMessage.id), { status: 'failed' }); + store.patchMessages(conversationId, patches); + // Outbox:失败消息落库 + this.persistMessage(failedMessage); return null; } catch (error) { - console.error('[MessageSendService] 发送消息失败:', error); + errorHandler(error); // 6. 发送失败:更新消息状态为失败 const failedMessage: MessageResponse = { ...tempMessage, status: 'failed', }; - store.addOrReplaceMessage(conversationId, failedMessage); + // addOrReplaceMessage 按 id 幂等不更新,改用 patchMessages 更新 status + const patches = new Map>(); + patches.set(String(failedMessage.id), { status: 'failed' }); + store.patchMessages(conversationId, patches); + // Outbox:失败消息落库 + this.persistMessage(failedMessage); return null; } @@ -161,6 +162,10 @@ export class MessageSendService implements IMessageSendService { const store = useMessageStore.getState(); // 移除旧的失败消息,避免 sendMessage 新插入的 pending 消息与之并存 store.removeMessage(conversationId, String(failedMessage.id)); + // Outbox:同步删除 DB 中旧的 failed 记录,避免重启后 getPendingOrFailed 重复恢复 + // (sendMessage 会用新 tempId 重新落库 pending,重试成功后正常落库 normal 并删新 tempId) + messageRepository.delete(String(failedMessage.id)) + .catch((e: unknown) => errorHandler(e)); // 重新发送:sendMessage 会重新插入 pending 临时消息并完成后续流程 return this.sendMessage(conversationId, failedMessage.segments, { @@ -168,6 +173,33 @@ export class MessageSendService implements IMessageSendService { }); } + /** + * 持久化消息到本地数据库(Outbox 机制)。 + * 把 MessageResponse(DTO)映射为 CachedMessage(DB 行)并落库(INSERT OR REPLACE 按 id 幂等)。 + * 用于 pending/failed/normal 三态消息的持久化,保证 App 重启后发送状态可恢复。 + * 落库失败不阻断 UI(fire-and-forget + 日志),最坏情况是重启后丢失该条未完成发送。 + */ + private persistMessage(message: MessageResponse): void { + const segments = message.segments || []; + const textContent = segments + .filter((s: any) => s.type === 'text') + .map((s: any) => s.data?.text || '') + .join('') || ''; + + messageRepository.saveMessage({ + id: String(message.id), + conversationId: message.conversation_id, + senderId: message.sender_id, + content: textContent, + type: segments.find((s: any) => s.type === 'image') ? 'image' : 'text', + isRead: true, + createdAt: message.created_at, + seq: message.seq, + status: message.status, + segments, + }).catch((e: unknown) => errorHandler(e)); + } + /** * 更新会话的最后消息信息 */ diff --git a/src/stores/message/services/MessageSyncService.ts b/src/stores/message/services/MessageSyncService.ts index 74c6e94..f8c05ad 100644 --- a/src/stores/message/services/MessageSyncService.ts +++ b/src/stores/message/services/MessageSyncService.ts @@ -20,6 +20,10 @@ import { MESSAGES_PAGE_SIZE, MAX_GAP_FILL_ROUNDS, } from '../constants'; +import { createErrorHandler } from '@/services/core'; + +/** 本模块统一的错误处理器(带 context 分类,替代散落的 console.error) */ +const errorHandler = createErrorHandler('MessageSyncService'); export class MessageSyncService implements IMessageSyncService { private getCurrentUserId: () => string | null; @@ -134,7 +138,7 @@ export class MessageSyncService implements IMessageSyncService { this.persistConversationListCache(); } catch (error) { - console.error('[MessageSyncService] 获取会话列表失败:', error); + errorHandler(error); if (store.getConversations().length === 0) { const recovered = await this.hydrateConversationsFromLocalSource(); if (recovered) { @@ -171,7 +175,7 @@ export class MessageSyncService implements IMessageSyncService { this.recomputeConversationTotalUnread(); this.persistConversationListCache(); } catch (error) { - console.error('[MessageSyncService] 加载更多会话失败:', error); + errorHandler(error); } finally { this.loadingMoreConversations = false; } @@ -193,7 +197,7 @@ export class MessageSyncService implements IMessageSyncService { } return null; } catch (error) { - console.error('[MessageSyncService] 获取会话详情失败:', error); + errorHandler(error); return null; } } @@ -235,6 +239,12 @@ export class MessageSyncService implements IMessageSyncService { store.setLoadingMessages(conversationId, true); + // 同步状态机:未 hydrated 的会话进入 hydrating 态(供 UI 展示) + const wasHydratedAtStart = store.isHydrated(conversationId); + if (!wasHydratedAtStart) { + store.setConversationSyncState(conversationId, 'hydrating'); + } + try { if (!afterSeq) { const wasHydrated = store.isHydrated(conversationId); @@ -242,7 +252,7 @@ export class MessageSyncService implements IMessageSyncService { // 已完整加载过:baseline 可信,仅做轻量增量同步 await this.incrementalSyncAbove(conversationId); } else { - // 完整 hydration 管线 + // 完整 hydration 管线(成功后内部会 markHydrated 并设 syncState='synced') await this.runHydrationPipeline(conversationId); } } else { @@ -256,12 +266,16 @@ export class MessageSyncService implements IMessageSyncService { messageRepository.saveMessagesBatch( MessageMapper.toCachedMessages(newMessages, conversationId) ).catch(error => { - console.error('[MessageSyncService] 保存消息到本地失败:', error); + errorHandler(error); }); } } } catch (error) { - console.error('[MessageSyncService] 获取消息失败:', error); + errorHandler(error); + // hydration 失败:标记 error 态(供 UI 展示错误/重试) + if (!wasHydratedAtStart) { + useMessageStore.getState().setConversationSyncState(conversationId, 'error'); + } } finally { // 异步填充用户信息(不阻塞消息显示) // 使用 merge 而非覆盖,防止异步期间新到的 WS 消息丢失 @@ -310,11 +324,11 @@ export class MessageSyncService implements IMessageSyncService { messageRepository.saveMessagesBatch( MessageMapper.toCachedMessages(newMessages, conversationId) ).catch(error => { - console.error('[MessageSyncService] 增量同步保存失败:', error); + errorHandler(error); }); } } catch (error) { - console.error('[MessageSyncService] 增量同步失败:', error); + errorHandler(error); } } @@ -349,7 +363,7 @@ export class MessageSyncService implements IMessageSyncService { useMessageStore.getState().mergeMessages(conversationId, formattedMessages); } } catch (error) { - console.warn('[MessageSyncService] hydration 读取本地消息失败:', error); + errorHandler(error); } // ---- Step 2:上缺口增量同步(baselineMaxSeq 之后的新消息)---- @@ -365,11 +379,11 @@ export class MessageSyncService implements IMessageSyncService { messageRepository.saveMessagesBatch( MessageMapper.toCachedMessages(newMessages, conversationId) ).catch(error => { - console.error('[MessageSyncService] hydration 增量同步保存失败:', error); + errorHandler(error); }); } } catch (error) { - console.error('[MessageSyncService] hydration 增量同步失败:', error); + errorHandler(error); } } else { // 本地与内存均空:冷启动快照兜底 @@ -383,7 +397,10 @@ export class MessageSyncService implements IMessageSyncService { // ---- Step 4(方案 C):标记 hydrated ---- // 无论回填是否完全到顶都标记,避免反复重试卡首屏;用户上滑仍可触发 loadMoreHistory。 - useMessageStore.getState().markHydrated(conversationId); + const finalStore = useMessageStore.getState(); + finalStore.markHydrated(conversationId); + // 同步状态机:hydration 成功完成 → synced + finalStore.setConversationSyncState(conversationId, 'synced'); } /** @@ -399,11 +416,11 @@ export class MessageSyncService implements IMessageSyncService { messageRepository.saveMessagesBatch( MessageMapper.toCachedMessages(snapshotMessages, conversationId) ).catch(error => { - console.error('[MessageSyncService] 保存快照消息到本地失败:', error); + errorHandler(error); }); } } catch (error) { - console.error('[MessageSyncService] 快照同步失败:', error); + errorHandler(error); } } @@ -431,7 +448,7 @@ export class MessageSyncService implements IMessageSyncService { // 复用现有 loadMoreMessages:本地不足则请求服务端,内部已 merge + 落库 await this.loadMoreMessages(conversationId, contiguousMinSeq, MESSAGES_PAGE_SIZE); } catch (error) { - console.error('[MessageSyncService] 回填历史缺口失败:', error); + errorHandler(error); return; // 本轮失败,停止回填(已加载部分正常展示) } @@ -444,11 +461,18 @@ export class MessageSyncService implements IMessageSyncService { /** * 计算内存消息数组从最大 seq 往下的连续区间下沿(contiguousMinSeq)。 * 遇到第一个缺口即停。返回值 <= 1 表示已连续到顶。 + * + * 注意:seq<=0 的消息(发送中/失败的临时消息)不参与历史连续性判断, + * 它们不代表真实的历史序号,过滤掉避免干扰回填决策。 */ private computeContiguousMinSeq(messages: MessageResponse[]): number { if (messages.length === 0) return 0; - // 按 seq 降序遍历找连续段 - const sorted = [...messages].sort((a, b) => b.seq - a.seq); + // 按 seq 降序遍历找连续段,过滤掉 seq<=0 的临时消息 + const sorted = messages + .filter(m => (m.seq || 0) > 0) + .sort((a, b) => b.seq - a.seq); + if (sorted.length === 0) return 0; + let expected = sorted[0].seq; for (let i = 1; i < sorted.length; i++) { if (sorted[i].seq === expected - 1) { @@ -505,7 +529,7 @@ export class MessageSyncService implements IMessageSyncService { return []; } catch (error) { - console.error('[MessageSyncService] 加载更多消息失败:', error); + errorHandler(error); return []; } } @@ -554,7 +578,7 @@ export class MessageSyncService implements IMessageSyncService { // 服务端汇总未读为 0:不主动清零本地未读,等 fetchConversations 提供权威数据 // 避免后端缓存竞态导致的未读数抖动 } catch (error) { - console.error('[MessageSyncService] 获取未读数失败:', error); + errorHandler(error); } } @@ -629,7 +653,67 @@ export class MessageSyncService implements IMessageSyncService { return true; } catch (error) { - console.error('[MessageSyncService] syncBySeq 失败:', error); + errorHandler(error); + return false; + } + } + + /** + * 基于版本号的精确增量同步(对标 Matrix resume 游标)。 + * + * 用持久化的 syncVersion 调用 getSyncByVersion,服务端返回该版本之后的会话变更, + * 按需 fetchConversationDetail/fetchMessages 补齐,比 syncBySeq 的全量轮询更精确。 + * + * @returns true=成功并推进游标;false=需退化到 syncBySeq 或全量(full_sync 或异常) + */ + async syncByVersion(version: number): Promise { + if (!version || version <= 0) return false; + + try { + const resp = await messageService.getSyncByVersion(version); + if (!resp) return false; + + // 服务端要求全量同步(版本太旧或游标失效):退化 + if (resp.full_sync) { + console.warn('[MessageSyncService] syncByVersion 服务端要求全量同步,退化'); + return false; + } + + const changes = resp.changes || []; + const store = useMessageStore.getState(); + + // 按 change_type 处理每个会话的变更 + for (const change of changes) { + try { + const conv = store.getConversation(change.conversation_id); + const localMaxSeq = conv?.last_seq || 0; + + // 会话有新消息(change_type 含 message/seq 变更,或本地落后于服务端 max_seq) + if (change.max_seq && change.max_seq > localMaxSeq) { + await this.fetchConversationDetail(change.conversation_id); + // 活动会话拉取新增消息 + if (store.getActiveConversation() === change.conversation_id) { + await this.fetchMessages(change.conversation_id, localMaxSeq); + } + } + } catch (e) { + // 单个会话变更处理失败不阻断整体(其余会话仍可推进游标) + errorHandler(e); + } + } + + // 推进持久化游标(has_more 表示还有更多变更,下次继续) + const nextVersion = resp.current_version; + if (nextVersion && nextVersion > version) { + store.setSyncVersion(nextVersion); + } + + // 刷新未读数(变更可能影响未读) + await this.fetchUnreadCount(); + + return true; + } catch (error) { + errorHandler(error); return false; } } @@ -692,7 +776,7 @@ export class MessageSyncService implements IMessageSyncService { .filter((group): group is NonNullable => Boolean(group)); conversationRepository.saveWithRelated(list, users, groups).catch(error => { - console.error('[MessageSyncService] 持久化会话列表失败:', error); + errorHandler(error); }); } diff --git a/src/stores/message/services/ReadReceiptManager.ts b/src/stores/message/services/ReadReceiptManager.ts index d0a93c3..ce5f3c0 100644 --- a/src/stores/message/services/ReadReceiptManager.ts +++ b/src/stores/message/services/ReadReceiptManager.ts @@ -9,6 +9,9 @@ import { messageRepository, conversationRepository } from '@/database'; import type { ReadStateRecord, IReadReceiptManager } from '../types'; import { READ_STATE_PROTECTION_DELAY } from '../constants'; import { useMessageStore, normalizeConversationId } from '../store'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('ReadReceiptManager'); export class ReadReceiptManager implements IReadReceiptManager { /** @@ -75,13 +78,13 @@ export class ReadReceiptManager implements IReadReceiptManager { store.updateConversationWithUnread(updatedConv, newTotalUnread, currentUnread.system); // 3. 更新本地数据库 - messageRepository.markConversationAsRead(normalizedId).catch(console.error); + messageRepository.markConversationAsRead(normalizedId).catch((e: unknown) => errorHandler(e)); // 4. 调用 API,完成后设置延迟清除保护 try { await messageService.markAsRead(normalizedId, seq); } catch (error) { - console.error('[ReadReceiptManager] 标记已读API失败:', error); + errorHandler(error); // 条件回滚(含 totalUnreadCount):仅在 my_last_read_seq 仍是本次写入值时回滚, // 避免覆盖并发 markAsRead 已写入的更大读游标或更小未读 @@ -160,7 +163,7 @@ export class ReadReceiptManager implements IReadReceiptManager { } } } catch (error) { - console.error('[ReadReceiptManager] 标记所有已读失败:', error); + errorHandler(error); // 条件回滚:仅在 conversations 的关键字段未被其他操作修改时还原, // 避免覆盖期间到达的 markAsRead / WS 推送造成的新未读 const current = useMessageStore.getState().conversations; diff --git a/src/stores/message/services/RealtimeIngestionPipeline.ts b/src/stores/message/services/RealtimeIngestionPipeline.ts index e643791..6be9a1a 100644 --- a/src/stores/message/services/RealtimeIngestionPipeline.ts +++ b/src/stores/message/services/RealtimeIngestionPipeline.ts @@ -42,6 +42,9 @@ import { } from '../constants'; import { useMessageStore, normalizeConversationId } from '../store'; import type { IUserCacheService } from '../types'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('RealtimeIngestionPipeline'); /** 实时事件统一缓冲类型(bootstrap 阶段暂存,初始化完成后回放) */ type BufferedEvent = @@ -108,7 +111,7 @@ export class RealtimeIngestionPipeline { return; } this.ingestChatMessage(message).catch(error => { - console.error('[RealtimeIngestionPipeline] ingest chat 失败:', error); + errorHandler(error); }); }) ); @@ -120,7 +123,7 @@ export class RealtimeIngestionPipeline { return; } this.ingestChatMessage(message).catch(error => { - console.error('[RealtimeIngestionPipeline] ingest group_message 失败:', error); + errorHandler(error); }); }) ); @@ -170,7 +173,7 @@ export class RealtimeIngestionPipeline { wsService.onConnect(() => { useMessageStore.getState().setSSEConnected(true); this.triggerSync('sse-reconnect').catch(error => { - console.error('[RealtimeIngestionPipeline] 重连同步失败:', error); + errorHandler(error); }); }) ); @@ -178,7 +181,7 @@ export class RealtimeIngestionPipeline { push( wsService.on('sync_required', () => { this.triggerSync('sync_required').catch(error => { - console.error('[RealtimeIngestionPipeline] sync_required 同步失败:', error); + errorHandler(error); }); }) ); @@ -198,7 +201,7 @@ export class RealtimeIngestionPipeline { try { unsub(); } catch (error) { - console.error('[RealtimeIngestionPipeline] 注销监听器失败:', error); + errorHandler(error); } } this.unsubscribers = []; @@ -320,7 +323,7 @@ export class RealtimeIngestionPipeline { // 活动会话且非本人消息:自动标记已读 if (isActiveConversation && !isCurrentUserMessage && !suppressAutoMarkAsRead) { this.deps.onActiveConvMessage(normalizedConversationId, seq).catch(error => { - console.error('[RealtimeIngestionPipeline] 自动标记已读失败:', error); + errorHandler(error); }); } @@ -340,7 +343,7 @@ export class RealtimeIngestionPipeline { useMessageStore.getState().patchMessages(normalizedConversationId, patches); }) .catch(error => { - console.error('[RealtimeIngestionPipeline] 获取发送者信息失败:', { userId: sender_id, error }); + errorHandler(error); }); } @@ -364,7 +367,7 @@ export class RealtimeIngestionPipeline { segments, }) .catch(error => { - console.error('[RealtimeIngestionPipeline] 保存消息到本地失败:', error); + errorHandler(error); }); } @@ -389,7 +392,7 @@ export class RealtimeIngestionPipeline { this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id); messageRepository.updateStatus(message_id, 'recalled', true).catch(error => { - console.error('[RealtimeIngestionPipeline] 更新本地撤回状态失败:', error); + errorHandler(error); }); } @@ -402,7 +405,7 @@ export class RealtimeIngestionPipeline { this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id); messageRepository.updateStatus(message_id, 'recalled', true).catch(error => { - console.error('[RealtimeIngestionPipeline] 更新本地撤回状态失败:', error); + errorHandler(error); }); } @@ -516,7 +519,7 @@ export class RealtimeIngestionPipeline { try { await this.deps.triggerSync(source); } catch (error) { - console.error('[RealtimeIngestionPipeline] 同步失败:', error); + errorHandler(error); } finally { this.syncInProgress = false; } diff --git a/src/stores/message/services/UserCacheService.ts b/src/stores/message/services/UserCacheService.ts index 6023f78..82f1214 100644 --- a/src/stores/message/services/UserCacheService.ts +++ b/src/stores/message/services/UserCacheService.ts @@ -8,6 +8,9 @@ import { api } from '@/services/core'; import { userCacheRepository } from '@/database'; import { useMessageStore } from '../store'; import type { IUserCacheService } from '../types'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('UserCacheService'); export class UserCacheService implements IUserCacheService { // 正在获取用户信息的请求映射(用于去重) @@ -55,7 +58,7 @@ export class UserCacheService implements IUserCacheService { } return null; } catch (error) { - console.error(`[UserCacheService] 获取用户信息失败: ${userId}`, error); + errorHandler(error); return null; } } @@ -109,7 +112,7 @@ export class UserCacheService implements IUserCacheService { notifyUpdate(conversationId, updated); }) .catch(error => { - console.error('[UserCacheService] 批量获取 sender 信息失败:', error); + errorHandler(error); }); } diff --git a/src/stores/message/sources.ts b/src/stores/message/sources.ts index 0683147..7c05e86 100644 --- a/src/stores/message/sources.ts +++ b/src/stores/message/sources.ts @@ -5,6 +5,9 @@ import { ConversationResponse } from '../../types/dto'; import { messageService } from '@/services/message'; import { conversationRepository } from '@/database'; +import { createErrorHandler } from '@/services/core'; + +const errorHandler = createErrorHandler('ConversationListSource'); export const CONVERSATION_LIST_PAGE_SIZE = 20; @@ -97,7 +100,7 @@ export class SqliteConversationListPagedSource implements IConversationListPaged const items = await conversationRepository.getListCache(); return { items, hasMore: false }; } catch (e) { - console.warn('[SqliteConversationListPagedSource] 读取会话列表缓存失败:', e); + errorHandler(e); return { items: [], hasMore: false }; } } diff --git a/src/stores/message/store.ts b/src/stores/message/store.ts index 32e0e65..d9ad636 100644 --- a/src/stores/message/store.ts +++ b/src/stores/message/store.ts @@ -11,6 +11,8 @@ import type { } from '../../types/dto'; const LAST_SYSTEM_MESSAGE_AT_KEY = 'last_system_message_at'; +/** 同步游标持久化 key(WS resume / syncByVersion 用) */ +const SYNC_VERSION_KEY = 'sync_version'; // ==================== 状态接口 ==================== @@ -53,6 +55,19 @@ export interface MessageState { */ hydratedConversations: Set; + /** + * 会话同步状态机(收敛 hydration/loading/error 的派生上层视图)。 + * + * - 'idle' 初始 / 未激活 + * - 'hydrating' 正在执行完整 hydration 管线(对应 loadingMessagesSet && !hydrated) + * - 'synced' 已完成完整 hydration(对应 hydrated) + * - 'error' hydration 失败(原无此概念,新增供 UI 展示错误态/重试) + * + * 设计为「派生层」,不删除底层 loadingMessagesSet/hydratedConversations + * (UI 仍订阅后者),由 hydration 管线同步维护,零回归风险。 + */ + conversationSyncStateMap: Map; + // 初始化状态 isInitialized: boolean; @@ -144,6 +159,10 @@ export interface MessageActions { markHydrated: (conversationId: string) => void; /** 清除某会话的 hydrated 标志(删除会话/清空记录时调用,避免残留)。 */ clearHydrated: (conversationId: string) => void; + /** 设置某会话的同步状态(hydration 管线维护)。 */ + setConversationSyncState: (conversationId: string, state: ConversationSyncState) => void; + /** 查询某会话的同步状态。 */ + getConversationSyncState: (conversationId: string) => ConversationSyncState; setTypingUsers: (groupId: string, users: string[]) => void; setMutedStatus: (groupId: string, isMuted: boolean) => void; setNotificationMuted: (conversationId: string, notificationMuted: boolean) => void; @@ -186,6 +205,7 @@ const initialState: MessageState = { isLoadingConversations: false, loadingMessagesSet: new Set(), hydratedConversations: new Set(), + conversationSyncStateMap: new Map(), isInitialized: false, typingUsersMap: new Map(), mutedStatusMap: new Map(), @@ -206,6 +226,11 @@ export function normalizeConversationId(conversationId: string | number | null | * 更新会话列表排序 * 排序规则:置顶优先,再按最后消息时间排序 */ +/** + * 会话同步状态机枚举。详见 store 的 conversationSyncStateMap 字段注释。 + */ +export type ConversationSyncState = 'idle' | 'hydrating' | 'synced' | 'error'; + function sortConversationList(conversations: Map): ConversationResponse[] { return Array.from(conversations.values()).sort((a, b) => { const aPinned = a.is_pinned ? 1 : 0; @@ -256,6 +281,7 @@ export const useMessageStore = create((set, get) => ({ isLoadingConversations: state.isLoadingConversations, loadingMessagesSet: state.loadingMessagesSet, hydratedConversations: state.hydratedConversations, + conversationSyncStateMap: state.conversationSyncStateMap, isInitialized: state.isInitialized, typingUsersMap: state.typingUsersMap, mutedStatusMap: state.mutedStatusMap, @@ -373,6 +399,10 @@ export const useMessageStore = create((set, get) => ({ const newHydratedConversations = new Set(state.hydratedConversations); newHydratedConversations.delete(normalizedId); + // 同步重置同步状态机,会话删除后回到 idle + const newSyncStateMap = new Map(state.conversationSyncStateMap); + newSyncStateMap.delete(normalizedId); + const newTotalUnreadCount = Math.max(0, state.totalUnreadCount - removedUnread); const newCurrentConversationId = state.currentConversationId === normalizedId @@ -389,6 +419,7 @@ export const useMessageStore = create((set, get) => ({ currentConversationId: newCurrentConversationId, loadingMessagesSet: newLoadingMessagesSet, hydratedConversations: newHydratedConversations, + conversationSyncStateMap: newSyncStateMap, }; }); }, @@ -467,10 +498,14 @@ export const useMessageStore = create((set, get) => ({ // 同步清除 loading 锁,避免残留锁阻断后续 fetchMessages const newLoadingMessagesSet = new Set(state.loadingMessagesSet); newLoadingMessagesSet.delete(normalizedId); + // 同步重置同步状态机,回到 idle + const newSyncStateMap = new Map(state.conversationSyncStateMap); + newSyncStateMap.delete(normalizedId); return { messagesMap: newMessagesMap, hydratedConversations: newHydratedConversations, loadingMessagesSet: newLoadingMessagesSet, + conversationSyncStateMap: newSyncStateMap, }; }); }, @@ -591,6 +626,20 @@ export const useMessageStore = create((set, get) => ({ }); }, + setConversationSyncState: (conversationId: string, syncState: ConversationSyncState) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + if (state.conversationSyncStateMap.get(normalizedId) === syncState) return state; + const next = new Map(state.conversationSyncStateMap); + next.set(normalizedId, syncState); + return { conversationSyncStateMap: next }; + }); + }, + + getConversationSyncState: (conversationId: string) => { + return get().conversationSyncStateMap.get(normalizeConversationId(conversationId)) || 'idle'; + }, + setTypingUsers: (groupId: string, users: string[]) => { set(state => { const newTypingUsersMap = new Map(state.typingUsersMap); @@ -617,6 +666,8 @@ export const useMessageStore = create((set, get) => ({ setSyncVersion: (version: number) => { set({ syncVersion: version }); + // 持久化同步游标,App 重启后用于 syncByVersion 精确增量恢复(替代全量轮询) + AsyncStorage.setItem(SYNC_VERSION_KEY, String(version)).catch(() => {}); }, setInitialized: (initialized: boolean) => { @@ -709,11 +760,13 @@ export const useMessageStore = create((set, get) => ({ messagesMap: new Map(), loadingMessagesSet: new Set(), hydratedConversations: new Set(), + conversationSyncStateMap: new Map(), typingUsersMap: new Map(), mutedStatusMap: new Map(), notificationMutedMap: new Map(), }); AsyncStorage.removeItem(LAST_SYSTEM_MESSAGE_AT_KEY).catch(() => {}); + AsyncStorage.removeItem(SYNC_VERSION_KEY).catch(() => {}); }, })); @@ -724,3 +777,19 @@ export async function loadPersistedLastSystemMessageAt(): Promise { } } +/** + * 从 AsyncStorage 恢复持久化的同步游标。 + * App 启动时调用,供 syncByVersion 做精确增量恢复(替代全量 syncBySeq 轮询)。 + */ +export async function loadPersistedSyncVersion(): Promise { + const raw = await AsyncStorage.getItem(SYNC_VERSION_KEY); + if (raw) { + const version = Number(raw); + if (!Number.isNaN(version) && version > 0) { + useMessageStore.setState({ syncVersion: version }); + return version; + } + } + return null; +} + diff --git a/src/stores/message/types.ts b/src/stores/message/types.ts index 5f36753..45b39bc 100644 --- a/src/stores/message/types.ts +++ b/src/stores/message/types.ts @@ -67,6 +67,11 @@ export interface IMessageSyncService { loadMoreMessages(conversationId: string, beforeSeq: number, limit?: number): Promise; fetchUnreadCount(): Promise; syncBySeq(): Promise; + /** + * 基于版本号的精确增量同步(优先于 syncBySeq,对标 Matrix resume 游标)。 + * @returns true=成功同步并推进游标;false=需退化到 syncBySeq 或全量 + */ + syncByVersion(version: number): Promise; canLoadMoreConversations(): boolean; restartSources(): void; }