From 259de04f3ef12f1fe6f6fef39029c398af3bfb43 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Tue, 31 Mar 2026 19:15:13 +0800 Subject: [PATCH] refactor(message): replace SubscriptionManager with Zustand selectors Remove custom SubscriptionManager class and adopt Zustand's built-in selector mechanism for reactive state updates. This simplifies the architecture by eliminating manual subscription management and relying on Zustand's automatic dependency tracking for component re-renders. Key changes: - Remove SubscriptionManager class from store.ts - Replace all notifySubscribers calls with direct store updates - Update hooks to use Zustand selectors with useShallow for complex data - Remove subscribe/unsubscribe patterns from MessageManager - Simplify EmbeddedChat to use selector instead of local state - Rename requestConversationListRefresh to refreshConversations --- src/hooks/usePrefetch.ts | 5 +- src/screens/message/ChatScreen.tsx | 7 +- .../message/components/EmbeddedChat.tsx | 28 +- src/stores/index.ts | 2 +- src/stores/message/MessageManager.ts | 98 +----- src/stores/message/hooks.ts | 233 +++++--------- src/stores/message/index.ts | 7 +- .../services/ConversationOperations.ts | 116 +------ .../message/services/MessageSendService.ts | 30 +- .../message/services/MessageSyncService.ts | 151 +-------- .../message/services/ReadReceiptManager.ts | 56 +--- .../message/services/WSMessageHandler.ts | 113 +------ src/stores/message/store.ts | 57 +--- src/stores/messageManagerHooks.ts | 288 +++++------------- 14 files changed, 198 insertions(+), 993 deletions(-) diff --git a/src/hooks/usePrefetch.ts b/src/hooks/usePrefetch.ts index dcbb3e8..a4aa89c 100644 --- a/src/hooks/usePrefetch.ts +++ b/src/hooks/usePrefetch.ts @@ -153,10 +153,7 @@ function prefetchConversations(): void { key: 'conversations:list', executor: async () => { await messageManager.initialize(); - await messageManager.requestConversationListRefresh('prefetch', { - force: true, - allowDefer: true, - }); + await messageManager.refreshConversations(true, 'prefetch'); return messageManager.getConversations(); }, priority: Priority.HIGH, diff --git a/src/screens/message/ChatScreen.tsx b/src/screens/message/ChatScreen.tsx index 1e5f5ff..fe4e238 100644 --- a/src/screens/message/ChatScreen.tsx +++ b/src/screens/message/ChatScreen.tsx @@ -12,7 +12,7 @@ * - useChatScreen.ts: 自定义Hook，管理所有状态和逻辑 * - EmojiPanel.tsx: 表情面板组件 * - MorePanel.tsx: 更多功能面板组件 - * - MentionPanel.tsx: @成员选择面板组件 + * - MentionPanel.tsx: @成员选择 面板组件 * - LongPressMenu.tsx: 长按菜单组件 * - ChatHeader.tsx: 聊天头部组件 * - MessageBubble.tsx: 消息气泡组件 @@ -258,10 +258,7 @@ export const ChatScreen: React.FC = () => { useEffect(() => { const unsubscribe = navigation.addListener('beforeRemove', () => { // 刷新会话列表，确保已读状态正确显示 - messageManager.requestConversationListRefresh('chat-before-remove', { - force: true, - allowDefer: false, - }); + messageManager.refreshConversations(true, 'chat-before-remove'); }); return unsubscribe; }, [navigation]); diff --git a/src/screens/message/components/EmbeddedChat.tsx b/src/screens/message/components/EmbeddedChat.tsx index 0e078a1..84b3973 100644 --- a/src/screens/message/components/EmbeddedChat.tsx +++ b/src/screens/message/components/EmbeddedChat.tsx @@ -22,7 +22,7 @@ import { Image as ExpoImage } from 'expo-image'; import { spacing, fontSizes, shadows, useAppColors, type AppColors } from '../../../theme'; import { ConversationResponse, MessageResponse, MessageSegment, GroupMemberResponse, ImageSegmentData } from '../../../types/dto'; import { Avatar, Text, ImageGallery, AppBackButton } from '../../../components/common'; -import { useAuthStore, messageManager, MessageEvent, MessageSubscriber } from '../../../stores'; +import { useAuthStore, messageManager, useMessageStore } from '../../../stores'; import { formatDistanceToNow } from 'date-fns'; import { zhCN } from 'date-fns/locale'; import { extractTextFromSegments } from '../../../types/dto'; @@ -302,8 +302,8 @@ export const EmbeddedChat: React.FC = ({ conversation, onBack // 使用 markAsRead hook const { markAsRead } = useMarkAsRead(String(conversation.id)); - // 状态 - const [messages, setMessages] = useState([]); + // 状态 - 使用 Zustand selector 直接订阅消息 + const messages = useMessageStore(state => state.messagesMap.get(String(conversation.id)) || []); const [loading, setLoading] = useState(true); const [sending, setSending] = useState(false); const [inputText, setInputText] = useState(''); @@ -490,7 +490,6 @@ export const EmbeddedChat: React.FC = ({ conversation, onBack try { setLoading(true); await messageManager.fetchMessages(String(conversation.id)); - setMessages(messageManager.getMessages(String(conversation.id))); } catch (error: any) { // 检查是否是权限错误或会话不存在错误 const errorMessage = error?.message || String(error); @@ -501,8 +500,6 @@ export const EmbeddedChat: React.FC = ({ conversation, onBack error?.code === 'FORBIDDEN' ) { console.warn('[EmbeddedChat] 会话不存在或无权限访问:', conversation.id); - // 设置空消息列表，让 UI 显示"暂无消息"而不是崩溃 - setMessages([]); } else { console.error('[EmbeddedChat] 加载消息失败:', error); } @@ -514,25 +511,6 @@ export const EmbeddedChat: React.FC = ({ conversation, onBack // 初始加载 useEffect(() => { loadMessages(); - - // 订阅消息事件 - const subscriber: MessageSubscriber = (event: MessageEvent) => { - if ( - event.type === 'messages_updated' && - String(event.payload?.conversationId) === String(conversation.id) - ) { - setMessages(event.payload.messages || []); - setTimeout(() => { - flatListRef.current?.scrollToEnd({ animated: true }); - }, 100); - } - }; - - const unsubscribe = messageManager.subscribe(subscriber); - - return () => { - unsubscribe(); - }; }, [conversation.id, loadMessages]); // 自动标记已读 - 当有新消息时 diff --git a/src/stores/index.ts b/src/stores/index.ts index 3db970f..8205c0d 100644 --- a/src/stores/index.ts +++ b/src/stores/index.ts @@ -14,7 +14,7 @@ export { // MessageManager 新架构导出（已替换 conversationStore） // 重构：从 ./message 目录导入，移除了 messageManager.ts 重导出文件 -export { messageManager, MessageManager } from './message'; +export { messageManager, MessageManager, useMessageStore } from './message'; export type { MessageEvent, MessageEventType, diff --git a/src/stores/message/MessageManager.ts b/src/stores/message/MessageManager.ts index c4b409b..a9753f6 100644 --- a/src/stores/message/MessageManager.ts +++ b/src/stores/message/MessageManager.ts @@ -8,6 +8,7 @@ * - 移除了 MessageStateManager 包装层 * - 直接使用 zustand store * - 服务层移至 services/ 目录 + * - 移除了 SubscriptionManager，改用 Zustand selector 进行响应式订阅 */ import type { ConversationResponse, MessageResponse, MessageSegment, UserDTO } from '../../types/dto'; @@ -24,7 +25,7 @@ import { MessageSyncService } from './services/MessageSyncService'; import { WSMessageHandler } from './services/WSMessageHandler'; // 导入 store -import { useMessageStore, subscriptionManager, normalizeConversationId } from './store'; +import { useMessageStore, normalizeConversationId } from './store'; // 重新导出类型，保持兼容性 export type { @@ -164,23 +165,12 @@ class MessageManager { this.wsHandler.setBootstrapping(false); - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: this.getConversations() }, - timestamp: Date.now(), - }); - if (useMessageStore.getState().currentConversationId) { await this.fetchMessages(useMessageStore.getState().currentConversationId!); } } catch (error) { console.error('[MessageManager] 初始化失败:', error); this.wsHandler.setBootstrapping(false); - subscriptionManager.notifySubscribers({ - type: 'error', - payload: { error, context: 'initialize' }, - timestamp: Date.now(), - }); } })(); @@ -194,7 +184,6 @@ class MessageManager { destroy(): void { this.wsHandler.disconnect(); useMessageStore.getState().reset(); - subscriptionManager.clear(); this.syncService.restartSources(); this.readReceiptManager.reset(); this.deduplication.reset(); @@ -339,18 +328,6 @@ class MessageManager { await this.initialize(); this.setActiveConversation(normalizedId); - // 先下发当前内存快照 - const snapshot = this.getMessages(normalizedId); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId: normalizedId, - messages: [...snapshot], - source: 'activation_snapshot', - }, - timestamp: Date.now(), - }); - const existingTask = this.activatingConversationTasks.get(normalizedId); if (existingTask && !options?.forceSync) { await existingTask; @@ -371,81 +348,14 @@ class MessageManager { } } - // ==================== 订阅机制 ==================== - - subscribe(subscriber: import('./types').MessageSubscriber): () => void { - const unsubscribe = subscriptionManager.subscribe(subscriber); - - // 立即发送当前状态给新订阅者（与原始版本保持一致） - subscriber({ - type: 'conversations_updated', - payload: { conversations: this.getConversations() }, - timestamp: Date.now(), - }); - - const unreadCount = this.getUnreadCount(); - subscriber({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: unreadCount.total, - systemUnreadCount: unreadCount.system, - }, - timestamp: Date.now(), - }); - - subscriber({ - type: 'connection_changed', - payload: { connected: this.isConnected() }, - timestamp: Date.now(), - }); - - // 如果当前有活动会话，立即发送该会话的消息给新订阅者 - const activeConversationId = this.getActiveConversation(); - if (activeConversationId) { - const currentMessages = this.getMessages(activeConversationId); - if (currentMessages.length > 0) { - subscriber({ - type: 'messages_updated', - payload: { - conversationId: activeConversationId, - messages: [...currentMessages], - source: 'initial_sync', - }, - timestamp: Date.now(), - }); - } - } - - return unsubscribe; - } - - subscribeToMessages(conversationId: string, callback: (messages: MessageResponse[]) => void): () => void { - return subscriptionManager.subscribe((event) => { - if (event.type === 'messages_updated' && event.payload.conversationId === conversationId) { - callback(event.payload.messages); - } - }); - } - // ==================== 其他方法 ==================== canLoadMoreConversations(): boolean { return this.syncService.canLoadMoreConversations(); } - invalidateCache(_type?: 'all' | 'list' | 'unread'): void { - // 兼容性方法，新架构不需要外部缓存 - } - - async requestConversationListRefresh( - source: string, - options?: { force?: boolean; allowDefer?: boolean } - ): Promise { - const forceRefresh = options?.force ?? true; - const allowDefer = options?.allowDefer ?? true; - - if (allowDefer && this.shouldDeferConversationRefresh(source, forceRefresh)) { - // 延迟刷新逻辑在 syncService 中处理 + async refreshConversations(forceRefresh: boolean, source: string): Promise { + if (this.shouldDeferConversationRefresh(source, forceRefresh)) { return; } diff --git a/src/stores/message/hooks.ts b/src/stores/message/hooks.ts index 6078c2e..4638d03 100644 --- a/src/stores/message/hooks.ts +++ b/src/stores/message/hooks.ts @@ -2,16 +2,19 @@ * 消息模块 React Hooks * * 提供React组件与消息状态管理的集成 - * 所有hooks都基于zustand store和订阅机制 + * 所有hooks都基于zustand store的selector机制 * 确保组件能实时获取状态更新 * - * 重构说明：直接使用 zustand store，移除了对 messageManager 的直接依赖 + * 重构说明： + * - 移除了 SubscriptionManager，改用 Zustand selector + * - Zustand 会自动处理依赖追踪和组件重渲染 + * - 不需要手动管理订阅/取消订阅 */ import { useState, useEffect, useCallback, useRef } from 'react'; +import { useShallow } from 'zustand/react/shallow'; import { ConversationResponse, MessageResponse, MessageSegment } from '../../types/dto'; -import { useMessageStore, subscriptionManager, normalizeConversationId } from './store'; -import type { MessageEvent, MessageSubscriber } from './types'; +import { useMessageStore, normalizeConversationId } from './store'; // ==================== useConversations - 获取会话列表 ==================== @@ -26,6 +29,7 @@ interface UseConversationsReturn { /** * 获取会话列表 * 用于 MessageListScreen 等需要显示会话列表的组件 + * 使用 Zustand selector 自动订阅状态变化 */ export function useConversations( fetchConversations: (forceRefresh: boolean, source: string) => Promise, @@ -33,28 +37,12 @@ export function useConversations( canLoadMore: () => boolean, initialize: () => Promise ): UseConversationsReturn { - const store = useMessageStore(); - const [conversations, setConversations] = useState(() => store.getConversations()); - const [isLoading, setIsLoading] = useState(() => store.isLoading()); + // 使用 Zustand selector 直接订阅状态 + const conversations = useMessageStore(state => state.conversationList); + const isLoading = useMessageStore(state => state.isLoadingConversations); 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); @@ -69,10 +57,14 @@ export function useConversations( return () => { clearTimeout(coldStartSyncTimer); - unsubscribe(); }; }, []); + // 监听 hasMore 变化 + useEffect(() => { + setHasMore(canLoadMore()); + }, [conversations, canLoadMore]); + const refresh = useCallback(async () => { await fetchConversations(true, 'hooks-manual-refresh'); }, [fetchConversations]); @@ -102,30 +94,24 @@ interface UseMessagesReturn { /** * 获取指定会话的消息 + * 使用 Zustand selector 自动订阅消息变化 */ 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 store = useMessageStore(); - const [messages, setMessages] = useState(() => - store.getMessages(normalizedConversationId) - ); - const [isLoading, setIsLoading] = useState(() => - store.isLoadingMessages(normalizedConversationId) + // 使用 useShallow 避免每次返回新数组引用导致的无限循环 + const messages = useMessageStore( + useShallow(state => state.messagesMap.get(normalizedConversationId) || []) ); + const isLoadingMessages = useMessageStore(state => state.loadingMessagesSet.has(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); @@ -136,12 +122,11 @@ export function useMessages( if (store.getActiveConversation() === normalizedConversationId) { store.setCurrentConversation(null); } - unsubscribe(); }; }, [normalizedConversationId]); const loadMore = useCallback(async () => { - const currentMessages = store.getMessages(conversationId); + const currentMessages = messages; if (currentMessages.length === 0) return; // 消息数组在 store 内保持 seq 升序，首项即最早消息 @@ -150,16 +135,15 @@ export function useMessages( const loadedMessages = await loadMoreMessages(conversationId, minSeq, 20); setHasMore(loadedMessages.length > 0); - }, [conversationId, loadMoreMessages]); + }, [conversationId, messages, loadMoreMessages]); const refresh = useCallback(async () => { await fetchMessages(conversationId); - setMessages(store.getMessages(conversationId)); }, [conversationId, fetchMessages]); return { messages, - isLoading, + isLoading: isLoadingMessages, hasMore, loadMore, refresh, @@ -240,22 +224,16 @@ interface UseUnreadCountReturn { /** * 获取未读数 + * 使用 Zustand selector 直接订阅未读数变化 */ export function useUnreadCount(fetchUnreadCount: () => Promise): UseUnreadCountReturn { - const store = useMessageStore(); - const [counts, setCounts] = useState(() => store.getUnreadCount()); + // 使用 Zustand selector 直接订阅未读数 + const totalUnreadCount = useMessageStore(state => state.totalUnreadCount); + const systemUnreadCount = useMessageStore(state => state.systemUnreadCount); useEffect(() => { - const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { - if (event.type === 'unread_count_updated') { - setCounts(store.getUnreadCount()); - } - }); - // 初始获取 fetchUnreadCount(); - - return unsubscribe; }, []); const refresh = useCallback(async () => { @@ -263,8 +241,8 @@ export function useUnreadCount(fetchUnreadCount: () => Promise): UseUnread }, [fetchUnreadCount]); return { - totalUnreadCount: counts.total, - systemUnreadCount: counts.system, + totalUnreadCount, + systemUnreadCount, refresh, }; } @@ -280,26 +258,15 @@ interface UseSystemUnreadCountReturn { /** * 获取系统消息未读数 + * 使用 Zustand selector 直接订阅 */ 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; - }, []); + // 使用 Zustand selector 直接订阅 + const systemUnreadCount = useMessageStore(state => state.systemUnreadCount); const setSystemUnreadCount = useCallback((count: number) => { setSystemUnreadCountService(count); @@ -330,32 +297,19 @@ interface UseConversationReturn { /** * 获取单个会话 + * 使用 Zustand selector 直接订阅会话变化 */ export function useConversation( conversationId: string, fetchConversationDetail: (conversationId: string) => Promise ): UseConversationReturn { - const store = useMessageStore(); - const [conversation, setConversation] = useState(() => - store.getConversation(conversationId) + // 使用 useShallow 避免每次返回新对象引用导致的无限循环 + const conversation = useMessageStore( + useShallow(state => state.conversations.get(conversationId) || null) ); - 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 { @@ -364,43 +318,28 @@ export function useConversation( }; } -// ==================== useMessageManager - 通用订阅 ==================== +// ==================== useMessageManager - 通用状态 ==================== interface UseMessageManagerReturn { isConnected: boolean; - subscribe: (subscriber: MessageSubscriber) => () => void; } /** - * 通用消息订阅 + * 通用消息管理状态 + * 使用 Zustand selector 直接订阅连接状态 */ export function useMessageManager(initialize: () => Promise): UseMessageManagerReturn { - const store = useMessageStore(); - const [isConnected, setIsConnected] = useState(() => store.isConnected()); + // 使用 Zustand selector 直接订阅连接状态 + const isConnected = useMessageStore(state => state.isWSConnected); useEffect(() => { - const unsubscribe = subscriptionManager.subscribe((event: MessageEvent) => { - if (event.type === 'connection_changed') { - setIsConnected(event.payload.connected); - } - }); - - initialize().then(() => { - setIsConnected(store.isConnected()); - }).catch(error => { + initialize().catch(error => { console.error('[useMessageManager] 初始化失败:', error); }); - - return unsubscribe; - }, []); - - const subscribe = useCallback((subscriber: MessageSubscriber) => { - return subscriptionManager.subscribe(subscriber); }, []); return { isConnected, - subscribe, }; } @@ -467,25 +406,14 @@ interface UseGroupTypingReturn { /** * 获取群聊输入状态 + * 使用 Zustand selector 直接订阅 */ export function useGroupTyping(groupId: string): UseGroupTypingReturn { - const store = useMessageStore(); - const [typingUsers, setTypingUsers] = useState(() => - store.getTypingUsers(groupId) + // 使用 useShallow 避免每次返回新数组引用导致的无限循环 + const typingUsers = useMessageStore( + useShallow(state => state.typingUsersMap.get(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, }; @@ -500,28 +428,16 @@ interface UseGroupMutedReturn { /** * 获取群聊禁言状态 + * 使用 Zustand selector 直接订阅 */ export function useGroupMuted(groupId: string): UseGroupMutedReturn { + // 使用 Zustand selector 直接订阅禁言状态 + const isMuted = useMessageStore(state => state.mutedStatusMap.get(groupId) || false); 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]); + }, [groupId, store]); return { isMuted, @@ -529,29 +445,38 @@ export function useGroupMuted(groupId: string): UseGroupMutedReturn { }; } -// ==================== useActiveConversation - 活动会话 ==================== +// ==================== useConnectionStatus - 连接状态 ==================== -interface UseActiveConversationReturn { - activeConversationId: string | null; - setActiveConversation: (conversationId: string | null) => void; +interface UseConnectionStatusReturn { + isConnected: boolean; } /** - * 获取/设置活动会话 + * 获取连接状态 + * 使用 Zustand selector 直接订阅 */ -export function useActiveConversation(): UseActiveConversationReturn { - const store = useMessageStore(); - const [activeConversationId, setActiveConversationId] = useState(() => - store.getActiveConversation() - ); - - const setActiveConversation = useCallback((conversationId: string | null) => { - store.setCurrentConversation(conversationId); - setActiveConversationId(conversationId); - }, []); +export function useConnectionStatus(): UseConnectionStatusReturn { + const isConnected = useMessageStore(state => state.isWSConnected); return { - activeConversationId, - setActiveConversation, + isConnected, }; -} \ No newline at end of file +} + +// ==================== useIsInitialized - 初始化状态 ==================== + +interface UseIsInitializedReturn { + isInitialized: boolean; +} + +/** + * 获取初始化状态 + * 使用 Zustand selector 直接订阅 + */ +export function useIsInitialized(): UseIsInitializedReturn { + const isInitialized = useMessageStore(state => state.isInitialized); + + return { + isInitialized, + }; +} diff --git a/src/stores/message/index.ts b/src/stores/message/index.ts index 4297e6b..cb064a5 100644 --- a/src/stores/message/index.ts +++ b/src/stores/message/index.ts @@ -6,6 +6,7 @@ * - 直接使用 zustand store 进行状态管理 * - 服务层移至 services/ 目录 * - 新增 hooks.ts 封装常用逻辑 + * - 移除了 SubscriptionManager，改用 Zustand selector 进行响应式订阅 */ // ==================== 主管理器 ==================== @@ -14,11 +15,8 @@ export { messageManager, MessageManager } from './MessageManager'; // ==================== 状态管理 ==================== export { useMessageStore, - subscriptionManager, normalizeConversationId, mergeMessagesById, - notifySubscribers, - subscribeToMessages, } from './store'; export type { MessageState, MessageActions, MessageStore } from './store'; @@ -86,7 +84,6 @@ export { useConversation, useCreateConversation, useUpdateConversation, - useActiveConversation, // 消息相关 useMessages, @@ -103,6 +100,8 @@ export { // 通用 useMessageManager, + useConnectionStatus, + useIsInitialized, } from './hooks'; // ==================== 默认导出 ==================== diff --git a/src/stores/message/services/ConversationOperations.ts b/src/stores/message/services/ConversationOperations.ts index c9b69a3..92aa375 100644 --- a/src/stores/message/services/ConversationOperations.ts +++ b/src/stores/message/services/ConversationOperations.ts @@ -2,14 +2,16 @@ * 会话操作服务 * 处理会话的 CRUD 操作 * - * 重构说明：直接使用 zustand store 替代 IMessageStateManager + * 重构说明： + * - 直接使用 zustand store 替代 IMessageStateManager + * - 移除了 SubscriptionManager 调用，Zustand 的 set() 会自动触发组件重渲染 */ 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'; +import { useMessageStore, normalizeConversationId } from '../store'; export class ConversationOperations implements IConversationOperations { /** @@ -24,21 +26,9 @@ export class ConversationOperations implements IConversationOperations { // 添加到会话列表 store.updateConversation(conversation); - // 通知更新 - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); - return conversation; } catch (error) { console.error('[ConversationOperations] 创建会话失败:', error); - subscriptionManager.notifySubscribers({ - type: 'error', - payload: { error, context: 'createConversation' }, - timestamp: Date.now(), - }); return null; } } @@ -60,13 +50,6 @@ export class ConversationOperations implements IConversationOperations { }; store.updateConversation(updatedConv); - - // 通知更新 - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); } /** @@ -76,23 +59,6 @@ export class ConversationOperations implements IConversationOperations { const store = useMessageStore.getState(); store.removeConversation(conversationId); - // 通知更新 - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); - - const unreadCount = store.getUnreadCount(); - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: unreadCount.total, - systemUnreadCount: unreadCount.system, - }, - timestamp: Date.now(), - }); - // 删除本地数据库中的会话 deleteConversation(conversationId).catch(error => { console.error('[ConversationOperations] 删除本地会话失败:', error); @@ -105,21 +71,6 @@ export class ConversationOperations implements IConversationOperations { clearConversations(): void { const store = useMessageStore.getState(); store.reset(); - - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: [] }, - timestamp: Date.now(), - }); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: 0, - systemUnreadCount: 0, - }, - timestamp: Date.now(), - }); } /** @@ -149,22 +100,6 @@ export class ConversationOperations implements IConversationOperations { Math.max(0, currentUnread.total + diff), currentUnread.system ); - - // 通知更新 - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: store.getUnreadCount().total, - systemUnreadCount: store.getUnreadCount().system, - }, - timestamp: Date.now(), - }); - - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); } /** @@ -191,22 +126,6 @@ export class ConversationOperations implements IConversationOperations { currentUnread.total + 1, currentUnread.system ); - - // 通知更新 - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: store.getUnreadCount().total, - systemUnreadCount: store.getUnreadCount().system, - }, - timestamp: Date.now(), - }); - - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); } /** @@ -216,15 +135,6 @@ export class ConversationOperations implements IConversationOperations { const store = useMessageStore.getState(); const currentUnread = store.getUnreadCount(); store.setUnreadCount(currentUnread.total, count); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: currentUnread.total, - systemUnreadCount: count, - }, - timestamp: Date.now(), - }); } /** @@ -234,15 +144,6 @@ export class ConversationOperations implements IConversationOperations { const store = useMessageStore.getState(); const currentUnread = store.getUnreadCount(); store.setUnreadCount(currentUnread.total, currentUnread.system + 1); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: currentUnread.total, - systemUnreadCount: currentUnread.system + 1, - }, - timestamp: Date.now(), - }); } /** @@ -253,14 +154,5 @@ export class ConversationOperations implements IConversationOperations { const currentUnread = store.getUnreadCount(); const newSystemCount = Math.max(0, currentUnread.system - count); store.setUnreadCount(currentUnread.total, newSystemCount); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: currentUnread.total, - systemUnreadCount: newSystemCount, - }, - timestamp: Date.now(), - }); } } \ No newline at end of file diff --git a/src/stores/message/services/MessageSendService.ts b/src/stores/message/services/MessageSendService.ts index 3a05940..7347a4e 100644 --- a/src/stores/message/services/MessageSendService.ts +++ b/src/stores/message/services/MessageSendService.ts @@ -2,14 +2,16 @@ * 消息发送服务 * 处理消息发送相关逻辑 * - * 重构说明：直接使用 zustand store 替代 IMessageStateManager + * 重构说明： + * - 直接使用 zustand store 替代 IMessageStateManager + * - 移除了 SubscriptionManager 调用，Zustand 的 set() 会自动触发组件重渲染 */ 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'; +import { useMessageStore, mergeMessagesById } from '../store'; export class MessageSendService implements IMessageSendService { private getCurrentUserId: () => string | null; @@ -53,28 +55,9 @@ export class MessageSendService implements IMessageSendService { const updatedMessages = mergeMessagesById(existingMessages, [newMessage]); store.setMessages(conversationId, updatedMessages); - // 关键：立即广播消息列表更新 - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: updatedMessages, - newMessage, - source: 'send', - }, - timestamp: Date.now(), - }); - // 更新会话最后消息 this.updateConversationWithNewMessage(conversationId, newMessage); - // 通知消息已发送 - subscriptionManager.notifySubscribers({ - type: 'message_sent', - payload: { conversationId, message: newMessage }, - timestamp: Date.now(), - }); - // 保存到本地数据库 const textContent = segments ?.filter((s: any) => s.type === 'text') @@ -100,11 +83,6 @@ export class MessageSendService implements IMessageSendService { return null; } catch (error) { console.error('[MessageSendService] 发送消息失败:', error); - subscriptionManager.notifySubscribers({ - type: 'error', - payload: { error, context: 'sendMessage' }, - timestamp: Date.now(), - }); return null; } } diff --git a/src/stores/message/services/MessageSyncService.ts b/src/stores/message/services/MessageSyncService.ts index c24a8d9..2c1a605 100644 --- a/src/stores/message/services/MessageSyncService.ts +++ b/src/stores/message/services/MessageSyncService.ts @@ -2,7 +2,9 @@ * 消息同步服务 * 处理消息和会话的服务器同步逻辑 * - * 重构说明：直接使用 zustand store 替代 IMessageStateManager + * 重构说明： + * - 直接使用 zustand store 替代 IMessageStateManager + * - 移除了 SubscriptionManager 调用，Zustand 的 set() 会自动触发组件重渲染 */ import type { ConversationResponse, MessageResponse } from '../../../types/dto'; @@ -21,7 +23,7 @@ import { CONVERSATION_LIST_PAGE_SIZE, } from '../../conversationListSources'; import type { IMessageSyncService, MessageManagerConversationListDeps, IUserCacheService } from '../types'; -import { useMessageStore, subscriptionManager, normalizeConversationId, mergeMessagesById } from '../store'; +import { useMessageStore, normalizeConversationId, mergeMessagesById } from '../store'; import { ReadReceiptManager } from './ReadReceiptManager'; export class MessageSyncService implements IMessageSyncService { @@ -94,19 +96,11 @@ export class MessageSyncService implements IMessageSyncService { if (!forceRefresh && currentState.conversations.size === 0) { const warmed = await this.hydrateConversationsFromLocalSource(); if (warmed) { - this.emitConversationListAndUnreadUpdates(); + this.persistConversationListCache(); } } store.setLoading(true); - const emitLoadingToUi = store.getConversations().length === 0; - if (emitLoadingToUi) { - subscriptionManager.notifySubscribers({ - type: 'conversations_loading', - payload: { loading: true }, - timestamp: Date.now(), - }); - } try { this.remoteConversationListSource.restart(); @@ -129,29 +123,17 @@ export class MessageSyncService implements IMessageSyncService { ); store.setUnreadCount(totalUnread, store.getUnreadCount().system); - this.emitConversationListAndUnreadUpdates(); + this.persistConversationListCache(); } catch (error) { console.error('[MessageSyncService] 获取会话列表失败:', error); if (store.getConversations().length === 0) { const recovered = await this.hydrateConversationsFromLocalSource(); if (recovered) { - this.emitConversationListAndUnreadUpdates(); + this.persistConversationListCache(); } } - subscriptionManager.notifySubscribers({ - type: 'error', - payload: { error, context: 'fetchConversations' }, - timestamp: Date.now(), - }); } finally { store.setLoading(false); - if (emitLoadingToUi) { - subscriptionManager.notifySubscribers({ - type: 'conversations_loading', - payload: { loading: false }, - timestamp: Date.now(), - }); - } } } @@ -178,14 +160,9 @@ export class MessageSyncService implements IMessageSyncService { }); this.recomputeConversationTotalUnread(); - this.emitConversationListAndUnreadUpdates(); + this.persistConversationListCache(); } catch (error) { console.error('[MessageSyncService] 加载更多会话失败:', error); - subscriptionManager.notifySubscribers({ - type: 'error', - payload: { error, context: 'loadMoreConversations' }, - timestamp: Date.now(), - }); } finally { this.loadingMoreConversations = false; } @@ -252,27 +229,9 @@ export class MessageSyncService implements IMessageSyncService { })); store.setMessages(conversationId, formattedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: formattedMessages, - source: 'local', - }, - timestamp: Date.now(), - }); } else { - // 冷启动兜底：先下发空列表事件 + // 冷启动兜底：先设置空列表 store.setMessages(conversationId, []); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: [], - source: 'local_empty', - }, - timestamp: Date.now(), - }); } } catch (error) { console.warn('[MessageSyncService] 读取本地消息失败:', error); @@ -288,17 +247,6 @@ export class MessageSyncService implements IMessageSyncService { const existingMessages = store.getMessages(conversationId); const mergedSnapshot = mergeMessagesById(existingMessages, snapshotMessages); store.setMessages(conversationId, mergedSnapshot); - - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: mergedSnapshot, - newMessages: snapshotMessages, - source: 'server_snapshot', - }, - timestamp: Date.now(), - }); // 持久化到本地 saveMessagesBatch(snapshotMessages.map((m: any) => ({ @@ -327,17 +275,6 @@ export class MessageSyncService implements IMessageSyncService { const existingMessages = store.getMessages(conversationId); const mergedMessages = mergeMessagesById(existingMessages, newMessages); store.setMessages(conversationId, mergedMessages); - - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: mergedMessages, - newMessages, - source: 'server_incremental', - }, - timestamp: Date.now(), - }); saveMessagesBatch(newMessages.map((m: any) => ({ id: m.id, @@ -368,16 +305,6 @@ export class MessageSyncService implements IMessageSyncService { const mergedMessages = mergeMessagesById(existingMessages, newMessages); store.setMessages(conversationId, mergedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: mergedMessages, - newMessages, - source: 'server', - }, - timestamp: Date.now(), - }); saveMessagesBatch(newMessages.map((m: any) => ({ id: m.id, @@ -397,11 +324,6 @@ export class MessageSyncService implements IMessageSyncService { } } catch (error) { console.error('[MessageSyncService] 获取消息失败:', error); - subscriptionManager.notifySubscribers({ - type: 'error', - payload: { error, context: 'fetchMessages', conversationId }, - timestamp: Date.now(), - }); } finally { // 异步填充用户信息（不阻塞消息显示） const currentMessages = store.getMessages(conversationId); @@ -411,15 +333,6 @@ export class MessageSyncService implements IMessageSyncService { currentMessages, (convId, enrichedMessages) => { store.setMessages(convId, enrichedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId: convId, - messages: enrichedMessages, - source: 'sender_enriched', - }, - timestamp: Date.now(), - }); } ); } @@ -452,12 +365,6 @@ export class MessageSyncService implements IMessageSyncService { const mergedMessages = this.mergeOlderMessages(existingMessages, formattedMessages); store.setMessages(conversationId, mergedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { conversationId, messages: mergedMessages, source: 'local_history' }, - timestamp: Date.now(), - }); - return formattedMessages; } @@ -484,12 +391,6 @@ export class MessageSyncService implements IMessageSyncService { const mergedMessages = this.mergeOlderMessages(existingMessages, serverMessages); store.setMessages(conversationId, mergedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { conversationId, messages: mergedMessages, source: 'server_history' }, - timestamp: Date.now(), - }); - return serverMessages; } @@ -532,22 +433,8 @@ export class MessageSyncService implements IMessageSyncService { // 使用 zustand set 函数正确更新状态 store.setConversations(newConversations); this.persistConversationListCache(); - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); } } - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: totalUnread, - systemUnreadCount: systemUnread, - }, - timestamp: Date.now(), - }); } catch (error) { console.error('[MessageSyncService] 获取未读数失败:', error); } @@ -598,26 +485,6 @@ export class MessageSyncService implements IMessageSyncService { store.setUnreadCount(totalUnread, store.getUnreadCount().system); } - private emitConversationListAndUnreadUpdates(): void { - const store = useMessageStore.getState(); - this.persistConversationListCache(); - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); - - const unreadCount = store.getUnreadCount(); - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: unreadCount.total, - systemUnreadCount: unreadCount.system, - }, - timestamp: Date.now(), - }); - } - private persistConversationListCache(): void { const store = useMessageStore.getState(); const currentState = store.getState(); diff --git a/src/stores/message/services/ReadReceiptManager.ts b/src/stores/message/services/ReadReceiptManager.ts index 3430d04..0592d20 100644 --- a/src/stores/message/services/ReadReceiptManager.ts +++ b/src/stores/message/services/ReadReceiptManager.ts @@ -2,7 +2,9 @@ * 已读回执管理器 * 管理消息已读状态的同步和保护 * - * 重构说明：直接使用 zustand store 替代 IMessageStateManager + * 重构说明： + * - 直接使用 zustand store 替代 IMessageStateManager + * - 移除了 SubscriptionManager 调用，Zustand 的 set() 会自动触发组件重渲染 */ import type { ConversationResponse } from '../../../types/dto'; @@ -10,7 +12,7 @@ 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'; +import { useMessageStore, normalizeConversationId } from '../store'; export class ReadReceiptManager implements IReadReceiptManager { /** @@ -78,27 +80,7 @@ export class ReadReceiptManager implements IReadReceiptManager { // 3. 更新本地数据库 markConversationAsRead(normalizedId).catch(console.error); - // 4. 立即通知所有订阅者 - subscriptionManager.notifySubscribers({ - type: 'message_read', - payload: { - conversationId: normalizedId, - unreadCount: 0, - totalUnreadCount: newTotalUnread, - }, - timestamp: Date.now(), - }); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: newTotalUnread, - systemUnreadCount: currentUnread.system, - }, - timestamp: Date.now(), - }); - - // 5. 调用 API，完成后设置延迟清除保护 + // 4. 调用 API，完成后设置延迟清除保护 try { await messageService.markAsRead(normalizedId, seq); } catch (error) { @@ -107,16 +89,6 @@ export class ReadReceiptManager implements IReadReceiptManager { // 失败时回滚状态 store.updateConversation(conversation); store.setUnreadCount(currentUnread.total, currentUnread.system); - - subscriptionManager.notifySubscribers({ - type: 'message_read', - payload: { - conversationId: normalizedId, - unreadCount: prevUnreadCount, - totalUnreadCount: currentUnread.total, - }, - timestamp: Date.now(), - }); // API 失败时立即清除保护 this.pendingReadMap.delete(normalizedId); @@ -156,15 +128,6 @@ export class ReadReceiptManager implements IReadReceiptManager { store.setUnreadCount(0, currentSystem); - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: 0, - systemUnreadCount: currentSystem, - }, - timestamp: Date.now(), - }); - try { // 标记系统消息已读 await messageService.markAllSystemMessagesRead(); @@ -182,15 +145,6 @@ export class ReadReceiptManager implements IReadReceiptManager { console.error('[ReadReceiptManager] 标记所有已读失败:', error); // 回滚 store.setUnreadCount(prevTotalUnread, currentSystem); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: prevTotalUnread, - systemUnreadCount: currentSystem, - }, - timestamp: Date.now(), - }); } } diff --git a/src/stores/message/services/WSMessageHandler.ts b/src/stores/message/services/WSMessageHandler.ts index 383f842..ff06395 100644 --- a/src/stores/message/services/WSMessageHandler.ts +++ b/src/stores/message/services/WSMessageHandler.ts @@ -2,7 +2,9 @@ * WebSocket 消息处理器 * 处理所有来自 WebSocket 的消息事件 * - * 重构说明：直接使用 zustand store 替代 IMessageStateManager + * 重构说明： + * - 直接使用 zustand store 替代 IMessageStateManager + * - 移除了 SubscriptionManager 调用，Zustand 的 set() 会自动触发组件重渲染 */ import type { @@ -26,7 +28,7 @@ import type { HandleNewMessageOptions, } from '../types'; import { MAX_FLUSH_ITERATIONS } from '../constants'; -import { useMessageStore, subscriptionManager, normalizeConversationId, mergeMessagesById } from '../store'; +import { useMessageStore, normalizeConversationId, mergeMessagesById } from '../store'; export class WSMessageHandler implements IWSMessageHandler { private deduplication: IMessageDeduplication; @@ -134,11 +136,6 @@ export class WSMessageHandler implements IWSMessageHandler { // 监听连接状态 wsService.onConnect(() => { store.setSSEConnected(true); - subscriptionManager.notifySubscribers({ - type: 'connection_changed', - payload: { connected: true }, - timestamp: Date.now(), - }); // 冷启动/重连兜底 const now = Date.now(); @@ -166,11 +163,6 @@ export class WSMessageHandler implements IWSMessageHandler { wsService.onDisconnect(() => { store.setSSEConnected(false); - subscriptionManager.notifySubscribers({ - type: 'connection_changed', - payload: { connected: false }, - timestamp: Date.now(), - }); }); } @@ -268,14 +260,6 @@ export class WSMessageHandler implements IWSMessageHandler { m.id === id ? { ...m, sender: user } : m ); store.setMessages(normalizedConversationId, updatedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId: normalizedConversationId, - messages: [...updatedMessages], - }, - timestamp: Date.now(), - }); } } }).catch(error => { @@ -301,16 +285,6 @@ export class WSMessageHandler implements IWSMessageHandler { if (!messageExists) { const updatedMessages = mergeMessagesById(existingMessages, [newMessage]); store.setMessages(normalizedConversationId, updatedMessages); - - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId: normalizedConversationId, - messages: updatedMessages, - newMessage, - }, - timestamp: Date.now(), - }); } // 更新会话信息 @@ -358,17 +332,6 @@ export class WSMessageHandler implements IWSMessageHandler { }).catch(error => { console.error('[WSMessageHandler] 保存消息到本地失败:', error); }); - - // 通知收到新消息 - subscriptionManager.notifySubscribers({ - type: 'message_received', - payload: { - conversationId: conversation_id, - message: newMessage, - isCurrentUser: isCurrentUserMessage, - }, - timestamp: Date.now(), - }); } /** @@ -395,12 +358,6 @@ export class WSMessageHandler implements IWSMessageHandler { this.markMessageAsRecalled(normalizedConversationId, message_id); this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id); - subscriptionManager.notifySubscribers({ - type: 'message_recalled', - payload: { conversationId: normalizedConversationId, messageId: message_id }, - timestamp: Date.now(), - }); - updateMessageStatus(message_id, 'recalled', true).catch(error => { console.error('[WSMessageHandler] 更新本地消息撤回状态失败:', error); }); @@ -416,12 +373,6 @@ export class WSMessageHandler implements IWSMessageHandler { this.markMessageAsRecalled(normalizedConversationId, message_id); this.syncConversationLastMessageOnRecall(normalizedConversationId, message_id); - subscriptionManager.notifySubscribers({ - type: 'message_recalled', - payload: { conversationId: normalizedConversationId, messageId: message_id, isGroup: true }, - timestamp: Date.now(), - }); - updateMessageStatus(message_id, 'recalled', true).catch(error => { console.error('[WSMessageHandler] 更新本地消息撤回状态失败:', error); }); @@ -450,11 +401,6 @@ export class WSMessageHandler implements IWSMessageHandler { if (updatedTypingUsers.length !== currentTypingUsers.length) { store.setTypingUsers(groupIdStr, updatedTypingUsers); - subscriptionManager.notifySubscribers({ - type: 'typing_status', - payload: { groupId: groupIdStr, typingUsers: updatedTypingUsers }, - timestamp: Date.now(), - }); } } @@ -500,34 +446,9 @@ export class WSMessageHandler implements IWSMessageHandler { const updatedMessages = [...existingMessages, systemNoticeMessage].sort((a, b) => a.seq - b.seq); store.setMessages(conversationId, updatedMessages); - - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { - conversationId, - messages: updatedMessages, - newMessage: systemNoticeMessage, - source: 'group_notice', - }, - timestamp: Date.now(), - }); } } } - - // 通知订阅者群通知 - subscriptionManager.notifySubscribers({ - type: 'group_notice', - payload: { - groupId: groupIdStr, - noticeType: notice_type, - data, - timestamp, - messageId: message_id, - seq, - }, - timestamp: Date.now(), - }); } // ==================== 私有工具方法 ==================== @@ -550,12 +471,6 @@ export class WSMessageHandler implements IWSMessageHandler { }; store.updateConversation(updatedConv); } - - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); } private incrementUnreadCount(conversationId: string): void { @@ -571,21 +486,6 @@ export class WSMessageHandler implements IWSMessageHandler { const currentUnread = store.getUnreadCount(); store.setUnreadCount(currentUnread.total + 1, currentUnread.system); - - subscriptionManager.notifySubscribers({ - type: 'unread_count_updated', - payload: { - totalUnreadCount: store.getUnreadCount().total, - systemUnreadCount: store.getUnreadCount().system, - }, - timestamp: Date.now(), - }); - - subscriptionManager.notifySubscribers({ - type: 'conversations_updated', - payload: { conversations: store.getConversations() }, - timestamp: Date.now(), - }); } } @@ -610,11 +510,6 @@ export class WSMessageHandler implements IWSMessageHandler { if (!changed) return; store.setMessages(conversationId, updatedMessages); - subscriptionManager.notifySubscribers({ - type: 'messages_updated', - payload: { conversationId, messages: updatedMessages }, - timestamp: Date.now(), - }); } private syncConversationLastMessageOnRecall(conversationId: string, messageId: string): void { diff --git a/src/stores/message/store.ts b/src/stores/message/store.ts index 730ab52..443f017 100644 --- a/src/stores/message/store.ts +++ b/src/stores/message/store.ts @@ -1,11 +1,11 @@ /** * 消息状态 Zustand Store * 使用 zustand 进行状态管理，提供响应式状态更新 - * + * * 重构说明： * - 移除了 MessageStateManager 包装层 * - 直接使用 zustand store 进行状态管理 - * - 订阅管理器集成到 store 中 + * - 移除了 SubscriptionManager，改用 Zustand selector 进行响应式订阅 */ import { create } from 'zustand'; @@ -13,10 +13,6 @@ import type { ConversationResponse, MessageResponse, } from '../../types/dto'; -import type { - MessageSubscriber, - MessageEvent, -} from './types'; // ==================== 状态接口 ==================== @@ -377,52 +373,3 @@ export const useMessageStore = create((set, get) => ({ }, })); -// ==================== 订阅管理器 ==================== - -/** - * 订阅管理器类 - * 管理消息事件的订阅和通知 - */ -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/messageManagerHooks.ts b/src/stores/messageManagerHooks.ts index 9f455ea..35fc326 100644 --- a/src/stores/messageManagerHooks.ts +++ b/src/stores/messageManagerHooks.ts @@ -2,13 +2,20 @@ * MessageManager React Hooks * * 提供React组件与MessageManager的集成 - * 所有hooks都基于MessageManager的订阅机制 + * 所有hooks都基于Zustand selector机制 * 确保组件能实时获取状态更新 + * + * 重构说明： + * - 移除了手动订阅机制，改用 Zustand selector + * - Zustand 会自动处理依赖追踪和组件重渲染 + * - 不需要手动管理订阅/取消订阅 */ import { useState, useEffect, useCallback, useRef } from 'react'; +import { useShallow } from 'zustand/react/shallow'; import { ConversationResponse, MessageResponse, MessageSegment } from '../types/dto'; -import { messageManager, MessageEvent } from './messageManager'; +import { messageManager } from './messageManager'; +import { useMessageStore } from './message'; // ==================== useConversations - 获取会话列表 ==================== @@ -21,33 +28,16 @@ interface UseConversationsReturn { } /** - * 获取会话列表（仅消费 MessageManager；网络游标与 SQLite 回退均在 Manager 内完成，调用方无感） - * 用于 MessageListScreen 等需要显示会话列表的组件 + * 获取会话列表 + * 使用 Zustand selector 自动订阅状态变化 */ export function useConversations(): UseConversationsReturn { - const [conversations, setConversations] = useState( - () => messageManager.getConversations() - ); - const [isLoading, setIsLoading] = useState(() => messageManager.isLoading()); + // 使用 Zustand selector 直接订阅状态 + const conversations = useMessageStore(state => state.conversationList); + const isLoading = useMessageStore(state => state.isLoadingConversations); const [hasMore, setHasMore] = useState(() => messageManager.canLoadMoreConversations()); useEffect(() => { - // 订阅MessageManager的事件 - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - switch (event.type) { - case 'conversations_updated': - setConversations(messageManager.getConversations()); - setHasMore(messageManager.canLoadMoreConversations()); - break; - case 'conversations_loading': - setIsLoading(!!event.payload?.loading); - break; - case 'connection_changed': - // 连接状态变化时可能需要刷新 - break; - } - }); - // 初始化时确保 MessageManager 已初始化 messageManager.initialize().catch(error => { console.error('[useConversations] 初始化失败:', error); @@ -55,25 +45,23 @@ export function useConversations(): UseConversationsReturn { // 冷启动兜底：延迟做一次强制刷新，避免首次因时序问题拿到空列表 const coldStartSyncTimer = setTimeout(() => { - messageManager.requestConversationListRefresh('hooks-initial-refresh', { - force: true, - allowDefer: true, - }).catch(error => { + messageManager.refreshConversations(true, 'hooks-initial-refresh').catch(error => { console.error('[useConversations] 冷启动强制刷新失败:', error); }); }, 1200); return () => { clearTimeout(coldStartSyncTimer); - unsubscribe(); }; }, []); + // 监听 hasMore 变化 + useEffect(() => { + setHasMore(messageManager.canLoadMoreConversations()); + }, [conversations]); + const refresh = useCallback(async () => { - await messageManager.requestConversationListRefresh('hooks-manual-refresh', { - force: true, - allowDefer: false, - }); + await messageManager.refreshConversations(true, 'hooks-manual-refresh'); }, []); const loadMore = useCallback(async () => { @@ -101,66 +89,40 @@ interface UseMessagesReturn { /** * 获取指定会话的消息 - * 用于ChatScreen等需要显示消息的组件 - * - * @param conversationId 会话ID + * 使用 Zustand selector 自动订阅消息变化 */ export function useMessages(conversationId: string | null): UseMessagesReturn { - const [messages, setMessages] = useState([]); - const [isLoading, setIsLoading] = useState(false); + const normalizedConversationId = conversationId ? String(conversationId) : null; + + // 使用 useShallow 避免每次返回新数组引用导致的无限循环 + const messages = useMessageStore( + useShallow(state => + normalizedConversationId ? (state.messagesMap.get(normalizedConversationId) || []) : [] + ) + ); + const isLoadingMessages = useMessageStore(state => + normalizedConversationId ? state.loadingMessagesSet.has(normalizedConversationId) : false + ); const [hasMore, setHasMore] = useState(true); const loadMoreInFlightRef = useRef(false); useEffect(() => { - if (!conversationId) { - setMessages([]); + if (!normalizedConversationId) { return; } - const normalizedConversationId = String(conversationId); - - setIsLoading(true); - setHasMore(true); - let isUnmounted = false; - - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if ( - event.type === 'messages_updated' && - String(event.payload.conversationId) === normalizedConversationId - ) { - setMessages(event.payload.messages); - if (!isUnmounted) { - setIsLoading(false); - } - } - }); - - // 先读取内存快照，确保页面切换时无闪烁 - const cachedMessages = messageManager.getMessages(normalizedConversationId); - if (cachedMessages.length > 0) { - setMessages(cachedMessages); - } - // 架构入口：激活会话并完成初始化/同步 messageManager.activateConversation(normalizedConversationId).catch(error => { console.error('[useMessages] 激活会话失败:', error); - if (!isUnmounted) { - setIsLoading(false); - } - }).finally(() => { - if (!isUnmounted) { - setIsLoading(false); - } }); return () => { - isUnmounted = true; - unsubscribe(); - if (String(messageManager.getActiveConversation()) === normalizedConversationId) { + // 清理活动会话 + if (messageManager.getActiveConversation() === normalizedConversationId) { messageManager.setActiveConversation(null); } }; - }, [conversationId]); + }, [normalizedConversationId]); const loadMore = useCallback(async () => { if (!conversationId || !hasMore || loadMoreInFlightRef.current) return; @@ -184,16 +146,13 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { const refresh = useCallback(async () => { if (!conversationId) return; - setIsLoading(true); await messageManager.fetchMessages(conversationId); - setMessages(messageManager.getMessages(conversationId)); - setIsLoading(false); setHasMore(true); }, [conversationId]); return { messages, - isLoading, + isLoading: isLoadingMessages, hasMore, loadMore, refresh, @@ -209,7 +168,6 @@ interface UseSendMessageReturn { /** * 发送消息 - * 返回发送函数和发送状态 */ export function useSendMessage(conversationId: string | null): UseSendMessageReturn { const [isSending, setIsSending] = useState(false); @@ -245,7 +203,6 @@ interface UseMarkAsReadReturn { /** * 标记已读 - * 用于ChatScreen标记当前会话已读 */ export function useMarkAsRead(conversationId: string | null): UseMarkAsReadReturn { const [isMarking, setIsMarking] = useState(false); @@ -289,27 +246,21 @@ interface UseUnreadCountReturn { /** * 获取未读消息数 - * 用于TabBar角标等 + * 使用 Zustand selector 直接订阅未读数变化 */ export function useUnreadCount(): UseUnreadCountReturn { - const [counts, setCounts] = useState(() => messageManager.getUnreadCount()); + // 使用 Zustand selector 直接订阅未读数 + const totalUnreadCount = useMessageStore(state => state.totalUnreadCount); + const systemUnreadCount = useMessageStore(state => state.systemUnreadCount); useEffect(() => { - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if (event.type === 'unread_count_updated') { - setCounts(messageManager.getUnreadCount()); - } - }); - // 初始化时获取最新未读数 messageManager.fetchUnreadCount(); - - return unsubscribe; }, []); return { - totalUnreadCount: counts.total, - systemUnreadCount: counts.system, + totalUnreadCount, + systemUnreadCount, }; } @@ -317,29 +268,18 @@ export function useUnreadCount(): UseUnreadCountReturn { /** * 获取总未读消息数（会话未读 + 系统消息未读） - * 专用于TabBar徽章显示 + * 使用 Zustand selector 直接订阅 */ export function useTotalUnreadCount(): number { - const [totalCount, setTotalCount] = useState(() => { - const counts = messageManager.getUnreadCount(); - return counts.total + counts.system; - }); + const totalUnreadCount = useMessageStore(state => state.totalUnreadCount); + const systemUnreadCount = useMessageStore(state => state.systemUnreadCount); useEffect(() => { - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if (event.type === 'unread_count_updated') { - const counts = messageManager.getUnreadCount(); - setTotalCount(counts.total + counts.system); - } - }); - // 初始化时获取最新未读数 messageManager.fetchUnreadCount(); - - return unsubscribe; }, []); - return totalCount; + return totalUnreadCount + systemUnreadCount; } // ==================== useConversation - 获取单个会话信息 ==================== @@ -352,37 +292,22 @@ interface UseConversationReturn { /** * 获取单个会话信息 + * 使用 Zustand selector 直接订阅会话变化 */ export function useConversation(conversationId: string | null): UseConversationReturn { - const [conversation, setConversation] = useState(null); + // 使用 useShallow 避免每次返回新对象引用导致的无限循环 + const conversation = useMessageStore( + useShallow(state => + conversationId ? (state.conversations.get(conversationId) || null) : null + ) + ); const [isLoading, setIsLoading] = useState(false); - useEffect(() => { - if (!conversationId) { - setConversation(null); - return; - } - - // 获取当前会话信息 - setConversation(messageManager.getConversation(conversationId)); - - // 订阅更新 - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if (event.type === 'conversations_updated') { - const updated = messageManager.getConversation(conversationId); - setConversation(updated); - } - }); - - return unsubscribe; - }, [conversationId]); - const refresh = useCallback(async () => { if (!conversationId) return; setIsLoading(true); await messageManager.fetchConversationDetail(conversationId); - setConversation(messageManager.getConversation(conversationId)); setIsLoading(false); }, [conversationId]); @@ -393,7 +318,7 @@ export function useConversation(conversationId: string | null): UseConversationR }; } -// ==================== useMessageManager - 通用订阅 ==================== +// ==================== useMessageManager - 通用状态 ==================== interface UseMessageManagerReturn { isConnected: boolean; @@ -401,26 +326,19 @@ interface UseMessageManagerReturn { } /** - * 通用MessageManager订阅 - * 用于获取连接状态等全局信息 + * 通用MessageManager状态 + * 使用 Zustand selector 直接订阅连接状态 */ export function useMessageManager(): UseMessageManagerReturn { - const [isConnected, setIsConnected] = useState(() => messageManager.isConnected()); - const [isInitialized, setIsInitialized] = useState(false); + // 使用 Zustand selector 直接订阅状态 + const isConnected = useMessageStore(state => state.isWSConnected); + const isInitialized = useMessageStore(state => state.isInitialized); useEffect(() => { - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if (event.type === 'connection_changed') { - setIsConnected(event.payload.connected); - } - }); - // 初始化 - messageManager.initialize().then(() => { - setIsInitialized(true); + messageManager.initialize().catch(error => { + console.error('[useMessageManager] 初始化失败:', error); }); - - return unsubscribe; }, []); return { @@ -517,10 +435,6 @@ interface UseMessageListRefreshReturn { /** * MessageListScreen焦点刷新Hook - * 从 ChatScreen 返回时，MessageManager 内存状态已经是最新的（markAsRead 乐观更新）， - * 无需再请求服务器（避免时序竞争覆盖已读状态）。 - * 仅在用户主动下拉刷新时才触发服务器请求。 - * 注意：isFocused 参数已移除，不再触发焦点刷新 */ export function useMessageListRefresh(): UseMessageListRefreshReturn { const [isRefreshing, setIsRefreshing] = useState(false); @@ -536,10 +450,7 @@ export function useMessageListRefresh(): UseMessageListRefreshReturn { setIsRefreshing(true); try { - await messageManager.requestConversationListRefresh('hooks-load-more-fallback', { - force: true, - allowDefer: false, - }); + await messageManager.refreshConversations(true, 'hooks-load-more-fallback'); } catch (error) { console.error('[useMessageListRefresh] 刷新失败:', error); } finally { @@ -547,9 +458,6 @@ export function useMessageListRefresh(): UseMessageListRefreshReturn { } }, []); - // 获得焦点时不主动拉服务器——内存状态已是最新，避免时序竞争 - // refresh 仅供下拉刷新手动调用 - return { refresh, isRefreshing, @@ -560,7 +468,6 @@ export function useMessageListRefresh(): UseMessageListRefreshReturn { /** * ChatScreen专用Hook组合 - * 整合所有ChatScreen需要的功能 */ interface UseChatReturn { // 消息相关 @@ -608,29 +515,15 @@ interface UseGroupTypingReturn { /** * 获取群聊输入状态 - * @param groupId 群组ID + * 使用 Zustand selector 直接订阅 */ export function useGroupTyping(groupId: string | null): UseGroupTypingReturn { - const [typingUsers, setTypingUsers] = useState([]); - - useEffect(() => { - if (!groupId) { - setTypingUsers([]); - return; - } - - // 初始值 - setTypingUsers(messageManager.getTypingUsers(groupId)); - - // 订阅输入状态变化 - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if (event.type === 'typing_status' && event.payload.groupId === groupId) { - setTypingUsers(event.payload.typingUsers); - } - }); - - return unsubscribe; - }, [groupId]); + // 使用 useShallow 避免每次返回新数组引用导致的无限循环 + const typingUsers = useMessageStore( + useShallow(state => + groupId ? (state.typingUsersMap.get(groupId) || []) : [] + ) + ); return { typingUsers }; } @@ -644,42 +537,17 @@ interface UseGroupMutedReturn { /** * 获取群聊禁言状态 - * @param groupId 群组ID - * @param currentUserId 当前用户ID（用于判断禁言通知是否针对当前用户） + * 使用 Zustand selector 直接订阅 */ export function useGroupMuted(groupId: string | null, currentUserId?: string): UseGroupMutedReturn { - const [isMuted, setIsMuted] = useState(false); - - useEffect(() => { - if (!groupId) { - setIsMuted(false); - return; - } - - // 初始值 - setIsMuted(messageManager.isMuted(groupId)); - - // 订阅群通知变化 - const unsubscribe = messageManager.subscribe((event: MessageEvent) => { - if (event.type === 'group_notice' && event.payload.groupId === groupId) { - const { noticeType, data } = event.payload; - - // 如果提供了当前用户ID，只响应当前用户的禁言/解禁通知 - if ((noticeType === 'muted' || noticeType === 'unmuted')) { - if (!currentUserId || data?.user_id === currentUserId) { - setIsMuted(noticeType === 'muted'); - } - } - } - }); - - return unsubscribe; - }, [groupId, currentUserId]); + // 使用 Zustand selector 直接订阅禁言状态 + const isMuted = useMessageStore(state => + groupId ? (state.mutedStatusMap.get(groupId) || false) : false + ); const setMuted = useCallback((muted: boolean) => { if (groupId) { messageManager.setMutedStatus(groupId, muted); - setIsMuted(muted); } }, [groupId]); @@ -698,13 +566,12 @@ interface UseMessageListReturn { totalUnreadCount: number; systemUnreadCount: number; markAllAsRead: () => Promise; - isMarking: boolean; } export function useMessageList(): UseMessageListReturn { const { conversations, isLoading, refresh, loadMore, hasMore } = useConversations(); const { totalUnreadCount, systemUnreadCount } = useUnreadCount(); - const { markAllAsRead, isMarking } = useMarkAsRead(null); + const { markAllAsRead } = useMarkAsRead(null); return { conversations, @@ -715,6 +582,5 @@ export function useMessageList(): UseMessageListReturn { totalUnreadCount, systemUnreadCount, markAllAsRead, - isMarking, }; }