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(token: string | null): string { return `${WS_URL}?token=${encodeURIComponent(token || '')}`; } async connect(): Promise { this.shouldRun = true; if (this.isConnecting || this.isConnected()) { console.log('[WebSocket] Already connected or connecting'); return true; } this.isConnecting = true; console.log('[WebSocket] Connecting...'); try { const token = await api.getToken(); if (!token) { console.error('[WebSocket] No token available'); this.isConnecting = false; this.scheduleReconnect(); return false; } const url = this.getWSUrl(token); console.log('[WebSocket] Connecting to:', url.replace(token, '***')); this.ws = new WebSocket(url); this.ws.onopen = () => { console.log('[WebSocket] Connected successfully'); 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 = (error) => { console.error('[WebSocket] Error:', error); }; this.ws.onclose = (event) => { console.log('[WebSocket] Closed:', event.code, event.reason); this.handleDisconnected(); }; return true; } catch (error) { console.error('[WebSocket] Connection error:', error); 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(); console.log(`[WebSocket] Reconnecting in ${this.reconnectDelay}ms (attempt ${this.reconnectAttempts + 1}/${this.maxReconnectAttempts})`); 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 { // 防止重复调用 if (!this.connected && !this.ws && !this.isConnecting) { console.log('[WebSocket] Already disconnected, ignoring'); return; } console.log('[WebSocket] Handling disconnect...'); 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(); } else { console.log('[WebSocket] shouldRun is false, not reconnecting'); } } } export const wsService = new WebSocketService();