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
This commit is contained in:
lafay
681
src/stores/message/services/MessageSyncService.ts
Normal file
681
src/stores/message/services/MessageSyncService.ts
Normal file
@@ -0,0 +1,681 @@
|
||||
/**
|
||||
* 消息同步服务
|
||||
* 处理消息和会话的服务器同步逻辑
|
||||
*
|
||||
* 重构说明：直接使用 zustand store 替代 IMessageStateManager
|
||||
*/
|
||||
|
||||
import type { ConversationResponse, MessageResponse } from '../../../types/dto';
|
||||
import { messageService } from '../../../services/messageService';
|
||||
import {
|
||||
getMessagesByConversation,
|
||||
getMaxSeq,
|
||||
saveMessagesBatch,
|
||||
saveConversationsWithRelatedCache,
|
||||
getMessagesBeforeSeq,
|
||||
} from '../../../services/database';
|
||||
import {
|
||||
type IConversationListPagedSource,
|
||||
SqliteConversationListPagedSource,
|
||||
createRemoteConversationListSource,
|
||||
CONVERSATION_LIST_PAGE_SIZE,
|
||||
} 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 getCurrentUserId: () => string | null;
|
||||
private readReceiptManager: ReadReceiptManager;
|
||||
private userCacheService: IUserCacheService;
|
||||
|
||||
/** 远端会话列表（游标或页码） */
|
||||
private readonly remoteConversationListSource: IConversationListPagedSource;
|
||||
/** 本地会话列表缓存（SQLite） */
|
||||
private readonly localConversationListSource: IConversationListPagedSource;
|
||||
|
||||
/** 正在加载会话列表下一页 */
|
||||
private loadingMoreConversations = false;
|
||||
/** 聊天页活跃期间延迟的会话列表刷新 */
|
||||
private deferredConversationRefresh = false;
|
||||
|
||||
constructor(
|
||||
getCurrentUserId: () => string | null,
|
||||
readReceiptManager: ReadReceiptManager,
|
||||
userCacheService: IUserCacheService,
|
||||
deps?: MessageManagerConversationListDeps
|
||||
) {
|
||||
this.getCurrentUserId = getCurrentUserId;
|
||||
this.readReceiptManager = readReceiptManager;
|
||||
this.userCacheService = userCacheService;
|
||||
|
||||
const pageSize = deps?.remoteListPageSize ?? CONVERSATION_LIST_PAGE_SIZE;
|
||||
this.remoteConversationListSource =
|
||||
deps?.remoteConversationListSource ??
|
||||
createRemoteConversationListSource(
|
||||
deps?.remoteListKind === 'offset' ? 'offset' : 'cursor',
|
||||
pageSize
|
||||
);
|
||||
this.localConversationListSource =
|
||||
deps?.localConversationListSource ?? new SqliteConversationListPagedSource();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话列表
|
||||
*/
|
||||
async fetchConversations(forceRefresh = false, source: string = 'unknown'): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
if (store.isLoading() && !forceRefresh) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.shouldDeferConversationRefresh(source, forceRefresh)) {
|
||||
this.deferredConversationRefresh = true;
|
||||
if (__DEV__) {
|
||||
console.log('[MessageSyncService] defer fetchConversations', {
|
||||
source,
|
||||
activeConversationId: store.getActiveConversation(),
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (__DEV__) {
|
||||
console.log('[MessageSyncService] fetchConversations', {
|
||||
source,
|
||||
forceRefresh,
|
||||
activeConversationId: store.getActiveConversation(),
|
||||
});
|
||||
}
|
||||
|
||||
// 非强制刷新且内存为空：先用本地源暖机
|
||||
const currentState = store.getState();
|
||||
if (!forceRefresh && currentState.conversations.size === 0) {
|
||||
const warmed = await this.hydrateConversationsFromLocalSource();
|
||||
if (warmed) {
|
||||
this.emitConversationListAndUnreadUpdates();
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
const page = await this.remoteConversationListSource.loadNext();
|
||||
|
||||
// 使用 zustand set 函数正确更新状态
|
||||
const newConversations = new Map<string, ConversationResponse>();
|
||||
page.items.forEach(conv => {
|
||||
const normalizedConv = this.normalizeConversationFromFetch(conv);
|
||||
newConversations.set(normalizedConv.id, normalizedConv);
|
||||
});
|
||||
|
||||
// 使用 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 (store.getConversations().length === 0) {
|
||||
const recovered = await this.hydrateConversationsFromLocalSource();
|
||||
if (recovered) {
|
||||
this.emitConversationListAndUnreadUpdates();
|
||||
}
|
||||
}
|
||||
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(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 会话列表加载下一页
|
||||
*/
|
||||
async loadMoreConversations(): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
if (!this.remoteConversationListSource.hasMore) {
|
||||
return;
|
||||
}
|
||||
if (this.loadingMoreConversations || store.isLoading()) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.loadingMoreConversations = true;
|
||||
try {
|
||||
const page = await this.remoteConversationListSource.loadNext();
|
||||
|
||||
page.items.forEach(conv => {
|
||||
const normalizedConv = this.normalizeConversationFromFetch(conv);
|
||||
store.updateConversation(normalizedConv);
|
||||
});
|
||||
|
||||
this.recomputeConversationTotalUnread();
|
||||
this.emitConversationListAndUnreadUpdates();
|
||||
} catch (error) {
|
||||
console.error('[MessageSyncService] 加载更多会话失败:', error);
|
||||
subscriptionManager.notifySubscribers({
|
||||
type: 'error',
|
||||
payload: { error, context: 'loadMoreConversations' },
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
} finally {
|
||||
this.loadingMoreConversations = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取单个会话详情
|
||||
*/
|
||||
async fetchConversationDetail(conversationId: string): Promise<ConversationResponse | null> {
|
||||
const store = useMessageStore.getState();
|
||||
const normalizedId = normalizeConversationId(conversationId);
|
||||
|
||||
try {
|
||||
const detail = await messageService.getConversationById(normalizedId);
|
||||
if (detail) {
|
||||
const normalizedDetail = this.normalizeConversationFromFetch(detail as ConversationResponse);
|
||||
store.updateConversation(normalizedDetail);
|
||||
return normalizedDetail;
|
||||
}
|
||||
return null;
|
||||
} catch (error) {
|
||||
console.error('[MessageSyncService] 获取会话详情失败:', error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话消息（增量同步）
|
||||
*/
|
||||
async fetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
// 防止重复加载
|
||||
if (store.isLoadingMessages(conversationId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
store.setLoadingMessages(conversationId, true);
|
||||
|
||||
try {
|
||||
const existingMessagesAtStart = store.getMessages(conversationId);
|
||||
const hasInMemoryMessages = existingMessagesAtStart.length > 0;
|
||||
let baselineMaxSeq = hasInMemoryMessages
|
||||
? existingMessagesAtStart.reduce((max, m) => Math.max(max, m.seq || 0), 0)
|
||||
: 0;
|
||||
|
||||
if (!afterSeq) {
|
||||
// 先从本地数据库加载
|
||||
if (!hasInMemoryMessages) {
|
||||
try {
|
||||
const localMessages = await getMessagesByConversation(conversationId, 20);
|
||||
const localMaxSeq = await getMaxSeq(conversationId);
|
||||
baselineMaxSeq = localMaxSeq;
|
||||
|
||||
if (localMessages.length > 0) {
|
||||
const formattedMessages: MessageResponse[] = localMessages.map(m => ({
|
||||
id: m.id,
|
||||
conversation_id: m.conversationId,
|
||||
sender_id: m.senderId,
|
||||
seq: m.seq,
|
||||
segments: m.segments || [],
|
||||
status: m.status as any,
|
||||
created_at: m.createdAt,
|
||||
}));
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
// 服务端快照 + 增量同步
|
||||
try {
|
||||
const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
|
||||
const snapshotMessages = snapshotResp?.messages || [];
|
||||
|
||||
if (snapshotMessages.length > 0) {
|
||||
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) => ({
|
||||
id: m.id,
|
||||
conversationId: m.conversation_id || conversationId,
|
||||
senderId: m.sender_id,
|
||||
content: m.content,
|
||||
type: m.type || 'text',
|
||||
isRead: m.is_read || false,
|
||||
createdAt: m.created_at,
|
||||
seq: m.seq,
|
||||
status: m.status || 'normal',
|
||||
segments: m.segments,
|
||||
}))).catch(error => {
|
||||
console.error('[MessageSyncService] 保存快照消息到本地失败:', error);
|
||||
});
|
||||
}
|
||||
|
||||
// 增量补齐
|
||||
const snapshotMaxSeq = snapshotMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
|
||||
if (snapshotMaxSeq > baselineMaxSeq) {
|
||||
const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
|
||||
const newMessages = incrementalResp?.messages || [];
|
||||
|
||||
if (newMessages.length > 0) {
|
||||
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,
|
||||
conversationId: m.conversation_id || conversationId,
|
||||
senderId: m.sender_id,
|
||||
content: m.content,
|
||||
type: m.type || 'text',
|
||||
isRead: m.is_read || false,
|
||||
createdAt: m.created_at,
|
||||
seq: m.seq,
|
||||
status: m.status || 'normal',
|
||||
segments: m.segments,
|
||||
}))).catch(error => {
|
||||
console.error('[MessageSyncService] 保存增量消息到本地失败:', error);
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[MessageSyncService] 快照/增量同步失败:', error);
|
||||
}
|
||||
} else {
|
||||
// 指定了 afterSeq
|
||||
const response = await messageService.getMessages(conversationId, afterSeq);
|
||||
|
||||
if (response?.messages && response.messages.length > 0) {
|
||||
const newMessages = response.messages;
|
||||
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',
|
||||
},
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
|
||||
saveMessagesBatch(newMessages.map((m: any) => ({
|
||||
id: m.id,
|
||||
conversationId: m.conversation_id || conversationId,
|
||||
senderId: m.sender_id,
|
||||
content: m.content,
|
||||
type: m.type || 'text',
|
||||
isRead: m.is_read || false,
|
||||
createdAt: m.created_at,
|
||||
seq: m.seq,
|
||||
status: m.status || 'normal',
|
||||
segments: m.segments,
|
||||
}))).catch(error => {
|
||||
console.error('[MessageSyncService] 保存消息到本地失败:', error);
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[MessageSyncService] 获取消息失败:', error);
|
||||
subscriptionManager.notifySubscribers({
|
||||
type: 'error',
|
||||
payload: { error, context: 'fetchMessages', conversationId },
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
} finally {
|
||||
// 异步填充用户信息（不阻塞消息显示）
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 加载更多历史消息
|
||||
*/
|
||||
async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise<MessageResponse[]> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
try {
|
||||
// 先从本地获取
|
||||
const localMessages = await getMessagesBeforeSeq(conversationId, beforeSeq, limit);
|
||||
|
||||
if (localMessages.length >= limit) {
|
||||
const formattedMessages: MessageResponse[] = [...localMessages].reverse().map(m => ({
|
||||
id: m.id,
|
||||
conversation_id: m.conversationId,
|
||||
sender_id: m.senderId,
|
||||
seq: m.seq,
|
||||
segments: m.segments || [],
|
||||
status: m.status as any,
|
||||
created_at: m.createdAt,
|
||||
}));
|
||||
|
||||
const existingMessages = store.getMessages(conversationId);
|
||||
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;
|
||||
}
|
||||
|
||||
// 本地数据不足，从服务端获取
|
||||
const response = await messageService.getMessages(conversationId, undefined, beforeSeq, limit);
|
||||
|
||||
if (response?.messages && response.messages.length > 0) {
|
||||
const serverMessages = response.messages;
|
||||
|
||||
await saveMessagesBatch(serverMessages.map((m: any) => ({
|
||||
id: m.id,
|
||||
conversationId: m.conversation_id || conversationId,
|
||||
senderId: m.sender_id,
|
||||
content: m.content,
|
||||
type: m.type || 'text',
|
||||
isRead: m.is_read || false,
|
||||
createdAt: m.created_at,
|
||||
seq: m.seq,
|
||||
status: m.status || 'normal',
|
||||
segments: m.segments,
|
||||
})));
|
||||
|
||||
const existingMessages = store.getMessages(conversationId);
|
||||
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;
|
||||
}
|
||||
|
||||
return [];
|
||||
} catch (error) {
|
||||
console.error('[MessageSyncService] 加载更多消息失败:', error);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取未读数
|
||||
*/
|
||||
async fetchUnreadCount(): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
try {
|
||||
const [unreadData, systemUnreadData] = await Promise.all([
|
||||
messageService.getUnreadCount(),
|
||||
messageService.getSystemUnreadCount(),
|
||||
]);
|
||||
|
||||
const totalUnread = unreadData?.total_unread_count ?? 0;
|
||||
const systemUnread = systemUnreadData?.unread_count ?? 0;
|
||||
|
||||
store.setUnreadCount(totalUnread, systemUnread);
|
||||
|
||||
// 服务端汇总未读为 0 时，清掉内存中残留的红点
|
||||
if (totalUnread === 0) {
|
||||
const currentConversations = store.getState().conversations;
|
||||
let anyCleared = false;
|
||||
const newConversations = new Map(currentConversations);
|
||||
for (const [cid, conv] of newConversations) {
|
||||
if ((conv.unread_count || 0) > 0) {
|
||||
newConversations.set(cid, { ...conv, unread_count: 0 });
|
||||
anyCleared = true;
|
||||
}
|
||||
}
|
||||
if (anyCleared) {
|
||||
// 使用 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);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查是否可加载更多会话
|
||||
*/
|
||||
canLoadMoreConversations(): boolean {
|
||||
return this.remoteConversationListSource.hasMore;
|
||||
}
|
||||
|
||||
/**
|
||||
* 重启数据源
|
||||
*/
|
||||
restartSources(): void {
|
||||
this.remoteConversationListSource.restart();
|
||||
this.localConversationListSource.restart();
|
||||
this.loadingMoreConversations = false;
|
||||
}
|
||||
|
||||
// ==================== 私有工具方法 ====================
|
||||
|
||||
private shouldDeferConversationRefresh(source: string, forceRefresh: boolean): boolean {
|
||||
const store = useMessageStore.getState();
|
||||
if (!store.getActiveConversation()) return false;
|
||||
if (!forceRefresh) return false;
|
||||
// hooks-initial-refresh 是冷启动兜底刷新，不应该被延迟
|
||||
return source === 'sse-reconnect' || source === 'prefetch';
|
||||
}
|
||||
|
||||
private normalizeConversationFromFetch(conv: ConversationResponse): ConversationResponse {
|
||||
const id = normalizeConversationId(conv.id);
|
||||
// 应用已读保护逻辑
|
||||
return this.readReceiptManager.applyConversationFromFetch({
|
||||
...conv,
|
||||
id,
|
||||
});
|
||||
}
|
||||
|
||||
private recomputeConversationTotalUnread(): void {
|
||||
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();
|
||||
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();
|
||||
const list = Array.from(currentState.conversations.values());
|
||||
const users = list.flatMap(conv => [
|
||||
...(conv.participants || []),
|
||||
...(conv.last_message?.sender ? [conv.last_message.sender] : []),
|
||||
]);
|
||||
const groups = list
|
||||
.map(conv => conv.group)
|
||||
.filter((group): group is NonNullable<ConversationResponse['group']> => Boolean(group));
|
||||
|
||||
saveConversationsWithRelatedCache(list, users, groups).catch(error => {
|
||||
console.error('[MessageSyncService] 持久化会话列表失败:', error);
|
||||
});
|
||||
}
|
||||
|
||||
private async hydrateConversationsFromLocalSource(): Promise<boolean> {
|
||||
const store = useMessageStore.getState();
|
||||
this.localConversationListSource.restart();
|
||||
const page = await this.localConversationListSource.loadNext();
|
||||
if (!page.items.length) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 使用 zustand set 函数正确更新状态
|
||||
const newConversations = new Map<string, ConversationResponse>();
|
||||
page.items.forEach(conv => {
|
||||
const normalizedConv = this.normalizeConversationFromFetch(conv);
|
||||
newConversations.set(normalizedConv.id, normalizedConv);
|
||||
});
|
||||
|
||||
// 使用 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 mergeOlderMessages(existing: MessageResponse[], incoming: MessageResponse[]): MessageResponse[] {
|
||||
if (incoming.length === 0) return existing;
|
||||
|
||||
const incomingAsc = [...incoming].sort((a, b) => a.seq - b.seq);
|
||||
if (existing.length === 0) return incomingAsc;
|
||||
|
||||
const existingMinSeq = existing[0]?.seq ?? Number.MAX_SAFE_INTEGER;
|
||||
const incomingMaxSeq = incomingAsc[incomingAsc.length - 1]?.seq ?? Number.MIN_SAFE_INTEGER;
|
||||
|
||||
if (incomingMaxSeq < existingMinSeq) {
|
||||
return [...incomingAsc, ...existing];
|
||||
}
|
||||
|
||||
return mergeMessagesById(existing, incomingAsc);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user