diff --git a/src/services/core/wsService.ts b/src/services/core/wsService.ts index 04392e4..e6deb02 100644 --- a/src/services/core/wsService.ts +++ b/src/services/core/wsService.ts @@ -357,7 +357,7 @@ class WebSocketService { this.markActivity(); this.startHeartbeat(); this.connectionHandlers.forEach(h => h()); - + // 发送队列中的消息 this.flushPendingMessages(); }; @@ -776,10 +776,19 @@ class WebSocketService { } private handleMessageSent(payload: any): void { - // 找到对应的发送请求并resolve - const pendingMsg = this.pendingMessages.find(p => - p.type === 'chat' && p.payload.conversation_id === payload.conversation_id - ); + // 使用后端回带的 client_msg_id 精确匹配单条消息 + // 避免同会话连续发送时，第一条 ACK 误清掉第二条的 pending 记录 + const clientId = payload?.client_msg_id; + let pendingMsg: PendingMessage | undefined; + if (clientId) { + pendingMsg = this.pendingMessages.find(p => p.id === clientId); + } + // 兜底：旧版后端不回带 client_msg_id 时，按 conversation_id + 类型匹配最旧一条 + if (!pendingMsg) { + pendingMsg = this.pendingMessages.find(p => + p.type === 'chat' && p.payload.conversation_id === payload.conversation_id + ); + } if (pendingMsg) { pendingMsg.resolve(payload); this.removePendingMessage(pendingMsg.id); @@ -787,9 +796,16 @@ class WebSocketService { } private handleMessageRecalled(payload: any): void { - const pendingMsg = this.pendingMessages.find(p => - p.type === 'recall' && p.payload.message_id === payload.message_id - ); + const clientId = payload?.client_msg_id; + let pendingMsg: PendingMessage | undefined; + if (clientId) { + pendingMsg = this.pendingMessages.find(p => p.id === clientId); + } + if (!pendingMsg) { + pendingMsg = this.pendingMessages.find(p => + p.type === 'recall' && p.payload.message_id === payload.message_id + ); + } if (pendingMsg) { pendingMsg.resolve(payload); this.removePendingMessage(pendingMsg.id); @@ -880,16 +896,18 @@ class WebSocketService { } const messageId = `msg_${++this.messageIdCounter}_${Date.now()}`; + // 携带 client_msg_id：用于服务器回 ACK / message_sent 时精确匹配单条消息 + const payloadWithId = { ...payload, client_msg_id: messageId }; const message = { type, - payload, + payload: payloadWithId, }; // 添加到待处理队列 const pendingMsg: PendingMessage = { id: messageId, type, - payload, + payload: payloadWithId, resolve, reject, timestamp: Date.now(), @@ -929,6 +947,11 @@ class WebSocketService { } // 刷新待发送消息队列 + // 过期判定： + // - 单条 sendViaWebSocket 自带 10s 超时，正常情况下 pending 消息不会超过这么久 + // - 重连后链路刚恢复时，距离"刚入队"的消息一律视为有效（避免把刚刚挂起的消息误清） + // - 兜底：超过 STALE_PENDING_MS（60s）一定视为过期丢弃 + private static readonly STALE_PENDING_MS = 60000; private flushPendingMessages(): void { if (this.pendingMessages.length === 0) return; @@ -938,7 +961,9 @@ class WebSocketService { // 分离过期和有效的消息 for (const msg of this.pendingMessages) { - if (now - msg.timestamp > 30000) { + const ageMs = now - msg.timestamp; + const isStale = ageMs > WebSocketService.STALE_PENDING_MS; + if (isStale) { expiredMessages.push(msg); } else { validMessages.push(msg); @@ -950,13 +975,32 @@ class WebSocketService { msg.reject(new Error('Message expired')); }); - // 重新发送有效消息 - this.pendingMessages = []; - validMessages.forEach(msg => { - this.sendViaWebSocket(msg.type, msg.payload) - .then(msg.resolve) - .catch(msg.reject); - }); + // 重新发送有效消息：直接复用原 client_msg_id，否则重连后服务端 ACK 无法回带到原 pending 记录 + // 注意：不能调 sendViaWebSocket（会重新生成 id 并入队），必须直接 ws.send + this.pendingMessages = validMessages; + for (let i = 0; i < validMessages.length; i++) { + const msg = validMessages[i]; + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { + // 链路断开，剩余消息留在队列等下次重连 + break; + } + try { + this.ws.send(JSON.stringify({ type: msg.type, payload: msg.payload })); + } catch { + // ws.send 抛异常时 reject 并移出队列，避免无限残留 + msg.reject(new Error('WebSocket send failed on flush')); + this.removePendingMessage(msg.id); + continue; + } + // 重置 10s 超时 + setTimeout(() => { + const index = this.pendingMessages.findIndex(p => p.id === msg.id); + if (index >= 0) { + this.pendingMessages[index].reject(new Error('Message timeout')); + this.removePendingMessage(msg.id); + } + }, 10000); + } } private removePendingMessage(id: string): void { diff --git a/src/stores/message/MessageManager.ts b/src/stores/message/MessageManager.ts index be9f678..ea8a5b1 100644 --- a/src/stores/message/MessageManager.ts +++ b/src/stores/message/MessageManager.ts @@ -154,10 +154,10 @@ class MessageManager { useMessageStore.getState().setInitialized(true); - // 处理缓冲的 SSE 事件 - await this.wsHandler.flushBufferedSSEEvents(); - + // 先关闭 bootstrap（事件流转入正常处理路径），再做尾部 flush + // 避免 flush 期间新到的事件被压入 buffer 后无人消费 this.wsHandler.setBootstrapping(false); + await this.wsHandler.flushBufferedSSEEvents(); if (useMessageStore.getState().currentConversationId) { await this.fetchMessages(useMessageStore.getState().currentConversationId!); diff --git a/src/stores/message/hooks.ts b/src/stores/message/hooks.ts index d949c59..ace59b8 100644 --- a/src/stores/message/hooks.ts +++ b/src/stores/message/hooks.ts @@ -100,9 +100,14 @@ interface UseMessagesReturn { * * 核心修复： * 1. useFocusEffect：屏幕重新获得焦点时强制同步消息，解决"需要退出重进才能看到新消息"的问题 - * 2. 清理 isLoadingMessages：组件卸载时清除残留的 loading 锁，防止下次进入被阻断 + * 2. 引用计数清理 isLoadingMessages：仅在最后一个订阅者卸载时清除 loading 锁， + * 避免 StrictMode 双调用或快速切换会话时，第一次 fetch 还没完成就被清锁 * 3. 重入时 forceSync：相同 conversationId 重复进入时强制拉取最新消息 */ + +// 记录每个 conversationId 上的活跃订阅数，仅在归零时清锁 +const loadingLockRefCount: Map = new Map(); + export function useMessages(conversationId: string | null): UseMessagesReturn { const normalizedConversationId = conversationId ? String(conversationId) : null; @@ -128,6 +133,10 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { return; } + // 引用计数 +1 + const prevCount = loadingLockRefCount.get(normalizedConversationId) ?? 0; + loadingLockRefCount.set(normalizedConversationId, prevCount + 1); + lastActivatedAtRef.current = Date.now(); const isReentry = lastActivatedIdRef.current === normalizedConversationId; @@ -141,11 +150,17 @@ export function useMessages(conversationId: string | null): UseMessagesReturn { lastActivatedIdRef.current = normalizedConversationId; return () => { - // 清理活动会话 + 清除残留的 loading 锁 - if (messageManager.getActiveConversation() === normalizedConversationId) { - messageManager.setActiveConversation(null); + // 引用计数 -1，仅在最后一个订阅者卸载时清活动会话 + 清 loading 锁 + const nextCount = (loadingLockRefCount.get(normalizedConversationId) ?? 1) - 1; + if (nextCount <= 0) { + loadingLockRefCount.delete(normalizedConversationId); + if (messageManager.getActiveConversation() === normalizedConversationId) { + messageManager.setActiveConversation(null); + } + useMessageStore.getState().setLoadingMessages(normalizedConversationId, false); + } else { + loadingLockRefCount.set(normalizedConversationId, nextCount); } - useMessageStore.getState().setLoadingMessages(normalizedConversationId, false); }; }, [normalizedConversationId]); diff --git a/src/stores/message/services/MessageSyncService.ts b/src/stores/message/services/MessageSyncService.ts index 366b651..1c80137 100644 --- a/src/stores/message/services/MessageSyncService.ts +++ b/src/stores/message/services/MessageSyncService.ts @@ -197,9 +197,10 @@ export class MessageSyncService implements IMessageSyncService { /** * 获取会话消息（增量同步）— 去重入口 + * key 包含 force 维度，强制刷新会跳过非强制的 in-flight 合并 */ async fetchMessages(conversationId: string, afterSeq?: number, force?: boolean): Promise { - const key = `${conversationId}:${afterSeq ?? 'all'}`; + const key = `${conversationId}:${afterSeq ?? 'all'}:${force ? 'force' : 'normal'}`; const existing = this.inflightFetches.get(key); if (existing) return existing; diff --git a/src/stores/message/services/ReadReceiptManager.ts b/src/stores/message/services/ReadReceiptManager.ts index 5e18605..f9e6d0d 100644 --- a/src/stores/message/services/ReadReceiptManager.ts +++ b/src/stores/message/services/ReadReceiptManager.ts @@ -30,7 +30,7 @@ export class ReadReceiptManager implements IReadReceiptManager { const normalizedId = normalizeConversationId(conversationId); const store = useMessageStore.getState(); const conversation = store.getConversation(normalizedId); - + if (!conversation) { console.warn('[ReadReceiptManager] 会话不存在:', normalizedId); return; @@ -53,6 +53,9 @@ export class ReadReceiptManager implements IReadReceiptManager { this.readStateVersion++; const currentVersion = this.readStateVersion; + // 记录本次写入的 my_last_read_seq，用于失败时条件回滚 + const writtenReadSeq = Math.max(Number(conversation.my_last_read_seq || 0), seq); + // 1. 标记此会话有进行中的已读请求 this.pendingReadMap.set(normalizedId, { timestamp: Date.now(), @@ -64,7 +67,7 @@ export class ReadReceiptManager implements IReadReceiptManager { const updatedConv: ConversationResponse = { ...conversation, unread_count: 0, - my_last_read_seq: Math.max(Number(conversation.my_last_read_seq || 0), seq), + my_last_read_seq: writtenReadSeq, }; const currentUnread = store.getUnreadCount(); @@ -79,11 +82,11 @@ export class ReadReceiptManager implements IReadReceiptManager { await messageService.markAsRead(normalizedId, seq); } catch (error) { console.error('[ReadReceiptManager] 标记已读API失败:', error); - - // 失败时回滚状态 - store.updateConversation(conversation); - store.setUnreadCount(currentUnread.total, currentUnread.system); - + + // 条件回滚（含 totalUnreadCount）：仅在 my_last_read_seq 仍是本次写入值时回滚， + // 避免覆盖并发 markAsRead 已写入的更大读游标或更小未读 + store.rollbackReadIfUnchanged(normalizedId, writtenReadSeq, prevUnreadCount); + // API 失败时立即清除保护 this.pendingReadMap.delete(normalizedId); return; @@ -137,6 +140,9 @@ export class ReadReceiptManager implements IReadReceiptManager { }); store.setConversationsWithUnread(newConversations, 0, currentSystem); + // 记录本次乐观写入的"目标快照"，用于失败时条件回滚 + const optimisticSnapshot = new Map(newConversations); + try { // 标记系统消息已读 await messageService.markAllSystemMessagesRead(); @@ -155,8 +161,22 @@ export class ReadReceiptManager implements IReadReceiptManager { } } catch (error) { console.error('[ReadReceiptManager] 标记所有已读失败:', error); - // 回滚：恢复原始会话和未读数 - store.setConversationsWithUnread(new Map(conversationsMap), prevTotalUnread, currentSystem); + // 条件回滚：仅在 conversations 的关键字段未被其他操作修改时还原， + // 避免覆盖期间到达的 markAsRead / WS 推送造成的新未读 + const current = useMessageStore.getState().conversations; + let stillMatches = current.size === optimisticSnapshot.size; + if (stillMatches) { + for (const [id, conv] of optimisticSnapshot.entries()) { + const cur = current.get(id); + if (!cur || cur.unread_count !== conv.unread_count || cur.my_last_read_seq !== conv.my_last_read_seq) { + stillMatches = false; + break; + } + } + } + if (stillMatches) { + store.setConversationsWithUnread(new Map(conversationsMap), prevTotalUnread, currentSystem); + } } } diff --git a/src/stores/message/services/UserCacheService.ts b/src/stores/message/services/UserCacheService.ts index 9b95cb1..8c0f774 100644 --- a/src/stores/message/services/UserCacheService.ts +++ b/src/stores/message/services/UserCacheService.ts @@ -6,6 +6,7 @@ import type { MessageResponse, UserDTO } from '../../../types/dto'; import { api } from '@/services/core'; import { userCacheRepository } from '@/database'; +import { useMessageStore } from '../store'; import type { IUserCacheService } from '../types'; export class UserCacheService implements IUserCacheService { @@ -61,16 +62,20 @@ export class UserCacheService implements IUserCacheService { /** * 批量异步填充消息的 sender 信息 - * 填充完成后调用 notifyUpdate 更新内存并通知订阅者 + * 注意：fetch 期间新消息可能已到达，回调内必须重读最新 messages 列表， + * 否则新消息的 sender 字段不会被填充。 */ enrichMessagesWithSenderInfo( conversationId: string, - messages: MessageResponse[], + _messages: MessageResponse[], notifyUpdate: (conversationId: string, messages: MessageResponse[]) => void ): void { + // 取一次最新列表作为本次填充目标 + const latestMessages = useMessageStore.getState().getMessages(conversationId); + // 收集所有需要查询的唯一 sender_id（排除系统用户） const senderIds = [...new Set( - messages + latestMessages .filter(m => m.sender_id && m.sender_id !== '10000' && !m.sender) .map(m => m.sender_id) )]; @@ -87,8 +92,10 @@ export class UserCacheService implements IUserCacheService { if (senderMap.size === 0) return; + // 重新读取最新消息列表（期间可能又有新消息到达） + const currentMessages = useMessageStore.getState().getMessages(conversationId); let changed = false; - const updated = messages.map(m => { + const updated = currentMessages.map(m => { if (!m.sender && m.sender_id && senderMap.has(m.sender_id)) { changed = true; return { ...m, sender: senderMap.get(m.sender_id) }; diff --git a/src/stores/message/services/WSMessageHandler.ts b/src/stores/message/services/WSMessageHandler.ts index 67ec2cc..ee17cfa 100644 --- a/src/stores/message/services/WSMessageHandler.ts +++ b/src/stores/message/services/WSMessageHandler.ts @@ -212,10 +212,13 @@ export class WSMessageHandler implements IWSMessageHandler { /** * 统一同步入口 — 节流 + 去重 * onConnect 和 sync_required 共用此入口，防止并发触发重复同步 + * 启动阶段（isBootstrapping=true）跳过，由 MessageManager.initialize 完成后统一触发 */ private async triggerSync(source: string): Promise { + if (this.isBootstrapping) return; + const now = Date.now(); - // 最小间隔节流 + // 最小间隔节流：避免连续 sync_required 事件短时间内重复触发全量同步 if (now - this.lastSyncTriggerAt < MIN_RECONNECT_SYNC_INTERVAL) return; if (this.syncInProgress) return; @@ -246,14 +249,22 @@ export class WSMessageHandler implements IWSMessageHandler { /** * 初始化完成后，处理缓冲的 SSE 事件 + * 每次循环：取走当前 buffer 后立即清空（避免在 fetch 期间新事件再被覆盖） + * 退出条件：连续一轮没有新事件，确保 flush 期间到达的事件也被消费 */ async flushBufferedSSEEvents(): Promise { + let lastChatLen = -1; + let lastReadLen = -1; for (let i = 0; i < MAX_FLUSH_ITERATIONS; i++) { - const hasBuffered = this.bufferedChatMessages.length > 0 || this.bufferedReadReceipts.length > 0; - if (!hasBuffered) return; - const chatEvents = this.bufferedChatMessages; const readEvents = this.bufferedReadReceipts; + // 上一轮长度未变且都已清空，说明已稳定 + if (chatEvents.length === lastChatLen && readEvents.length === lastReadLen) { + return; + } + lastChatLen = chatEvents.length; + lastReadLen = readEvents.length; + // 立即清空 buffer；flush 期间新到的事件将走正常处理路径（isBootstrapping 已是 false） this.bufferedChatMessages = []; this.bufferedReadReceipts = []; @@ -528,9 +539,8 @@ export class WSMessageHandler implements IWSMessageHandler { } private incrementSystemUnread(): void { - const store = useMessageStore.getState(); - const currentUnread = store.getUnreadCount(); - store.setUnreadCount(currentUnread.total + 1, currentUnread.system + 1); + // 原子递增：避免外部 read-modify-write 造成丢计数 + useMessageStore.getState().incrementSystemUnreadAtomic(); } private markMessageAsRecalled(conversationId: string, messageId: string): void { diff --git a/src/stores/message/store.ts b/src/stores/message/store.ts index 00a1252..cfaf840 100644 --- a/src/stores/message/store.ts +++ b/src/stores/message/store.ts @@ -88,6 +88,10 @@ export interface MessageActions { removeMessage: (conversationId: string, messageId: string) => void; patchMessages: (conversationId: string, patches: Map>) => void; setUnreadCount: (total: number, system: number) => void; + /** + * 原子递增系统未读数：避免外部 RMW 造成丢失 + */ + incrementSystemUnreadAtomic: () => void; setLastSystemMessageAt: (time: string | null) => void; setSSEConnected: (connected: boolean) => void; setCurrentConversation: (conversationId: string | null) => void; @@ -107,6 +111,16 @@ export interface MessageActions { ) => void; incrementUnreadAtomic: (conversationId: string) => void; + /** + * 原子回滚乐观已读：仅在当前会话的 my_last_read_seq 仍等于 expectedReadSeq 时才回滚 + * 避免覆盖后续并发 markAsRead 已更新的更大值 + */ + rollbackReadIfUnchanged: ( + conversationId: string, + expectedReadSeq: number, + prevUnreadCount: number, + ) => void; + // 重置 reset: () => void; } @@ -393,6 +407,13 @@ export const useMessageStore = create((set, get) => ({ set({ totalUnreadCount: total, systemUnreadCount: system }); }, + incrementSystemUnreadAtomic: () => { + set(state => ({ + totalUnreadCount: state.totalUnreadCount + 1, + systemUnreadCount: state.systemUnreadCount + 1, + })); + }, + setLastSystemMessageAt: (time: string | null) => { set({ lastSystemMessageAt: time }); if (time) { @@ -506,6 +527,36 @@ export const useMessageStore = create((set, get) => ({ }); }, + rollbackReadIfUnchanged: ( + conversationId: string, + expectedReadSeq: number, + prevUnreadCount: number, + ) => { + const normalizedId = normalizeConversationId(conversationId); + set(state => { + const conversation = state.conversations.get(normalizedId); + if (!conversation) return state; + // 仅当本地乐观已读游标仍是本次写入的值时，才回滚未读 + // 否则说明已被更新的 markAsRead 覆盖，不应回滚 + if (Number(conversation.my_last_read_seq || 0) !== expectedReadSeq) { + return state; + } + const newConversations = new Map(state.conversations); + newConversations.set(normalizedId, { + ...conversation, + unread_count: prevUnreadCount, + }); + const conversationList = sortConversationList(newConversations); + // 同步回滚 totalUnreadCount：加回本次扣减的未读数 + const unreadDelta = prevUnreadCount - (conversation.unread_count || 0); + return { + conversations: newConversations, + conversationList, + totalUnreadCount: state.totalUnreadCount + unreadDelta, + }; + }); + }, + // ==================== 重置 ==================== reset: () => {