lan
7e0436a799
Refactor the message module to streamline state management by removing the `MessageStateManager` and `SubscriptionManager` layers, relying instead on Zustand's built-in selector mechanism for reactive updates. - Remove redundant `setConversations`, `addMessage`, and `updateMessage` actions from the Zustand store to favor more specialized update methods. - Clean up `MessageService` by removing deprecated conversation and message sending convenience methods. - Update service interfaces to support new synchronization capabilities including `syncBySeq` and `restartSources`. - Simplify documentation and comments across the message store and its associated services.
571 lines
20 KiB
TypeScript
571 lines
20 KiB
TypeScript
/**
|
||
* 消息同步服务
|
||
* 处理消息和会话的服务器同步逻辑
|
||
*/
|
||
|
||
import type { ConversationResponse, MessageResponse } from '../../../types/dto';
|
||
import { messageService } from '@/services/message';
|
||
import { messageRepository, conversationRepository } from '@/database';
|
||
import { MessageMapper } from '@/data/mappers';
|
||
import {
|
||
type IConversationListPagedSource,
|
||
SqliteConversationListPagedSource,
|
||
createRemoteConversationListSource,
|
||
CONVERSATION_LIST_PAGE_SIZE,
|
||
} from '../sources';
|
||
import type { IMessageSyncService, MessageManagerConversationListDeps, IUserCacheService } from '../types';
|
||
import { useMessageStore, 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;
|
||
/** fetchUnreadCount 去重：复用 in-flight promise */
|
||
private fetchUnreadCountPromise: Promise<void> | null = null;
|
||
|
||
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)) {
|
||
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.persistConversationListCache();
|
||
}
|
||
}
|
||
|
||
store.setLoading(true);
|
||
|
||
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);
|
||
});
|
||
|
||
// 合并更新：conversations + totalUnread 在同一次 set 中完成，消除中间状态窗口
|
||
const totalUnread = Array.from(newConversations.values()).reduce(
|
||
(sum, conv) => sum + (conv.unread_count || 0),
|
||
0,
|
||
);
|
||
store.setConversationsWithUnread(newConversations, totalUnread, store.getUnreadCount().system);
|
||
|
||
this.persistConversationListCache();
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] 获取会话列表失败:', error);
|
||
if (store.getConversations().length === 0) {
|
||
const recovered = await this.hydrateConversationsFromLocalSource();
|
||
if (recovered) {
|
||
this.persistConversationListCache();
|
||
}
|
||
}
|
||
} finally {
|
||
store.setLoading(false);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 会话列表加载下一页
|
||
*/
|
||
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.persistConversationListCache();
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] 加载更多会话失败:', error);
|
||
} 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 messageRepository.getByConversation(conversationId, 20);
|
||
|
||
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);
|
||
} else {
|
||
store.setMessages(conversationId, []);
|
||
}
|
||
} catch (error) {
|
||
console.warn('[MessageSyncService] 读取本地消息失败:', error);
|
||
}
|
||
}
|
||
|
||
// 更新 baseline：合并本地DB后重新计算
|
||
const currentMessages = store.getMessages(conversationId);
|
||
baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
|
||
|
||
// 策略：有本地数据时仅增量同步，无数据时拉快照
|
||
if (baselineMaxSeq > 0) {
|
||
// 增量同步：只拉 baselineMaxSeq 之后的消息
|
||
try {
|
||
const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
|
||
const newMessages = incrementalResp?.messages || [];
|
||
|
||
if (newMessages.length > 0) {
|
||
const existingMsgs = store.getMessages(conversationId);
|
||
const mergedMessages = mergeMessagesById(existingMsgs, newMessages);
|
||
store.setMessages(conversationId, mergedMessages);
|
||
|
||
messageRepository.saveMessagesBatch(
|
||
MessageMapper.toCachedMessages(newMessages, conversationId)
|
||
).catch(error => {
|
||
console.error('[MessageSyncService] 保存增量消息到本地失败:', error);
|
||
});
|
||
}
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] 增量同步失败:', error);
|
||
}
|
||
} else {
|
||
// 冷启动：本地无数据，拉快照
|
||
try {
|
||
const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
|
||
const snapshotMessages = snapshotResp?.messages || [];
|
||
|
||
if (snapshotMessages.length > 0) {
|
||
store.setMessages(conversationId, snapshotMessages);
|
||
|
||
messageRepository.saveMessagesBatch(
|
||
MessageMapper.toCachedMessages(snapshotMessages, conversationId)
|
||
).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);
|
||
|
||
messageRepository.saveMessagesBatch(
|
||
MessageMapper.toCachedMessages(newMessages, conversationId)
|
||
).catch(error => {
|
||
console.error('[MessageSyncService] 保存消息到本地失败:', error);
|
||
});
|
||
}
|
||
}
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] 获取消息失败:', error);
|
||
} finally {
|
||
// 异步填充用户信息（不阻塞消息显示）
|
||
const currentMessages = store.getMessages(conversationId);
|
||
if (currentMessages.length > 0) {
|
||
this.userCacheService.enrichMessagesWithSenderInfo(
|
||
conversationId,
|
||
currentMessages,
|
||
(convId, enrichedMessages) => {
|
||
store.setMessages(convId, enrichedMessages);
|
||
}
|
||
);
|
||
}
|
||
store.setLoadingMessages(conversationId, false);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 加载更多历史消息
|
||
*/
|
||
async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise<MessageResponse[]> {
|
||
const store = useMessageStore.getState();
|
||
|
||
try {
|
||
// 先从本地获取
|
||
const localMessages = await messageRepository.getBeforeSeq(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);
|
||
|
||
return formattedMessages;
|
||
}
|
||
|
||
// 本地数据不足，从服务端获取
|
||
const response = await messageService.getMessages(conversationId, undefined, beforeSeq, limit);
|
||
|
||
if (response?.messages && response.messages.length > 0) {
|
||
const serverMessages = response.messages;
|
||
|
||
await messageRepository.saveMessagesBatch(
|
||
MessageMapper.toCachedMessages(serverMessages, conversationId)
|
||
);
|
||
|
||
const existingMessages = store.getMessages(conversationId);
|
||
const mergedMessages = this.mergeOlderMessages(existingMessages, serverMessages);
|
||
store.setMessages(conversationId, mergedMessages);
|
||
|
||
return serverMessages;
|
||
}
|
||
|
||
return [];
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] 加载更多消息失败:', error);
|
||
return [];
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取未读数（去重：复用 in-flight promise）
|
||
*/
|
||
async fetchUnreadCount(): Promise<void> {
|
||
if (this.fetchUnreadCountPromise) {
|
||
return this.fetchUnreadCountPromise;
|
||
}
|
||
this.fetchUnreadCountPromise = this.doFetchUnreadCount();
|
||
try {
|
||
await this.fetchUnreadCountPromise;
|
||
} finally {
|
||
this.fetchUnreadCountPromise = null;
|
||
}
|
||
}
|
||
|
||
private async doFetchUnreadCount(): 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);
|
||
|
||
// 获取最后一条系统通知的时间
|
||
if (systemUnread > 0) {
|
||
try {
|
||
const latestSystemMsg = await messageService.getSystemMessagesCursor({ cursor: '', page_size: 1 });
|
||
if (latestSystemMsg?.list?.[0]?.created_at) {
|
||
store.setLastSystemMessageAt(latestSystemMsg.list[0].created_at);
|
||
}
|
||
} catch {
|
||
// 获取最后系统通知时间失败，不影响主流程
|
||
}
|
||
}
|
||
|
||
// 服务端汇总未读为 0：不主动清零本地未读，等 fetchConversations 提供权威数据
|
||
// 避免后端缓存竞态导致的未读数抖动
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] 获取未读数失败:', error);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 基于 seq 的轻量增量同步（用于重连场景，替代全量刷新）
|
||
* 返回 true 表示同步成功，false 表示需要退化为全量刷新
|
||
*/
|
||
async syncBySeq(): Promise<boolean> {
|
||
try {
|
||
const serverItems = await messageService.getSyncData();
|
||
if (!serverItems || serverItems.length === 0) return false;
|
||
|
||
const store = useMessageStore.getState();
|
||
const localConversations = store.conversations;
|
||
|
||
// 没有本地数据，无法增量同步
|
||
if (localConversations.size === 0) return false;
|
||
|
||
// 构建 serverMap
|
||
const serverMap = new Map<string, { max_seq: number; last_message_at: string }>();
|
||
for (const item of serverItems) {
|
||
serverMap.set(normalizeConversationId(item.id), item);
|
||
}
|
||
|
||
// 对比差异
|
||
const staleIds: string[] = [];
|
||
for (const [id, conv] of localConversations) {
|
||
const serverItem = serverMap.get(id);
|
||
if (!serverItem) continue;
|
||
const localMaxSeq = Number(conv.last_seq || 0);
|
||
const serverMaxSeq = Number(serverItem.max_seq || 0);
|
||
if (serverMaxSeq > localMaxSeq) {
|
||
staleIds.push(id);
|
||
}
|
||
}
|
||
|
||
// 差异过大（>50%），退化为全量刷新
|
||
if (staleIds.length > localConversations.size * 0.5) {
|
||
return false;
|
||
}
|
||
|
||
// 对有差异的会话更新数据
|
||
if (staleIds.length > 0) {
|
||
const detailPromises = staleIds.map(id =>
|
||
this.fetchConversationDetail(id).catch(() => {})
|
||
);
|
||
await Promise.all(detailPromises);
|
||
|
||
// 重新读取 store，避免 stale 引用
|
||
const activeId = useMessageStore.getState().getActiveConversation();
|
||
if (activeId && staleIds.includes(activeId)) {
|
||
const localMaxSeq = useMessageStore.getState().getMessages(activeId).reduce(
|
||
(max, m) => Math.max(max, m.seq || 0), 0
|
||
);
|
||
if (localMaxSeq > 0) {
|
||
await this.fetchMessages(activeId, localMaxSeq);
|
||
}
|
||
}
|
||
}
|
||
|
||
return true;
|
||
} catch (error) {
|
||
console.error('[MessageSyncService] syncBySeq 失败:', error);
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 检查是否可加载更多会话
|
||
*/
|
||
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 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));
|
||
|
||
conversationRepository.saveWithRelated(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);
|
||
});
|
||
|
||
// 合并更新：conversations + totalUnread 在同一次 set 中完成
|
||
const totalUnread = Array.from(newConversations.values()).reduce(
|
||
(sum, conv) => sum + (conv.unread_count || 0),
|
||
0,
|
||
);
|
||
store.setConversationsWithUnread(newConversations, 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);
|
||
}
|
||
} |