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
This commit is contained in:
15
package-lock.json
generated
15
package-lock.json
generated
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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<typeof setTimeout> | 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<boolean> {
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -466,6 +466,65 @@ export class MessageSyncService implements IMessageSyncService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 基于版本号的增量同步(替代 syncBySeq,减少带宽)
|
||||
* 需要本地存储 syncVersion,首次同步时 version=0 会触发 full_sync
|
||||
*/
|
||||
async syncByVersion(): Promise<boolean> {
|
||||
try {
|
||||
const store = useMessageStore.getState();
|
||||
const version = store.syncVersion ?? 0;
|
||||
|
||||
const result = await messageService.getSyncByVersion(version, 100);
|
||||
if (!result) return false;
|
||||
|
||||
// full_sync 表示需要全量刷新
|
||||
if (result.full_sync) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (result.changes.length === 0) {
|
||||
// 保存当前版本号,下次增量同步时使用
|
||||
useMessageStore.getState().setSyncVersion(result.current_version);
|
||||
return true;
|
||||
}
|
||||
|
||||
// 按变更类型处理
|
||||
const staleIds: string[] = [];
|
||||
for (const change of result.changes) {
|
||||
const normId = normalizeConversationId(change.conversation_id);
|
||||
if (change.change_type === 'hide') {
|
||||
// 用户隐藏了会话,从本地移除
|
||||
store.removeConversation(normId);
|
||||
} else {
|
||||
// new_msg, read, pin, unpin, mute, unmute, group_update
|
||||
staleIds.push(normId);
|
||||
}
|
||||
}
|
||||
|
||||
// 刷新有变更的会话
|
||||
if (staleIds.length > 0) {
|
||||
const detailPromises = staleIds.map(id =>
|
||||
this.fetchConversationDetail(id).catch(() => {})
|
||||
);
|
||||
await Promise.all(detailPromises);
|
||||
}
|
||||
|
||||
// 保存版本号
|
||||
useMessageStore.getState().setSyncVersion(result.current_version);
|
||||
|
||||
// has_more 时继续拉取
|
||||
if (result.has_more) {
|
||||
return this.syncByVersion();
|
||||
}
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
console.error('[MessageSyncService] syncByVersion 失败:', error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查是否可加载更多会话
|
||||
*/
|
||||
|
||||
@@ -53,6 +53,9 @@ export interface MessageState {
|
||||
|
||||
// 会话免打扰状态 - 按会话ID存储
|
||||
notificationMutedMap: Map<string, boolean>;
|
||||
|
||||
// 版本日志同步游标
|
||||
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<MessageStore>((set, get) => ({
|
||||
typingUsersMap: state.typingUsersMap,
|
||||
mutedStatusMap: state.mutedStatusMap,
|
||||
notificationMutedMap: state.notificationMutedMap,
|
||||
syncVersion: state.syncVersion,
|
||||
};
|
||||
},
|
||||
|
||||
@@ -387,6 +393,10 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
|
||||
});
|
||||
},
|
||||
|
||||
setSyncVersion: (version: number) => {
|
||||
set({ syncVersion: version });
|
||||
},
|
||||
|
||||
setInitialized: (initialized: boolean) => {
|
||||
set({ isInitialized: initialized });
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user