carrot_bbs/frontend
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(WebSocket): migrate from SSE to WebSocket for real-time messaging

Browse Source 
- Replaced SSEClient with WSClient for handling real-time messaging.
- Updated ProcessMessageUseCase to initialize WebSocket listeners.
- Refactored messageService to prioritize WebSocket for sending messages, with fallback to HTTP.
- Removed sseService and adjusted imports across the application to utilize wsService.
- Enhanced message handling and connection management for improved performance and reliability.
This commit is contained in:
lafay
2026-03-26 22:04:46 +08:00
parent 4b89b50006
commit ba99900624
16 changed files with 986 additions and 659 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/core/usecases/ProcessMessageUseCase.ts
View File

@@ -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 },

66
src/data/datasources/SSEClient.ts → src/data/datasources/WSClient.ts
View File

@@ -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<T> = (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<SSEEventType, Set<EventHandler<any>>> = new Map();
class WSClient {
private handlers: Map<WSEventType, Set<EventHandler<any>>> = 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<T extends SSEEventType>(
on<T extends WSEventType>(
event: T,
handler: EventHandler<SSEEvents[T]>
handler: EventHandler<WSEvents[T]>
): () => void {
if (!this.handlers.has(event)) {
this.handlers.set(event, new Set());
@@ -142,9 +142,9 @@ class SSEClient {
/**
* 触发事件
*/
private emit<T extends SSEEventType>(
private emit<T extends WSEventType>(
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<boolean> {
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;

4
src/services/api.ts
View File

@@ -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 };

14
src/services/backgroundService.ts
View File

@@ -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();
}
}
});

8
src/services/index.ts
View File

@@ -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';

49
src/services/messageService.ts
View File

@@ -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<void> {
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<void> {
// 使用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<void> {
// 使用WebSocket发送输入状态
wsService.sendTyping(conversationId, true);
// 同时通过HTTP确保可靠性
await api.post(`/conversations/${encodeURIComponent(conversationId)}/typing`);
}

517
src/services/sseService.ts
View File

@@ -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<T extends WSMessage = WSMessage> = (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<WSMessageType, MessageHandler[]> = 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<boolean> {
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<T extends WSMessageType>(type: T, message: Extract<WSMessage, { type: T }>) {
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<T extends WSMessageType>(type: T, handler: MessageHandler<Extract<WSMessage, { type: T }>>): () => 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<boolean> {
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();

2
src/services/systemNotificationService.ts
View File

@@ -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';

807
src/services/wsService.ts Normal file
View File

@@ -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<T extends WSMessage = WSMessage> = (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<WSMessageType, MessageHandler[]> = 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<boolean> {
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<any> {
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<void> {
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<any> {
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<any> {
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<any> {
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<T extends WSMessageType>(type: T, message: Extract<WSMessage, { type: T }>) {
const handlers = this.messageHandlers.get(type) || [];
handlers.forEach(h => h(message as WSMessage));
}
on<T extends WSMessageType>(type: T, handler: MessageHandler<Extract<WSMessage, { type: T }>>): () => 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<boolean> {
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();

6
src/stores/authStore.ts
View File

@@ -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<void> {
try {
await sseService.start();
await wsService.start();
} catch (error) {
console.error('[AuthStore] 启动 SSE 服务失败:', error);
}
@@ -212,7 +212,7 @@ export const useAuthStore = create<AuthState>((set) => ({
// 1. 通知服务端（Token 清理在 authService 内部完成）
await authService.logout();
// 2. 停止 SSE
sseService.stop();
wsService.stop();
// 3. 清除 DB 中的用户缓存（DB 此时一定已初始化）
await clearCurrentUserCache().catch(() => {});
// 4. 关闭数据库连接

26
src/stores/message/MessageManager.ts
View File

@@ -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();
}

46
src/stores/message/SSEMessageHandler.ts → src/stores/message/WSMessageHandler.ts
View File

@@ -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<SSEEventHandler> = new Set();
private handlers: Set<WSEventHandler> = 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();
export const wsMessageHandler = new WSMessageHandler();

4
src/stores/message/index.ts
View File

@@ -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';

36
src/stores/messageManager.ts
View File

@@ -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(),
});

4
src/theme/palettes.ts
View File

@@ -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',

30
src/types/dto.ts
View File

@@ -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<string, any>;
}
// 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中提取纯文本内容
* 用于消息列表显示、通知等场景