refactor(message): improve message synchronization and hook reliability
Refactor the message management system to address synchronization issues and improve performance through request deduplication and enhanced lifecycle management. - Implement in-flight promise tracking in `MessageSyncService` to prevent redundant network requests for conversations and messages. - Enhance `useMessages` hook with `useFocusEffect` to trigger automatic synchronization when a chat screen regains focus. - Add `forceSync` capability to `MessageManager` and `MessageSyncService` to allow manual overrides of loading states during re-entry. - Consolidate WebSocket synchronization logic in `WSMessageHandler` using a throttled and deduplicated trigger mechanism. - Clean up unused styles and deprecated hooks (`baseHooks.ts`, `bubbleStyles.ts`, `inputStyles.ts`). - Update `metro.config.js` and `package.json` to include `@expo/ui` and optimize resolver settings.
This commit is contained in:
@@ -200,8 +200,8 @@ class MessageManager {
|
||||
return this.syncService.fetchConversationDetail(conversationId);
|
||||
}
|
||||
|
||||
async fetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
|
||||
return this.syncService.fetchMessages(conversationId, afterSeq);
|
||||
async fetchMessages(conversationId: string, afterSeq?: number, force?: boolean): Promise<void> {
|
||||
return this.syncService.fetchMessages(conversationId, afterSeq, force);
|
||||
}
|
||||
|
||||
async loadMoreMessages(conversationId: string, beforeSeq: number, limit = 20): Promise<MessageResponse[]> {
|
||||
@@ -343,6 +343,11 @@ class MessageManager {
|
||||
return;
|
||||
}
|
||||
|
||||
// forceSync 时清除可能残留的 loading 锁，防止 fetchMessages 被阻断
|
||||
if (options?.forceSync) {
|
||||
useMessageStore.getState().setLoadingMessages(normalizedId, false);
|
||||
}
|
||||
|
||||
const task = (async () => {
|
||||
// 确保会话数据在 store 中（JPUSH 通知进入时 store 可能还没有此会话）
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
@@ -1,477 +0,0 @@
|
||||
/**
|
||||
* 消息模块 React Hooks
|
||||
*
|
||||
* 提供React组件与消息状态管理的集成
|
||||
* 所有hooks都基于zustand store的selector机制
|
||||
* 确保组件能实时获取状态更新
|
||||
*/
|
||||
|
||||
import { useState, useEffect, useCallback, useRef } from 'react';
|
||||
import { useShallow } from 'zustand/react/shallow';
|
||||
import { ConversationResponse, MessageResponse, MessageSegment } from '../../types/dto';
|
||||
import { useMessageStore, normalizeConversationId } from './store';
|
||||
|
||||
// ==================== useConversations - 获取会话列表 ====================
|
||||
|
||||
interface UseConversationsReturn {
|
||||
conversations: ConversationResponse[];
|
||||
isLoading: boolean;
|
||||
refresh: () => Promise<void>;
|
||||
loadMore: () => Promise<void>;
|
||||
hasMore: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话列表
|
||||
* 用于 MessageListScreen 等需要显示会话列表的组件
|
||||
* 使用 Zustand selector 自动订阅状态变化
|
||||
*/
|
||||
export function useConversations(
|
||||
fetchConversations: (forceRefresh: boolean, source: string) => Promise<void>,
|
||||
loadMoreConversations: () => Promise<void>,
|
||||
canLoadMore: () => boolean,
|
||||
initialize: () => Promise<void>
|
||||
): UseConversationsReturn {
|
||||
// 使用 Zustand selector 直接订阅状态
|
||||
const conversations = useMessageStore(state => state.conversationList);
|
||||
const isLoading = useMessageStore(state => state.isLoadingConversations);
|
||||
const [hasMore, setHasMore] = useState(() => canLoadMore());
|
||||
|
||||
useEffect(() => {
|
||||
// 初始化
|
||||
initialize().catch(error => {
|
||||
console.error('[useConversations] 初始化失败:', error);
|
||||
});
|
||||
|
||||
// 冷启动兜底：延迟做一次强制刷新
|
||||
const coldStartSyncTimer = setTimeout(() => {
|
||||
fetchConversations(true, 'hooks-initial-refresh').catch(error => {
|
||||
console.error('[useConversations] 冷启动强制刷新失败:', error);
|
||||
});
|
||||
}, 1200);
|
||||
|
||||
return () => {
|
||||
clearTimeout(coldStartSyncTimer);
|
||||
};
|
||||
}, []);
|
||||
|
||||
// 监听 hasMore 变化
|
||||
useEffect(() => {
|
||||
setHasMore(canLoadMore());
|
||||
}, [conversations, canLoadMore]);
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
await fetchConversations(true, 'hooks-manual-refresh');
|
||||
}, [fetchConversations]);
|
||||
|
||||
const loadMore = useCallback(async () => {
|
||||
await loadMoreConversations();
|
||||
}, [loadMoreConversations]);
|
||||
|
||||
return {
|
||||
conversations,
|
||||
isLoading,
|
||||
refresh,
|
||||
loadMore,
|
||||
hasMore,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useMessages - 获取指定会话的消息 ====================
|
||||
|
||||
interface UseMessagesReturn {
|
||||
messages: MessageResponse[];
|
||||
isLoading: boolean;
|
||||
hasMore: boolean;
|
||||
loadMore: () => Promise<void>;
|
||||
refresh: () => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取指定会话的消息
|
||||
* 使用 Zustand selector 自动订阅消息变化
|
||||
*/
|
||||
export function useMessages(
|
||||
conversationId: string,
|
||||
fetchMessages: (conversationId: string) => Promise<void>,
|
||||
loadMoreMessages: (conversationId: string, beforeSeq: number, limit?: number) => Promise<MessageResponse[]>
|
||||
): UseMessagesReturn {
|
||||
const normalizedConversationId = normalizeConversationId(conversationId);
|
||||
|
||||
// 使用 useShallow 避免每次返回新数组引用导致的无限循环
|
||||
const messages = useMessageStore(
|
||||
useShallow(state => state.messagesMap.get(normalizedConversationId) || [])
|
||||
);
|
||||
const isLoadingMessages = useMessageStore(state => state.loadingMessagesSet.has(normalizedConversationId));
|
||||
const getActiveConversation = useMessageStore.getState().getActiveConversation;
|
||||
const setCurrentConversation = useMessageStore.getState().setCurrentConversation;
|
||||
const [hasMore, setHasMore] = useState(true);
|
||||
|
||||
useEffect(() => {
|
||||
// 激活会话
|
||||
fetchMessages(normalizedConversationId).catch(error => {
|
||||
console.error('[useMessages] 激活会话失败:', error);
|
||||
});
|
||||
|
||||
return () => {
|
||||
// 清理活动会话
|
||||
if (getActiveConversation() === normalizedConversationId) {
|
||||
setCurrentConversation(null);
|
||||
}
|
||||
};
|
||||
}, [normalizedConversationId]);
|
||||
|
||||
const loadMore = useCallback(async () => {
|
||||
const currentMessages = messages;
|
||||
if (currentMessages.length === 0) return;
|
||||
|
||||
// 消息数组在 store 内保持 seq 升序，首项即最早消息
|
||||
const minSeq = currentMessages[0]?.seq;
|
||||
if (minSeq === undefined) return;
|
||||
|
||||
const loadedMessages = await loadMoreMessages(conversationId, minSeq, 20);
|
||||
setHasMore(loadedMessages.length > 0);
|
||||
}, [conversationId, messages, loadMoreMessages]);
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
await fetchMessages(conversationId);
|
||||
}, [conversationId, fetchMessages]);
|
||||
|
||||
return {
|
||||
messages,
|
||||
isLoading: isLoadingMessages,
|
||||
hasMore,
|
||||
loadMore,
|
||||
refresh,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useSendMessage - 发送消息 ====================
|
||||
|
||||
interface UseSendMessageReturn {
|
||||
sendMessage: (segments: MessageSegment[], options?: { replyToId?: string }) => Promise<MessageResponse | null>;
|
||||
isSending: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 发送消息
|
||||
*/
|
||||
export function useSendMessage(
|
||||
conversationId: string,
|
||||
sendMessageService: (conversationId: string, segments: MessageSegment[], options?: { replyToId?: string }) => Promise<MessageResponse | null>
|
||||
): UseSendMessageReturn {
|
||||
const [isSending, setIsSending] = useState(false);
|
||||
|
||||
const sendMessage = useCallback(async (segments: MessageSegment[], options?: { replyToId?: string }) => {
|
||||
setIsSending(true);
|
||||
try {
|
||||
const message = await sendMessageService(conversationId, segments, options);
|
||||
return message;
|
||||
} catch (error) {
|
||||
console.error('[useSendMessage] 发送消息失败:', error);
|
||||
return null;
|
||||
} finally {
|
||||
setIsSending(false);
|
||||
}
|
||||
}, [conversationId, sendMessageService]);
|
||||
|
||||
return {
|
||||
sendMessage,
|
||||
isSending,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useMarkAsRead - 标记已读 ====================
|
||||
|
||||
interface UseMarkAsReadReturn {
|
||||
markAsRead: (seq: number) => Promise<void>;
|
||||
markAllAsRead: () => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记已读
|
||||
*/
|
||||
export function useMarkAsRead(
|
||||
conversationId: string,
|
||||
markAsReadService: (conversationId: string, seq: number) => Promise<void>,
|
||||
markAllAsReadService: () => Promise<void>
|
||||
): UseMarkAsReadReturn {
|
||||
const markAsRead = useCallback(async (seq: number) => {
|
||||
await markAsReadService(conversationId, seq);
|
||||
}, [conversationId, markAsReadService]);
|
||||
|
||||
const markAllAsRead = useCallback(async () => {
|
||||
await markAllAsReadService();
|
||||
}, [markAllAsReadService]);
|
||||
|
||||
return {
|
||||
markAsRead,
|
||||
markAllAsRead,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useUnreadCount - 获取未读数 ====================
|
||||
|
||||
interface UseUnreadCountReturn {
|
||||
totalUnreadCount: number;
|
||||
systemUnreadCount: number;
|
||||
refresh: () => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取未读数
|
||||
* 使用 Zustand selector 直接订阅未读数变化
|
||||
*/
|
||||
export function useUnreadCount(fetchUnreadCount: () => Promise<void>): UseUnreadCountReturn {
|
||||
// 使用 Zustand selector 直接订阅未读数
|
||||
const totalUnreadCount = useMessageStore(state => state.totalUnreadCount);
|
||||
const systemUnreadCount = useMessageStore(state => state.systemUnreadCount);
|
||||
|
||||
useEffect(() => {
|
||||
// 初始获取
|
||||
fetchUnreadCount();
|
||||
}, []);
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
await fetchUnreadCount();
|
||||
}, [fetchUnreadCount]);
|
||||
|
||||
return {
|
||||
totalUnreadCount,
|
||||
systemUnreadCount,
|
||||
refresh,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useSystemUnreadCount - 获取系统消息未读数 ====================
|
||||
|
||||
interface UseSystemUnreadCountReturn {
|
||||
systemUnreadCount: number;
|
||||
setSystemUnreadCount: (count: number) => void;
|
||||
incrementSystemUnreadCount: () => void;
|
||||
decrementSystemUnreadCount: (count?: number) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取系统消息未读数
|
||||
* 使用 Zustand selector 直接订阅
|
||||
*/
|
||||
export function useSystemUnreadCount(
|
||||
setSystemUnreadCountService: (count: number) => void,
|
||||
incrementSystemUnreadCountService: () => void,
|
||||
decrementSystemUnreadCountService: (count?: number) => void
|
||||
): UseSystemUnreadCountReturn {
|
||||
// 使用 Zustand selector 直接订阅
|
||||
const systemUnreadCount = useMessageStore(state => state.systemUnreadCount);
|
||||
|
||||
const setSystemUnreadCount = useCallback((count: number) => {
|
||||
setSystemUnreadCountService(count);
|
||||
}, [setSystemUnreadCountService]);
|
||||
|
||||
const incrementSystemUnreadCount = useCallback(() => {
|
||||
incrementSystemUnreadCountService();
|
||||
}, [incrementSystemUnreadCountService]);
|
||||
|
||||
const decrementSystemUnreadCount = useCallback((count = 1) => {
|
||||
decrementSystemUnreadCountService(count);
|
||||
}, [decrementSystemUnreadCountService]);
|
||||
|
||||
return {
|
||||
systemUnreadCount,
|
||||
setSystemUnreadCount,
|
||||
incrementSystemUnreadCount,
|
||||
decrementSystemUnreadCount,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useConversation - 获取单个会话 ====================
|
||||
|
||||
interface UseConversationReturn {
|
||||
conversation: ConversationResponse | null;
|
||||
refresh: () => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取单个会话
|
||||
* 使用 Zustand selector 直接订阅会话变化
|
||||
*/
|
||||
export function useConversation(
|
||||
conversationId: string,
|
||||
fetchConversationDetail: (conversationId: string) => Promise<ConversationResponse | null>
|
||||
): UseConversationReturn {
|
||||
// 使用 useShallow 避免每次返回新对象引用导致的无限循环
|
||||
const conversation = useMessageStore(
|
||||
useShallow(state => state.conversations.get(conversationId) || null)
|
||||
);
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
await fetchConversationDetail(conversationId);
|
||||
}, [conversationId, fetchConversationDetail]);
|
||||
|
||||
return {
|
||||
conversation,
|
||||
refresh,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useMessageManager - 通用状态 ====================
|
||||
|
||||
interface UseMessageManagerReturn {
|
||||
isConnected: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 通用消息管理状态
|
||||
* 使用 Zustand selector 直接订阅连接状态
|
||||
*/
|
||||
export function useMessageManager(initialize: () => Promise<void>): UseMessageManagerReturn {
|
||||
// 使用 Zustand selector 直接订阅连接状态
|
||||
const isConnected = useMessageStore(state => state.isWSConnected);
|
||||
|
||||
useEffect(() => {
|
||||
initialize().catch(error => {
|
||||
console.error('[useMessageManager] 初始化失败:', error);
|
||||
});
|
||||
}, []);
|
||||
|
||||
return {
|
||||
isConnected,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useCreateConversation - 创建会话 ====================
|
||||
|
||||
interface UseCreateConversationReturn {
|
||||
createConversation: (userId: string) => Promise<ConversationResponse | null>;
|
||||
isCreating: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建会话
|
||||
*/
|
||||
export function useCreateConversation(
|
||||
createConversationService: (userId: string) => Promise<ConversationResponse | null>
|
||||
): UseCreateConversationReturn {
|
||||
const [isCreating, setIsCreating] = useState(false);
|
||||
|
||||
const createConversation = useCallback(async (userId: string) => {
|
||||
setIsCreating(true);
|
||||
try {
|
||||
const conversation = await createConversationService(userId);
|
||||
return conversation;
|
||||
} catch (error) {
|
||||
console.error('[useCreateConversation] 创建会话失败:', error);
|
||||
return null;
|
||||
} finally {
|
||||
setIsCreating(false);
|
||||
}
|
||||
}, [createConversationService]);
|
||||
|
||||
return {
|
||||
createConversation,
|
||||
isCreating,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useUpdateConversation - 更新会话 ====================
|
||||
|
||||
interface UseUpdateConversationReturn {
|
||||
updateConversation: (conversationId: string, updates: Partial<ConversationResponse>) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新会话
|
||||
*/
|
||||
export function useUpdateConversation(
|
||||
updateConversationService: (conversationId: string, updates: Partial<ConversationResponse>) => void
|
||||
): UseUpdateConversationReturn {
|
||||
const updateConversation = useCallback((conversationId: string, updates: Partial<ConversationResponse>) => {
|
||||
updateConversationService(conversationId, updates);
|
||||
}, [updateConversationService]);
|
||||
|
||||
return {
|
||||
updateConversation,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useGroupTyping - 群聊输入状态 ====================
|
||||
|
||||
interface UseGroupTypingReturn {
|
||||
typingUsers: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取群聊输入状态
|
||||
* 使用 Zustand selector 直接订阅
|
||||
*/
|
||||
export function useGroupTyping(groupId: string): UseGroupTypingReturn {
|
||||
// 使用 useShallow 避免每次返回新数组引用导致的无限循环
|
||||
const typingUsers = useMessageStore(
|
||||
useShallow(state => state.typingUsersMap.get(groupId) || [])
|
||||
);
|
||||
|
||||
return {
|
||||
typingUsers,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useGroupMuted - 群聊禁言状态 ====================
|
||||
|
||||
interface UseGroupMutedReturn {
|
||||
isMuted: boolean;
|
||||
setMutedStatus: (muted: boolean) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取群聊禁言状态
|
||||
* 使用 Zustand selector 直接订阅
|
||||
*/
|
||||
export function useGroupMuted(groupId: string): UseGroupMutedReturn {
|
||||
// 使用 Zustand selector 直接订阅禁言状态
|
||||
const isMuted = useMessageStore(state => state.mutedStatusMap.get(groupId) || false);
|
||||
|
||||
const setMutedStatus = useCallback((muted: boolean) => {
|
||||
useMessageStore.getState().setMutedStatus(groupId, muted);
|
||||
}, [groupId]);
|
||||
|
||||
return {
|
||||
isMuted,
|
||||
setMutedStatus,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useConnectionStatus - 连接状态 ====================
|
||||
|
||||
interface UseConnectionStatusReturn {
|
||||
isConnected: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取连接状态
|
||||
* 使用 Zustand selector 直接订阅
|
||||
*/
|
||||
export function useConnectionStatus(): UseConnectionStatusReturn {
|
||||
const isConnected = useMessageStore(state => state.isWSConnected);
|
||||
|
||||
return {
|
||||
isConnected,
|
||||
};
|
||||
}
|
||||
|
||||
// ==================== useIsInitialized - 初始化状态 ====================
|
||||
|
||||
interface UseIsInitializedReturn {
|
||||
isInitialized: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取初始化状态
|
||||
* 使用 Zustand selector 直接订阅
|
||||
*/
|
||||
export function useIsInitialized(): UseIsInitializedReturn {
|
||||
const isInitialized = useMessageStore(state => state.isInitialized);
|
||||
|
||||
return {
|
||||
isInitialized,
|
||||
};
|
||||
}
|
||||
@@ -12,6 +12,7 @@
|
||||
*/
|
||||
|
||||
import { useState, useEffect, useCallback, useRef } from 'react';
|
||||
import { useFocusEffect } from 'expo-router';
|
||||
import { useShallow } from 'zustand/react/shallow';
|
||||
import { ConversationResponse, MessageResponse, MessageSegment } from '../../types/dto';
|
||||
import { messageManager } from './MessageManager';
|
||||
@@ -96,10 +97,15 @@ interface UseMessagesReturn {
|
||||
/**
|
||||
* 获取指定会话的消息
|
||||
* 使用 Zustand selector 自动订阅消息变化
|
||||
*
|
||||
* 核心修复：
|
||||
* 1. useFocusEffect：屏幕重新获得焦点时强制同步消息，解决"需要退出重进才能看到新消息"的问题
|
||||
* 2. 清理 isLoadingMessages：组件卸载时清除残留的 loading 锁，防止下次进入被阻断
|
||||
* 3. 重入时 forceSync：相同 conversationId 重复进入时强制拉取最新消息
|
||||
*/
|
||||
export function useMessages(conversationId: string | null): UseMessagesReturn {
|
||||
const normalizedConversationId = conversationId ? String(conversationId) : null;
|
||||
|
||||
|
||||
// 使用 useShallow 避免每次返回新数组引用导致的无限循环
|
||||
const messages = useMessageStore(
|
||||
useShallow(state => {
|
||||
@@ -107,30 +113,68 @@ export function useMessages(conversationId: string | null): UseMessagesReturn {
|
||||
return state.messagesMap.get(normalizedConversationId) ?? EMPTY_MESSAGES;
|
||||
})
|
||||
);
|
||||
const isLoadingMessages = useMessageStore(state =>
|
||||
const isLoadingMessages = useMessageStore(state =>
|
||||
normalizedConversationId ? state.loadingMessagesSet.has(normalizedConversationId) : false
|
||||
);
|
||||
const [hasMore, setHasMore] = useState(true);
|
||||
const loadMoreInFlightRef = useRef(false);
|
||||
// 追踪上次激活的 conversationId，避免同一会话重复激活时不必要的 forceSync
|
||||
const lastActivatedIdRef = useRef<string | null>(null);
|
||||
// 追踪 useEffect 激活时间，与 useFocusEffect 去重
|
||||
const lastActivatedAtRef = useRef<number>(0);
|
||||
|
||||
useEffect(() => {
|
||||
if (!normalizedConversationId) {
|
||||
return;
|
||||
}
|
||||
|
||||
lastActivatedAtRef.current = Date.now();
|
||||
|
||||
const isReentry = lastActivatedIdRef.current === normalizedConversationId;
|
||||
|
||||
// 架构入口：激活会话并完成初始化/同步
|
||||
messageManager.activateConversation(normalizedConversationId).catch(error => {
|
||||
// 重入同一会话时 forceSync，确保拿到最新消息
|
||||
messageManager.activateConversation(normalizedConversationId, { forceSync: isReentry }).catch(error => {
|
||||
console.error('[useMessages] 激活会话失败:', error);
|
||||
});
|
||||
|
||||
lastActivatedIdRef.current = normalizedConversationId;
|
||||
|
||||
return () => {
|
||||
// 清理活动会话
|
||||
// 清理活动会话 + 清除残留的 loading 锁
|
||||
if (messageManager.getActiveConversation() === normalizedConversationId) {
|
||||
messageManager.setActiveConversation(null);
|
||||
}
|
||||
useMessageStore.getState().setLoadingMessages(normalizedConversationId, false);
|
||||
};
|
||||
}, [normalizedConversationId]);
|
||||
|
||||
// 屏幕重新获得焦点时，同步活动会话的最新消息
|
||||
useFocusEffect(
|
||||
useCallback(() => {
|
||||
if (!normalizedConversationId) return;
|
||||
|
||||
// 如果刚被 useEffect 激活（500ms 内），跳过焦点同步，避免重复
|
||||
if (Date.now() - lastActivatedAtRef.current < 500) return;
|
||||
|
||||
// 先清除可能残留的 loading 锁，防止上次退出时未正常清理
|
||||
useMessageStore.getState().setLoadingMessages(normalizedConversationId, false);
|
||||
|
||||
// 增量同步最新消息（从当前内存最高 seq 开始拉取）
|
||||
const currentMessages = useMessageStore.getState().getMessages(normalizedConversationId);
|
||||
const maxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
|
||||
if (maxSeq > 0) {
|
||||
messageManager.fetchMessages(normalizedConversationId, maxSeq).catch(error => {
|
||||
console.error('[useMessages] 焦点同步消息失败:', error);
|
||||
});
|
||||
} else {
|
||||
messageManager.fetchMessages(normalizedConversationId).catch(error => {
|
||||
console.error('[useMessages] 焦点同步消息（冷启动）失败:', error);
|
||||
});
|
||||
}
|
||||
}, [normalizedConversationId])
|
||||
);
|
||||
|
||||
const loadMore = useCallback(async () => {
|
||||
if (!conversationId || !hasMore || loadMoreInFlightRef.current) return;
|
||||
|
||||
|
||||
@@ -29,9 +29,15 @@ export class MessageSyncService implements IMessageSyncService {
|
||||
|
||||
/** 正在加载会话列表下一页 */
|
||||
private loadingMoreConversations = false;
|
||||
/** fetchUnreadCount 去重：复用 in-flight promise */
|
||||
/** fetchUnreadCount 去重：复用 in-flight promise */
|
||||
private fetchUnreadCountPromise: Promise<void> | null = null;
|
||||
|
||||
// ---- 请求去重：in-flight promise maps ----
|
||||
private inflightFetches: Map<string, Promise<void>> = new Map();
|
||||
private inflightFetchConversations: Promise<void> | null = null;
|
||||
private inflightSyncBySeq: Promise<boolean> | null = null;
|
||||
private inflightSyncByVersion: Promise<boolean> | null = null;
|
||||
|
||||
constructor(
|
||||
getCurrentUserId: () => string | null,
|
||||
readReceiptManager: ReadReceiptManager,
|
||||
@@ -54,9 +60,22 @@ export class MessageSyncService implements IMessageSyncService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话列表
|
||||
* 获取会话列表 — 去重入口
|
||||
*/
|
||||
async fetchConversations(forceRefresh = false, source: string = 'unknown'): Promise<void> {
|
||||
if (this.inflightFetchConversations) return this.inflightFetchConversations;
|
||||
this.inflightFetchConversations = this._doFetchConversations(forceRefresh, source);
|
||||
try {
|
||||
await this.inflightFetchConversations;
|
||||
} finally {
|
||||
this.inflightFetchConversations = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话列表 — 核心实现
|
||||
*/
|
||||
private async _doFetchConversations(forceRefresh: boolean, source: string): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
if (store.isLoading() && !forceRefresh) {
|
||||
@@ -177,15 +196,29 @@ export class MessageSyncService implements IMessageSyncService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话消息（增量同步）
|
||||
* 获取会话消息（增量同步）— 去重入口
|
||||
*/
|
||||
async fetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
async fetchMessages(conversationId: string, afterSeq?: number, force?: boolean): Promise<void> {
|
||||
const key = `${conversationId}:${afterSeq ?? 'all'}`;
|
||||
const existing = this.inflightFetches.get(key);
|
||||
if (existing) return existing;
|
||||
|
||||
// 防止重复加载
|
||||
if (store.isLoadingMessages(conversationId)) {
|
||||
return;
|
||||
if (!force && useMessageStore.getState().isLoadingMessages(conversationId)) return;
|
||||
|
||||
const promise = this._doFetchMessages(conversationId, afterSeq);
|
||||
this.inflightFetches.set(key, promise);
|
||||
try {
|
||||
await promise;
|
||||
} finally {
|
||||
this.inflightFetches.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取会话消息（增量同步）— 核心实现
|
||||
*/
|
||||
private async _doFetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
|
||||
const store = useMessageStore.getState();
|
||||
|
||||
store.setLoadingMessages(conversationId, true);
|
||||
|
||||
@@ -398,10 +431,22 @@ export class MessageSyncService implements IMessageSyncService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 基于 seq 的轻量增量同步（用于重连场景，替代全量刷新）
|
||||
* 返回 true 表示同步成功，false 表示需要退化为全量刷新
|
||||
* 基于 seq 的轻量增量同步 — 去重入口
|
||||
*/
|
||||
async syncBySeq(): Promise<boolean> {
|
||||
if (this.inflightSyncBySeq) return this.inflightSyncBySeq;
|
||||
this.inflightSyncBySeq = this._doSyncBySeq();
|
||||
try {
|
||||
return await this.inflightSyncBySeq;
|
||||
} finally {
|
||||
this.inflightSyncBySeq = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 基于 seq 的轻量增量同步 — 核心实现
|
||||
*/
|
||||
private async _doSyncBySeq(): Promise<boolean> {
|
||||
try {
|
||||
const serverItems = await messageService.getSyncData();
|
||||
if (!serverItems || serverItems.length === 0) return false;
|
||||
@@ -462,10 +507,22 @@ export class MessageSyncService implements IMessageSyncService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 基于版本号的增量同步（替代 syncBySeq，减少带宽）
|
||||
* 需要本地存储 syncVersion，首次同步时 version=0 会触发 full_sync
|
||||
* 基于版本号的增量同步 — 去重入口
|
||||
*/
|
||||
async syncByVersion(): Promise<boolean> {
|
||||
if (this.inflightSyncByVersion) return this.inflightSyncByVersion;
|
||||
this.inflightSyncByVersion = this._doSyncByVersion();
|
||||
try {
|
||||
return await this.inflightSyncByVersion;
|
||||
} finally {
|
||||
this.inflightSyncByVersion = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 基于版本号的增量同步 — 核心实现
|
||||
*/
|
||||
private async _doSyncByVersion(): Promise<boolean> {
|
||||
try {
|
||||
const store = useMessageStore.getState();
|
||||
const version = store.syncVersion ?? 0;
|
||||
|
||||
@@ -24,7 +24,7 @@ import type {
|
||||
IUserCacheService,
|
||||
HandleNewMessageOptions,
|
||||
} from '../types';
|
||||
import { MAX_FLUSH_ITERATIONS } from '../constants';
|
||||
import { MAX_FLUSH_ITERATIONS, MIN_RECONNECT_SYNC_INTERVAL } from '../constants';
|
||||
import { useMessageStore, normalizeConversationId } from '../store';
|
||||
|
||||
export class WSMessageHandler implements IWSMessageHandler {
|
||||
@@ -41,6 +41,10 @@ export class WSMessageHandler implements IWSMessageHandler {
|
||||
private isBootstrapping: boolean = false;
|
||||
private lastReconnectSyncAt: number = 0;
|
||||
|
||||
// 同步触发器状态：防止 onConnect 和 sync_required 并发
|
||||
private syncInProgress: boolean = false;
|
||||
private lastSyncTriggerAt: number = 0;
|
||||
|
||||
// 系统通知去重（防止重连时重复递增系统未读）
|
||||
private processedNotificationIds: Set<string> = new Set();
|
||||
|
||||
@@ -166,67 +170,13 @@ export class WSMessageHandler implements IWSMessageHandler {
|
||||
// 监听连接状态
|
||||
wsService.onConnect(() => {
|
||||
useMessageStore.getState().setSSEConnected(true);
|
||||
|
||||
// 冷启动/重连兜底
|
||||
const now = Date.now();
|
||||
if (now - this.lastReconnectSyncAt > 1500) {
|
||||
this.lastReconnectSyncAt = now;
|
||||
|
||||
// 优先尝试 seq 驱动的轻量同步（内部会为 stale 的活动会话拉取消息）
|
||||
this.syncBySeqCallback().then(ok => {
|
||||
if (!ok) {
|
||||
// 增量同步失败或差异过大，退化为全量刷新
|
||||
this.fetchConversationsCallback(true, 'sse-reconnect').catch(error => {
|
||||
console.error('[WSMessageHandler] 连接后同步会话失败:', error);
|
||||
});
|
||||
// 退化场景下仍需同步活动会话消息
|
||||
const activeConv = useMessageStore.getState().getActiveConversation();
|
||||
if (activeConv) {
|
||||
this.fetchMessagesCallback(activeConv).catch(() => {});
|
||||
}
|
||||
}
|
||||
this.fetchUnreadCountCallback().catch(error => {
|
||||
console.error('[WSMessageHandler] 连接后同步未读数失败:', error);
|
||||
});
|
||||
}).catch(() => {
|
||||
// syncBySeq 异常，退化为全量
|
||||
this.fetchConversationsCallback(true, 'sse-reconnect').catch(() => {});
|
||||
const activeConv = useMessageStore.getState().getActiveConversation();
|
||||
if (activeConv) {
|
||||
this.fetchMessagesCallback(activeConv).catch(() => {});
|
||||
}
|
||||
this.fetchUnreadCountCallback().catch(() => {});
|
||||
});
|
||||
}
|
||||
this.triggerSync('sse-reconnect');
|
||||
});
|
||||
|
||||
// 监听 sync_required 事件，触发增量同步
|
||||
wsService.on('sync_required', () => {
|
||||
console.log('[WSMessageHandler] 收到 sync_required，开始增量同步');
|
||||
|
||||
// 先尝试 seq 驱动的轻量同步（内部会为 stale 的活动会话拉取消息）
|
||||
this.syncBySeqCallback().then(ok => {
|
||||
if (!ok) {
|
||||
this.fetchConversationsCallback(true, 'sync_required').catch(error => {
|
||||
console.error('[WSMessageHandler] sync_required 同步会话失败:', error);
|
||||
});
|
||||
// 退化场景下仍需同步活动会话消息
|
||||
const activeConv = useMessageStore.getState().getActiveConversation();
|
||||
if (activeConv) {
|
||||
this.fetchMessagesCallback(activeConv).catch(() => {});
|
||||
}
|
||||
}
|
||||
this.fetchUnreadCountCallback().catch(error => {
|
||||
console.error('[WSMessageHandler] sync_required 同步未读数失败:', error);
|
||||
});
|
||||
}).catch(() => {
|
||||
this.fetchConversationsCallback(true, 'sync_required').catch(() => {});
|
||||
const activeConv = useMessageStore.getState().getActiveConversation();
|
||||
if (activeConv) {
|
||||
this.fetchMessagesCallback(activeConv).catch(() => {});
|
||||
}
|
||||
this.fetchUnreadCountCallback().catch(() => {});
|
||||
});
|
||||
this.triggerSync('sync_required');
|
||||
});
|
||||
|
||||
wsService.onDisconnect(() => {
|
||||
@@ -259,6 +209,41 @@ export class WSMessageHandler implements IWSMessageHandler {
|
||||
this.isBootstrapping = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* 统一同步入口 — 节流 + 去重
|
||||
* onConnect 和 sync_required 共用此入口，防止并发触发重复同步
|
||||
*/
|
||||
private async triggerSync(source: string): Promise<void> {
|
||||
const now = Date.now();
|
||||
// 最小间隔节流
|
||||
if (now - this.lastSyncTriggerAt < MIN_RECONNECT_SYNC_INTERVAL) return;
|
||||
if (this.syncInProgress) return;
|
||||
|
||||
this.syncInProgress = true;
|
||||
this.lastSyncTriggerAt = now;
|
||||
|
||||
try {
|
||||
const ok = await this.syncBySeqCallback();
|
||||
if (!ok) {
|
||||
await this.fetchConversationsCallback(true, source);
|
||||
const activeConv = useMessageStore.getState().getActiveConversation();
|
||||
if (activeConv) {
|
||||
await this.fetchMessagesCallback(activeConv).catch(() => {});
|
||||
}
|
||||
}
|
||||
await this.fetchUnreadCountCallback().catch(() => {});
|
||||
} catch {
|
||||
await this.fetchConversationsCallback(true, source).catch(() => {});
|
||||
const activeConv = useMessageStore.getState().getActiveConversation();
|
||||
if (activeConv) {
|
||||
await this.fetchMessagesCallback(activeConv).catch(() => {});
|
||||
}
|
||||
await this.fetchUnreadCountCallback().catch(() => {});
|
||||
} finally {
|
||||
this.syncInProgress = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 初始化完成后，处理缓冲的 SSE 事件
|
||||
*/
|
||||
|
||||
@@ -70,7 +70,7 @@ export interface IMessageSyncService {
|
||||
fetchConversations(forceRefresh?: boolean, source?: string): Promise<void>;
|
||||
loadMoreConversations(): Promise<void>;
|
||||
fetchConversationDetail(conversationId: string): Promise<ConversationResponse | null>;
|
||||
fetchMessages(conversationId: string, afterSeq?: number): Promise<void>;
|
||||
fetchMessages(conversationId: string, afterSeq?: number, force?: boolean): Promise<void>;
|
||||
loadMoreMessages(conversationId: string, beforeSeq: number, limit?: number): Promise<MessageResponse[]>;
|
||||
fetchUnreadCount(): Promise<void>;
|
||||
syncBySeq(): Promise<boolean>;
|
||||
|
||||
Reference in New Issue
Block a user