diff --git a/src/core/usecases/ProcessMessageUseCase.ts b/src/core/usecases/ProcessMessageUseCase.ts index aaab9a0..45c7f26 100644 --- a/src/core/usecases/ProcessMessageUseCase.ts +++ b/src/core/usecases/ProcessMessageUseCase.ts @@ -5,7 +5,7 @@ */ import { messageRepository } from '../../data/repositories/MessageRepository'; -import { sseClient, SSEEventType } from '../../data/datasources/SSEClient'; +import { wsClient, WSEventType } from '../../data/datasources/WSClient'; import { Message, Conversation, @@ -62,7 +62,7 @@ class ProcessMessageUseCase { */ initialize(currentUserId: string | null): void { this.currentUserId = currentUserId; - this.setupSSEListeners(); + this.setupWSListeners(); } /** @@ -81,49 +81,49 @@ class ProcessMessageUseCase { /** * 设置SSE监听器 */ - private setupSSEListeners(): void { + private setupWSListeners(): void { // 监听私聊消息 - const unsubChat = sseClient.on('chat', (message) => { + const unsubChat = wsClient.on('chat', (message) => { this.handleNewMessage(message); }); // 监听群聊消息 - const unsubGroupMessage = sseClient.on('group_message', (message) => { + const unsubGroupMessage = wsClient.on('group_message', (message) => { this.handleNewMessage(message); }); // 监听私聊已读回执 - const unsubRead = sseClient.on('read', (message) => { + const unsubRead = wsClient.on('read', (message) => { this.handleReadReceipt(message); }); // 监听群聊已读回执 - const unsubGroupRead = sseClient.on('group_read', (message) => { + const unsubGroupRead = wsClient.on('group_read', (message) => { this.handleGroupReadReceipt(message); }); // 监听私聊消息撤回 - const unsubRecall = sseClient.on('recall', (message) => { + const unsubRecall = wsClient.on('recall', (message) => { this.handleRecallMessage(message); }); // 监听群聊消息撤回 - const unsubGroupRecall = sseClient.on('group_recall', (message) => { + const unsubGroupRecall = wsClient.on('group_recall', (message) => { this.handleGroupRecallMessage(message); }); // 监听群聊输入状态 - const unsubGroupTyping = sseClient.on('group_typing', (message) => { + const unsubGroupTyping = wsClient.on('group_typing', (message) => { this.handleGroupTyping(message); }); // 监听群通知 - const unsubGroupNotice = sseClient.on('group_notice', (message) => { + const unsubGroupNotice = wsClient.on('group_notice', (message) => { this.handleGroupNotice(message); }); // 监听连接状态 - const unsubConnected = sseClient.on('connected', () => { + const unsubConnected = wsClient.on('connected', () => { this.notifySubscribers({ type: 'connection_changed', payload: { connected: true }, @@ -131,7 +131,7 @@ class ProcessMessageUseCase { }); }); - const unsubDisconnected = sseClient.on('disconnected', () => { + const unsubDisconnected = wsClient.on('disconnected', () => { this.notifySubscribers({ type: 'connection_changed', payload: { connected: false }, diff --git a/src/data/datasources/SSEClient.ts b/src/data/datasources/WSClient.ts similarity index 64% rename from src/data/datasources/SSEClient.ts rename to src/data/datasources/WSClient.ts index ae955ac..598bf04 100644 --- a/src/data/datasources/SSEClient.ts +++ b/src/data/datasources/WSClient.ts @@ -1,10 +1,10 @@ /** - * SSEClient - SSE连接管理 - * 只负责SSE连接管理，提供事件驱动架构 + * WSClient - WebSocket连接管理 + * 只负责WebSocket连接管理，提供事件驱动架构 */ import { - sseService, + wsService, WSChatMessage, WSGroupChatMessage, WSReadMessage, @@ -14,13 +14,13 @@ import { WSGroupTypingMessage, WSGroupNoticeMessage, WSMessageType, -} from '../../services/sseService'; +} from '../../services/wsService'; // 事件处理器类型 type EventHandler = (data: T) => void; -// SSE事件类型 -export interface SSEEvents { +// WS事件类型 +export interface WSEvents { 'chat': WSChatMessage; 'group_message': WSGroupChatMessage; 'read': WSReadMessage; @@ -34,65 +34,65 @@ export interface SSEEvents { 'error': Error; } -export type SSEEventType = keyof SSEEvents; +export type WSEventType = keyof WSEvents; -class SSEClient { - private handlers: Map>> = new Map(); +class WSClient { + private handlers: Map>> = new Map(); private unsubscribeFns: Array<() => void> = []; private isInitialized = false; /** - * 初始化SSE监听 + * 初始化WebSocket监听 */ initialize(): void { if (this.isInitialized) return; // 监听私聊消息 - const unsubChat = sseService.on('chat', (message) => { + const unsubChat = wsService.on('chat', (message) => { this.emit('chat', message); }); // 监听群聊消息 - const unsubGroupMessage = sseService.on('group_message', (message) => { + const unsubGroupMessage = wsService.on('group_message', (message) => { this.emit('group_message', message); }); // 监听私聊已读回执 - const unsubRead = sseService.on('read', (message) => { + const unsubRead = wsService.on('read', (message) => { this.emit('read', message); }); // 监听群聊已读回执 - const unsubGroupRead = sseService.on('group_read', (message) => { + const unsubGroupRead = wsService.on('group_read', (message) => { this.emit('group_read', message); }); // 监听私聊消息撤回 - const unsubRecall = sseService.on('recall', (message) => { + const unsubRecall = wsService.on('recall', (message) => { this.emit('recall', message); }); // 监听群聊消息撤回 - const unsubGroupRecall = sseService.on('group_recall', (message) => { + const unsubGroupRecall = wsService.on('group_recall', (message) => { this.emit('group_recall', message); }); // 监听群聊输入状态 - const unsubGroupTyping = sseService.on('group_typing', (message) => { + const unsubGroupTyping = wsService.on('group_typing', (message) => { this.emit('group_typing', message); }); // 监听群通知 - const unsubGroupNotice = sseService.on('group_notice', (message) => { + const unsubGroupNotice = wsService.on('group_notice', (message) => { this.emit('group_notice', message); }); // 监听连接状态 - const unsubConnect = sseService.onConnect(() => { + const unsubConnect = wsService.onConnect(() => { this.emit('connected', undefined); }); - const unsubDisconnect = sseService.onDisconnect(() => { + const unsubDisconnect = wsService.onDisconnect(() => { this.emit('disconnected', undefined); }); @@ -113,7 +113,7 @@ class SSEClient { } /** - * 销毁SSE监听 + * 销毁WebSocket监听 */ destroy(): void { this.unsubscribeFns.forEach((fn) => fn()); @@ -125,9 +125,9 @@ class SSEClient { /** * 订阅事件 */ - on( + on( event: T, - handler: EventHandler + handler: EventHandler ): () => void { if (!this.handlers.has(event)) { this.handlers.set(event, new Set()); @@ -142,9 +142,9 @@ class SSEClient { /** * 触发事件 */ - private emit( + private emit( event: T, - data: SSEEvents[T] + data: WSEvents[T] ): void { const handlers = this.handlers.get(event); if (handlers) { @@ -152,7 +152,7 @@ class SSEClient { try { handler(data); } catch (error) { - console.error(`[SSEClient] 事件处理器执行失败: ${event}`, error); + console.error(`[WSClient] 事件处理器执行失败: ${event}`, error); } }); } @@ -162,23 +162,23 @@ class SSEClient { * 检查连接状态 */ isConnected(): boolean { - return sseService.isConnected(); + return wsService.isConnected(); } /** - * 启动SSE连接 + * 启动WebSocket连接 */ async connect(): Promise { - return sseService.start(); + return wsService.start(); } /** - * 断开SSE连接 + * 断开WebSocket连接 */ disconnect(): void { - sseService.stop(); + wsService.stop(); } } -export const sseClient = new SSEClient(); -export default sseClient; +export const wsClient = new WSClient(); +export default wsClient; \ No newline at end of file diff --git a/src/services/api.ts b/src/services/api.ts index 75e475c..74a02a2 100644 --- a/src/services/api.ts +++ b/src/services/api.ts @@ -20,7 +20,7 @@ const getBaseUrl = () => { }; const BASE_URL = getBaseUrl(); -const SSE_URL = `${BASE_URL.replace(/\/+$/, '')}/realtime/sse`; +const WS_URL = `${BASE_URL.replace(/^https?:\/\//, 'wss://').replace(/\/api\/v1$/, '/api/v1')}/realtime/ws`; // Token 存储键 const TOKEN_KEY = 'auth_token'; @@ -441,4 +441,4 @@ class ApiClient { // 导出 API 客户端实例 export const api = new ApiClient(BASE_URL); -export { SSE_URL, TOKEN_KEY, REFRESH_TOKEN_KEY }; +export { WS_URL, TOKEN_KEY, REFRESH_TOKEN_KEY }; diff --git a/src/services/backgroundService.ts b/src/services/backgroundService.ts index c473084..9850a02 100644 --- a/src/services/backgroundService.ts +++ b/src/services/backgroundService.ts @@ -11,7 +11,7 @@ import { AppState, AppStateStatus, Platform } from 'react-native'; import * as BackgroundFetch from 'expo-background-fetch'; import * as TaskManager from 'expo-task-manager'; -import { sseService } from './sseService'; +import { wsService } from './wsService'; import { triggerVibration, vibrateOnMessage, @@ -36,8 +36,8 @@ let appStateSubscription: any = null; TaskManager.defineTask(BACKGROUND_FETCH_TASK, async () => { try { // 检查 SSE 连接状态 - if (!sseService.isConnected()) { - await sseService.connect(); + if (!wsService.isConnected()) { + await wsService.connect(); } // 返回收到新数据 @@ -51,8 +51,8 @@ TaskManager.defineTask(BACKGROUND_FETCH_TASK, async () => { // SSE 保活任务 TaskManager.defineTask(REALTIME_KEEPALIVE_TASK, async () => { try { - if (!sseService.isConnected()) { - await sseService.connect(); + if (!wsService.isConnected()) { + await wsService.connect(); } return BackgroundFetch.BackgroundFetchResult.NewData; } catch (error) { @@ -120,8 +120,8 @@ function setupAppStateListener(): void { void nextAppState; if (nextAppState === 'active') { // App 回到前台，确保连接 - if (!sseService.isConnected()) { - sseService.connect(); + if (!wsService.isConnected()) { + wsService.connect(); } } }); diff --git a/src/services/index.ts b/src/services/index.ts index 51a2a01..0982e3c 100644 --- a/src/services/index.ts +++ b/src/services/index.ts @@ -4,7 +4,7 @@ */ // API 客户端 -export { api, SSE_URL, TOKEN_KEY, REFRESH_TOKEN_KEY } from './api'; +export { api, WS_URL, TOKEN_KEY, REFRESH_TOKEN_KEY } from './api'; export type { ApiResponse, PaginatedData, ApiError } from './api'; // 认证服务 @@ -47,8 +47,8 @@ export { voteService } from './voteService'; export { channelService } from './channelService'; export type { ChannelItem } from './channelService'; -// SSE 实时服务 -export { sseService } from './sseService'; +// WebSocket 实时服务 +export { wsService } from './wsService'; export type { WSMessage, WSMessageType, @@ -65,7 +65,7 @@ export type { WSGroupMentionMessage, WSGroupReadMessage, WSGroupRecallMessage -} from './sseService'; +} from './wsService'; // 系统通知服务 export { systemNotificationService, getNotificationTitle } from './systemNotificationService'; diff --git a/src/services/messageService.ts b/src/services/messageService.ts index 631b18c..48d165f 100644 --- a/src/services/messageService.ts +++ b/src/services/messageService.ts @@ -5,6 +5,7 @@ */ import { api } from './api'; +import { wsService } from './wsService'; import { ConversationListResponse, ConversationResponse, @@ -344,6 +345,7 @@ class MessageService { /** * 发送消息（新格式） + * 优先使用WebSocket发送，失败时降级到HTTP * POST /api/v1/conversations/:id/messages * Body: { detail_type, segments } */ @@ -357,12 +359,25 @@ class MessageService { throw new Error('消息内容不能为空'); } - return this.sendMessageByAction( - detailType, - conversationId, - data.segments, - data.reply_to_id - ); + // 优先使用WebSocket发送 + try { + const result = await wsService.sendMessage( + conversationId, + detailType, + data.segments, + data.reply_to_id + ); + return result; + } catch (error) { + console.log('WebSocket发送失败，降级到HTTP:', error); + // 降级到HTTP发送 + return this.sendMessageByAction( + detailType, + conversationId, + data.segments, + data.reply_to_id + ); + } } /** @@ -494,13 +509,19 @@ class MessageService { /** * 撤回/删除消息 + * 优先使用WebSocket，失败时降级到HTTP * POST /api/v1/messages/delete_msg * Body: { "message_id": "xxx" } */ async recallMessage(messageId: string): Promise { - await api.post('/messages/delete_msg', { - message_id: messageId, - }); + try { + await wsService.recallMessage(messageId); + } catch (error) { + console.log('WebSocket撤回失败，降级到HTTP:', error); + await api.post('/messages/delete_msg', { + message_id: messageId, + }); + } } /** @@ -515,11 +536,16 @@ class MessageService { /** * 标记已读 + * 优先使用WebSocket，失败时降级到HTTP * POST /api/v1/conversations/:id/read * @param conversationId 会话ID * @param seq 已读到的消息序号 */ async markAsRead(conversationId: string, seq: number): Promise { + // 使用WebSocket发送已读标记（不等待响应） + wsService.markRead(conversationId, seq).catch(() => {}); + + // 同时通过HTTP确保可靠性 await api.post(`/conversations/${encodeURIComponent(conversationId)}/read`, { last_read_seq: Number(seq), }); @@ -531,9 +557,14 @@ class MessageService { /** * 上报输入状态 + * 优先使用WebSocket，失败时降级到HTTP * POST /api/v1/conversations/:id/typing */ async sendTyping(conversationId: string): Promise { + // 使用WebSocket发送输入状态 + wsService.sendTyping(conversationId, true); + + // 同时通过HTTP确保可靠性 await api.post(`/conversations/${encodeURIComponent(conversationId)}/typing`); } diff --git a/src/services/sseService.ts b/src/services/sseService.ts deleted file mode 100644 index 64bbe44..0000000 --- a/src/services/sseService.ts +++ /dev/null @@ -1,517 +0,0 @@ -import { AppState, AppStateStatus } from 'react-native'; -import EventSource from 'react-native-sse'; - -import { api, SSE_URL } from './api'; -import { MessageCategory, SystemMessageType, SystemMessageExtraData, MessageSegment } from '../types/dto'; -import { systemNotificationService } from './systemNotificationService'; - -export type WSMessageType = - | 'chat' - | 'message' - | 'read' - | 'typing' - | 'recall' - | 'notification' - | 'announcement' - | 'group_message' - | 'group_typing' - | 'group_notice' - | 'group_mention' - | 'group_read' - | 'group_recall' - | 'notice' - | 'request' - | 'meta' - | 'private' - | 'group' - | 'follow' - | 'like' - | 'comment' - | 'heartbeat'; - -export interface WSChatMessage { - type: 'chat'; - conversation_id: string; - id: string; - sender_id: string; - seq: number; - segments?: MessageSegment[]; - created_at: string; -} - -export interface WSReadMessage { - type: 'read'; - conversation_id: string; - user_id: string; - seq: number; -} - -export interface WSTypingMessage { - type: 'typing'; - conversation_id: string; - user_id: string; - is_typing: boolean; -} - -export interface WSRecallMessage { - type: 'recall'; - conversation_id: string; - message_id: string; -} - -export interface WSNotificationMessage { - type: 'notification'; - id: string; - sender_id?: string; - receiver_id?: string; - content: string; - category?: MessageCategory; - system_type?: SystemMessageType; - extra_data?: SystemMessageExtraData; - created_at: string; -} - -export interface WSAnnouncementMessage { - type: 'announcement'; - id: string; - sender_id?: string; - receiver_id?: string; - content: string; - category?: MessageCategory; - system_type?: SystemMessageType; - extra_data?: SystemMessageExtraData; - created_at: string; -} - -export interface WSGroupMentionMessage { - type: 'group_mention'; - group_id: string; - conversation_id: string; - message_id: string; - from_user_id: string; - content: string; - mention_all: boolean; - created_at: string; -} - -export interface WSGroupChatMessage { - type: 'group_message'; - conversation_id: string; - group_id: string; - id: string; - sender_id: string; - seq: number; - segments?: MessageSegment[]; - created_at: string; -} - -export interface WSGroupTypingMessage { - type: 'group_typing'; - group_id: string; - user_id: string; - is_typing: boolean; -} - -export type GroupNoticeType = 'member_join' | 'member_leave' | 'member_removed' | 'role_changed' | 'muted' | 'unmuted'; - -export interface WSGroupNoticeMessage { - type: 'group_notice'; - notice_type: GroupNoticeType; - group_id: string; - data: { - user_id?: string; - operator_id?: string; - role?: string; - [key: string]: any; - }; - timestamp: number; - message_id?: string; - seq?: number; -} - -export interface WSGroupReadMessage { - type: 'group_read'; - group_id: string; - conversation_id: string; - user_id: string; - seq: number; -} - -export interface WSGroupRecallMessage { - type: 'group_recall'; - group_id: string; - conversation_id: string; - message_id: string; -} - -export type WSMessage = - | WSChatMessage - | WSReadMessage - | WSTypingMessage - | WSRecallMessage - | WSNotificationMessage - | WSAnnouncementMessage - | WSGroupChatMessage - | WSGroupTypingMessage - | WSGroupNoticeMessage - | WSGroupMentionMessage - | WSGroupReadMessage - | WSGroupRecallMessage; - -type MessageHandler = (message: T) => void; -type ConnectionHandler = () => void; - -interface SSEEnvelope { - event_id?: number; - event?: string; - ts?: number; - payload?: any; -} - -class SSEService { - private source: EventSource | null = null; - private isConnecting = false; - private connected = false; - private reconnectAttempts = 0; - private maxReconnectAttempts = 20; - private reconnectDelay = 3000; - private reconnectTimer: NodeJS.Timeout | null = null; - private connectionWatchdogTimer: NodeJS.Timeout | null = null; - private readonly CONNECTION_IDLE_TIMEOUT_MS = 70000; - private messageHandlers: Map = new Map(); - private connectionHandlers: ConnectionHandler[] = []; - private disconnectionHandlers: ConnectionHandler[] = []; - private appStateSubscription: any = null; - private lastAppState: AppStateStatus = 'active'; - private lastEventId = ''; - private lastActivityAt = 0; - private shouldRun = false; - - private toSSEUrl(): string { - return `${SSE_URL}?last_event_id=${encodeURIComponent(this.lastEventId)}`; - } - - async connect(): Promise { - this.shouldRun = true; - if (this.isConnecting || this.isConnected()) return true; - this.isConnecting = true; - try { - const token = await api.getToken(); - if (!token) { - this.isConnecting = false; - return false; - } - const url = this.toSSEUrl(); - this.source = new EventSource(url, { - headers: { - Authorization: `Bearer ${token}`, - }, - // Use one reconnect strategy only (manual), avoid library auto-reconnect fighting with us. - pollingInterval: 0, - // Be explicit to prevent platform-level short idle timeout behavior. - timeout: this.CONNECTION_IDLE_TIMEOUT_MS, - }); - - this.source.addEventListener('open', () => { - this.isConnecting = false; - this.connected = true; - this.reconnectAttempts = 0; - this.markActivity(); - this.startConnectionWatchdog(); - this.connectionHandlers.forEach(h => h()); - }); - - this.source.addEventListener('error', () => { - this.handleDisconnected(); - }); - - this.source.addEventListener('done' as any, () => { - this.handleDisconnected(); - }); - - this.source.addEventListener('close' as any, () => { - this.handleDisconnected(); - }); - - const events = ['chat_message', 'message_read', 'typing', 'system_notification', 'group_notice', 'message_recall', 'heartbeat']; - events.forEach(eventName => { - this.source?.addEventListener(eventName as any, (evt: any) => this.handleIncoming(eventName, evt)); - }); - - return true; - } catch { - this.isConnecting = false; - this.scheduleReconnect(); - return false; - } - } - - private handleIncoming(eventName: string, evt: any): void { - this.markActivity(); - const rawData = typeof evt?.data === 'string' ? evt.data : '{}'; - const lastEventId = evt?.lastEventId; - if (lastEventId) { - this.lastEventId = String(lastEventId); - } - let payload: any = {}; - try { - payload = JSON.parse(rawData); - } catch { - payload = {}; - } - this.dispatchEvent(eventName, payload); - } - - private dispatchEvent(eventName: string, payload: any): void { - if (eventName === 'chat_message') { - const detailType = payload?.detail_type || 'private'; - const m = payload?.message || payload; - if (detailType === 'group') { - const gm: WSGroupChatMessage = { - type: 'group_message', - conversation_id: m.conversation_id, - group_id: String(m.group_id ?? ''), - id: m.id, - sender_id: m.sender_id, - seq: Number(m.seq || 0), - segments: m.segments || [], - created_at: m.created_at || new Date().toISOString(), - }; - this.emit('group_message', gm); - } else { - const cm: WSChatMessage = { - type: 'chat', - conversation_id: m.conversation_id, - id: m.id, - sender_id: m.sender_id, - seq: Number(m.seq || 0), - segments: m.segments || [], - created_at: m.created_at || new Date().toISOString(), - }; - this.emit('chat', cm); - } - return; - } - - if (eventName === 'message_read') { - const detailType = payload?.detail_type || 'private'; - if (detailType === 'group') { - const m: WSGroupReadMessage = { - type: 'group_read', - group_id: String(payload.group_id ?? ''), - conversation_id: payload.conversation_id, - user_id: payload.user_id, - seq: Number(payload.seq || 0), - }; - this.emit('group_read', m); - } else { - const m: WSReadMessage = { - type: 'read', - conversation_id: payload.conversation_id, - user_id: payload.user_id, - seq: Number(payload.seq || 0), - }; - this.emit('read', m); - } - return; - } - - if (eventName === 'typing') { - const detailType = payload?.detail_type || 'private'; - if (detailType === 'group') { - const m: WSGroupTypingMessage = { - type: 'group_typing', - group_id: String(payload.group_id ?? ''), - user_id: payload.user_id, - is_typing: payload.is_typing !== false, - }; - this.emit('group_typing', m); - } else { - const m: WSTypingMessage = { - type: 'typing', - conversation_id: payload.conversation_id, - user_id: payload.user_id, - is_typing: payload.is_typing !== false, - }; - this.emit('typing', m); - } - return; - } - - if (eventName === 'message_recall') { - const detailType = payload?.detail_type || 'private'; - if (detailType === 'group') { - const m: WSGroupRecallMessage = { - type: 'group_recall', - group_id: String(payload.group_id ?? ''), - conversation_id: payload.conversation_id, - message_id: payload.message_id, - }; - this.emit('group_recall', m); - } else { - const m: WSRecallMessage = { - type: 'recall', - conversation_id: payload.conversation_id, - message_id: payload.message_id, - }; - this.emit('recall', m); - } - return; - } - - if (eventName === 'group_notice') { - const m: WSGroupNoticeMessage = { - type: 'group_notice', - notice_type: payload.notice_type, - group_id: String(payload.group_id ?? ''), - data: payload.data || {}, - timestamp: payload.timestamp || Date.now(), - message_id: payload.message_id, - seq: payload.seq, - }; - this.emit('group_notice', m); - return; - } - - if (eventName === 'system_notification') { - const m: WSNotificationMessage = { - type: 'notification', - id: String(payload.id ?? ''), - content: payload.content || '', - created_at: payload.created_at || new Date().toISOString(), - }; - this.emit('notification', m); - systemNotificationService.handleWSMessage(m as any).catch(() => {}); - } - } - - private emit(type: T, message: Extract) { - const handlers = this.messageHandlers.get(type) || []; - handlers.forEach(h => h(message as WSMessage)); - } - - disconnect(): void { - this.shouldRun = false; - this.connected = false; - this.isConnecting = false; - if (this.source) { - this.source.close(); - this.source = null; - } - this.stopConnectionWatchdog(); - this.stopReconnect(); - } - - private scheduleReconnect(): void { - if (this.reconnectAttempts >= this.maxReconnectAttempts) return; - this.stopReconnect(); - this.reconnectTimer = setTimeout(() => { - this.reconnectAttempts += 1; - this.connect(); - }, this.reconnectDelay); - } - - private stopReconnect() { - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } - } - - isConnected(): boolean { - return this.connected; - } - - on(type: T, handler: MessageHandler>): () => void { - const list = this.messageHandlers.get(type) || []; - list.push(handler as MessageHandler); - this.messageHandlers.set(type, list); - return () => { - const current = this.messageHandlers.get(type) || []; - const idx = current.indexOf(handler as MessageHandler); - if (idx >= 0) current.splice(idx, 1); - }; - } - - onConnect(handler: ConnectionHandler): () => void { - this.connectionHandlers.push(handler); - return () => { - const i = this.connectionHandlers.indexOf(handler); - if (i >= 0) this.connectionHandlers.splice(i, 1); - }; - } - - onDisconnect(handler: ConnectionHandler): () => void { - this.disconnectionHandlers.push(handler); - return () => { - const i = this.disconnectionHandlers.indexOf(handler); - if (i >= 0) this.disconnectionHandlers.splice(i, 1); - }; - } - - private setupAppStateListener(): void { - if (this.appStateSubscription) return; - this.lastAppState = AppState.currentState; - this.appStateSubscription = AppState.addEventListener('change', (nextState: AppStateStatus) => { - if (this.lastAppState.match(/inactive|background/) && nextState === 'active' && !this.isConnected()) { - this.reconnectAttempts = 0; - this.connect(); - } - this.lastAppState = nextState; - }); - } - - async start(): Promise { - this.setupAppStateListener(); - return this.connect(); - } - - stop(): void { - if (this.appStateSubscription) { - this.appStateSubscription.remove(); - this.appStateSubscription = null; - } - this.disconnect(); - } - - private markActivity(): void { - this.lastActivityAt = Date.now(); - } - - private startConnectionWatchdog(): void { - this.stopConnectionWatchdog(); - this.connectionWatchdogTimer = setInterval(() => { - // During active app usage, if no SSE activity for too long, reconnect proactively. - if (!this.connected || this.lastActivityAt <= 0) return; - if (Date.now() - this.lastActivityAt > this.CONNECTION_IDLE_TIMEOUT_MS) { - this.handleDisconnected(); - } - }, 10000); - } - - private stopConnectionWatchdog(): void { - if (this.connectionWatchdogTimer) { - clearInterval(this.connectionWatchdogTimer); - this.connectionWatchdogTimer = null; - } - } - - private handleDisconnected(): void { - const wasActive = this.connected || this.source != null || this.isConnecting; - this.isConnecting = false; - this.connected = false; - if (this.source) { - this.source.close(); - this.source = null; - } - this.stopConnectionWatchdog(); - if (wasActive) { - this.disconnectionHandlers.forEach(h => h()); - } - if (this.shouldRun) { - this.scheduleReconnect(); - } - } -} - -export const sseService = new SSEService(); diff --git a/src/services/systemNotificationService.ts b/src/services/systemNotificationService.ts index 2467fce..d617e3a 100644 --- a/src/services/systemNotificationService.ts +++ b/src/services/systemNotificationService.ts @@ -6,7 +6,7 @@ import * as Notifications from 'expo-notifications'; import { Platform, AppState, AppStateStatus } from 'react-native'; -import type { WSChatMessage, WSNotificationMessage, WSAnnouncementMessage } from './sseService'; +import type { WSChatMessage, WSNotificationMessage, WSAnnouncementMessage } from './wsService'; import { extractTextFromSegments } from '../types/dto'; import { getNotificationPreferencesSync } from './notificationPreferences'; import { vibrateOnMessage } from './messageVibrationService'; diff --git a/src/services/wsService.ts b/src/services/wsService.ts new file mode 100644 index 0000000..3cecee4 --- /dev/null +++ b/src/services/wsService.ts @@ -0,0 +1,807 @@ +import { AppState, AppStateStatus } from 'react-native'; + +import { api, WS_URL } from './api'; +import { MessageCategory, SystemMessageType, SystemMessageExtraData, MessageSegment } from '../types/dto'; +import { systemNotificationService } from './systemNotificationService'; + +export type WSMessageType = + | 'chat' + | 'message' + | 'read' + | 'typing' + | 'recall' + | 'notification' + | 'announcement' + | 'group_message' + | 'group_typing' + | 'group_notice' + | 'group_mention' + | 'group_read' + | 'group_recall' + | 'notice' + | 'request' + | 'meta' + | 'private' + | 'group' + | 'follow' + | 'like' + | 'comment' + | 'heartbeat' + | 'pong' + | 'message_sent' + | 'message_recalled' + | 'error'; + +export interface WSChatMessage { + type: 'chat'; + conversation_id: string; + id: string; + sender_id: string; + seq: number; + segments?: MessageSegment[]; + created_at: string; +} + +export interface WSReadMessage { + type: 'read'; + conversation_id: string; + user_id: string; + seq: number; +} + +export interface WSTypingMessage { + type: 'typing'; + conversation_id: string; + user_id: string; + is_typing: boolean; +} + +export interface WSRecallMessage { + type: 'recall'; + conversation_id: string; + message_id: string; +} + +export interface WSNotificationMessage { + type: 'notification'; + id: string; + sender_id?: string; + receiver_id?: string; + content: string; + category?: MessageCategory; + system_type?: SystemMessageType; + extra_data?: SystemMessageExtraData; + created_at: string; +} + +export interface WSAnnouncementMessage { + type: 'announcement'; + id: string; + sender_id?: string; + receiver_id?: string; + content: string; + category?: MessageCategory; + system_type?: SystemMessageType; + extra_data?: SystemMessageExtraData; + created_at: string; +} + +export interface WSGroupMentionMessage { + type: 'group_mention'; + group_id: string; + conversation_id: string; + message_id: string; + from_user_id: string; + content: string; + mention_all: boolean; + created_at: string; +} + +export interface WSGroupChatMessage { + type: 'group_message'; + conversation_id: string; + group_id: string; + id: string; + sender_id: string; + seq: number; + segments?: MessageSegment[]; + created_at: string; +} + +export interface WSGroupTypingMessage { + type: 'group_typing'; + group_id: string; + user_id: string; + is_typing: boolean; +} + +export type GroupNoticeType = 'member_join' | 'member_leave' | 'member_removed' | 'role_changed' | 'muted' | 'unmuted'; + +export interface WSGroupNoticeMessage { + type: 'group_notice'; + notice_type: GroupNoticeType; + group_id: string; + data: { + user_id?: string; + operator_id?: string; + role?: string; + [key: string]: any; + }; + timestamp: number; + message_id?: string; + seq?: number; +} + +export interface WSGroupReadMessage { + type: 'group_read'; + group_id: string; + conversation_id: string; + user_id: string; + seq: number; +} + +export interface WSGroupRecallMessage { + type: 'group_recall'; + group_id: string; + conversation_id: string; + message_id: string; +} + +export interface WSServerMessage { + event_id?: number; + type: string; + ts?: number; + payload?: any; +} + +export type WSMessage = + | WSChatMessage + | WSReadMessage + | WSTypingMessage + | WSRecallMessage + | WSNotificationMessage + | WSAnnouncementMessage + | WSGroupChatMessage + | WSGroupTypingMessage + | WSGroupNoticeMessage + | WSGroupMentionMessage + | WSGroupReadMessage + | WSGroupRecallMessage; + +type MessageHandler = (message: T) => void; +type ConnectionHandler = () => void; + +// 发送队列中的消息 +interface PendingMessage { + id: string; + type: string; + payload: any; + resolve: (value: any) => void; + reject: (reason: any) => void; + timestamp: number; +} + +class WebSocketService { + private ws: WebSocket | null = null; + private isConnecting = false; + private connected = false; + private reconnectAttempts = 0; + private maxReconnectAttempts = 20; + private reconnectDelay = 3000; + private reconnectTimer: NodeJS.Timeout | null = null; + private heartbeatTimer: NodeJS.Timeout | null = null; + private heartbeatInterval = 30000; // 30秒心跳 + private messageHandlers: Map = new Map(); + private connectionHandlers: ConnectionHandler[] = []; + private disconnectionHandlers: ConnectionHandler[] = []; + private appStateSubscription: any = null; + private lastAppState: AppStateStatus = 'active'; + private shouldRun = false; + private lastEventId = ''; + private lastActivityAt = 0; + + // 发送队列 + private pendingMessages: PendingMessage[] = []; + private messageIdCounter = 0; + + // 降级状态 + private isFallbackMode = false; + private fallbackCheckTimer: NodeJS.Timeout | null = null; + + private getWSUrl(): string { + const token = api.getToken(); + return `${WS_URL}?token=${encodeURIComponent(token || '')}`; + } + + async connect(): Promise { + this.shouldRun = true; + if (this.isConnecting || this.isConnected()) return true; + this.isConnecting = true; + + try { + const token = await api.getToken(); + if (!token) { + this.isConnecting = false; + return false; + } + + const url = this.getWSUrl(); + this.ws = new WebSocket(url); + + this.ws.onopen = () => { + this.isConnecting = false; + this.connected = true; + this.reconnectAttempts = 0; + this.isFallbackMode = false; + this.markActivity(); + this.startHeartbeat(); + this.connectionHandlers.forEach(h => h()); + + // 发送队列中的消息 + this.flushPendingMessages(); + }; + + this.ws.onmessage = (event) => { + this.markActivity(); + this.handleIncoming(event.data); + }; + + this.ws.onerror = () => { + this.handleDisconnected(); + }; + + this.ws.onclose = () => { + this.handleDisconnected(); + }; + + return true; + } catch { + this.isConnecting = false; + this.scheduleReconnect(); + return false; + } + } + + private handleIncoming(data: string): void { + try { + const msg: WSServerMessage = JSON.parse(data); + + // 处理pong响应 + if (msg.type === 'pong') { + return; + } + + // 处理错误消息 + if (msg.type === 'error') { + console.error('WebSocket error:', msg.payload); + return; + } + + // 处理消息发送成功响应 + if (msg.type === 'message_sent') { + this.handleMessageSent(msg.payload); + return; + } + + // 处理消息撤回响应 + if (msg.type === 'message_recalled') { + this.handleMessageRecalled(msg.payload); + return; + } + + // 保存事件ID用于断线重连 + if (msg.event_id) { + this.lastEventId = String(msg.event_id); + } + + // 分发事件 + this.dispatchServerMessage(msg); + } catch (error) { + console.error('Failed to parse WebSocket message:', error); + } + } + + private dispatchServerMessage(msg: WSServerMessage): void { + const { type, payload } = msg; + + switch (type) { + case 'chat_message': + this.handleChatMessage(payload); + break; + case 'message_read': + this.handleMessageRead(payload); + break; + case 'typing': + this.handleTyping(payload); + break; + case 'message_recall': + this.handleMessageRecall(payload); + break; + case 'group_notice': + this.handleGroupNotice(payload); + break; + case 'system_notification': + this.handleSystemNotification(payload); + break; + default: + console.log('Unknown message type:', type); + } + } + + private handleChatMessage(payload: any): void { + const detailType = payload?.detail_type || 'private'; + const m = payload?.message || payload; + + if (detailType === 'group') { + const gm: WSGroupChatMessage = { + type: 'group_message', + conversation_id: m.conversation_id, + group_id: String(m.group_id ?? ''), + id: m.id, + sender_id: m.sender_id, + seq: Number(m.seq || 0), + segments: m.segments || [], + created_at: m.created_at || new Date().toISOString(), + }; + this.emit('group_message', gm); + } else { + const cm: WSChatMessage = { + type: 'chat', + conversation_id: m.conversation_id, + id: m.id, + sender_id: m.sender_id, + seq: Number(m.seq || 0), + segments: m.segments || [], + created_at: m.created_at || new Date().toISOString(), + }; + this.emit('chat', cm); + } + } + + private handleMessageRead(payload: any): void { + const detailType = payload?.detail_type || 'private'; + + if (detailType === 'group') { + const m: WSGroupReadMessage = { + type: 'group_read', + group_id: String(payload.group_id ?? ''), + conversation_id: payload.conversation_id, + user_id: payload.user_id, + seq: Number(payload.seq || 0), + }; + this.emit('group_read', m); + } else { + const m: WSReadMessage = { + type: 'read', + conversation_id: payload.conversation_id, + user_id: payload.user_id, + seq: Number(payload.seq || 0), + }; + this.emit('read', m); + } + } + + private handleTyping(payload: any): void { + const detailType = payload?.detail_type || 'private'; + + if (detailType === 'group') { + const m: WSGroupTypingMessage = { + type: 'group_typing', + group_id: String(payload.group_id ?? ''), + user_id: payload.user_id, + is_typing: payload.is_typing !== false, + }; + this.emit('group_typing', m); + } else { + const m: WSTypingMessage = { + type: 'typing', + conversation_id: payload.conversation_id, + user_id: payload.user_id, + is_typing: payload.is_typing !== false, + }; + this.emit('typing', m); + } + } + + private handleMessageRecall(payload: any): void { + const detailType = payload?.detail_type || 'private'; + + if (detailType === 'group') { + const m: WSGroupRecallMessage = { + type: 'group_recall', + group_id: String(payload.group_id ?? ''), + conversation_id: payload.conversation_id, + message_id: payload.message_id, + }; + this.emit('group_recall', m); + } else { + const m: WSRecallMessage = { + type: 'recall', + conversation_id: payload.conversation_id, + message_id: payload.message_id, + }; + this.emit('recall', m); + } + } + + private handleGroupNotice(payload: any): void { + const m: WSGroupNoticeMessage = { + type: 'group_notice', + notice_type: payload.notice_type, + group_id: String(payload.group_id ?? ''), + data: payload.data || {}, + timestamp: payload.timestamp || Date.now(), + message_id: payload.message_id, + seq: payload.seq, + }; + this.emit('group_notice', m); + } + + private handleSystemNotification(payload: any): void { + const m: WSNotificationMessage = { + type: 'notification', + id: String(payload.id ?? ''), + content: payload.content || '', + created_at: payload.created_at || new Date().toISOString(), + }; + this.emit('notification', m); + systemNotificationService.handleWSMessage(m as any).catch(() => {}); + } + + private handleMessageSent(payload: any): void { + // 找到对应的发送请求并resolve + const pendingMsg = this.pendingMessages.find(p => + p.type === 'chat' && p.payload.conversation_id === payload.conversation_id + ); + if (pendingMsg) { + pendingMsg.resolve(payload); + this.removePendingMessage(pendingMsg.id); + } + } + + private handleMessageRecalled(payload: any): void { + const pendingMsg = this.pendingMessages.find(p => + p.type === 'recall' && p.payload.message_id === payload.message_id + ); + if (pendingMsg) { + pendingMsg.resolve(payload); + this.removePendingMessage(pendingMsg.id); + } + } + + // 发送消息（支持降级到HTTP） + async sendMessage( + conversationId: string, + detailType: 'private' | 'group', + segments: MessageSegment[], + replyToId?: string + ): Promise { + const payload = { + conversation_id: conversationId, + detail_type: detailType, + segments, + reply_to_id: replyToId, + }; + + // 如果WebSocket连接正常，优先使用WebSocket发送 + if (this.isConnected() && !this.isFallbackMode) { + return this.sendViaWebSocket('chat', payload); + } + + // 降级到HTTP发送 + return this.sendViaHTTP(conversationId, detailType, segments, replyToId); + } + + // 标记已读 + async markRead(conversationId: string, seq: number): Promise { + const payload = { + conversation_id: conversationId, + seq, + }; + + if (this.isConnected() && !this.isFallbackMode) { + this.sendViaWebSocket('read', payload).catch(() => {}); + } + // 已读不需要等待响应，失败也不影响 + } + + // 发送输入状态 + sendTyping(conversationId: string, isTyping: boolean): void { + const payload = { + conversation_id: conversationId, + is_typing: isTyping, + }; + + if (this.isConnected() && !this.isFallbackMode) { + this.sendViaWebSocket('typing', payload).catch(() => {}); + } + } + + // 撤回消息 + async recallMessage(messageId: string): Promise { + const payload = { message_id: messageId }; + + if (this.isConnected() && !this.isFallbackMode) { + return this.sendViaWebSocket('recall', payload); + } + + // 降级到HTTP + const { api } = await import('./api'); + return api.post('/messages/delete_msg', { message_id: messageId }); + } + + // 通过WebSocket发送 + private sendViaWebSocket(type: string, payload: any): Promise { + return new Promise((resolve, reject) => { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { + reject(new Error('WebSocket not connected')); + return; + } + + const messageId = `msg_${++this.messageIdCounter}_${Date.now()}`; + const message = { + type, + payload, + }; + + // 添加到待处理队列 + const pendingMsg: PendingMessage = { + id: messageId, + type, + payload, + resolve, + reject, + timestamp: Date.now(), + }; + this.pendingMessages.push(pendingMsg); + + // 发送消息 + this.ws.send(JSON.stringify(message)); + + // 设置超时 + setTimeout(() => { + const index = this.pendingMessages.findIndex(p => p.id === messageId); + if (index >= 0) { + this.pendingMessages[index].reject(new Error('Message timeout')); + this.removePendingMessage(messageId); + } + }, 10000); + }); + } + + // 通过HTTP发送（降级方案） + private async sendViaHTTP( + conversationId: string, + detailType: 'private' | 'group', + segments: MessageSegment[], + replyToId?: string + ): Promise { + const { api } = await import('./api'); + const body: any = { + detail_type: detailType, + segments, + }; + if (replyToId) { + body.reply_to_id = replyToId; + } + return api.post(`/conversations/${conversationId}/messages`, body); + } + + // 刷新待发送消息队列 + private flushPendingMessages(): void { + if (this.pendingMessages.length === 0) return; + + const now = Date.now(); + const expiredMessages: PendingMessage[] = []; + const validMessages: PendingMessage[] = []; + + // 分离过期和有效的消息 + for (const msg of this.pendingMessages) { + if (now - msg.timestamp > 30000) { + expiredMessages.push(msg); + } else { + validMessages.push(msg); + } + } + + // 拒绝过期消息 + expiredMessages.forEach(msg => { + msg.reject(new Error('Message expired')); + }); + + // 重新发送有效消息 + this.pendingMessages = []; + validMessages.forEach(msg => { + this.sendViaWebSocket(msg.type, msg.payload) + .then(msg.resolve) + .catch(msg.reject); + }); + } + + private removePendingMessage(id: string): void { + const index = this.pendingMessages.findIndex(p => p.id === id); + if (index >= 0) { + this.pendingMessages.splice(index, 1); + } + } + + disconnect(): void { + this.shouldRun = false; + this.connected = false; + this.isConnecting = false; + this.isFallbackMode = false; + + if (this.ws) { + this.ws.close(); + this.ws = null; + } + + this.stopHeartbeat(); + this.stopReconnect(); + this.stopFallbackCheck(); + + // 拒绝所有待处理消息 + this.pendingMessages.forEach(msg => { + msg.reject(new Error('WebSocket disconnected')); + }); + this.pendingMessages = []; + } + + private scheduleReconnect(): void { + if (this.reconnectAttempts >= this.maxReconnectAttempts) { + // 超过最大重试次数，进入降级模式 + this.enterFallbackMode(); + return; + } + + this.stopReconnect(); + this.reconnectTimer = setTimeout(() => { + this.reconnectAttempts += 1; + this.connect(); + }, this.reconnectDelay); + } + + private stopReconnect() { + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } + + isConnected(): boolean { + return this.connected && this.ws?.readyState === WebSocket.OPEN; + } + + isInFallbackMode(): boolean { + return this.isFallbackMode; + } + + // 进入降级模式 + private enterFallbackMode(): void { + if (this.isFallbackMode) return; + + this.isFallbackMode = true; + console.log('WebSocket entering fallback mode, using HTTP API'); + + // 启动降级模式检查定时器，定期尝试重连WebSocket + this.startFallbackCheck(); + } + + // 启动降级模式检查 + private startFallbackCheck(): void { + this.stopFallbackCheck(); + this.fallbackCheckTimer = setInterval(() => { + if (!this.isConnected()) { + this.reconnectAttempts = 0; + this.connect(); + } + }, 30000); // 每30秒尝试重连 + } + + private stopFallbackCheck(): void { + if (this.fallbackCheckTimer) { + clearInterval(this.fallbackCheckTimer); + this.fallbackCheckTimer = null; + } + } + + private emit(type: T, message: Extract) { + const handlers = this.messageHandlers.get(type) || []; + handlers.forEach(h => h(message as WSMessage)); + } + + on(type: T, handler: MessageHandler>): () => void { + const list = this.messageHandlers.get(type) || []; + list.push(handler as MessageHandler); + this.messageHandlers.set(type, list); + return () => { + const current = this.messageHandlers.get(type) || []; + const idx = current.indexOf(handler as MessageHandler); + if (idx >= 0) current.splice(idx, 1); + }; + } + + onConnect(handler: ConnectionHandler): () => void { + this.connectionHandlers.push(handler); + return () => { + const i = this.connectionHandlers.indexOf(handler); + if (i >= 0) this.connectionHandlers.splice(i, 1); + }; + } + + onDisconnect(handler: ConnectionHandler): () => void { + this.disconnectionHandlers.push(handler); + return () => { + const i = this.disconnectionHandlers.indexOf(handler); + if (i >= 0) this.disconnectionHandlers.splice(i, 1); + }; + } + + private setupAppStateListener(): void { + if (this.appStateSubscription) return; + this.lastAppState = AppState.currentState; + this.appStateSubscription = AppState.addEventListener('change', (nextState: AppStateStatus) => { + if (this.lastAppState.match(/inactive|background/) && nextState === 'active' && !this.isConnected()) { + this.reconnectAttempts = 0; + this.connect(); + } + this.lastAppState = nextState; + }); + } + + async start(): Promise { + this.setupAppStateListener(); + return this.connect(); + } + + stop(): void { + if (this.appStateSubscription) { + this.appStateSubscription.remove(); + this.appStateSubscription = null; + } + this.disconnect(); + } + + private markActivity(): void { + this.lastActivityAt = Date.now(); + } + + private startHeartbeat(): void { + this.stopHeartbeat(); + this.heartbeatTimer = setInterval(() => { + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify({ type: 'ping' })); + } + }, this.heartbeatInterval); + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + } + + private handleDisconnected(): void { + const wasActive = this.connected || this.ws != null || this.isConnecting; + this.isConnecting = false; + this.connected = false; + + if (this.ws) { + this.ws.close(); + this.ws = null; + } + + this.stopHeartbeat(); + + if (wasActive) { + this.disconnectionHandlers.forEach(h => h()); + } + + if (this.shouldRun) { + this.scheduleReconnect(); + } + } +} + +export const wsService = new WebSocketService(); \ No newline at end of file diff --git a/src/stores/authStore.ts b/src/stores/authStore.ts index 7fbc59e..d4ad174 100644 --- a/src/stores/authStore.ts +++ b/src/stores/authStore.ts @@ -14,7 +14,7 @@ import AsyncStorage from '@react-native-async-storage/async-storage'; import { create } from 'zustand'; import { User } from '../types'; import { authService, resolveAuthApiError, LoginRequest, RegisterRequest } from '../services'; -import { sseService } from '../services/sseService'; +import { wsService } from '../services/wsService'; import { initDatabase, closeDatabase, @@ -95,7 +95,7 @@ function resolveLoginError(error: any): string { // ── 启动 SSE 实时服务 ── async function startRealtime(): Promise { try { - await sseService.start(); + await wsService.start(); } catch (error) { console.error('[AuthStore] 启动 SSE 服务失败:', error); } @@ -212,7 +212,7 @@ export const useAuthStore = create((set) => ({ // 1. 通知服务端（Token 清理在 authService 内部完成） await authService.logout(); // 2. 停止 SSE - sseService.stop(); + wsService.stop(); // 3. 清除 DB 中的用户缓存（DB 此时一定已初始化） await clearCurrentUserCache().catch(() => {}); // 4. 关闭数据库连接 diff --git a/src/stores/message/MessageManager.ts b/src/stores/message/MessageManager.ts index 62aad2c..e7acb26 100644 --- a/src/stores/message/MessageManager.ts +++ b/src/stores/message/MessageManager.ts @@ -4,7 +4,7 @@ */ import { messageStateManager, MessageStateManager } from './MessageStateManager'; -import { sseMessageHandler, SSEMessageHandler } from './SSEMessageHandler'; +import { wsMessageHandler, WSMessageHandler } from './WSMessageHandler'; import { messageSyncService, MessageSyncService } from './MessageSyncService'; import { readReceiptManager, ReadReceiptManager } from './ReadReceiptManager'; import { messageRepository } from '../../data/repositories/MessageRepository'; @@ -20,23 +20,23 @@ export type { export class MessageManager { private stateManager: MessageStateManager; - private sseHandler: SSEMessageHandler; + private wsHandler: WSMessageHandler; private syncService: MessageSyncService; private readManager: ReadReceiptManager; private initialized: boolean = false; private authUnsubscribe: (() => void) | null = null; - private sseUnsubscribe: (() => void) | null = null; + private wsUnsubscribe: (() => void) | null = null; private currentUserId: string | null = null; constructor() { this.stateManager = messageStateManager; - this.sseHandler = sseMessageHandler; + this.wsHandler = wsMessageHandler; this.syncService = messageSyncService; this.readManager = readReceiptManager; this.readManager.setStateManager(this.stateManager); - this.setupSSEHandlers(); + this.setupWSHandlers(); this.initAuthListener(); } @@ -58,8 +58,8 @@ export class MessageManager { }); } - private setupSSEHandlers(): void { - this.sseUnsubscribe = this.sseHandler.subscribe((event) => { + private setupWSHandlers(): void { + this.wsUnsubscribe = this.wsHandler.subscribe((event) => { switch (event.type) { case 'chat_message': case 'group_message': @@ -89,7 +89,7 @@ export class MessageManager { try { console.log('[MessageManager] Initializing...'); - this.sseHandler.connect(); + this.wsHandler.connect(); const conversations = await this.syncService.syncConversations(); this.stateManager.setConversations(conversations); @@ -224,14 +224,14 @@ export class MessageManager { } isConnected(): boolean { - return this.sseHandler.isConnected(); + return this.wsHandler.isConnected(); } reset(): void { this.initialized = false; this.stateManager.reset(); this.readManager.reset(); - this.sseHandler.disconnect(); + this.wsHandler.disconnect(); } destroy(): void { @@ -239,9 +239,9 @@ export class MessageManager { this.authUnsubscribe(); this.authUnsubscribe = null; } - if (this.sseUnsubscribe) { - this.sseUnsubscribe(); - this.sseUnsubscribe = null; + if (this.wsUnsubscribe) { + this.wsUnsubscribe(); + this.wsUnsubscribe = null; } this.reset(); } diff --git a/src/stores/message/SSEMessageHandler.ts b/src/stores/message/WSMessageHandler.ts similarity index 66% rename from src/stores/message/SSEMessageHandler.ts rename to src/stores/message/WSMessageHandler.ts index afbad45..a05f05e 100644 --- a/src/stores/message/SSEMessageHandler.ts +++ b/src/stores/message/WSMessageHandler.ts @@ -1,12 +1,12 @@ /** - * SSE 消息处理器 - * 只负责处理 SSE 消息，不管理状态 + * WS 消息处理器 + * 只负责处理 WS 消息，不管理状态 */ -import { sseService, WSChatMessage, WSGroupChatMessage, WSReadMessage, WSGroupReadMessage, WSRecallMessage, WSGroupRecallMessage, WSGroupTypingMessage, WSGroupNoticeMessage, GroupNoticeType, WSMessage } from '../../services/sseService'; +import { wsService, WSChatMessage, WSGroupChatMessage, WSReadMessage, WSGroupReadMessage, WSRecallMessage, WSGroupRecallMessage, WSGroupTypingMessage, WSGroupNoticeMessage, GroupNoticeType, WSMessage } from '../../services/wsService'; import type { Message, GroupNotice } from '../../core/entities/Message'; -export type SSEEventType = +export type WSEventType = | 'chat_message' | 'group_message' | 'read_receipt' @@ -16,58 +16,58 @@ export type SSEEventType = | 'typing' | 'group_notice'; -export interface SSEEvent { - type: SSEEventType; +export interface WSEvent { + type: WSEventType; payload: any; raw: any; } -export type SSEEventHandler = (event: SSEEvent) => void; +export type WSEventHandler = (event: WSEvent) => void; -export class SSEMessageHandler { +export class WSMessageHandler { private unsubscribeFns: Array<() => void> = []; - private handlers: Set = new Set(); + private handlers: Set = new Set(); connect(): void { if (this.unsubscribeFns.length > 0) return; // 监听私聊消息 - const unsubChat = sseService.on('chat', (message) => { + const unsubChat = wsService.on('chat', (message) => { this.emit('chat_message', this.parseChatMessage(message), message); }); // 监听群聊消息 - const unsubGroupMessage = sseService.on('group_message', (message) => { + const unsubGroupMessage = wsService.on('group_message', (message) => { this.emit('group_message', this.parseGroupMessage(message), message); }); // 监听私聊已读回执 - const unsubRead = sseService.on('read', (message) => { + const unsubRead = wsService.on('read', (message) => { this.emit('read_receipt', message, message); }); // 监听群聊已读回执 - const unsubGroupRead = sseService.on('group_read', (message) => { + const unsubGroupRead = wsService.on('group_read', (message) => { this.emit('group_read_receipt', message, message); }); // 监听私聊消息撤回 - const unsubRecall = sseService.on('recall', (message) => { + const unsubRecall = wsService.on('recall', (message) => { this.emit('message_recalled', message, message); }); // 监听群聊消息撤回 - const unsubGroupRecall = sseService.on('group_recall', (message) => { + const unsubGroupRecall = wsService.on('group_recall', (message) => { this.emit('group_message_recalled', message, message); }); // 监听群聊输入状态 - const unsubGroupTyping = sseService.on('group_typing', (message) => { + const unsubGroupTyping = wsService.on('group_typing', (message) => { this.emit('typing', message, message); }); // 监听群通知 - const unsubGroupNotice = sseService.on('group_notice', (message) => { + const unsubGroupNotice = wsService.on('group_notice', (message) => { this.emit('group_notice', this.parseGroupNotice(message), message); }); @@ -89,18 +89,18 @@ export class SSEMessageHandler { this.handlers.clear(); } - subscribe(handler: SSEEventHandler): () => void { + subscribe(handler: WSEventHandler): () => void { this.handlers.add(handler); return () => this.handlers.delete(handler); } - private emit(type: SSEEventType, payload: any, raw: any): void { - const event: SSEEvent = { type, payload, raw }; + private emit(type: WSEventType, payload: any, raw: any): void { + const event: WSEvent = { type, payload, raw }; this.handlers.forEach(handler => { try { handler(event); } catch (error) { - console.error('[SSEMessageHandler] Handler error:', error); + console.error('[WSMessageHandler] Handler error:', error); } }); } @@ -141,8 +141,8 @@ export class SSEMessageHandler { } isConnected(): boolean { - return sseService.isConnected(); + return wsService.isConnected(); } } -export const sseMessageHandler = new SSEMessageHandler(); \ No newline at end of file +export const wsMessageHandler = new WSMessageHandler(); \ No newline at end of file diff --git a/src/stores/message/index.ts b/src/stores/message/index.ts index 0961ba2..37f1baa 100644 --- a/src/stores/message/index.ts +++ b/src/stores/message/index.ts @@ -12,8 +12,8 @@ export type { MessageState, MessageEvent, MessageSubscriber, MessageEventType } // 导出同步服务 export { messageSyncService, MessageSyncService } from './MessageSyncService'; -// 导出SSE处理器 -export { sseMessageHandler, SSEMessageHandler } from './SSEMessageHandler'; +// 导出WS处理器 +export { wsMessageHandler, WSMessageHandler } from './WSMessageHandler'; // 导出已读管理器 export { readReceiptManager, ReadReceiptManager } from './ReadReceiptManager'; \ No newline at end of file diff --git a/src/stores/messageManager.ts b/src/stores/messageManager.ts index 5aae26e..ccf806d 100644 --- a/src/stores/messageManager.ts +++ b/src/stores/messageManager.ts @@ -18,7 +18,7 @@ import { ConversationResponse, MessageResponse, MessageSegment, UserDTO } from '../types/dto'; import { messageService } from '../services/messageService'; import { - sseService, + wsService, WSChatMessage, WSGroupChatMessage, WSReadMessage, @@ -28,7 +28,7 @@ import { WSGroupTypingMessage, WSGroupNoticeMessage, GroupNoticeType, -} from '../services/sseService'; +} from '../services/wsService'; import { saveMessage, saveMessagesBatch, @@ -94,7 +94,7 @@ interface MessageManagerState { systemUnreadCount: number; // 连接状态 - isSSEConnected: boolean; + isWSConnected: boolean; // 当前活动会话ID（用户正在查看的会话） currentConversationId: string | null; @@ -210,7 +210,7 @@ class MessageManager { messagesMap: new Map(), totalUnreadCount: 0, systemUnreadCount: 0, - isSSEConnected: false, + isWSConnected: false, currentConversationId: null, isLoadingConversations: false, loadingMessagesSet: new Set(), @@ -582,7 +582,7 @@ class MessageManager { // 监听私聊消息 - sseService.on('chat', (message: WSChatMessage) => { + wsService.on('chat', (message: WSChatMessage) => { if (!this.state.isInitialized || this.isBootstrapping) { this.bufferedChatMessages.push(message); return; @@ -591,7 +591,7 @@ class MessageManager { }); // 监听群聊消息 - sseService.on('group_message', (message: WSGroupChatMessage) => { + wsService.on('group_message', (message: WSGroupChatMessage) => { if (!this.state.isInitialized || this.isBootstrapping) { this.bufferedChatMessages.push(message); return; @@ -600,7 +600,7 @@ class MessageManager { }); // 监听私聊已读回执 - sseService.on('read', (message: WSReadMessage) => { + wsService.on('read', (message: WSReadMessage) => { if (!this.state.isInitialized || this.isBootstrapping) { this.bufferedReadReceipts.push(message); return; @@ -609,7 +609,7 @@ class MessageManager { }); // 监听群聊已读回执 - sseService.on('group_read', (message: WSGroupReadMessage) => { + wsService.on('group_read', (message: WSGroupReadMessage) => { if (!this.state.isInitialized || this.isBootstrapping) { this.bufferedReadReceipts.push(message); return; @@ -618,28 +618,28 @@ class MessageManager { }); // 监听私聊消息撤回 - sseService.on('recall', (message: WSRecallMessage) => { + wsService.on('recall', (message: WSRecallMessage) => { this.handleRecallMessage(message); }); // 监听群聊消息撤回 - sseService.on('group_recall', (message: WSGroupRecallMessage) => { + wsService.on('group_recall', (message: WSGroupRecallMessage) => { this.handleGroupRecallMessage(message); }); // 监听群聊输入状态 - sseService.on('group_typing', (message: WSGroupTypingMessage) => { + wsService.on('group_typing', (message: WSGroupTypingMessage) => { this.handleGroupTyping(message); }); // 监听群通知 - sseService.on('group_notice', (message: WSGroupNoticeMessage) => { + wsService.on('group_notice', (message: WSGroupNoticeMessage) => { this.handleGroupNotice(message); }); // 监听连接状态 - sseService.onConnect(() => { - this.state.isSSEConnected = true; + wsService.onConnect(() => { + this.state.isWSConnected = true; this.notifySubscribers({ type: 'connection_changed', payload: { connected: true }, @@ -668,8 +668,8 @@ class MessageManager { } }); - sseService.onDisconnect(() => { - this.state.isSSEConnected = false; + wsService.onDisconnect(() => { + this.state.isWSConnected = false; this.notifySubscribers({ type: 'connection_changed', payload: { connected: false }, @@ -2375,7 +2375,7 @@ class MessageManager { * 获取SSE连接状态 */ isConnected(): boolean { - return this.state.isSSEConnected; + return this.state.isWSConnected; } /** @@ -2486,7 +2486,7 @@ class MessageManager { subscriber({ type: 'connection_changed', - payload: { connected: this.state.isSSEConnected }, + payload: { connected: this.state.isWSConnected }, timestamp: Date.now(), }); diff --git a/src/theme/palettes.ts b/src/theme/palettes.ts index 9a14cc3..91fff9f 100644 --- a/src/theme/palettes.ts +++ b/src/theme/palettes.ts @@ -76,7 +76,7 @@ export const lightColors = { link: '#4A88C7', success: '#34C759', danger: '#FF3B30', - bubbleOutgoing: '#7CB9FF', // 柔和的淡蓝色 + bubbleOutgoing: '#E8E8E8', // 柔和的淡灰色 bubbleIncoming: '#FFFFFF', // 对方发的消息 - 纯白 replyTint: '#DFF2FF', replyTintActive: '#CFEAFF', @@ -168,7 +168,7 @@ export const darkColors = { link: '#5AC8FA', success: '#32D74B', danger: '#FF453A', - bubbleOutgoing: '#2E5A8C', // 暗色模式：深蓝气泡 + bubbleOutgoing: '#3A3A3C', // 暗色模式：深灰色 bubbleIncoming: '#2C2C2E', // 暗色模式：灰色气泡 replyTint: '#1A3A52', replyTintActive: '#224060', diff --git a/src/types/dto.ts b/src/types/dto.ts index 65ff8fc..e91851a 100644 --- a/src/types/dto.ts +++ b/src/types/dto.ts @@ -684,31 +684,37 @@ export interface GroupAnnouncementListResponse { total_pages: number; } -// ==================== SSE Event Types ==================== +// ==================== WS Event Types ==================== -// SSE 事件类型 -export type SSEEventType = 'message' | 'notice' | 'request' | 'meta'; +// WS 事件类型 +export type WSEventType = 'message' | 'notice' | 'request' | 'meta'; -// SSE 消息详细类型 -export type SSEDetailType = 'private' | 'group' | 'follow' | 'like' | 'comment' | 'request' | 'typing' | 'heartbeat' | 'read'; +// WS 消息详细类型 +export type WSDetailType = 'private' | 'group' | 'follow' | 'like' | 'comment' | 'request' | 'typing' | 'heartbeat' | 'read'; -// SSE 事件消息段（与后端 message 字段一致） -export interface SSESegment { +// WS 事件消息段（与后端 message 字段一致） +export interface WSSegment { type: string; data: Record; } -// SSE 事件（后端推送的事件格式） -export interface SSEEvent { +// WS 事件（后端推送的事件格式） +export interface WSEvent { id: string; // 事件唯一ID time: number; // 事件时间戳（毫秒） - type: SSEEventType; // 事件类型: message(消息), notice(通知), request(请求), meta(元事件) - detail_type: SSEDetailType; // 详细类型: private(私聊), group(群聊), follow(关注), like(点赞)等 + type: WSEventType; // 事件类型: message(消息), notice(通知), request(请求), meta(元事件) + detail_type: WSDetailType; // 详细类型: private(私聊), group(群聊), follow(关注), like(点赞)等 seq: string; // 消息序号 - message?: SSESegment[]; // 消息内容数组 + message?: WSSegment[]; // 消息内容数组 conversation_id: string; // 会话ID } +// 旧的类型名称，保持向后兼容 +export type SSEEventType = WSEventType; +export type SSEDetailType = WSDetailType; +export interface SSESegment extends WSSegment {} +export interface SSEEvent extends WSEvent {} + /** * 从消息segments中提取纯文本内容 * 用于消息列表显示、通知等场景