From fb67fb6d5b3d4f2fcd471b8da22ed2339833b5c7 Mon Sep 17 00:00:00 2001 From: lan Date: Sun, 17 May 2026 23:37:38 +0800 Subject: [PATCH] feat(message): implement version-based incremental sync and WS compression Implement a more efficient conversation synchronization mechanism using version numbers instead of sequence numbers to reduce bandwidth usage. This includes adding support for Gzip decompression on WebSocket messages to optimize data transfer. - Add `pako` for WebSocket message decompression - Implement `getSyncByVersion` in `MessageService` - Implement `syncByVersion` in `MessageSyncService` to handle incremental updates and conversation state changes - Update `WebSocketService` to support binary frames and Gzip inflation - Add `syncVersion` to `MessageStore` for tracking synchronization state --- package-lock.json | 15 +++++ package.json | 2 + src/services/core/wsService.ts | 19 ++++-- src/services/message/messageService.ts | 31 ++++++++++ .../message/services/MessageSyncService.ts | 59 +++++++++++++++++++ src/stores/message/store.ts | 10 ++++ 6 files changed, 132 insertions(+), 4 deletions(-) diff --git a/package-lock.json b/package-lock.json index d331536..61a3463 100644 --- a/package-lock.json +++ b/package-lock.json @@ -43,6 +43,7 @@ "jpush-react-native": "^3.2.6", "katex": "^0.16.42", "markdown-it": "^14.1.1", + "pako": "^2.1.0", "prismjs": "^1.30.0", "react": "19.2.0", "react-dom": "19.2.0", @@ -64,6 +65,7 @@ }, "devDependencies": { "@react-native-community/cli": "^20.1.2", + "@types/pako": "^2.0.4", "@types/react": "~19.2.2", "@types/react-native-vector-icons": "^6.4.18", "typescript": "~5.9.2" @@ -3845,6 +3847,13 @@ "undici-types": "~7.18.0" } }, + "node_modules/@types/pako": { + "version": "2.0.4", + "resolved": "https://registry.npmmirror.com/@types/pako/-/pako-2.0.4.tgz", + "integrity": "sha512-VWDCbrLeVXJM9fihYodcLiIv0ku+AlOa/TQ1SvYOaBuyrSKgEcro95LJyIsJ4vSo6BXIxOKxiJAat04CmST9Fw==", + "dev": true, + "license": "MIT" + }, "node_modules/@types/react": { "version": "19.2.14", "resolved": "https://registry.npmmirror.com/@types/react/-/react-19.2.14.tgz", @@ -9082,6 +9091,12 @@ "node": ">=6" } }, + "node_modules/pako": { + "version": "2.1.0", + "resolved": "https://registry.npmmirror.com/pako/-/pako-2.1.0.tgz", + "integrity": "sha512-w+eufiZ1WuJYgPXbV/PO3NCMEc3xqylkKHzp8bxp1uW4qaSNQUkwmLLEc3kKsfz8lpV1F8Ht3U1Cm+9Srog2ug==", + "license": "(MIT AND Zlib)" + }, "node_modules/parent-module": { "version": "1.0.1", "resolved": "https://registry.npmmirror.com/parent-module/-/parent-module-1.0.1.tgz", diff --git a/package.json b/package.json index 0286628..7b81c05 100644 --- a/package.json +++ b/package.json @@ -49,6 +49,7 @@ "jpush-react-native": "^3.2.6", "katex": "^0.16.42", "markdown-it": "^14.1.1", + "pako": "^2.1.0", "prismjs": "^1.30.0", "react": "19.2.0", "react-dom": "19.2.0", @@ -70,6 +71,7 @@ }, "devDependencies": { "@react-native-community/cli": "^20.1.2", + "@types/pako": "^2.0.4", "@types/react": "~19.2.2", "@types/react-native-vector-icons": "^6.4.18", "typescript": "~5.9.2" diff --git a/src/services/core/wsService.ts b/src/services/core/wsService.ts index 5d41b12..b2e243e 100644 --- a/src/services/core/wsService.ts +++ b/src/services/core/wsService.ts @@ -1,3 +1,5 @@ +import pako from 'pako'; + import { AppState, AppStateStatus } from 'react-native'; import { api, WS_URL } from './api'; @@ -329,7 +331,7 @@ class WebSocketService { private fallbackCheckTimer: ReturnType | null = null; private getWSUrl(token: string | null): string { - return `${WS_URL}?token=${encodeURIComponent(token || '')}`; + return `${WS_URL}?token=${encodeURIComponent(token || '')}&compress=1`; } async connect(): Promise { @@ -354,6 +356,7 @@ class WebSocketService { console.log('[WebSocket] Connecting to:', url.replace(token, '***')); this.ws = new WebSocket(url); + this.ws.binaryType = 'arraybuffer'; this.ws.onopen = () => { console.log('[WebSocket] Connected successfully'); @@ -392,9 +395,17 @@ class WebSocketService { } } - private handleIncoming(data: string): void { + private handleIncoming(data: string | ArrayBuffer): void { + let textData: string; + if (typeof data === 'string') { + textData = data; + } else { + // Binary frame — gzip decompress + const decompressed = pako.inflate(new Uint8Array(data as ArrayBuffer)); + textData = new TextDecoder('utf-8').decode(decompressed); + } // 后端 writePump 会将多条消息以 \n 分隔合并发送，需要逐条解析 - const lines = data.split('\n'); + const lines = textData.split('\n'); for (const line of lines) { const trimmed = line.trim(); if (!trimmed) continue; @@ -402,7 +413,7 @@ class WebSocketService { const msg: WSServerMessage = JSON.parse(trimmed); this.processMessage(msg); } catch (error) { - console.error('Failed to parse WebSocket message:', error, 'raw:', trimmed); + console.error('Failed to parse WebSocket message:', error); } } } diff --git a/src/services/message/messageService.ts b/src/services/message/messageService.ts index 2cacf0f..e42ff60 100644 --- a/src/services/message/messageService.ts +++ b/src/services/message/messageService.ts @@ -476,6 +476,37 @@ class MessageService { return response.data?.conversations || []; } + /** + * 增量同步（基于版本号） + * GET /api/v1/conversations/sync?version=N&limit=100 + */ + async getSyncByVersion(version: number, limit: number = 100): Promise<{ + current_version: number; + changes: Array<{ + conversation_id: string; + change_type: string; + max_seq?: number; + last_message_at?: number; + version: number; + }>; + full_sync: boolean; + has_more: boolean; + }> { + const response = await api.get<{ + current_version: number; + changes: Array<{ + conversation_id: string; + change_type: string; + max_seq?: number; + last_message_at?: number; + version: number; + }>; + full_sync: boolean; + has_more: boolean; + }>('/conversations/sync', { version, limit }); + return response.data; + } + /** * 获取系统消息未读数 * GET /api/v1/messages/system/unread-count diff --git a/src/stores/message/services/MessageSyncService.ts b/src/stores/message/services/MessageSyncService.ts index 117b99e..7ec4043 100644 --- a/src/stores/message/services/MessageSyncService.ts +++ b/src/stores/message/services/MessageSyncService.ts @@ -466,6 +466,65 @@ export class MessageSyncService implements IMessageSyncService { } } + /** + * 基于版本号的增量同步（替代 syncBySeq，减少带宽） + * 需要本地存储 syncVersion，首次同步时 version=0 会触发 full_sync + */ + async syncByVersion(): Promise { + try { + const store = useMessageStore.getState(); + const version = store.syncVersion ?? 0; + + const result = await messageService.getSyncByVersion(version, 100); + if (!result) return false; + + // full_sync 表示需要全量刷新 + if (result.full_sync) { + return false; + } + + if (result.changes.length === 0) { + // 保存当前版本号，下次增量同步时使用 + useMessageStore.getState().setSyncVersion(result.current_version); + return true; + } + + // 按变更类型处理 + const staleIds: string[] = []; + for (const change of result.changes) { + const normId = normalizeConversationId(change.conversation_id); + if (change.change_type === 'hide') { + // 用户隐藏了会话，从本地移除 + store.removeConversation(normId); + } else { + // new_msg, read, pin, unpin, mute, unmute, group_update + staleIds.push(normId); + } + } + + // 刷新有变更的会话 + if (staleIds.length > 0) { + const detailPromises = staleIds.map(id => + this.fetchConversationDetail(id).catch(() => {}) + ); + await Promise.all(detailPromises); + } + + // 保存版本号 + useMessageStore.getState().setSyncVersion(result.current_version); + + // has_more 时继续拉取 + if (result.has_more) { + return this.syncByVersion(); + } + + return true; + } catch (error) { + console.error('[MessageSyncService] syncByVersion 失败:', error); + return false; + } + } + /** * 检查是否可加载更多会话 */ diff --git a/src/stores/message/store.ts b/src/stores/message/store.ts index 1530c4a..8a9523c 100644 --- a/src/stores/message/store.ts +++ b/src/stores/message/store.ts @@ -53,6 +53,9 @@ export interface MessageState { // 会话免打扰状态 - 按会话ID存储 notificationMutedMap: Map; + + // 版本日志同步游标 + syncVersion: number | null; } // ==================== Actions 接口 ==================== @@ -90,6 +93,7 @@ export interface MessageActions { setTypingUsers: (groupId: string, users: string[]) => void; setMutedStatus: (groupId: string, isMuted: boolean) => void; setNotificationMuted: (conversationId: string, notificationMuted: boolean) => void; + setSyncVersion: (version: number) => void; setInitialized: (initialized: boolean) => void; // 原子性更新（避免中间渲染状态） @@ -121,6 +125,7 @@ const initialState: MessageState = { typingUsersMap: new Map(), mutedStatusMap: new Map(), notificationMutedMap: new Map(), + syncVersion: null, }; // ==================== 工具函数 ==================== @@ -189,6 +194,7 @@ export const useMessageStore = create((set, get) => ({ typingUsersMap: state.typingUsersMap, mutedStatusMap: state.mutedStatusMap, notificationMutedMap: state.notificationMutedMap, + syncVersion: state.syncVersion, }; }, @@ -387,6 +393,10 @@ export const useMessageStore = create((set, get) => ({ }); }, + setSyncVersion: (version: number) => { + set({ syncVersion: version }); + }, + setInitialized: (initialized: boolean) => { set({ isInitialized: initialized }); },