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
This commit is contained in:
lafay
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user