From 94c11062f071459a58ad3f03bb90caf6d19286b9 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Tue, 31 Mar 2026 18:22:24 +0800 Subject: [PATCH] refactor(message): replace MessageStateManager with zustand store - Remove MessageStateManager wrapper layer in favor of direct zustand store - Move service modules to services/ subdirectory (ConversationOperations, MessageDeduplication, MessageSendService, MessageSyncService, ReadReceiptManager, UserCacheService, WSMessageHandler) - Add new zustand-based store with subscriptionManager for event handling - Introduce React hooks for message state (useConversations, useMessages, useUnreadCount, etc.) - Update exports in index.ts to reflect new module structure - Deprecate messageManager.ts entry point in favor of message/index.ts - Maintain backward compatibility with existing MessageManager API --- src/stores/index.ts | 5 +- src/stores/message/MessageManager.ts | 163 +++-- src/stores/message/MessageStateManager.ts | 260 -------- src/stores/message/hooks.ts | 557 ++++++++++++++++++ src/stores/message/index.ts | 91 ++- .../{ => services}/ConversationOperations.ts | 115 ++-- .../{ => services}/MessageDeduplication.ts | 6 +- .../{ => services}/MessageSendService.ts | 48 +- .../{ => services}/MessageSyncService.ts | 258 ++++---- .../{ => services}/ReadReceiptManager.ts | 66 +-- .../{ => services}/UserCacheService.ts | 10 +- .../{ => services}/WSMessageHandler.ts | 148 +++-- src/stores/message/services/index.ts | 26 + src/stores/message/store.ts | 428 ++++++++++++++ src/stores/messageManager.ts | 13 +- 15 files changed, 1488 insertions(+), 706 deletions(-) delete mode 100644 src/stores/message/MessageStateManager.ts create mode 100644 src/stores/message/hooks.ts rename src/stores/message/{ => services}/ConversationOperations.ts (60%) rename src/stores/message/{ => services}/MessageDeduplication.ts (91%) rename src/stores/message/{ => services}/MessageSendService.ts (69%) rename src/stores/message/{ => services}/MessageSyncService.ts (70%) rename src/stores/message/{ => services}/ReadReceiptManager.ts (80%) rename src/stores/message/{ => services}/UserCacheService.ts (91%) rename src/stores/message/{ => services}/WSMessageHandler.ts (80%) create mode 100644 src/stores/message/services/index.ts create mode 100644 src/stores/message/store.ts diff --git a/src/stores/index.ts b/src/stores/index.ts index e5d5a30..3db970f 100644 --- a/src/stores/index.ts +++ b/src/stores/index.ts @@ -13,13 +13,14 @@ export { } from './userStore'; // MessageManager 新架构导出（已替换 conversationStore） -export { messageManager } from './messageManager'; +// 重构：从 ./message 目录导入，移除了 messageManager.ts 重导出文件 +export { messageManager, MessageManager } from './message'; export type { MessageEvent, MessageEventType, MessageSubscriber, MessageManagerConversationListDeps, -} from './messageManager'; +} from './message'; export { CONVERSATION_LIST_PAGE_SIZE, type ConversationListPage, diff --git a/src/stores/message/MessageManager.ts b/src/stores/message/MessageManager.ts index dd1aab7..c4b409b 100644 --- a/src/stores/message/MessageManager.ts +++ b/src/stores/message/MessageManager.ts @@ -1,59 +1,30 @@ /** * MessageManager - 消息管理核心模块（重构版） * - * 作为协调者，整合各个专注的服务模块 - * 保持与原始版本完全兼容的 API + * 作为轻量级协调者，整合各个专注的服务模块 + * 直接使用 zustand store 进行状态管理 + * + * 重构说明： + * - 移除了 MessageStateManager 包装层 + * - 直接使用 zustand store + * - 服务层移至 services/ 目录 */ import type { ConversationResponse, MessageResponse, MessageSegment, UserDTO } from '../../types/dto'; -import { messageService } from '../../services/messageService'; -import { - wsService, - WSChatMessage, - WSGroupChatMessage, - WSReadMessage, - WSGroupReadMessage, - WSRecallMessage, - WSGroupRecallMessage, - WSGroupTypingMessage, - WSGroupNoticeMessage, - GroupNoticeType, -} from '../../services/wsService'; -import { - saveMessage, - saveMessagesBatch, - getMessagesByConversation, - getMaxSeq, - getMinSeq, - getMessagesBeforeSeq, - markConversationAsRead, - updateConversationCacheUnreadCount, - CachedMessage, - getUserCache, - saveUserCache, - updateMessageStatus, - deleteConversation as deleteConversationFromDb, - saveConversationsWithRelatedCache, -} from '../../services/database'; -import { api } from '../../services/api'; -import { vibrateOnMessage } from '../../services/messageVibrationService'; import { useAuthStore } from '../authStore'; -import { - type IConversationListPagedSource, - SqliteConversationListPagedSource, - createRemoteConversationListSource, - CONVERSATION_LIST_PAGE_SIZE, -} from '../conversationListSources'; +import type { MessageManagerConversationListDeps } from './types'; -// 导入新模块 -import { MessageStateManager } from './MessageStateManager'; -import { MessageDeduplication } from './MessageDeduplication'; -import { UserCacheService } from './UserCacheService'; -import { ReadReceiptManager } from './ReadReceiptManager'; -import { ConversationOperations } from './ConversationOperations'; -import { MessageSendService } from './MessageSendService'; -import { MessageSyncService } from './MessageSyncService'; -import { WSMessageHandler } from './WSMessageHandler'; +// 导入服务 +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, subscriptionManager, normalizeConversationId } from './store'; // 重新导出类型，保持兼容性 export type { @@ -69,7 +40,6 @@ export type { class MessageManager { // 子模块 - private stateManager: MessageStateManager; private deduplication: MessageDeduplication; private userCacheService: UserCacheService; private readReceiptManager: ReadReceiptManager; @@ -84,21 +54,20 @@ class MessageManager { private initializePromise: Promise | null = null; private activatingConversationTasks: Map> = new Map(); - constructor(deps?: import('./types').MessageManagerConversationListDeps) { + constructor(deps?: MessageManagerConversationListDeps) { // 初始化子模块 - this.stateManager = new MessageStateManager(); this.deduplication = new MessageDeduplication(); this.userCacheService = new UserCacheService(); - this.readReceiptManager = new ReadReceiptManager(this.stateManager); - this.conversationOps = new ConversationOperations(this.stateManager); - this.sendService = new MessageSendService(this.stateManager, () => this.getCurrentUserId()); + this.readReceiptManager = new ReadReceiptManager(); + this.conversationOps = new ConversationOperations(); + this.sendService = new MessageSendService(() => this.getCurrentUserId()); this.syncService = new MessageSyncService( - this.stateManager, () => this.getCurrentUserId(), + this.readReceiptManager, + this.userCacheService, deps ); this.wsHandler = new WSMessageHandler( - this.stateManager, this.deduplication, this.userCacheService, () => this.getCurrentUserId(), @@ -119,8 +88,9 @@ class MessageManager { private initAuthListener() { const authState = useAuthStore.getState(); this.currentUserId = authState.currentUser?.id || null; + const store = useMessageStore.getState(); - if (this.currentUserId && !this.stateManager.getState().isInitialized) { + if (this.currentUserId && !store.isInitialized) { this.initialize(); } @@ -130,17 +100,17 @@ class MessageManager { const prevUserId = this.currentUserId; this.currentUserId = nextUserId; - if (nextUserId && !this.stateManager.getState().isInitialized) { + if (nextUserId && !useMessageStore.getState().isInitialized) { this.initialize(); } - if (nextUserId && this.stateManager.getActiveConversation()) { - this.activateConversation(this.stateManager.getActiveConversation()!, { forceSync: true }).catch(error => { + if (nextUserId && useMessageStore.getState().currentConversationId) { + this.activateConversation(useMessageStore.getState().currentConversationId!, { forceSync: true }).catch(error => { console.error('[MessageManager] 登录态就绪后重激活会话失败:', error); }); } - if (!nextUserId && prevUserId && this.stateManager.getState().isInitialized) { + if (!nextUserId && prevUserId && useMessageStore.getState().isInitialized) { this.destroy(); } }); @@ -154,14 +124,12 @@ class MessageManager { return this.currentUserId; } - private normalizeConversationId(conversationId: string | number | null | undefined): string { - return conversationId == null ? '' : String(conversationId); - } - // ==================== 初始化与销毁 ==================== async initialize(): Promise { - if (this.stateManager.getState().isInitialized) { + const store = useMessageStore.getState(); + + if (store.isInitialized) { return; } @@ -189,26 +157,26 @@ class MessageManager { // 加载未读数 await this.fetchUnreadCount(); - this.stateManager.setInitialized(true); + useMessageStore.getState().setInitialized(true); // 处理缓冲的 SSE 事件 await this.wsHandler.flushBufferedSSEEvents(); this.wsHandler.setBootstrapping(false); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: this.getConversations() }, timestamp: Date.now(), }); - if (this.stateManager.getActiveConversation()) { - await this.fetchMessages(this.stateManager.getActiveConversation()!); + if (useMessageStore.getState().currentConversationId) { + await this.fetchMessages(useMessageStore.getState().currentConversationId!); } } catch (error) { console.error('[MessageManager] 初始化失败:', error); this.wsHandler.setBootstrapping(false); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'error', payload: { error, context: 'initialize' }, timestamp: Date.now(), @@ -225,7 +193,8 @@ class MessageManager { destroy(): void { this.wsHandler.disconnect(); - this.stateManager.reset(); + useMessageStore.getState().reset(); + subscriptionManager.clear(); this.syncService.restartSources(); this.readReceiptManager.reset(); this.deduplication.reset(); @@ -310,61 +279,61 @@ class MessageManager { return this.conversationOps.decrementSystemUnreadCount(count); } - // ==================== 状态查询方法（代理到 stateManager）==================== + // ==================== 状态查询方法（直接使用 store）==================== getConversations(): ConversationResponse[] { - return this.stateManager.getConversations(); + return useMessageStore.getState().getConversations(); } getConversation(conversationId: string): ConversationResponse | null { - return this.stateManager.getConversation(conversationId); + return useMessageStore.getState().getConversation(conversationId); } getMessages(conversationId: string): MessageResponse[] { - return this.stateManager.getMessages(conversationId); + return useMessageStore.getState().getMessages(conversationId); } getUnreadCount(): { total: number; system: number } { - return this.stateManager.getUnreadCount(); + return useMessageStore.getState().getUnreadCount(); } isConnected(): boolean { - return this.stateManager.isConnected(); + return useMessageStore.getState().isConnected(); } isLoading(): boolean { - return this.stateManager.isLoading(); + return useMessageStore.getState().isLoading(); } isLoadingMessages(conversationId: string): boolean { - return this.stateManager.isLoadingMessages(conversationId); + return useMessageStore.getState().isLoadingMessages(conversationId); } getTypingUsers(groupId: string): string[] { - return this.stateManager.getTypingUsers(groupId); + return useMessageStore.getState().getTypingUsers(groupId); } isMuted(groupId: string): boolean { - return this.stateManager.isMuted(groupId); + return useMessageStore.getState().isMuted(groupId); } setMutedStatus(groupId: string, isMuted: boolean): void { - return this.stateManager.setMutedStatus(groupId, isMuted); + return useMessageStore.getState().setMutedStatus(groupId, isMuted); } // ==================== 活动会话管理 ==================== setActiveConversation(conversationId: string | null): void { - const normalizedId = conversationId ? this.normalizeConversationId(conversationId) : null; - this.stateManager.setCurrentConversation(normalizedId); + const normalizedId = conversationId ? normalizeConversationId(conversationId) : null; + useMessageStore.getState().setCurrentConversation(normalizedId); } getActiveConversation(): string | null { - return this.stateManager.getActiveConversation(); + return useMessageStore.getState().currentConversationId; } async activateConversation(conversationId: string, options?: { forceSync?: boolean }): Promise { - const normalizedId = this.normalizeConversationId(conversationId); + const normalizedId = normalizeConversationId(conversationId); if (!normalizedId) return; await this.initialize(); @@ -372,7 +341,7 @@ class MessageManager { // 先下发当前内存快照 const snapshot = this.getMessages(normalizedId); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId: normalizedId, @@ -405,16 +374,16 @@ class MessageManager { // ==================== 订阅机制 ==================== subscribe(subscriber: import('./types').MessageSubscriber): () => void { - const unsubscribe = this.stateManager.subscribe(subscriber); + const unsubscribe = subscriptionManager.subscribe(subscriber); // 立即发送当前状态给新订阅者（与原始版本保持一致） subscriber({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: this.getConversations() }, timestamp: Date.now(), }); - const unreadCount = this.stateManager.getUnreadCount(); + const unreadCount = this.getUnreadCount(); subscriber({ type: 'unread_count_updated', payload: { @@ -426,14 +395,14 @@ class MessageManager { subscriber({ type: 'connection_changed', - payload: { connected: this.stateManager.isConnected() }, + payload: { connected: this.isConnected() }, timestamp: Date.now(), }); // 如果当前有活动会话，立即发送该会话的消息给新订阅者 - const activeConversationId = this.stateManager.getActiveConversation(); + const activeConversationId = this.getActiveConversation(); if (activeConversationId) { - const currentMessages = this.stateManager.getMessages(activeConversationId); + const currentMessages = this.getMessages(activeConversationId); if (currentMessages.length > 0) { subscriber({ type: 'messages_updated', @@ -451,7 +420,7 @@ class MessageManager { } subscribeToMessages(conversationId: string, callback: (messages: MessageResponse[]) => void): () => void { - return this.stateManager.subscribe((event) => { + return subscriptionManager.subscribe((event) => { if (event.type === 'messages_updated' && event.payload.conversationId === conversationId) { callback(event.payload.messages); } @@ -484,7 +453,7 @@ class MessageManager { } private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean { - if (!this.stateManager.getActiveConversation()) return false; + if (!this.getActiveConversation()) return false; if (!forceRefresh) return false; return source === 'sse-reconnect' || source === 'prefetch' || source === 'hooks-initial-refresh'; } diff --git a/src/stores/message/MessageStateManager.ts b/src/stores/message/MessageStateManager.ts deleted file mode 100644 index eea7269..0000000 --- a/src/stores/message/MessageStateManager.ts +++ /dev/null @@ -1,260 +0,0 @@ -/** - * 消息状态管理器 - * 纯状态管理类，不包含业务逻辑 - * 负责管理所有消息相关的内存状态 - */ - -import type { - ConversationResponse, - MessageResponse, -} from '../../types/dto'; -import type { - MessageManagerState, - MessageSubscriber, - MessageEvent, - IMessageStateManager, -} from './types'; - -export class MessageStateManager implements IMessageStateManager { - private state: MessageManagerState; - - constructor() { - this.state = { - conversations: new Map(), - conversationList: [], - messagesMap: new Map(), - totalUnreadCount: 0, - systemUnreadCount: 0, - isWSConnected: false, - currentConversationId: null, - isLoadingConversations: false, - loadingMessagesSet: new Set(), - isInitialized: false, - subscribers: new Set(), - typingUsersMap: new Map(), - mutedStatusMap: new Map(), - }; - } - - // ==================== 获取状态 ==================== - - getState(): MessageManagerState { - return this.state; - } - - getConversations(): ConversationResponse[] { - return this.state.conversationList; - } - - getConversation(conversationId: string): ConversationResponse | null { - const normalizedId = this.normalizeConversationId(conversationId); - return this.state.conversations.get(normalizedId) || null; - } - - getMessages(conversationId: string): MessageResponse[] { - const normalizedId = this.normalizeConversationId(conversationId); - return this.state.messagesMap.get(normalizedId) || []; - } - - getUnreadCount(): { total: number; system: number } { - return { - total: this.state.totalUnreadCount, - system: this.state.systemUnreadCount, - }; - } - - isConnected(): boolean { - return this.state.isWSConnected; - } - - isLoading(): boolean { - return this.state.isLoadingConversations; - } - - isLoadingMessages(conversationId: string): boolean { - const normalizedId = this.normalizeConversationId(conversationId); - return this.state.loadingMessagesSet.has(normalizedId); - } - - getTypingUsers(groupId: string): string[] { - return this.state.typingUsersMap.get(groupId) || []; - } - - isMuted(groupId: string): boolean { - return this.state.mutedStatusMap.get(groupId) || false; - } - - getActiveConversation(): string | null { - return this.state.currentConversationId; - } - - // ==================== 设置状态 ==================== - - setConversations(conversations: Map): void { - this.state.conversations = conversations; - this.updateConversationList(); - } - - updateConversation(conversation: ConversationResponse): void { - const id = this.normalizeConversationId(conversation.id); - this.state.conversations.set(id, { ...conversation, id }); - this.updateConversationList(); - } - - removeConversation(conversationId: string): void { - const normalizedId = this.normalizeConversationId(conversationId); - const target = this.state.conversations.get(normalizedId); - const removedUnread = target?.unread_count || 0; - - this.state.conversations.delete(normalizedId); - this.state.messagesMap.delete(normalizedId); - this.state.totalUnreadCount = Math.max(0, this.state.totalUnreadCount - removedUnread); - - if (this.state.currentConversationId === normalizedId) { - this.state.currentConversationId = null; - } - - this.updateConversationList(); - } - - setMessages(conversationId: string, messages: MessageResponse[]): void { - const normalizedId = this.normalizeConversationId(conversationId); - this.state.messagesMap.set(normalizedId, messages); - } - - addMessage(conversationId: string, message: MessageResponse): void { - const normalizedId = this.normalizeConversationId(conversationId); - const existing = this.state.messagesMap.get(normalizedId) || []; - const exists = existing.some(m => String(m.id) === String(message.id)); - - if (!exists) { - const updated = this.mergeMessagesById(existing, [message]); - this.state.messagesMap.set(normalizedId, updated); - } - } - - updateMessage(conversationId: string, messageId: string, updates: Partial): void { - const normalizedId = this.normalizeConversationId(conversationId); - const messages = this.state.messagesMap.get(normalizedId); - if (!messages) return; - - const updated = messages.map(m => - String(m.id) === String(messageId) ? { ...m, ...updates } : m - ); - this.state.messagesMap.set(normalizedId, updated); - } - - setUnreadCount(total: number, system: number): void { - this.state.totalUnreadCount = total; - this.state.systemUnreadCount = system; - } - - setSSEConnected(connected: boolean): void { - this.state.isWSConnected = connected; - } - - setCurrentConversation(conversationId: string | null): void { - this.state.currentConversationId = conversationId; - } - - setLoading(loading: boolean): void { - this.state.isLoadingConversations = loading; - } - - setLoadingMessages(conversationId: string, loading: boolean): void { - const normalizedId = this.normalizeConversationId(conversationId); - if (loading) { - this.state.loadingMessagesSet.add(normalizedId); - } else { - this.state.loadingMessagesSet.delete(normalizedId); - } - } - - setTypingUsers(groupId: string, users: string[]): void { - this.state.typingUsersMap.set(groupId, users); - } - - setMutedStatus(groupId: string, isMuted: boolean): void { - this.state.mutedStatusMap.set(groupId, isMuted); - } - - setInitialized(initialized: boolean): void { - this.state.isInitialized = initialized; - } - - // ==================== 订阅机制 ==================== - - subscribe(subscriber: MessageSubscriber): () => void { - this.state.subscribers.add(subscriber); - return () => { - this.state.subscribers.delete(subscriber); - }; - } - - notifySubscribers(event: MessageEvent): void { - this.state.subscribers.forEach(subscriber => { - try { - subscriber(event); - } catch (error) { - console.error('[MessageStateManager] 订阅者执行失败:', error); - } - }); - } - - // ==================== 工具方法 ==================== - - private normalizeConversationId(conversationId: string | number | null | undefined): string { - return conversationId == null ? '' : String(conversationId); - } - - /** - * 更新会话列表排序 - * 排序规则：置顶优先，再按最后消息时间排序 - */ - private updateConversationList(): void { - const list = Array.from(this.state.conversations.values()).sort((a, b) => { - const aPinned = a.is_pinned ? 1 : 0; - const bPinned = b.is_pinned ? 1 : 0; - if (aPinned !== bPinned) { - return bPinned - aPinned; - } - const aTime = new Date(a.last_message_at || a.updated_at || 0).getTime(); - const bTime = new Date(b.last_message_at || b.updated_at || 0).getTime(); - return bTime - aTime; - }); - this.state.conversationList = list; - } - - /** - * 按 message_id 合并消息并按 seq 排序 - * 后者覆盖前者，可用来吸收更完整的同 ID 消息体 - */ - private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] { - if (incoming.length === 0) return base; - const merged = new Map(); - [...base, ...incoming].forEach(msg => { - merged.set(String(msg.id), msg); - }); - return Array.from(merged.values()).sort((a, b) => a.seq - b.seq); - } - - // ==================== 重置 ==================== - - reset(): void { - this.state.conversations.clear(); - this.state.conversationList = []; - this.state.messagesMap.clear(); - this.state.totalUnreadCount = 0; - this.state.systemUnreadCount = 0; - this.state.isWSConnected = false; - this.state.currentConversationId = null; - this.state.isLoadingConversations = false; - this.state.loadingMessagesSet.clear(); - this.state.isInitialized = false; - this.state.typingUsersMap.clear(); - this.state.mutedStatusMap.clear(); - } -} - -// 单例导出 -export const messageStateManager = new MessageStateManager(); diff --git a/src/stores/message/hooks.ts b/src/stores/message/hooks.ts new file mode 100644 index 0000000..6078c2e --- /dev/null +++ b/src/stores/message/hooks.ts @@ -0,0 +1,557 @@ +/** + * 消息模块 React Hooks + * + * 提供React组件与消息状态管理的集成 + * 所有hooks都基于zustand store和订阅机制 + * 确保组件能实时获取状态更新 + * + * 重构说明：直接使用 zustand store，移除了对 messageManager 的直接依赖 + */ + +import { useState, useEffect, useCallback, useRef } from 'react'; +import { ConversationResponse, MessageResponse, MessageSegment } from '../../types/dto'; +import { useMessageStore, subscriptionManager, normalizeConversationId } from './store'; +import type { MessageEvent, MessageSubscriber } from './types'; + +// ==================== useConversations - 获取会话列表 ==================== + +interface UseConversationsReturn { + conversations: ConversationResponse[]; + isLoading: boolean; + refresh: () => Promise; + loadMore: () => Promise; + hasMore: boolean; +} + +/** + * 获取会话列表 + * 用于 MessageListScreen 等需要显示会话列表的组件 + */ +export function useConversations( + fetchConversations: (forceRefresh: boolean, source: string) => Promise, + loadMoreConversations: () => Promise, + canLoadMore: () => boolean, + initialize: () => Promise +): UseConversationsReturn { + const store = useMessageStore(); + const [conversations, setConversations] = useState(() => store.getConversations()); + const [isLoading, setIsLoading] = useState(() => store.isLoading()); + const [hasMore, setHasMore] = useState(() => canLoadMore()); + + useEffect(() => { + // 订阅事件 + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + switch (event.type) { + case 'conversations_updated': + setConversations(store.getConversations()); + setHasMore(canLoadMore()); + break; + case 'conversations_loading': + setIsLoading(!!event.payload?.loading); + break; + case 'connection_changed': + // 连接状态变化时可能需要刷新 + break; + } + }); + + // 初始化 + initialize().catch(error => { + console.error('[useConversations] 初始化失败:', error); + }); + + // 冷启动兜底：延迟做一次强制刷新 + const coldStartSyncTimer = setTimeout(() => { + fetchConversations(true, 'hooks-initial-refresh').catch(error => { + console.error('[useConversations] 冷启动强制刷新失败:', error); + }); + }, 1200); + + return () => { + clearTimeout(coldStartSyncTimer); + unsubscribe(); + }; + }, []); + + const refresh = useCallback(async () => { + await fetchConversations(true, 'hooks-manual-refresh'); + }, [fetchConversations]); + + const loadMore = useCallback(async () => { + await loadMoreConversations(); + }, [loadMoreConversations]); + + return { + conversations, + isLoading, + refresh, + loadMore, + hasMore, + }; +} + +// ==================== useMessages - 获取指定会话的消息 ==================== + +interface UseMessagesReturn { + messages: MessageResponse[]; + isLoading: boolean; + hasMore: boolean; + loadMore: () => Promise; + refresh: () => Promise; +} + +/** + * 获取指定会话的消息 + */ +export function useMessages( + conversationId: string, + fetchMessages: (conversationId: string) => Promise, + loadMoreMessages: (conversationId: string, beforeSeq: number, limit?: number) => Promise +): UseMessagesReturn { + const store = useMessageStore(); + const normalizedConversationId = normalizeConversationId(conversationId); + + const [messages, setMessages] = useState(() => + store.getMessages(normalizedConversationId) + ); + const [isLoading, setIsLoading] = useState(() => + store.isLoadingMessages(normalizedConversationId) + ); + const [hasMore, setHasMore] = useState(true); + + useEffect(() => { + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'messages_updated' && event.payload.conversationId === normalizedConversationId) { + setMessages(event.payload.messages); + } + }); + + // 激活会话 + fetchMessages(normalizedConversationId).catch(error => { + console.error('[useMessages] 激活会话失败:', error); + }); + + return () => { + // 清理活动会话 + if (store.getActiveConversation() === normalizedConversationId) { + store.setCurrentConversation(null); + } + unsubscribe(); + }; + }, [normalizedConversationId]); + + const loadMore = useCallback(async () => { + const currentMessages = store.getMessages(conversationId); + if (currentMessages.length === 0) return; + + // 消息数组在 store 内保持 seq 升序，首项即最早消息 + const minSeq = currentMessages[0]?.seq; + if (minSeq === undefined) return; + + const loadedMessages = await loadMoreMessages(conversationId, minSeq, 20); + setHasMore(loadedMessages.length > 0); + }, [conversationId, loadMoreMessages]); + + const refresh = useCallback(async () => { + await fetchMessages(conversationId); + setMessages(store.getMessages(conversationId)); + }, [conversationId, fetchMessages]); + + return { + messages, + isLoading, + hasMore, + loadMore, + refresh, + }; +} + +// ==================== useSendMessage - 发送消息 ==================== + +interface UseSendMessageReturn { + sendMessage: (segments: MessageSegment[], options?: { replyToId?: string }) => Promise; + isSending: boolean; +} + +/** + * 发送消息 + */ +export function useSendMessage( + conversationId: string, + sendMessageService: (conversationId: string, segments: MessageSegment[], options?: { replyToId?: string }) => Promise +): UseSendMessageReturn { + const [isSending, setIsSending] = useState(false); + + const sendMessage = useCallback(async (segments: MessageSegment[], options?: { replyToId?: string }) => { + setIsSending(true); + try { + const message = await sendMessageService(conversationId, segments, options); + return message; + } catch (error) { + console.error('[useSendMessage] 发送消息失败:', error); + return null; + } finally { + setIsSending(false); + } + }, [conversationId, sendMessageService]); + + return { + sendMessage, + isSending, + }; +} + +// ==================== useMarkAsRead - 标记已读 ==================== + +interface UseMarkAsReadReturn { + markAsRead: (seq: number) => Promise; + markAllAsRead: () => Promise; +} + +/** + * 标记已读 + */ +export function useMarkAsRead( + conversationId: string, + markAsReadService: (conversationId: string, seq: number) => Promise, + markAllAsReadService: () => Promise +): UseMarkAsReadReturn { + const markAsRead = useCallback(async (seq: number) => { + await markAsReadService(conversationId, seq); + }, [conversationId, markAsReadService]); + + const markAllAsRead = useCallback(async () => { + await markAllAsReadService(); + }, [markAllAsReadService]); + + return { + markAsRead, + markAllAsRead, + }; +} + +// ==================== useUnreadCount - 获取未读数 ==================== + +interface UseUnreadCountReturn { + totalUnreadCount: number; + systemUnreadCount: number; + refresh: () => Promise; +} + +/** + * 获取未读数 + */ +export function useUnreadCount(fetchUnreadCount: () => Promise): UseUnreadCountReturn { + const store = useMessageStore(); + const [counts, setCounts] = useState(() => store.getUnreadCount()); + + useEffect(() => { + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'unread_count_updated') { + setCounts(store.getUnreadCount()); + } + }); + + // 初始获取 + fetchUnreadCount(); + + return unsubscribe; + }, []); + + const refresh = useCallback(async () => { + await fetchUnreadCount(); + }, [fetchUnreadCount]); + + return { + totalUnreadCount: counts.total, + systemUnreadCount: counts.system, + refresh, + }; +} + +// ==================== useSystemUnreadCount - 获取系统消息未读数 ==================== + +interface UseSystemUnreadCountReturn { + systemUnreadCount: number; + setSystemUnreadCount: (count: number) => void; + incrementSystemUnreadCount: () => void; + decrementSystemUnreadCount: (count?: number) => void; +} + +/** + * 获取系统消息未读数 + */ +export function useSystemUnreadCount( + setSystemUnreadCountService: (count: number) => void, + incrementSystemUnreadCountService: () => void, + decrementSystemUnreadCountService: (count?: number) => void +): UseSystemUnreadCountReturn { + const store = useMessageStore(); + const [systemUnreadCount, setSystemUnreadCountState] = useState(() => + store.getUnreadCount().system + ); + + useEffect(() => { + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'unread_count_updated') { + setSystemUnreadCountState(store.getUnreadCount().system); + } + }); + + return unsubscribe; + }, []); + + const setSystemUnreadCount = useCallback((count: number) => { + setSystemUnreadCountService(count); + }, [setSystemUnreadCountService]); + + const incrementSystemUnreadCount = useCallback(() => { + incrementSystemUnreadCountService(); + }, [incrementSystemUnreadCountService]); + + const decrementSystemUnreadCount = useCallback((count = 1) => { + decrementSystemUnreadCountService(count); + }, [decrementSystemUnreadCountService]); + + return { + systemUnreadCount, + setSystemUnreadCount, + incrementSystemUnreadCount, + decrementSystemUnreadCount, + }; +} + +// ==================== useConversation - 获取单个会话 ==================== + +interface UseConversationReturn { + conversation: ConversationResponse | null; + refresh: () => Promise; +} + +/** + * 获取单个会话 + */ +export function useConversation( + conversationId: string, + fetchConversationDetail: (conversationId: string) => Promise +): UseConversationReturn { + const store = useMessageStore(); + const [conversation, setConversation] = useState(() => + store.getConversation(conversationId) + ); + + useEffect(() => { + setConversation(store.getConversation(conversationId)); + + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'conversations_updated') { + const updated = store.getConversation(conversationId); + setConversation(updated); + } + }); + + return unsubscribe; + }, [conversationId]); + + const refresh = useCallback(async () => { + await fetchConversationDetail(conversationId); + setConversation(store.getConversation(conversationId)); + }, [conversationId, fetchConversationDetail]); + + return { + conversation, + refresh, + }; +} + +// ==================== useMessageManager - 通用订阅 ==================== + +interface UseMessageManagerReturn { + isConnected: boolean; + subscribe: (subscriber: MessageSubscriber) => () => void; +} + +/** + * 通用消息订阅 + */ +export function useMessageManager(initialize: () => Promise): UseMessageManagerReturn { + const store = useMessageStore(); + const [isConnected, setIsConnected] = useState(() => store.isConnected()); + + useEffect(() => { + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'connection_changed') { + setIsConnected(event.payload.connected); + } + }); + + initialize().then(() => { + setIsConnected(store.isConnected()); + }).catch(error => { + console.error('[useMessageManager] 初始化失败:', error); + }); + + return unsubscribe; + }, []); + + const subscribe = useCallback((subscriber: MessageSubscriber) => { + return subscriptionManager.subscribe(subscriber); + }, []); + + return { + isConnected, + subscribe, + }; +} + +// ==================== useCreateConversation - 创建会话 ==================== + +interface UseCreateConversationReturn { + createConversation: (userId: string) => Promise; + isCreating: boolean; +} + +/** + * 创建会话 + */ +export function useCreateConversation( + createConversationService: (userId: string) => Promise +): UseCreateConversationReturn { + const [isCreating, setIsCreating] = useState(false); + + const createConversation = useCallback(async (userId: string) => { + setIsCreating(true); + try { + const conversation = await createConversationService(userId); + return conversation; + } catch (error) { + console.error('[useCreateConversation] 创建会话失败:', error); + return null; + } finally { + setIsCreating(false); + } + }, [createConversationService]); + + return { + createConversation, + isCreating, + }; +} + +// ==================== useUpdateConversation - 更新会话 ==================== + +interface UseUpdateConversationReturn { + updateConversation: (conversationId: string, updates: Partial) => void; +} + +/** + * 更新会话 + */ +export function useUpdateConversation( + updateConversationService: (conversationId: string, updates: Partial) => void +): UseUpdateConversationReturn { + const updateConversation = useCallback((conversationId: string, updates: Partial) => { + updateConversationService(conversationId, updates); + }, [updateConversationService]); + + return { + updateConversation, + }; +} + +// ==================== useGroupTyping - 群聊输入状态 ==================== + +interface UseGroupTypingReturn { + typingUsers: string[]; +} + +/** + * 获取群聊输入状态 + */ +export function useGroupTyping(groupId: string): UseGroupTypingReturn { + const store = useMessageStore(); + const [typingUsers, setTypingUsers] = useState(() => + store.getTypingUsers(groupId) + ); + + useEffect(() => { + setTypingUsers(store.getTypingUsers(groupId)); + + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'typing_status' && event.payload.groupId === groupId) { + setTypingUsers(event.payload.typingUsers); + } + }); + + return unsubscribe; + }, [groupId]); + + return { + typingUsers, + }; +} + +// ==================== useGroupMuted - 群聊禁言状态 ==================== + +interface UseGroupMutedReturn { + isMuted: boolean; + setMutedStatus: (muted: boolean) => void; +} + +/** + * 获取群聊禁言状态 + */ +export function useGroupMuted(groupId: string): UseGroupMutedReturn { + const store = useMessageStore(); + const [isMuted, setIsMuted] = useState(() => store.isMuted(groupId)); + + useEffect(() => { + setIsMuted(store.isMuted(groupId)); + + const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { + if (event.type === 'group_notice' && event.payload.groupId === groupId) { + if (event.payload.noticeType === 'muted' || event.payload.noticeType === 'unmuted') { + setIsMuted(event.payload.noticeType === 'muted'); + } + } + }); + + return unsubscribe; + }, [groupId]); + + const setMutedStatus = useCallback((muted: boolean) => { + store.setMutedStatus(groupId, muted); + }, [groupId]); + + return { + isMuted, + setMutedStatus, + }; +} + +// ==================== useActiveConversation - 活动会话 ==================== + +interface UseActiveConversationReturn { + activeConversationId: string | null; + setActiveConversation: (conversationId: string | null) => void; +} + +/** + * 获取/设置活动会话 + */ +export function useActiveConversation(): UseActiveConversationReturn { + const store = useMessageStore(); + const [activeConversationId, setActiveConversationId] = useState(() => + store.getActiveConversation() + ); + + const setActiveConversation = useCallback((conversationId: string | null) => { + store.setCurrentConversation(conversationId); + setActiveConversationId(conversationId); + }, []); + + return { + activeConversationId, + setActiveConversation, + }; +} \ No newline at end of file diff --git a/src/stores/message/index.ts b/src/stores/message/index.ts index 82943fb..4297e6b 100644 --- a/src/stores/message/index.ts +++ b/src/stores/message/index.ts @@ -1,12 +1,27 @@ /** * 消息模块统一导出 * - * 提供向后兼容的 API 导出 + * 重构说明： + * - 移除了 MessageStateManager 包装层 + * - 直接使用 zustand store 进行状态管理 + * - 服务层移至 services/ 目录 + * - 新增 hooks.ts 封装常用逻辑 */ // ==================== 主管理器 ==================== export { messageManager, MessageManager } from './MessageManager'; +// ==================== 状态管理 ==================== +export { + useMessageStore, + subscriptionManager, + normalizeConversationId, + mergeMessagesById, + notifySubscribers, + subscribeToMessages, +} from './store'; +export type { MessageState, MessageActions, MessageStore } from './store'; + // ==================== 类型导出 ==================== // 从 types.ts 导出所有类型 export type { @@ -17,6 +32,7 @@ export type { ReadStateRecord, MessageManagerConversationListDeps, HandleNewMessageOptions, + // 服务接口类型（保留向后兼容，但服务层已不再使用） IMessageStateManager, IMessageDeduplication, IUserCacheService, @@ -37,30 +53,57 @@ export { MIN_RECONNECT_SYNC_INTERVAL, } from './constants'; -// ==================== 子模块导出（可选使用）==================== -// 状态管理器 -export { MessageStateManager, messageStateManager } from './MessageStateManager'; +// ==================== 服务导出 ==================== +export { + // 消息去重服务 + MessageDeduplication, + messageDeduplication, + + // 用户缓存服务 + UserCacheService, + userCacheService, + + // 已读回执管理器 + ReadReceiptManager, + + // 会话操作服务 + ConversationOperations, + + // 消息发送服务 + MessageSendService, + + // 消息同步服务 + MessageSyncService, + + // WebSocket 消息处理器 + WSMessageHandler, +} from './services'; -// 消息去重服务 -export { MessageDeduplication, messageDeduplication } from './MessageDeduplication'; - -// 用户缓存服务 -export { UserCacheService, userCacheService } from './UserCacheService'; - -// 已读回执管理器 -export { ReadReceiptManager } from './ReadReceiptManager'; - -// 会话操作服务 -export { ConversationOperations } from './ConversationOperations'; - -// 消息发送服务 -export { MessageSendService } from './MessageSendService'; - -// 消息同步服务 -export { MessageSyncService } from './MessageSyncService'; - -// WebSocket 消息处理器 -export { WSMessageHandler } from './WSMessageHandler'; +// ==================== Hooks 导出 ==================== +export { + // 会话相关 + useConversations, + useConversation, + useCreateConversation, + useUpdateConversation, + useActiveConversation, + + // 消息相关 + useMessages, + useSendMessage, + useMarkAsRead, + + // 未读数相关 + useUnreadCount, + useSystemUnreadCount, + + // 群聊相关 + useGroupTyping, + useGroupMuted, + + // 通用 + useMessageManager, +} from './hooks'; // ==================== 默认导出 ==================== export { default } from './MessageManager'; diff --git a/src/stores/message/ConversationOperations.ts b/src/stores/message/services/ConversationOperations.ts similarity index 60% rename from src/stores/message/ConversationOperations.ts rename to src/stores/message/services/ConversationOperations.ts index 30b1962..c9b69a3 100644 --- a/src/stores/message/ConversationOperations.ts +++ b/src/stores/message/services/ConversationOperations.ts @@ -1,41 +1,40 @@ /** * 会话操作服务 * 处理会话的 CRUD 操作 + * + * 重构说明：直接使用 zustand store 替代 IMessageStateManager */ -import type { ConversationResponse } from '../../types/dto'; -import { messageService } from '../../services/messageService'; -import { deleteConversation } from '../../services/database'; -import type { IConversationOperations, IMessageStateManager } from './types'; +import type { ConversationResponse } from '../../../types/dto'; +import { messageService } from '../../../services/messageService'; +import { deleteConversation } from '../../../services/database'; +import type { IConversationOperations } from '../types'; +import { useMessageStore, subscriptionManager, normalizeConversationId } from '../store'; export class ConversationOperations implements IConversationOperations { - private stateManager: IMessageStateManager; - - constructor(stateManager: IMessageStateManager) { - this.stateManager = stateManager; - } - /** * 创建私聊会话 */ async createConversation(userId: string): Promise { + const store = useMessageStore.getState(); + try { const conversation = await messageService.createConversation(userId); // 添加到会话列表 - this.stateManager.updateConversation(conversation); + store.updateConversation(conversation); // 通知更新 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); return conversation; } catch (error) { console.error('[ConversationOperations] 创建会话失败:', error); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'error', payload: { error, context: 'createConversation' }, timestamp: Date.now(), @@ -48,7 +47,8 @@ export class ConversationOperations implements IConversationOperations { * 本地更新会话信息 */ updateConversation(conversationId: string, updates: Partial): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); if (!conversation) { console.warn('[ConversationOperations] 更新会话失败，会话不存在:', conversationId); return; @@ -59,12 +59,12 @@ export class ConversationOperations implements IConversationOperations { ...updates, }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); // 通知更新 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); } @@ -73,17 +73,18 @@ export class ConversationOperations implements IConversationOperations { * 仅自己删除会话后的本地状态回收 */ removeConversation(conversationId: string): void { - this.stateManager.removeConversation(conversationId); + const store = useMessageStore.getState(); + store.removeConversation(conversationId); // 通知更新 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); - const unreadCount = this.stateManager.getUnreadCount(); - this.stateManager.notifySubscribers({ + const unreadCount = store.getUnreadCount(); + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: unreadCount.total, @@ -102,15 +103,16 @@ export class ConversationOperations implements IConversationOperations { * 清空会话列表 */ clearConversations(): void { - this.stateManager.reset(); + const store = useMessageStore.getState(); + store.reset(); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', payload: { conversations: [] }, timestamp: Date.now(), }); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: 0, @@ -124,7 +126,8 @@ export class ConversationOperations implements IConversationOperations { * 更新单个会话未读数 */ updateUnreadCount(conversationId: string, count: number): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); if (!conversation) { console.warn('[ConversationOperations] 更新未读数失败，会话不存在:', conversationId); return; @@ -138,28 +141,28 @@ export class ConversationOperations implements IConversationOperations { ...conversation, unread_count: count, }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); // 更新总未读数 - const currentUnread = this.stateManager.getUnreadCount(); - this.stateManager.setUnreadCount( + const currentUnread = store.getUnreadCount(); + store.setUnreadCount( Math.max(0, currentUnread.total + diff), currentUnread.system ); // 通知更新 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { - totalUnreadCount: this.stateManager.getUnreadCount().total, - systemUnreadCount: this.stateManager.getUnreadCount().system, + totalUnreadCount: store.getUnreadCount().total, + systemUnreadCount: store.getUnreadCount().system, }, timestamp: Date.now(), }); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); } @@ -168,7 +171,8 @@ export class ConversationOperations implements IConversationOperations { * 增加会话未读数 */ incrementUnreadCount(conversationId: string): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); if (!conversation) return; const prevUnreadCount = conversation.unread_count || 0; @@ -180,27 +184,27 @@ export class ConversationOperations implements IConversationOperations { unread_count: newUnreadCount, }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); - const currentUnread = this.stateManager.getUnreadCount(); - this.stateManager.setUnreadCount( + const currentUnread = store.getUnreadCount(); + store.setUnreadCount( currentUnread.total + 1, currentUnread.system ); // 通知更新 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { - totalUnreadCount: this.stateManager.getUnreadCount().total, - systemUnreadCount: this.stateManager.getUnreadCount().system, + totalUnreadCount: store.getUnreadCount().total, + systemUnreadCount: store.getUnreadCount().system, }, timestamp: Date.now(), }); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); } @@ -209,10 +213,11 @@ export class ConversationOperations implements IConversationOperations { * 设置系统消息未读数 */ setSystemUnreadCount(count: number): void { - const currentUnread = this.stateManager.getUnreadCount(); - this.stateManager.setUnreadCount(currentUnread.total, count); + const store = useMessageStore.getState(); + const currentUnread = store.getUnreadCount(); + store.setUnreadCount(currentUnread.total, count); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: currentUnread.total, @@ -226,10 +231,11 @@ export class ConversationOperations implements IConversationOperations { * 增加系统消息未读数 */ incrementSystemUnreadCount(): void { - const currentUnread = this.stateManager.getUnreadCount(); - this.stateManager.setUnreadCount(currentUnread.total, currentUnread.system + 1); + const store = useMessageStore.getState(); + const currentUnread = store.getUnreadCount(); + store.setUnreadCount(currentUnread.total, currentUnread.system + 1); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: currentUnread.total, @@ -243,11 +249,12 @@ export class ConversationOperations implements IConversationOperations { * 减少系统消息未读数 */ decrementSystemUnreadCount(count = 1): void { - const currentUnread = this.stateManager.getUnreadCount(); + const store = useMessageStore.getState(); + const currentUnread = store.getUnreadCount(); const newSystemCount = Math.max(0, currentUnread.system - count); - this.stateManager.setUnreadCount(currentUnread.total, newSystemCount); + store.setUnreadCount(currentUnread.total, newSystemCount); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: currentUnread.total, @@ -256,4 +263,4 @@ export class ConversationOperations implements IConversationOperations { timestamp: Date.now(), }); } -} +} \ No newline at end of file diff --git a/src/stores/message/MessageDeduplication.ts b/src/stores/message/services/MessageDeduplication.ts similarity index 91% rename from src/stores/message/MessageDeduplication.ts rename to src/stores/message/services/MessageDeduplication.ts index ce2ce3f..f8e215e 100644 --- a/src/stores/message/MessageDeduplication.ts +++ b/src/stores/message/services/MessageDeduplication.ts @@ -3,11 +3,11 @@ * 处理消息去重逻辑，防止重复处理相同的消息 */ -import type { IMessageDeduplication } from './types'; +import type { IMessageDeduplication } from '../types'; import { PROCESSED_MESSAGE_ID_EXPIRY, PROCESSED_MESSAGE_ID_MAX_SIZE, -} from './constants'; +} from '../constants'; export class MessageDeduplication implements IMessageDeduplication { // 已处理消息ID集合（用于去重，防止ACK消息和正常消息重复处理） @@ -58,4 +58,4 @@ export class MessageDeduplication implements IMessageDeduplication { } // 单例导出 -export const messageDeduplication = new MessageDeduplication(); +export const messageDeduplication = new MessageDeduplication(); \ No newline at end of file diff --git a/src/stores/message/MessageSendService.ts b/src/stores/message/services/MessageSendService.ts similarity index 69% rename from src/stores/message/MessageSendService.ts rename to src/stores/message/services/MessageSendService.ts index 40abb20..3a05940 100644 --- a/src/stores/message/MessageSendService.ts +++ b/src/stores/message/services/MessageSendService.ts @@ -1,19 +1,20 @@ /** * 消息发送服务 * 处理消息发送相关逻辑 + * + * 重构说明：直接使用 zustand store 替代 IMessageStateManager */ -import type { MessageResponse, MessageSegment, ConversationResponse } from '../../types/dto'; -import { messageService } from '../../services/messageService'; -import { saveMessage } from '../../services/database'; -import type { IMessageSendService, IMessageStateManager } from './types'; +import type { MessageResponse, MessageSegment, ConversationResponse } from '../../../types/dto'; +import { messageService } from '../../../services/messageService'; +import { saveMessage } from '../../../services/database'; +import type { IMessageSendService } from '../types'; +import { useMessageStore, subscriptionManager, mergeMessagesById } from '../store'; export class MessageSendService implements IMessageSendService { - private stateManager: IMessageStateManager; private getCurrentUserId: () => string | null; - constructor(stateManager: IMessageStateManager, getCurrentUserId: () => string | null) { - this.stateManager = stateManager; + constructor(getCurrentUserId: () => string | null) { this.getCurrentUserId = getCurrentUserId; } @@ -25,6 +26,8 @@ export class MessageSendService implements IMessageSendService { segments: MessageSegment[], options?: { replyToId?: string } ): Promise { + const store = useMessageStore.getState(); + try { // 调用API发送 const response = await messageService.sendMessage(conversationId, { @@ -36,7 +39,7 @@ export class MessageSendService implements IMessageSendService { const currentUserId = this.getCurrentUserId(); // 添加到本地消息列表 - const existingMessages = this.stateManager.getMessages(conversationId); + const existingMessages = store.getMessages(conversationId); const newMessage: MessageResponse = { id: response.id, conversation_id: conversationId, @@ -47,11 +50,11 @@ export class MessageSendService implements IMessageSendService { status: 'normal', }; - const updatedMessages = this.mergeMessagesById(existingMessages, [newMessage]); - this.stateManager.setMessages(conversationId, updatedMessages); + const updatedMessages = mergeMessagesById(existingMessages, [newMessage]); + store.setMessages(conversationId, updatedMessages); // 关键：立即广播消息列表更新 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -66,7 +69,7 @@ export class MessageSendService implements IMessageSendService { this.updateConversationWithNewMessage(conversationId, newMessage); // 通知消息已发送 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'message_sent', payload: { conversationId, message: newMessage }, timestamp: Date.now(), @@ -97,7 +100,7 @@ export class MessageSendService implements IMessageSendService { return null; } catch (error) { console.error('[MessageSendService] 发送消息失败:', error); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'error', payload: { error, context: 'sendMessage' }, timestamp: Date.now(), @@ -113,7 +116,8 @@ export class MessageSendService implements IMessageSendService { conversationId: string, message: MessageResponse ): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); const createdAt = message.created_at || new Date().toISOString(); if (conversation) { @@ -126,19 +130,7 @@ export class MessageSendService implements IMessageSendService { updated_at: createdAt, }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); } } - - /** - * 按 message_id 合并消息并按 seq 排序 - */ - private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] { - if (incoming.length === 0) return base; - const merged = new Map(); - [...base, ...incoming].forEach(msg => { - merged.set(String(msg.id), msg); - }); - return Array.from(merged.values()).sort((a, b) => a.seq - b.seq); - } -} +} \ No newline at end of file diff --git a/src/stores/message/MessageSyncService.ts b/src/stores/message/services/MessageSyncService.ts similarity index 70% rename from src/stores/message/MessageSyncService.ts rename to src/stores/message/services/MessageSyncService.ts index ba9a78a..c24a8d9 100644 --- a/src/stores/message/MessageSyncService.ts +++ b/src/stores/message/services/MessageSyncService.ts @@ -1,28 +1,33 @@ /** * 消息同步服务 * 处理消息和会话的服务器同步逻辑 + * + * 重构说明：直接使用 zustand store 替代 IMessageStateManager */ -import type { ConversationResponse, MessageResponse } from '../../types/dto'; -import { messageService } from '../../services/messageService'; +import type { ConversationResponse, MessageResponse } from '../../../types/dto'; +import { messageService } from '../../../services/messageService'; import { getMessagesByConversation, getMaxSeq, saveMessagesBatch, saveConversationsWithRelatedCache, getMessagesBeforeSeq, -} from '../../services/database'; +} from '../../../services/database'; import { type IConversationListPagedSource, SqliteConversationListPagedSource, createRemoteConversationListSource, CONVERSATION_LIST_PAGE_SIZE, -} from '../conversationListSources'; -import type { IMessageSyncService, IMessageStateManager, MessageManagerConversationListDeps } from './types'; +} from '../../conversationListSources'; +import type { IMessageSyncService, MessageManagerConversationListDeps, IUserCacheService } from '../types'; +import { useMessageStore, subscriptionManager, normalizeConversationId, mergeMessagesById } from '../store'; +import { ReadReceiptManager } from './ReadReceiptManager'; export class MessageSyncService implements IMessageSyncService { - private stateManager: IMessageStateManager; private getCurrentUserId: () => string | null; + private readReceiptManager: ReadReceiptManager; + private userCacheService: IUserCacheService; /** 远端会话列表（游标或页码） */ private readonly remoteConversationListSource: IConversationListPagedSource; @@ -35,12 +40,14 @@ export class MessageSyncService implements IMessageSyncService { private deferredConversationRefresh = false; constructor( - stateManager: IMessageStateManager, getCurrentUserId: () => string | null, + readReceiptManager: ReadReceiptManager, + userCacheService: IUserCacheService, deps?: MessageManagerConversationListDeps ) { - this.stateManager = stateManager; this.getCurrentUserId = getCurrentUserId; + this.readReceiptManager = readReceiptManager; + this.userCacheService = userCacheService; const pageSize = deps?.remoteListPageSize ?? CONVERSATION_LIST_PAGE_SIZE; this.remoteConversationListSource = @@ -57,7 +64,9 @@ export class MessageSyncService implements IMessageSyncService { * 获取会话列表 */ async fetchConversations(forceRefresh = false, source: string = 'unknown'): Promise { - if (this.stateManager.isLoading() && !forceRefresh) { + const store = useMessageStore.getState(); + + if (store.isLoading() && !forceRefresh) { return; } @@ -66,7 +75,7 @@ export class MessageSyncService implements IMessageSyncService { if (__DEV__) { console.log('[MessageSyncService] defer fetchConversations', { source, - activeConversationId: this.stateManager.getActiveConversation(), + activeConversationId: store.getActiveConversation(), }); } return; @@ -76,12 +85,12 @@ export class MessageSyncService implements IMessageSyncService { console.log('[MessageSyncService] fetchConversations', { source, forceRefresh, - activeConversationId: this.stateManager.getActiveConversation(), + activeConversationId: store.getActiveConversation(), }); } // 非强制刷新且内存为空：先用本地源暖机 - const currentState = this.stateManager.getState(); + const currentState = store.getState(); if (!forceRefresh && currentState.conversations.size === 0) { const warmed = await this.hydrateConversationsFromLocalSource(); if (warmed) { @@ -89,10 +98,10 @@ export class MessageSyncService implements IMessageSyncService { } } - this.stateManager.setLoading(true); - const emitLoadingToUi = this.stateManager.getConversations().length === 0; + store.setLoading(true); + const emitLoadingToUi = store.getConversations().length === 0; if (emitLoadingToUi) { - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_loading', payload: { loading: true }, timestamp: Date.now(), @@ -103,32 +112,41 @@ export class MessageSyncService implements IMessageSyncService { this.remoteConversationListSource.restart(); const page = await this.remoteConversationListSource.loadNext(); - currentState.conversations.clear(); + // 使用 zustand set 函数正确更新状态 + const newConversations = new Map(); page.items.forEach(conv => { const normalizedConv = this.normalizeConversationFromFetch(conv); - currentState.conversations.set(normalizedConv.id, normalizedConv); + newConversations.set(normalizedConv.id, normalizedConv); }); - - this.updateConversationList(); - this.recomputeConversationTotalUnread(); + + // 使用 store.setConversations 会自动排序并更新 conversationList + store.setConversations(newConversations); + + // 计算并更新未读数 + const totalUnread = Array.from(newConversations.values()).reduce( + (sum, conv) => sum + (conv.unread_count || 0), + 0 + ); + store.setUnreadCount(totalUnread, store.getUnreadCount().system); + this.emitConversationListAndUnreadUpdates(); } catch (error) { console.error('[MessageSyncService] 获取会话列表失败:', error); - if (this.stateManager.getConversations().length === 0) { + if (store.getConversations().length === 0) { const recovered = await this.hydrateConversationsFromLocalSource(); if (recovered) { this.emitConversationListAndUnreadUpdates(); } } - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'error', payload: { error, context: 'fetchConversations' }, timestamp: Date.now(), }); } finally { - this.stateManager.setLoading(false); + store.setLoading(false); if (emitLoadingToUi) { - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_loading', payload: { loading: false }, timestamp: Date.now(), @@ -141,10 +159,12 @@ export class MessageSyncService implements IMessageSyncService { * 会话列表加载下一页 */ async loadMoreConversations(): Promise { + const store = useMessageStore.getState(); + if (!this.remoteConversationListSource.hasMore) { return; } - if (this.loadingMoreConversations || this.stateManager.isLoading()) { + if (this.loadingMoreConversations || store.isLoading()) { return; } @@ -154,14 +174,14 @@ export class MessageSyncService implements IMessageSyncService { page.items.forEach(conv => { const normalizedConv = this.normalizeConversationFromFetch(conv); - this.stateManager.updateConversation(normalizedConv); + store.updateConversation(normalizedConv); }); this.recomputeConversationTotalUnread(); this.emitConversationListAndUnreadUpdates(); } catch (error) { console.error('[MessageSyncService] 加载更多会话失败:', error); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'error', payload: { error, context: 'loadMoreConversations' }, timestamp: Date.now(), @@ -175,12 +195,14 @@ export class MessageSyncService implements IMessageSyncService { * 获取单个会话详情 */ async fetchConversationDetail(conversationId: string): Promise { - const normalizedId = this.normalizeConversationId(conversationId); + const store = useMessageStore.getState(); + const normalizedId = normalizeConversationId(conversationId); + try { const detail = await messageService.getConversationById(normalizedId); if (detail) { const normalizedDetail = this.normalizeConversationFromFetch(detail as ConversationResponse); - this.stateManager.updateConversation(normalizedDetail); + store.updateConversation(normalizedDetail); return normalizedDetail; } return null; @@ -194,15 +216,17 @@ export class MessageSyncService implements IMessageSyncService { * 获取会话消息（增量同步） */ async fetchMessages(conversationId: string, afterSeq?: number): Promise { + const store = useMessageStore.getState(); + // 防止重复加载 - if (this.stateManager.isLoadingMessages(conversationId)) { + if (store.isLoadingMessages(conversationId)) { return; } - this.stateManager.setLoadingMessages(conversationId, true); + store.setLoadingMessages(conversationId, true); try { - const existingMessagesAtStart = this.stateManager.getMessages(conversationId); + const existingMessagesAtStart = store.getMessages(conversationId); const hasInMemoryMessages = existingMessagesAtStart.length > 0; let baselineMaxSeq = hasInMemoryMessages ? existingMessagesAtStart.reduce((max, m) => Math.max(max, m.seq || 0), 0) @@ -227,8 +251,8 @@ export class MessageSyncService implements IMessageSyncService { created_at: m.createdAt, })); - this.stateManager.setMessages(conversationId, formattedMessages); - this.stateManager.notifySubscribers({ + store.setMessages(conversationId, formattedMessages); + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -239,8 +263,8 @@ export class MessageSyncService implements IMessageSyncService { }); } else { // 冷启动兜底：先下发空列表事件 - this.stateManager.setMessages(conversationId, []); - this.stateManager.notifySubscribers({ + store.setMessages(conversationId, []); + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -261,11 +285,11 @@ export class MessageSyncService implements IMessageSyncService { const snapshotMessages = snapshotResp?.messages || []; if (snapshotMessages.length > 0) { - const existingMessages = this.stateManager.getMessages(conversationId); - const mergedSnapshot = this.mergeMessagesById(existingMessages, snapshotMessages); - this.stateManager.setMessages(conversationId, mergedSnapshot); + const existingMessages = store.getMessages(conversationId); + const mergedSnapshot = mergeMessagesById(existingMessages, snapshotMessages); + store.setMessages(conversationId, mergedSnapshot); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -300,11 +324,11 @@ export class MessageSyncService implements IMessageSyncService { const newMessages = incrementalResp?.messages || []; if (newMessages.length > 0) { - const existingMessages = this.stateManager.getMessages(conversationId); - const mergedMessages = this.mergeMessagesById(existingMessages, newMessages); - this.stateManager.setMessages(conversationId, mergedMessages); + const existingMessages = store.getMessages(conversationId); + const mergedMessages = mergeMessagesById(existingMessages, newMessages); + store.setMessages(conversationId, mergedMessages); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -340,11 +364,11 @@ export class MessageSyncService implements IMessageSyncService { if (response?.messages && response.messages.length > 0) { const newMessages = response.messages; - const existingMessages = this.stateManager.getMessages(conversationId); - const mergedMessages = this.mergeMessagesById(existingMessages, newMessages); + const existingMessages = store.getMessages(conversationId); + const mergedMessages = mergeMessagesById(existingMessages, newMessages); - this.stateManager.setMessages(conversationId, mergedMessages); - this.stateManager.notifySubscribers({ + store.setMessages(conversationId, mergedMessages); + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -373,13 +397,33 @@ export class MessageSyncService implements IMessageSyncService { } } catch (error) { console.error('[MessageSyncService] 获取消息失败:', error); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'error', payload: { error, context: 'fetchMessages', conversationId }, timestamp: Date.now(), }); } finally { - this.stateManager.setLoadingMessages(conversationId, false); + // 异步填充用户信息（不阻塞消息显示） + const currentMessages = store.getMessages(conversationId); + if (currentMessages.length > 0) { + this.userCacheService.enrichMessagesWithSenderInfo( + conversationId, + currentMessages, + (convId, enrichedMessages) => { + store.setMessages(convId, enrichedMessages); + subscriptionManager.notifySubscribers({ + type: 'messages_updated', + payload: { + conversationId: convId, + messages: enrichedMessages, + source: 'sender_enriched', + }, + timestamp: Date.now(), + }); + } + ); + } + store.setLoadingMessages(conversationId, false); } } @@ -387,6 +431,8 @@ export class MessageSyncService implements IMessageSyncService { * 加载更多历史消息 */ async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise { + const store = useMessageStore.getState(); + try { // 先从本地获取 const localMessages = await getMessagesBeforeSeq(conversationId, beforeSeq, limit); @@ -402,11 +448,11 @@ export class MessageSyncService implements IMessageSyncService { created_at: m.createdAt, })); - const existingMessages = this.stateManager.getMessages(conversationId); + const existingMessages = store.getMessages(conversationId); const mergedMessages = this.mergeOlderMessages(existingMessages, formattedMessages); - this.stateManager.setMessages(conversationId, mergedMessages); + store.setMessages(conversationId, mergedMessages); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, messages: mergedMessages, source: 'local_history' }, timestamp: Date.now(), @@ -434,11 +480,11 @@ export class MessageSyncService implements IMessageSyncService { segments: m.segments, }))); - const existingMessages = this.stateManager.getMessages(conversationId); + const existingMessages = store.getMessages(conversationId); const mergedMessages = this.mergeOlderMessages(existingMessages, serverMessages); - this.stateManager.setMessages(conversationId, mergedMessages); + store.setMessages(conversationId, mergedMessages); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, messages: mergedMessages, source: 'server_history' }, timestamp: Date.now(), @@ -458,6 +504,8 @@ export class MessageSyncService implements IMessageSyncService { * 获取未读数 */ async fetchUnreadCount(): Promise { + const store = useMessageStore.getState(); + try { const [unreadData, systemUnreadData] = await Promise.all([ messageService.getUnreadCount(), @@ -467,30 +515,32 @@ export class MessageSyncService implements IMessageSyncService { const totalUnread = unreadData?.total_unread_count ?? 0; const systemUnread = systemUnreadData?.unread_count ?? 0; - this.stateManager.setUnreadCount(totalUnread, systemUnread); + store.setUnreadCount(totalUnread, systemUnread); // 服务端汇总未读为 0 时，清掉内存中残留的红点 if (totalUnread === 0) { - const currentState = this.stateManager.getState(); + const currentConversations = store.getState().conversations; let anyCleared = false; - for (const [cid, conv] of currentState.conversations) { + const newConversations = new Map(currentConversations); + for (const [cid, conv] of newConversations) { if ((conv.unread_count || 0) > 0) { - currentState.conversations.set(cid, { ...conv, unread_count: 0 }); + newConversations.set(cid, { ...conv, unread_count: 0 }); anyCleared = true; } } if (anyCleared) { - this.updateConversationList(); + // 使用 zustand set 函数正确更新状态 + store.setConversations(newConversations); this.persistConversationListCache(); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); } } - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: totalUnread, @@ -522,56 +572,43 @@ export class MessageSyncService implements IMessageSyncService { // ==================== 私有工具方法 ==================== private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean { - if (!this.stateManager.getActiveConversation()) return false; + const store = useMessageStore.getState(); + if (!store.getActiveConversation()) return false; if (!forceRefresh) return false; - return source === 'sse-reconnect' || source === 'prefetch' || source === 'hooks-initial-refresh'; - } - - private normalizeConversationId(conversationId: string | number | null | undefined): string { - return conversationId == null ? '' : String(conversationId); + // hooks-initial-refresh 是冷启动兜底刷新，不应该被延迟 + return source === 'sse-reconnect' || source === 'prefetch'; } private normalizeConversationFromFetch(conv: ConversationResponse): ConversationResponse { - const id = this.normalizeConversationId(conv.id); - return { + const id = normalizeConversationId(conv.id); + // 应用已读保护逻辑 + return this.readReceiptManager.applyConversationFromFetch({ ...conv, id, - }; - } - - private updateConversationList(): void { - const currentState = this.stateManager.getState(); - const list = Array.from(currentState.conversations.values()).sort((a, b) => { - const aPinned = a.is_pinned ? 1 : 0; - const bPinned = b.is_pinned ? 1 : 0; - if (aPinned !== bPinned) { - return bPinned - aPinned; - } - const aTime = new Date(a.last_message_at || a.updated_at || 0).getTime(); - const bTime = new Date(b.last_message_at || b.updated_at || 0).getTime(); - return bTime - aTime; }); - currentState.conversationList = list; } private recomputeConversationTotalUnread(): void { - const currentState = this.stateManager.getState(); - currentState.totalUnreadCount = Array.from(currentState.conversations.values()).reduce( + const store = useMessageStore.getState(); + const totalUnread = Array.from(store.getState().conversations.values()).reduce( (sum, conv) => sum + (conv.unread_count || 0), 0 ); + // 使用 zustand set 函数正确更新状态 + store.setUnreadCount(totalUnread, store.getUnreadCount().system); } private emitConversationListAndUnreadUpdates(): void { + const store = useMessageStore.getState(); this.persistConversationListCache(); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); - const unreadCount = this.stateManager.getUnreadCount(); - this.stateManager.notifySubscribers({ + const unreadCount = store.getUnreadCount(); + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: unreadCount.total, @@ -582,7 +619,8 @@ export class MessageSyncService implements IMessageSyncService { } private persistConversationListCache(): void { - const currentState = this.stateManager.getState(); + const store = useMessageStore.getState(); + const currentState = store.getState(); const list = Array.from(currentState.conversations.values()); const users = list.flatMap(conv => [ ...(conv.participants || []), @@ -598,33 +636,33 @@ export class MessageSyncService implements IMessageSyncService { } private async hydrateConversationsFromLocalSource(): Promise { + const store = useMessageStore.getState(); this.localConversationListSource.restart(); const page = await this.localConversationListSource.loadNext(); if (!page.items.length) { return false; } - const currentState = this.stateManager.getState(); - currentState.conversations.clear(); + // 使用 zustand set 函数正确更新状态 + const newConversations = new Map(); page.items.forEach(conv => { const normalizedConv = this.normalizeConversationFromFetch(conv); - currentState.conversations.set(normalizedConv.id, normalizedConv); + newConversations.set(normalizedConv.id, normalizedConv); }); - this.updateConversationList(); - this.recomputeConversationTotalUnread(); + // 使用 store.setConversations 会自动排序并更新 conversationList + store.setConversations(newConversations); + + // 计算并更新未读数 + const totalUnread = Array.from(newConversations.values()).reduce( + (sum, conv) => sum + (conv.unread_count || 0), + 0 + ); + store.setUnreadCount(totalUnread, store.getUnreadCount().system); + return true; } - private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] { - if (incoming.length === 0) return base; - const merged = new Map(); - [...base, ...incoming].forEach(msg => { - merged.set(String(msg.id), msg); - }); - return Array.from(merged.values()).sort((a, b) => a.seq - b.seq); - } - private mergeOlderMessages(existing: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] { if (incoming.length === 0) return existing; @@ -638,6 +676,6 @@ export class MessageSyncService implements IMessageSyncService { return [...incomingAsc, ...existing]; } - return this.mergeMessagesById(existing, incomingAsc); + return mergeMessagesById(existing, incomingAsc); } -} +} \ No newline at end of file diff --git a/src/stores/message/ReadReceiptManager.ts b/src/stores/message/services/ReadReceiptManager.ts similarity index 80% rename from src/stores/message/ReadReceiptManager.ts rename to src/stores/message/services/ReadReceiptManager.ts index f424293..3430d04 100644 --- a/src/stores/message/ReadReceiptManager.ts +++ b/src/stores/message/services/ReadReceiptManager.ts @@ -1,17 +1,18 @@ /** * 已读回执管理器 * 管理消息已读状态的同步和保护 + * + * 重构说明：直接使用 zustand store 替代 IMessageStateManager */ -import type { ConversationResponse } from '../../types/dto'; -import { messageService } from '../../services/messageService'; -import { markConversationAsRead, updateConversationCacheUnreadCount } from '../../services/database'; -import type { ReadStateRecord, IReadReceiptManager, IMessageStateManager } from './types'; -import { READ_STATE_PROTECTION_DELAY } from './constants'; +import type { ConversationResponse } from '../../../types/dto'; +import { messageService } from '../../../services/messageService'; +import { markConversationAsRead, updateConversationCacheUnreadCount } from '../../../services/database'; +import type { ReadStateRecord, IReadReceiptManager } from '../types'; +import { READ_STATE_PROTECTION_DELAY } from '../constants'; +import { useMessageStore, subscriptionManager, normalizeConversationId } from '../store'; export class ReadReceiptManager implements IReadReceiptManager { - private stateManager: IMessageStateManager; - /** * 正在进行中的已读 API 请求集合 * 用于防止 fetchConversations 的服务器数据覆盖乐观已读状态 @@ -23,17 +24,14 @@ export class ReadReceiptManager implements IReadReceiptManager { */ private readStateVersion: number = 0; - constructor(stateManager: IMessageStateManager) { - this.stateManager = stateManager; - } - /** * 标记会话已读 * 关键方法：确保已读状态在所有组件间同步 */ async markAsRead(conversationId: string, seq: number): Promise { - const normalizedId = this.normalizeConversationId(conversationId); - const conversation = this.stateManager.getConversation(normalizedId); + const normalizedId = normalizeConversationId(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(normalizedId); if (!conversation) { console.warn('[ReadReceiptManager] 会话不存在:', normalizedId); @@ -71,17 +69,17 @@ export class ReadReceiptManager implements IReadReceiptManager { my_last_read_seq: Math.max(Number(conversation.my_last_read_seq || 0), seq), }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); - const currentUnread = this.stateManager.getUnreadCount(); + const currentUnread = store.getUnreadCount(); const newTotalUnread = Math.max(0, currentUnread.total - prevUnreadCount); - this.stateManager.setUnreadCount(newTotalUnread, currentUnread.system); + store.setUnreadCount(newTotalUnread, currentUnread.system); // 3. 更新本地数据库 markConversationAsRead(normalizedId).catch(console.error); // 4. 立即通知所有订阅者 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'message_read', payload: { conversationId: normalizedId, @@ -91,7 +89,7 @@ export class ReadReceiptManager implements IReadReceiptManager { timestamp: Date.now(), }); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: newTotalUnread, @@ -107,10 +105,10 @@ export class ReadReceiptManager implements IReadReceiptManager { console.error('[ReadReceiptManager] 标记已读API失败:', error); // 失败时回滚状态 - this.stateManager.updateConversation(conversation); - this.stateManager.setUnreadCount(currentUnread.total, currentUnread.system); + store.updateConversation(conversation); + store.setUnreadCount(currentUnread.total, currentUnread.system); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'message_read', payload: { conversationId: normalizedId, @@ -146,18 +144,19 @@ export class ReadReceiptManager implements IReadReceiptManager { * 标记所有消息已读 */ async markAllAsRead(): Promise { - const conversations = this.stateManager.getState().conversations; - const prevTotalUnread = this.stateManager.getUnreadCount().total; - const currentSystem = this.stateManager.getUnreadCount().system; + const store = useMessageStore.getState(); + const conversations = store.conversations; + const prevTotalUnread = store.getUnreadCount().total; + const currentSystem = store.getUnreadCount().system; // 乐观更新 conversations.forEach(conv => { conv.unread_count = 0; }); - this.stateManager.setUnreadCount(0, currentSystem); + store.setUnreadCount(0, currentSystem); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: 0, @@ -182,9 +181,9 @@ export class ReadReceiptManager implements IReadReceiptManager { } catch (error) { console.error('[ReadReceiptManager] 标记所有已读失败:', error); // 回滚 - this.stateManager.setUnreadCount(prevTotalUnread, currentSystem); + store.setUnreadCount(prevTotalUnread, currentSystem); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { totalUnreadCount: prevTotalUnread, @@ -258,18 +257,11 @@ export class ReadReceiptManager implements IReadReceiptManager { return this.pendingReadMap; } - /** - * 规范化会话ID - */ - private normalizeConversationId(conversationId: string | number | null | undefined): string { - return conversationId == null ? '' : String(conversationId); - } - /** * 根据已读保护状态智能合并会话 */ applyConversationFromFetch(conv: ConversationResponse): ConversationResponse { - const id = this.normalizeConversationId(conv.id); + const id = normalizeConversationId(conv.id); const readRecord = this.pendingReadMap.get(id); let next: ConversationResponse; @@ -310,4 +302,4 @@ export class ReadReceiptManager implements IReadReceiptManager { this.pendingReadMap.clear(); this.readStateVersion = 0; } -} +} \ No newline at end of file diff --git a/src/stores/message/UserCacheService.ts b/src/stores/message/services/UserCacheService.ts similarity index 91% rename from src/stores/message/UserCacheService.ts rename to src/stores/message/services/UserCacheService.ts index b1a2218..65a23d1 100644 --- a/src/stores/message/UserCacheService.ts +++ b/src/stores/message/services/UserCacheService.ts @@ -3,10 +3,10 @@ * 管理用户信息的缓存和获取 */ -import type { MessageResponse, UserDTO } from '../../types/dto'; -import { api } from '../../services/api'; -import { getUserCache, saveUserCache } from '../../services/database'; -import type { IUserCacheService } from './types'; +import type { MessageResponse, UserDTO } from '../../../types/dto'; +import { api } from '../../../services/api'; +import { getUserCache, saveUserCache } from '../../../services/database'; +import type { IUserCacheService } from '../types'; export class UserCacheService implements IUserCacheService { // 正在获取用户信息的请求映射（用于去重） @@ -116,4 +116,4 @@ export class UserCacheService implements IUserCacheService { } // 单例导出 -export const userCacheService = new UserCacheService(); +export const userCacheService = new UserCacheService(); \ No newline at end of file diff --git a/src/stores/message/WSMessageHandler.ts b/src/stores/message/services/WSMessageHandler.ts similarity index 80% rename from src/stores/message/WSMessageHandler.ts rename to src/stores/message/services/WSMessageHandler.ts index 5ec977c..383f842 100644 --- a/src/stores/message/WSMessageHandler.ts +++ b/src/stores/message/services/WSMessageHandler.ts @@ -1,6 +1,8 @@ /** * WebSocket 消息处理器 * 处理所有来自 WebSocket 的消息事件 + * + * 重构说明：直接使用 zustand store 替代 IMessageStateManager */ import type { @@ -12,22 +14,21 @@ import type { WSGroupRecallMessage, WSGroupTypingMessage, WSGroupNoticeMessage, -} from '../../services/wsService'; -import { wsService, GroupNoticeType } from '../../services/wsService'; -import { vibrateOnMessage } from '../../services/messageVibrationService'; -import { saveMessage, updateMessageStatus } from '../../services/database'; -import type { MessageResponse, ConversationResponse } from '../../types/dto'; +} from '../../../services/wsService'; +import { wsService, GroupNoticeType } from '../../../services/wsService'; +import { vibrateOnMessage } from '../../../services/messageVibrationService'; +import { saveMessage, updateMessageStatus } from '../../../services/database'; +import type { MessageResponse, ConversationResponse } from '../../../types/dto'; import type { IWSMessageHandler, - IMessageStateManager, IMessageDeduplication, IUserCacheService, HandleNewMessageOptions, -} from './types'; -import { MAX_FLUSH_ITERATIONS } from './constants'; +} from '../types'; +import { MAX_FLUSH_ITERATIONS } from '../constants'; +import { useMessageStore, subscriptionManager, normalizeConversationId, mergeMessagesById } from '../store'; export class WSMessageHandler implements IWSMessageHandler { - private stateManager: IMessageStateManager; private deduplication: IMessageDeduplication; private userCacheService: IUserCacheService; private getCurrentUserId: () => string | null; @@ -45,7 +46,6 @@ export class WSMessageHandler implements IWSMessageHandler { private bufferedReadReceipts: Array = []; constructor( - stateManager: IMessageStateManager, deduplication: IMessageDeduplication, userCacheService: IUserCacheService, getCurrentUserId: () => string | null, @@ -56,7 +56,6 @@ export class WSMessageHandler implements IWSMessageHandler { fetchMessages: (conversationId: string) => Promise; } ) { - this.stateManager = stateManager; this.deduplication = deduplication; this.userCacheService = userCacheService; this.getCurrentUserId = getCurrentUserId; @@ -74,9 +73,11 @@ export class WSMessageHandler implements IWSMessageHandler { this.sseUnsubscribe(); } + const store = useMessageStore.getState(); + // 监听私聊消息 wsService.on('chat', (message: WSChatMessage) => { - if (!this.stateManager.getState().isInitialized || this.isBootstrapping) { + if (!store.getState().isInitialized || this.isBootstrapping) { this.bufferedChatMessages.push(message); return; } @@ -85,7 +86,7 @@ export class WSMessageHandler implements IWSMessageHandler { // 监听群聊消息 wsService.on('group_message', (message: WSGroupChatMessage) => { - if (!this.stateManager.getState().isInitialized || this.isBootstrapping) { + if (!store.getState().isInitialized || this.isBootstrapping) { this.bufferedChatMessages.push(message); return; } @@ -94,7 +95,7 @@ export class WSMessageHandler implements IWSMessageHandler { // 监听私聊已读回执 wsService.on('read', (message: WSReadMessage) => { - if (!this.stateManager.getState().isInitialized || this.isBootstrapping) { + if (!store.getState().isInitialized || this.isBootstrapping) { this.bufferedReadReceipts.push(message); return; } @@ -103,7 +104,7 @@ export class WSMessageHandler implements IWSMessageHandler { // 监听群聊已读回执 wsService.on('group_read', (message: WSGroupReadMessage) => { - if (!this.stateManager.getState().isInitialized || this.isBootstrapping) { + if (!store.getState().isInitialized || this.isBootstrapping) { this.bufferedReadReceipts.push(message); return; } @@ -132,8 +133,8 @@ export class WSMessageHandler implements IWSMessageHandler { // 监听连接状态 wsService.onConnect(() => { - this.stateManager.setSSEConnected(true); - this.stateManager.notifySubscribers({ + store.setSSEConnected(true); + subscriptionManager.notifySubscribers({ type: 'connection_changed', payload: { connected: true }, timestamp: Date.now(), @@ -143,7 +144,7 @@ export class WSMessageHandler implements IWSMessageHandler { const now = Date.now(); if (now - this.lastReconnectSyncAt > 1500) { this.lastReconnectSyncAt = now; - const activeConversation = this.stateManager.getActiveConversation(); + const activeConversation = store.getActiveConversation(); if (!activeConversation) { this.fetchConversationsCallback(true, 'sse-reconnect').catch(error => { @@ -164,8 +165,8 @@ export class WSMessageHandler implements IWSMessageHandler { }); wsService.onDisconnect(() => { - this.stateManager.setSSEConnected(false); - this.stateManager.notifySubscribers({ + store.setSSEConnected(false); + subscriptionManager.notifySubscribers({ type: 'connection_changed', payload: { connected: false }, timestamp: Date.now(), @@ -187,7 +188,7 @@ export class WSMessageHandler implements IWSMessageHandler { * 检查是否已连接 */ isConnected(): boolean { - return this.stateManager.isConnected(); + return useMessageStore.getState().isConnected(); } /** @@ -242,8 +243,9 @@ export class WSMessageHandler implements IWSMessageHandler { message: (WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean }, options?: HandleNewMessageOptions ): Promise { + const store = useMessageStore.getState(); const { conversation_id, id, sender_id, seq, segments, created_at, _isAck } = message; - const normalizedConversationId = this.normalizeConversationId(conversation_id); + const normalizedConversationId = normalizeConversationId(conversation_id); const currentUserId = this.getCurrentUserId(); const suppressUnreadIncrement = options?.suppressUnreadIncrement ?? false; const suppressVibration = options?.suppressVibration ?? false; @@ -260,13 +262,13 @@ export class WSMessageHandler implements IWSMessageHandler { if (message.type === 'group_message' && sender_id && sender_id !== currentUserId && sender_id !== '10000') { this.userCacheService.getSenderInfo(sender_id).then(user => { if (user) { - const currentMessages = this.stateManager.getMessages(normalizedConversationId); + const currentMessages = store.getMessages(normalizedConversationId); if (currentMessages) { const updatedMessages = currentMessages.map(m => m.id === id ? { ...m, sender: user } : m ); - this.stateManager.setMessages(normalizedConversationId, updatedMessages); - this.stateManager.notifySubscribers({ + store.setMessages(normalizedConversationId, updatedMessages); + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId: normalizedConversationId, @@ -293,14 +295,14 @@ export class WSMessageHandler implements IWSMessageHandler { }; // 立即更新内存中的消息列表 - const existingMessages = this.stateManager.getMessages(normalizedConversationId); + const existingMessages = store.getMessages(normalizedConversationId); const messageExists = existingMessages.some(m => String(m.id) === String(id)); if (!messageExists) { - const updatedMessages = this.mergeMessagesById(existingMessages, [newMessage]); - this.stateManager.setMessages(normalizedConversationId, updatedMessages); + const updatedMessages = mergeMessagesById(existingMessages, [newMessage]); + store.setMessages(normalizedConversationId, updatedMessages); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId: normalizedConversationId, @@ -318,7 +320,7 @@ export class WSMessageHandler implements IWSMessageHandler { // 更新未读数 const isCurrentUserMessage = sender_id === currentUserId; - const isActiveConversation = normalizedConversationId === this.stateManager.getActiveConversation(); + const isActiveConversation = normalizedConversationId === store.getActiveConversation(); const shouldIncrementUnread = !suppressUnreadIncrement && @@ -358,7 +360,7 @@ export class WSMessageHandler implements IWSMessageHandler { }); // 通知收到新消息 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'message_received', payload: { conversationId: conversation_id, @@ -388,12 +390,12 @@ export class WSMessageHandler implements IWSMessageHandler { */ handleRecallMessage(message: WSRecallMessage): void { const { conversation_id, message_id } = message; - const normalizedConversationId = this.normalizeConversationId(conversation_id); + const normalizedConversationId = normalizeConversationId(conversation_id); this.markMessageAsRecalled(normalizedConversationId, message_id); this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'message_recalled', payload: { conversationId: normalizedConversationId, messageId: message_id }, timestamp: Date.now(), @@ -409,12 +411,12 @@ export class WSMessageHandler implements IWSMessageHandler { */ handleGroupRecallMessage(message: WSGroupRecallMessage): void { const { conversation_id, message_id } = message; - const normalizedConversationId = this.normalizeConversationId(conversation_id); + const normalizedConversationId = normalizeConversationId(conversation_id); this.markMessageAsRecalled(normalizedConversationId, message_id); this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'message_recalled', payload: { conversationId: normalizedConversationId, messageId: message_id, isGroup: true }, timestamp: Date.now(), @@ -429,10 +431,11 @@ export class WSMessageHandler implements IWSMessageHandler { * 处理群聊输入状态 */ handleGroupTyping(message: WSGroupTypingMessage): void { + const store = useMessageStore.getState(); const { group_id, user_id, is_typing } = message; const groupIdStr = String(group_id); - const currentTypingUsers = this.stateManager.getTypingUsers(groupIdStr); + const currentTypingUsers = store.getTypingUsers(groupIdStr); let updatedTypingUsers: string[]; if (is_typing) { @@ -446,8 +449,8 @@ export class WSMessageHandler implements IWSMessageHandler { } if (updatedTypingUsers.length !== currentTypingUsers.length) { - this.stateManager.setTypingUsers(groupIdStr, updatedTypingUsers); - this.stateManager.notifySubscribers({ + store.setTypingUsers(groupIdStr, updatedTypingUsers); + subscriptionManager.notifySubscribers({ type: 'typing_status', payload: { groupId: groupIdStr, typingUsers: updatedTypingUsers }, timestamp: Date.now(), @@ -459,6 +462,7 @@ export class WSMessageHandler implements IWSMessageHandler { * 处理群通知 */ handleGroupNotice(message: WSGroupNoticeMessage): void { + const store = useMessageStore.getState(); const { notice_type, group_id, data, timestamp, message_id, seq } = message; const groupIdStr = String(group_id); @@ -469,7 +473,7 @@ export class WSMessageHandler implements IWSMessageHandler { if (mutedUserId && mutedUserId === currentUserId) { const isMuted = notice_type === 'muted'; - this.stateManager.setMutedStatus(groupIdStr, isMuted); + store.setMutedStatus(groupIdStr, isMuted); } } @@ -477,7 +481,7 @@ export class WSMessageHandler implements IWSMessageHandler { if (message_id && typeof seq === 'number' && seq > 0) { const conversationId = this.findConversationIdByGroupId(groupIdStr); if (conversationId) { - const existingMessages = this.stateManager.getMessages(conversationId); + const existingMessages = store.getMessages(conversationId); const exists = existingMessages.some(m => String(m.id) === String(message_id)); if (!exists) { const noticeText = this.buildGroupNoticeText(notice_type, data); @@ -495,9 +499,9 @@ export class WSMessageHandler implements IWSMessageHandler { }; const updatedMessages = [...existingMessages, systemNoticeMessage].sort((a, b) => a.seq - b.seq); - this.stateManager.setMessages(conversationId, updatedMessages); + store.setMessages(conversationId, updatedMessages); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, @@ -512,7 +516,7 @@ export class WSMessageHandler implements IWSMessageHandler { } // 通知订阅者群通知 - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'group_notice', payload: { groupId: groupIdStr, @@ -528,25 +532,13 @@ export class WSMessageHandler implements IWSMessageHandler { // ==================== 私有工具方法 ==================== - private normalizeConversationId(conversationId: string | number | null | undefined): string { - return conversationId == null ? '' : String(conversationId); - } - - private mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] { - if (incoming.length === 0) return base; - const merged = new Map(); - [...base, ...incoming].forEach(msg => { - merged.set(String(msg.id), msg); - }); - return Array.from(merged.values()).sort((a, b) => a.seq - b.seq); - } - private updateConversationWithNewMessage( conversationId: string, message: MessageResponse, createdAt: string ): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); if (conversation) { const updatedConv: ConversationResponse = { @@ -556,48 +548,50 @@ export class WSMessageHandler implements IWSMessageHandler { last_message_at: createdAt, updated_at: createdAt, }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); } - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); } private incrementUnreadCount(conversationId: string): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); if (conversation) { const prevUnreadCount = conversation.unread_count || 0; const updatedConv: ConversationResponse = { ...conversation, unread_count: prevUnreadCount + 1, }; - this.stateManager.updateConversation(updatedConv); + store.updateConversation(updatedConv); - const currentUnread = this.stateManager.getUnreadCount(); - this.stateManager.setUnreadCount(currentUnread.total + 1, currentUnread.system); + const currentUnread = store.getUnreadCount(); + store.setUnreadCount(currentUnread.total + 1, currentUnread.system); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'unread_count_updated', payload: { - totalUnreadCount: this.stateManager.getUnreadCount().total, - systemUnreadCount: this.stateManager.getUnreadCount().system, + totalUnreadCount: store.getUnreadCount().total, + systemUnreadCount: store.getUnreadCount().system, }, timestamp: Date.now(), }); - this.stateManager.notifySubscribers({ + subscriptionManager.notifySubscribers({ type: 'conversations_updated', - payload: { conversations: this.stateManager.getConversations() }, + payload: { conversations: store.getConversations() }, timestamp: Date.now(), }); } } private markMessageAsRecalled(conversationId: string, messageId: string): void { - const messages = this.stateManager.getMessages(conversationId); + const store = useMessageStore.getState(); + const messages = store.getMessages(conversationId); if (!messages) return; let changed = false; @@ -615,8 +609,8 @@ export class WSMessageHandler implements IWSMessageHandler { if (!changed) return; - this.stateManager.setMessages(conversationId, updatedMessages); - this.stateManager.notifySubscribers({ + store.setMessages(conversationId, updatedMessages); + subscriptionManager.notifySubscribers({ type: 'messages_updated', payload: { conversationId, messages: updatedMessages }, timestamp: Date.now(), @@ -624,7 +618,8 @@ export class WSMessageHandler implements IWSMessageHandler { } private syncConversationLastMessageOnRecall(conversationId: string, messageId: string): void { - const conversation = this.stateManager.getConversation(conversationId); + const store = useMessageStore.getState(); + const conversation = store.getConversation(conversationId); if (!conversation?.last_message) return; if (String(conversation.last_message.id) !== String(messageId)) return; if (conversation.last_message.status === 'recalled') return; @@ -636,11 +631,12 @@ export class WSMessageHandler implements IWSMessageHandler { status: 'recalled', }, }; - this.stateManager.updateConversation(updatedConversation); + store.updateConversation(updatedConversation); } private findConversationIdByGroupId(groupId: string): string | null { - const conversations = this.stateManager.getConversations(); + const store = useMessageStore.getState(); + const conversations = store.getConversations(); for (const conv of conversations) { if (String(conv.group?.id || '') === groupId) { return conv.id; @@ -666,4 +662,4 @@ export class WSMessageHandler implements IWSMessageHandler { return ''; } } -} +} \ No newline at end of file diff --git a/src/stores/message/services/index.ts b/src/stores/message/services/index.ts new file mode 100644 index 0000000..a0e86b2 --- /dev/null +++ b/src/stores/message/services/index.ts @@ -0,0 +1,26 @@ +/** + * 消息服务层统一导出 + * + * 重构说明：所有服务直接使用 zustand store，移除了 IMessageStateManager 接口依赖 + */ + +// 消息去重服务 +export { MessageDeduplication, messageDeduplication } from './MessageDeduplication'; + +// 用户缓存服务 +export { UserCacheService, userCacheService } from './UserCacheService'; + +// 已读回执管理器 +export { ReadReceiptManager } from './ReadReceiptManager'; + +// 会话操作服务 +export { ConversationOperations } from './ConversationOperations'; + +// 消息发送服务 +export { MessageSendService } from './MessageSendService'; + +// 消息同步服务 +export { MessageSyncService } from './MessageSyncService'; + +// WebSocket 消息处理器 +export { WSMessageHandler } from './WSMessageHandler'; \ No newline at end of file diff --git a/src/stores/message/store.ts b/src/stores/message/store.ts new file mode 100644 index 0000000..730ab52 --- /dev/null +++ b/src/stores/message/store.ts @@ -0,0 +1,428 @@ +/** + * 消息状态 Zustand Store + * 使用 zustand 进行状态管理，提供响应式状态更新 + * + * 重构说明： + * - 移除了 MessageStateManager 包装层 + * - 直接使用 zustand store 进行状态管理 + * - 订阅管理器集成到 store 中 + */ + +import { create } from 'zustand'; +import type { + ConversationResponse, + MessageResponse, +} from '../../types/dto'; +import type { + MessageSubscriber, + MessageEvent, +} from './types'; + +// ==================== 状态接口 ==================== + +/** + * 消息状态接口（不包含订阅者，订阅者单独管理） + */ +export interface MessageState { + // 会话相关 + conversations: Map; + conversationList: ConversationResponse[]; + + // 消息相关 - 按会话ID存储 + messagesMap: Map; + + // 未读数 + totalUnreadCount: number; + systemUnreadCount: number; + + // 连接状态 + isWSConnected: boolean; + + // 当前活动会话ID（用户正在查看的会话） + currentConversationId: string | null; + + // 加载状态 + isLoadingConversations: boolean; + loadingMessagesSet: Set; + + // 初始化状态 + isInitialized: boolean; + + // 输入状态 - 按群组ID存储正在输入的用户ID列表 + typingUsersMap: Map; + + // 当前用户的禁言状态 - 按群组ID存储 + mutedStatusMap: Map; +} + +// ==================== Actions 接口 ==================== + +export interface MessageActions { + // 获取状态 + getState: () => MessageState; + getConversations: () => ConversationResponse[]; + getConversation: (conversationId: string) => ConversationResponse | null; + getMessages: (conversationId: string) => MessageResponse[]; + getUnreadCount: () => { total: number; system: number }; + isConnected: () => boolean; + isLoading: () => boolean; + isLoadingMessages: (conversationId: string) => boolean; + getTypingUsers: (groupId: string) => string[]; + isMuted: (groupId: string) => boolean; + getActiveConversation: () => string | null; + + // 设置状态 + setConversations: (conversations: Map) => void; + updateConversation: (conversation: ConversationResponse) => void; + removeConversation: (conversationId: string) => void; + setMessages: (conversationId: string, messages: MessageResponse[]) => void; + addMessage: (conversationId: string, message: MessageResponse) => void; + updateMessage: (conversationId: string, messageId: string, updates: Partial) => void; + setUnreadCount: (total: number, system: number) => void; + setSSEConnected: (connected: boolean) => void; + setCurrentConversation: (conversationId: string | null) => void; + setLoading: (loading: boolean) => void; + setLoadingMessages: (conversationId: string, loading: boolean) => void; + setTypingUsers: (groupId: string, users: string[]) => void; + setMutedStatus: (groupId: string, isMuted: boolean) => void; + setInitialized: (initialized: boolean) => void; + + // 重置 + reset: () => void; +} + +// ==================== 初始状态 ==================== + +const initialState: MessageState = { + conversations: new Map(), + conversationList: [], + messagesMap: new Map(), + totalUnreadCount: 0, + systemUnreadCount: 0, + isWSConnected: false, + currentConversationId: null, + isLoadingConversations: false, + loadingMessagesSet: new Set(), + isInitialized: false, + typingUsersMap: new Map(), + mutedStatusMap: new Map(), +}; + +// ==================== 工具函数 ==================== + +/** + * 规范化会话ID + */ +export function normalizeConversationId(conversationId: string | number | null | undefined): string { + return conversationId == null ? '' : String(conversationId); +} + +/** + * 更新会话列表排序 + * 排序规则：置顶优先，再按最后消息时间排序 + */ +function sortConversationList(conversations: Map): ConversationResponse[] { + return Array.from(conversations.values()).sort((a, b) => { + const aPinned = a.is_pinned ? 1 : 0; + const bPinned = b.is_pinned ? 1 : 0; + if (aPinned !== bPinned) { + return bPinned - aPinned; + } + const aTime = new Date(a.last_message_at || a.updated_at || 0).getTime(); + const bTime = new Date(b.last_message_at || b.updated_at || 0).getTime(); + return bTime - aTime; + }); +} + +/** + * 按 message_id 合并消息并按 seq 排序 + */ +export function mergeMessagesById(base: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] { + if (incoming.length === 0) return base; + const merged = new Map(); + [...base, ...incoming].forEach(msg => { + merged.set(String(msg.id), msg); + }); + return Array.from(merged.values()).sort((a, b) => a.seq - b.seq); +} + +// ==================== Store 创建 ==================== + +export type MessageStore = MessageState & MessageActions; + +export const useMessageStore = create((set, get) => ({ + // ==================== 初始状态 ==================== + ...initialState, + + // ==================== 获取状态 ==================== + + getState: () => { + const state = get(); + return { + conversations: state.conversations, + conversationList: state.conversationList, + messagesMap: state.messagesMap, + totalUnreadCount: state.totalUnreadCount, + systemUnreadCount: state.systemUnreadCount, + isWSConnected: state.isWSConnected, + currentConversationId: state.currentConversationId, + isLoadingConversations: state.isLoadingConversations, + loadingMessagesSet: state.loadingMessagesSet, + isInitialized: state.isInitialized, + typingUsersMap: state.typingUsersMap, + mutedStatusMap: state.mutedStatusMap, + }; + }, + + getConversations: () => { + return get().conversationList; + }, + + getConversation: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + return get().conversations.get(normalizedId) || null; + }, + + getMessages: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + return get().messagesMap.get(normalizedId) || []; + }, + + getUnreadCount: () => { + const state = get(); + return { + total: state.totalUnreadCount, + system: state.systemUnreadCount, + }; + }, + + isConnected: () => { + return get().isWSConnected; + }, + + isLoading: () => { + return get().isLoadingConversations; + }, + + isLoadingMessages: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + return get().loadingMessagesSet.has(normalizedId); + }, + + getTypingUsers: (groupId: string) => { + return get().typingUsersMap.get(groupId) || []; + }, + + isMuted: (groupId: string) => { + return get().mutedStatusMap.get(groupId) || false; + }, + + getActiveConversation: () => { + return get().currentConversationId; + }, + + // ==================== 设置状态 ==================== + + setConversations: (conversations: Map) => { + const conversationList = sortConversationList(conversations); + set({ conversations, conversationList }); + }, + + updateConversation: (conversation: ConversationResponse) => { + const id = normalizeConversationId(conversation.id); + set(state => { + const newConversations = new Map(state.conversations); + newConversations.set(id, { ...conversation, id }); + const conversationList = sortConversationList(newConversations); + return { conversations: newConversations, conversationList }; + }); + }, + + removeConversation: (conversationId: string) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const target = state.conversations.get(normalizedId); + const removedUnread = target?.unread_count || 0; + + const newConversations = new Map(state.conversations); + newConversations.delete(normalizedId); + + const newMessagesMap = new Map(state.messagesMap); + newMessagesMap.delete(normalizedId); + + const newLoadingMessagesSet = new Set(state.loadingMessagesSet); + newLoadingMessagesSet.delete(normalizedId); + + const newTotalUnreadCount = Math.max(0, state.totalUnreadCount - removedUnread); + + const newCurrentConversationId = state.currentConversationId === normalizedId + ? null + : state.currentConversationId; + + const conversationList = sortConversationList(newConversations); + + return { + conversations: newConversations, + conversationList, + messagesMap: newMessagesMap, + totalUnreadCount: newTotalUnreadCount, + currentConversationId: newCurrentConversationId, + loadingMessagesSet: newLoadingMessagesSet, + }; + }); + }, + + setMessages: (conversationId: string, messages: MessageResponse[]) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const newMessagesMap = new Map(state.messagesMap); + newMessagesMap.set(normalizedId, messages); + return { messagesMap: newMessagesMap }; + }); + }, + + addMessage: (conversationId: string, message: MessageResponse) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const existing = state.messagesMap.get(normalizedId) || []; + const exists = existing.some(m => String(m.id) === String(message.id)); + + if (exists) { + return state; + } + + const updated = mergeMessagesById(existing, [message]); + const newMessagesMap = new Map(state.messagesMap); + newMessagesMap.set(normalizedId, updated); + return { messagesMap: newMessagesMap }; + }); + }, + + updateMessage: (conversationId: string, messageId: string, updates: Partial) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const messages = state.messagesMap.get(normalizedId); + if (!messages) return state; + + const updated = messages.map(m => + String(m.id) === String(messageId) ? { ...m, ...updates } : m + ); + + const newMessagesMap = new Map(state.messagesMap); + newMessagesMap.set(normalizedId, updated); + return { messagesMap: newMessagesMap }; + }); + }, + + setUnreadCount: (total: number, system: number) => { + set({ totalUnreadCount: total, systemUnreadCount: system }); + }, + + setSSEConnected: (connected: boolean) => { + set({ isWSConnected: connected }); + }, + + setCurrentConversation: (conversationId: string | null) => { + set({ currentConversationId: conversationId }); + }, + + setLoading: (loading: boolean) => { + set({ isLoadingConversations: loading }); + }, + + setLoadingMessages: (conversationId: string, loading: boolean) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const newLoadingMessagesSet = new Set(state.loadingMessagesSet); + if (loading) { + newLoadingMessagesSet.add(normalizedId); + } else { + newLoadingMessagesSet.delete(normalizedId); + } + return { loadingMessagesSet: newLoadingMessagesSet }; + }); + }, + + setTypingUsers: (groupId: string, users: string[]) => { + set(state => { + const newTypingUsersMap = new Map(state.typingUsersMap); + newTypingUsersMap.set(groupId, users); + return { typingUsersMap: newTypingUsersMap }; + }); + }, + + setMutedStatus: (groupId: string, isMuted: boolean) => { + set(state => { + const newMutedStatusMap = new Map(state.mutedStatusMap); + newMutedStatusMap.set(groupId, isMuted); + return { mutedStatusMap: newMutedStatusMap }; + }); + }, + + setInitialized: (initialized: boolean) => { + set({ isInitialized: initialized }); + }, + + // ==================== 重置 ==================== + + reset: () => { + set({ + ...initialState, + conversations: new Map(), + messagesMap: new Map(), + loadingMessagesSet: new Set(), + typingUsersMap: new Map(), + mutedStatusMap: new Map(), + }); + }, +})); + +// ==================== 订阅管理器 ==================== + +/** + * 订阅管理器类 + * 管理消息事件的订阅和通知 + */ +export class SubscriptionManager { + private subscribers: Set = new Set(); + + subscribe(subscriber: MessageSubscriber): () => void { + this.subscribers.add(subscriber); + return () => { + this.subscribers.delete(subscriber); + }; + } + + notifySubscribers(event: MessageEvent): void { + this.subscribers.forEach(subscriber => { + try { + subscriber(event); + } catch (error) { + console.error('[SubscriptionManager] 订阅者执行失败:', error); + } + }); + } + + clear(): void { + this.subscribers.clear(); + } +} + +// 创建全局订阅管理器实例 +export const subscriptionManager = new SubscriptionManager(); + +// ==================== 便捷方法 ==================== + +/** + * 通知所有订阅者 + */ +export function notifySubscribers(event: MessageEvent): void { + subscriptionManager.notifySubscribers(event); +} + +/** + * 订阅消息事件 + */ +export function subscribeToMessages(subscriber: MessageSubscriber): () => void { + return subscriptionManager.subscribe(subscriber); +} diff --git a/src/stores/messageManager.ts b/src/stores/messageManager.ts index 39475d6..e5771c3 100644 --- a/src/stores/messageManager.ts +++ b/src/stores/messageManager.ts @@ -1,15 +1,13 @@ /** * MessageManager - 消息管理核心模块 * - * 向后兼容的重导出文件 - * 所有实现已迁移到 ./message 目录下的模块化组件 + * ⚠️ 此文件已废弃，保留仅用于向后兼容 * * 新代码应该使用: * import { messageManager } from '@/stores/message' - * import type { Message, Conversation, MessageEventManager } from '@/stores/message' + * import type { MessageEvent, MessageSubscriber } from '@/stores/message' * - * 此文件保持向后兼容性，原有导入方式仍然有效: - * import { messageManager } from '@/stores/messageManager' + * 此文件将在未来版本中移除 */ // 从新模块重导出所有内容 @@ -18,10 +16,6 @@ export { messageManager, MessageManager, - // 状态管理器 - MessageStateManager, - messageStateManager, - // 消息去重服务 MessageDeduplication, messageDeduplication, @@ -68,7 +62,6 @@ export type { HandleNewMessageOptions, // 服务接口类型 - IMessageStateManager, IMessageDeduplication, IUserCacheService, IReadReceiptManager,