feat(ws): implement sync_required event and message acknowledgement
Add support for a new `sync_required` WebSocket message type to trigger incremental synchronization of conversations, unread counts, and active messages. Also implement an `ack` mechanism to acknowledge received messages.
This commit is contained in:
@@ -41,7 +41,8 @@ export type WSMessageType =
|
|||||||
| 'call_ended'
|
| 'call_ended'
|
||||||
| 'call_peer_muted'
|
| 'call_peer_muted'
|
||||||
| 'call_invited'
|
| 'call_invited'
|
||||||
| 'call_answered_elsewhere';
|
| 'call_answered_elsewhere'
|
||||||
|
| 'sync_required';
|
||||||
|
|
||||||
export interface WSCallIncomingMessage {
|
export interface WSCallIncomingMessage {
|
||||||
type: 'call_incoming';
|
type: 'call_incoming';
|
||||||
@@ -249,6 +250,10 @@ export interface WSGroupRecallMessage {
|
|||||||
message_id: string;
|
message_id: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface WSSyncRequiredMessage {
|
||||||
|
type: 'sync_required';
|
||||||
|
}
|
||||||
|
|
||||||
export interface WSServerMessage {
|
export interface WSServerMessage {
|
||||||
event_id?: number;
|
event_id?: number;
|
||||||
type: string;
|
type: string;
|
||||||
@@ -279,7 +284,8 @@ export type WSMessage =
|
|||||||
| WSCallPeerMutedMessage
|
| WSCallPeerMutedMessage
|
||||||
| WSCallInvitedMessage
|
| WSCallInvitedMessage
|
||||||
| WSCallAnsweredElsewhereMessage
|
| WSCallAnsweredElsewhereMessage
|
||||||
| WSErrorMessage;
|
| WSErrorMessage
|
||||||
|
| WSSyncRequiredMessage;
|
||||||
|
|
||||||
type MessageHandler<T extends WSMessage = WSMessage> = (message: T) => void;
|
type MessageHandler<T extends WSMessage = WSMessage> = (message: T) => void;
|
||||||
type ConnectionHandler = () => void;
|
type ConnectionHandler = () => void;
|
||||||
@@ -496,6 +502,9 @@ class WebSocketService {
|
|||||||
case 'call_answered_elsewhere':
|
case 'call_answered_elsewhere':
|
||||||
this.handleCallAnsweredElsewhere(payload);
|
this.handleCallAnsweredElsewhere(payload);
|
||||||
break;
|
break;
|
||||||
|
case 'sync_required':
|
||||||
|
this.emit('sync_required', { type: 'sync_required' });
|
||||||
|
break;
|
||||||
default:
|
default:
|
||||||
console.log('Unknown message type:', type);
|
console.log('Unknown message type:', type);
|
||||||
}
|
}
|
||||||
@@ -886,6 +895,11 @@ class WebSocketService {
|
|||||||
this.ws.send(JSON.stringify(message));
|
this.ws.send(JSON.stringify(message));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 发送 ACK 确认(fire-and-forget)
|
||||||
|
sendAck(messageId: string): void {
|
||||||
|
this.sendFireAndForget('ack', { message_id: messageId });
|
||||||
|
}
|
||||||
|
|
||||||
// 通过WebSocket发送
|
// 通过WebSocket发送
|
||||||
private sendViaWebSocket(type: string, payload: any): Promise<any> {
|
private sendViaWebSocket(type: string, payload: any): Promise<any> {
|
||||||
return new Promise((resolve, reject) => {
|
return new Promise((resolve, reject) => {
|
||||||
|
|||||||
@@ -155,7 +155,7 @@ export class WSMessageHandler implements IWSMessageHandler {
|
|||||||
if (now - this.lastReconnectSyncAt > 1500) {
|
if (now - this.lastReconnectSyncAt > 1500) {
|
||||||
this.lastReconnectSyncAt = now;
|
this.lastReconnectSyncAt = now;
|
||||||
const activeConversation = useMessageStore.getState().getActiveConversation();
|
const activeConversation = useMessageStore.getState().getActiveConversation();
|
||||||
|
|
||||||
if (!activeConversation) {
|
if (!activeConversation) {
|
||||||
this.fetchConversationsCallback(true, 'sse-reconnect').catch(error => {
|
this.fetchConversationsCallback(true, 'sse-reconnect').catch(error => {
|
||||||
console.error('[WSMessageHandler] 连接后同步会话失败:', error);
|
console.error('[WSMessageHandler] 连接后同步会话失败:', error);
|
||||||
@@ -165,7 +165,7 @@ export class WSMessageHandler implements IWSMessageHandler {
|
|||||||
console.error('[WSMessageHandler] 连接后同步未读数失败:', error);
|
console.error('[WSMessageHandler] 连接后同步未读数失败:', error);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
if (activeConversation) {
|
if (activeConversation) {
|
||||||
this.fetchMessagesCallback(activeConversation).catch(error => {
|
this.fetchMessagesCallback(activeConversation).catch(error => {
|
||||||
console.error('[WSMessageHandler] 连接后同步当前会话消息失败:', error);
|
console.error('[WSMessageHandler] 连接后同步当前会话消息失败:', error);
|
||||||
@@ -174,6 +174,25 @@ export class WSMessageHandler implements IWSMessageHandler {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// 监听 sync_required 事件,触发增量同步
|
||||||
|
wsService.on('sync_required', () => {
|
||||||
|
console.log('[WSMessageHandler] 收到 sync_required,开始增量同步');
|
||||||
|
const activeConversation = useMessageStore.getState().getActiveConversation();
|
||||||
|
|
||||||
|
this.fetchConversationsCallback(true, 'sync_required').catch(error => {
|
||||||
|
console.error('[WSMessageHandler] sync_required 同步会话失败:', error);
|
||||||
|
});
|
||||||
|
this.fetchUnreadCountCallback().catch(error => {
|
||||||
|
console.error('[WSMessageHandler] sync_required 同步未读数失败:', error);
|
||||||
|
});
|
||||||
|
|
||||||
|
if (activeConversation) {
|
||||||
|
this.fetchMessagesCallback(activeConversation).catch(error => {
|
||||||
|
console.error('[WSMessageHandler] sync_required 同步消息失败:', error);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
wsService.onDisconnect(() => {
|
wsService.onDisconnect(() => {
|
||||||
useMessageStore.getState().setSSEConnected(false);
|
useMessageStore.getState().setSSEConnected(false);
|
||||||
});
|
});
|
||||||
@@ -263,6 +282,9 @@ export class WSMessageHandler implements IWSMessageHandler {
|
|||||||
}
|
}
|
||||||
this.deduplication.markMessageAsProcessed(id);
|
this.deduplication.markMessageAsProcessed(id);
|
||||||
|
|
||||||
|
// 发送 ACK 确认
|
||||||
|
wsService.sendAck(id);
|
||||||
|
|
||||||
// 对于群聊消息,获取发送者信息
|
// 对于群聊消息,获取发送者信息
|
||||||
if (message.type === 'group_message' && sender_id && sender_id !== currentUserId && sender_id !== '10000') {
|
if (message.type === 'group_message' && sender_id && sender_id !== currentUserId && sender_id !== '10000') {
|
||||||
this.userCacheService.getSenderInfo(sender_id).then(user => {
|
this.userCacheService.getSenderInfo(sender_id).then(user => {
|
||||||
|
|||||||
Reference in New Issue
Block a user