/** * MessageManager - 消息管理核心模块(重构版) * * 作为轻量级协调者,整合各个专注的服务模块 * 直接使用 zustand store 进行状态管理 * * 重构说明: * - 移除了 MessageStateManager 包装层 * - 直接使用 zustand store * - 服务层移至 services/ 目录 * - 移除了 SubscriptionManager,改用 Zustand selector 进行响应式订阅 */ import type { ConversationResponse, MessageResponse, MessageSegment, UserDTO } from '../../types/dto'; import { useAuthStore } from '../authStore'; import type { MessageManagerConversationListDeps } from './types'; // 导入服务 import { MessageDeduplication, messageDeduplication } from './services/MessageDeduplication'; import { UserCacheService, userCacheService } from './services/UserCacheService'; import { ReadReceiptManager } from './services/ReadReceiptManager'; import { ConversationOperations } from './services/ConversationOperations'; import { MessageSendService } from './services/MessageSendService'; import { MessageSyncService } from './services/MessageSyncService'; import { WSMessageHandler } from './services/WSMessageHandler'; // 导入 store import { useMessageStore, normalizeConversationId } from './store'; // 重新导出类型,保持兼容性 export type { MessageManagerState, ReadStateRecord, MessageManagerConversationListDeps, } from './types'; // ==================== MessageManager 类 ==================== class MessageManager { // 子模块 private deduplication: MessageDeduplication; private userCacheService: UserCacheService; private readReceiptManager: ReadReceiptManager; private conversationOps: ConversationOperations; private sendService: MessageSendService; private syncService: MessageSyncService; private wsHandler: WSMessageHandler; // 本地状态 private currentUserId: string | null = null; private authUnsubscribe: (() => void) | null = null; private initializePromise: Promise | null = null; private activatingConversationTasks: Map> = new Map(); constructor(deps?: MessageManagerConversationListDeps) { // 初始化子模块 this.deduplication = new MessageDeduplication(); this.userCacheService = new UserCacheService(); this.readReceiptManager = new ReadReceiptManager(); this.conversationOps = new ConversationOperations(); this.sendService = new MessageSendService(() => this.getCurrentUserId()); this.syncService = new MessageSyncService( () => this.getCurrentUserId(), this.readReceiptManager, this.userCacheService, deps ); this.wsHandler = new WSMessageHandler( this.deduplication, this.userCacheService, () => this.getCurrentUserId(), { markAsRead: (id, seq) => this.markAsRead(id, seq), fetchConversations: (force, source) => this.fetchConversations(force, source), fetchUnreadCount: () => this.fetchUnreadCount(), fetchMessages: (id) => this.fetchMessages(id), } ); // 监听认证状态变化 this.initAuthListener(); } // ==================== 私有工具方法 ==================== private initAuthListener() { const authState = useAuthStore.getState(); this.currentUserId = authState.currentUser?.id || null; const store = useMessageStore.getState(); if (this.currentUserId && !store.isInitialized) { this.initialize(); } if (!this.authUnsubscribe) { this.authUnsubscribe = useAuthStore.subscribe((state) => { const nextUserId = state.currentUser?.id || null; const prevUserId = this.currentUserId; this.currentUserId = nextUserId; if (nextUserId && !useMessageStore.getState().isInitialized) { this.initialize(); } if (nextUserId && useMessageStore.getState().currentConversationId) { this.activateConversation(useMessageStore.getState().currentConversationId!, { forceSync: true }).catch(error => { console.error('[MessageManager] 登录态就绪后重激活会话失败:', error); }); } if (!nextUserId && prevUserId && useMessageStore.getState().isInitialized) { this.destroy(); } }); } } private getCurrentUserId(): string | null { if (!this.currentUserId) { this.currentUserId = useAuthStore.getState().currentUser?.id || null; } return this.currentUserId; } // ==================== 初始化与销毁 ==================== async initialize(): Promise { const store = useMessageStore.getState(); if (store.isInitialized) { return; } if (this.initializePromise) { await this.initializePromise; return; } this.initializePromise = (async () => { try { this.currentUserId = this.getCurrentUserId(); if (!this.currentUserId) { console.warn('[MessageManager] 用户未登录,延迟初始化'); return; } this.wsHandler.setBootstrapping(true); // 初始化SSE监听 this.wsHandler.connect(); // 加载会话列表 await this.fetchConversations(false, 'initialize'); // 加载未读数 await this.fetchUnreadCount(); useMessageStore.getState().setInitialized(true); // 处理缓冲的 SSE 事件 await this.wsHandler.flushBufferedSSEEvents(); this.wsHandler.setBootstrapping(false); if (useMessageStore.getState().currentConversationId) { await this.fetchMessages(useMessageStore.getState().currentConversationId!); } } catch (error) { console.error('[MessageManager] 初始化失败:', error); this.wsHandler.setBootstrapping(false); } })(); try { await this.initializePromise; } finally { this.initializePromise = null; } } destroy(): void { this.wsHandler.disconnect(); useMessageStore.getState().reset(); this.syncService.restartSources(); this.readReceiptManager.reset(); this.deduplication.reset(); this.userCacheService.reset(); this.activatingConversationTasks.clear(); this.initializePromise = null; } // ==================== 数据获取方法(代理到 syncService)==================== async fetchConversations(forceRefresh = false, source: string = 'unknown'): Promise { return this.syncService.fetchConversations(forceRefresh, source); } async loadMoreConversations(): Promise { return this.syncService.loadMoreConversations(); } async fetchConversationDetail(conversationId: string): Promise { return this.syncService.fetchConversationDetail(conversationId); } async fetchMessages(conversationId: string, afterSeq?: number): Promise { return this.syncService.fetchMessages(conversationId, afterSeq); } async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise { return this.syncService.loadMoreMessages(conversationId, beforeSeq, limit); } async fetchUnreadCount(): Promise { return this.syncService.fetchUnreadCount(); } // ==================== 数据操作方法(代理到各服务)==================== async sendMessage( conversationId: string, segments: MessageSegment[], options?: { replyToId?: string } ): Promise { return this.sendService.sendMessage(conversationId, segments, options); } async markAsRead(conversationId: string, seq: number): Promise { return this.readReceiptManager.markAsRead(conversationId, seq); } async markAllAsRead(): Promise { return this.readReceiptManager.markAllAsRead(); } async createConversation(userId: string): Promise { return this.conversationOps.createConversation(userId); } updateConversation(conversationId: string, updates: Partial): void { return this.conversationOps.updateConversation(conversationId, updates); } removeConversation(conversationId: string): void { return this.conversationOps.removeConversation(conversationId); } clearConversations(): void { return this.conversationOps.clearConversations(); } updateUnreadCount(conversationId: string, count: number): void { return this.conversationOps.updateUnreadCount(conversationId, count); } setSystemUnreadCount(count: number): void { return this.conversationOps.setSystemUnreadCount(count); } incrementSystemUnreadCount(): void { return this.conversationOps.incrementSystemUnreadCount(); } decrementSystemUnreadCount(count?: number): void { return this.conversationOps.decrementSystemUnreadCount(count); } // ==================== 状态查询方法(直接使用 store)==================== getConversations(): ConversationResponse[] { return useMessageStore.getState().getConversations(); } getConversation(conversationId: string): ConversationResponse | null { return useMessageStore.getState().getConversation(conversationId); } getMessages(conversationId: string): MessageResponse[] { return useMessageStore.getState().getMessages(conversationId); } getUnreadCount(): { total: number; system: number } { return useMessageStore.getState().getUnreadCount(); } isConnected(): boolean { return useMessageStore.getState().isConnected(); } isLoading(): boolean { return useMessageStore.getState().isLoading(); } isLoadingMessages(conversationId: string): boolean { return useMessageStore.getState().isLoadingMessages(conversationId); } getTypingUsers(groupId: string): string[] { return useMessageStore.getState().getTypingUsers(groupId); } isMuted(groupId: string): boolean { return useMessageStore.getState().isMuted(groupId); } setMutedStatus(groupId: string, isMuted: boolean): void { return useMessageStore.getState().setMutedStatus(groupId, isMuted); } // ==================== 活动会话管理 ==================== setActiveConversation(conversationId: string | null): void { const normalizedId = conversationId ? normalizeConversationId(conversationId) : null; useMessageStore.getState().setCurrentConversation(normalizedId); } getActiveConversation(): string | null { return useMessageStore.getState().currentConversationId; } async activateConversation(conversationId: string, options?: { forceSync?: boolean }): Promise { const normalizedId = normalizeConversationId(conversationId); if (!normalizedId) return; await this.initialize(); this.setActiveConversation(normalizedId); const existingTask = this.activatingConversationTasks.get(normalizedId); if (existingTask && !options?.forceSync) { await existingTask; return; } const task = (async () => { await this.fetchMessages(normalizedId); })(); this.activatingConversationTasks.set(normalizedId, task); try { await task; } finally { if (this.activatingConversationTasks.get(normalizedId) === task) { this.activatingConversationTasks.delete(normalizedId); } } } // ==================== 其他方法 ==================== canLoadMoreConversations(): boolean { return this.syncService.canLoadMoreConversations(); } async refreshConversations(forceRefresh: boolean, source: string): Promise { if (this.shouldDeferConversationRefresh(source, forceRefresh)) { return; } await this.fetchConversations(forceRefresh, source); } private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean { if (!this.getActiveConversation()) return false; if (!forceRefresh) return false; return source === 'sse-reconnect' || source === 'prefetch' || source === 'hooks-initial-refresh'; } } // ==================== 单例导出 ==================== export const messageManager = new MessageManager(); // 导出类以支持类型检查和测试 export { MessageManager }; export default messageManager;