/** * MessageStore(规范 §131) * * 职责: * - 维护会话列表(Mutual Matches) * - 维护当前打开会话的消息历史(按 conversationId 索引) * - 维护总未读数(驱动 Tab 徽标) * - 发送消息(本地先入队、失败重试) * * 不变量(AGENTS.md): * - 第 3 条:Mutual Match 后才能聊天——send 失败时若返回 MATCH_REQUIRED, * UI 引导回推荐页;本 Store 不在客户端做硬阻断,由后端守门。 * - 第 4 条:Block/Unmatch 后会话 canSendMessage=false,send 直接 reject, * UI 显示"会话已停用"提示。 * - 第 7 条:普通工作人员不能查看私人聊天——本 Store 仅服务于用户端。 * - 第 9 条:PostgreSQL 是 Source of Truth——本地仅缓存最近活跃会话的最近 * N 条消息用于即时展示,不做完整离线持久化;冷启动从服务端拉取。 */ const BaseStore = require('./baseStore'); const { messageService } = require('../services'); const { STORAGE_KEYS, MESSAGE_SEND_STATUS, CHAT_CONNECTION } = require('../constants'); /** 单会话内存缓存上限(避免无限增长) */ const PER_CONVERSATION_CACHE = 30; class MessageStore extends BaseStore { constructor() { super({ /** @type {import('../types').Conversation[]} */ conversations: [], conversationsCursor: null, conversationsHasMore: true, conversationsLoading: false, /** 当前打开的会话 ID */ activeConversationId: null, /** @type {Record} 按 conversationId 索引的内存缓存 */ messagesByConversation: {}, /** @type {Record} 游标 */ cursorsByConversation: {}, hasMoreByConversation: {}, totalUnreadCount: 0, /** @type {CHAT_CONNECTION[keyof CHAT_CONNECTION]} WebSocket 连接状态 */ connectionState: CHAT_CONNECTION.DISCONNECTED, lastError: null, }); /** @type {Map void>} 收到服务端推送的监听器,按 conversationId 索引 */ this._realtimeListeners = new Map(); /** @type {Array<(count: number) => void>} 总未读数变化监听 */ this._unreadListeners = []; } /** 冷启动从 storage 还原总未读数(仅作兜底,真值以服务端为准) */ hydrate() { const snap = BaseStore.hydrate(STORAGE_KEYS.TOTAL_UNREAD_COUNT); if (typeof snap === 'number') this.setState({ totalUnreadCount: snap }); } /** 拉取会话列表 */ async fetchConversations() { if (this.state.conversationsLoading) return; this.setState({ conversationsLoading: true, lastError: null }); try { const result = await messageService.getConversations({ pageSize: 20 }); this.setState({ conversations: result.items || [], conversationsCursor: result.nextCursor || null, conversationsHasMore: !!result.nextCursor, conversationsLoading: false, }); await this.refreshTotalUnread(); return result; } catch (err) { this.setState({ conversationsLoading: false, lastError: err }); throw err; } } /** 加载更多会话 */ async loadMoreConversations() { if (this.state.conversationsLoading || !this.state.conversationsCursor) return; this.setState({ conversationsLoading: true }); try { const result = await messageService.getConversations({ cursor: this.state.conversationsCursor, pageSize: 20, }); this.setState({ conversations: [...this.state.conversations, ...(result.items || [])], conversationsCursor: result.nextCursor || null, conversationsHasMore: !!result.nextCursor, conversationsLoading: false, }); } catch (err) { this.setState({ conversationsLoading: false, lastError: err }); throw err; } } /** 打开会话:拉取最近消息 */ async openConversation(conversationId) { this.setState({ activeConversationId: conversationId }); if (!this.state.messagesByConversation[conversationId]) { await this.fetchMessages(conversationId); } // 打开会话时清除该会话未读 await this.markConversationRead(conversationId); } /** 关闭会话 */ closeConversation() { this.setState({ activeConversationId: null }); } /** 拉取消息历史(最新一页) */ async fetchMessages(conversationId) { try { const result = await messageService.getMessages(conversationId, { pageSize: 20 }); const items = result.items || []; this.setState({ messagesByConversation: { ...this.state.messagesByConversation, [conversationId]: items, }, cursorsByConversation: { ...this.state.cursorsByConversation, [conversationId]: result.nextCursor || null, }, hasMoreByConversation: { ...this.state.hasMoreByConversation, [conversationId]: !!result.nextCursor, }, }); return items; } catch (err) { this.setState({ lastError: err }); throw err; } } /** 加载更早的消息 */ async loadMoreMessages(conversationId) { const cursor = this.state.cursorsByConversation[conversationId]; if (!cursor) return; try { const result = await messageService.getMessages(conversationId, { cursor, pageSize: 20, }); const items = result.items || []; const prev = this.state.messagesByConversation[conversationId] || []; this.setState({ messagesByConversation: { ...this.state.messagesByConversation, [conversationId]: [...items, ...prev].slice(-PER_CONVERSATION_CACHE * 2), }, cursorsByConversation: { ...this.state.cursorsByConversation, [conversationId]: result.nextCursor || null, }, hasMoreByConversation: { ...this.state.hasMoreByConversation, [conversationId]: !!result.nextCursor, }, }); return items; } catch (err) { this.setState({ lastError: err }); throw err; } } /** * 发送消息(先入队、乐观更新,失败重试一次后标 failed) * @param {string} conversationId * @param {string} content * @returns {Promise} */ async sendMessage(conversationId, content) { const messageId = 'local-' + Date.now() + '-' + Math.random().toString(36).slice(2, 8); const clientSentAt = Date.now(); const optimistic = { messageId, conversationId, senderId: 'me', bubbleType: 'self', content, sentAt: clientSentAt, sendStatus: MESSAGE_SEND_STATUS.SENDING, }; this._appendMessage(conversationId, optimistic); try { const result = await messageService.sendMessage(conversationId, { content, messageId, clientSentAt, }); const confirmed = { ...optimistic, messageId: result.messageId || messageId, sentAt: result.sentAt || clientSentAt, sendStatus: MESSAGE_SEND_STATUS.SENT, }; this._replaceMessage(conversationId, messageId, confirmed); // 更新会话列表里 lastMessage this._patchConversationLastMessage(conversationId, content, confirmed.sentAt); return confirmed; } catch (err) { // 标记失败,用户可手动重试 this._patchMessage(conversationId, messageId, { sendStatus: MESSAGE_SEND_STATUS.FAILED, }); throw err; } } /** * 重试失败的本地消息 * @param {string} conversationId * @param {string} localMessageId 原 messageId */ async retrySend(conversationId, localMessageId) { const list = this.state.messagesByConversation[conversationId] || []; const msg = list.find((m) => m.messageId === localMessageId); if (!msg || msg.sendStatus !== MESSAGE_SEND_STATUS.FAILED) return; this._patchMessage(conversationId, localMessageId, { sendStatus: MESSAGE_SEND_STATUS.SENDING, }); try { const result = await messageService.sendMessage(conversationId, { content: msg.content, messageId: localMessageId, clientSentAt: msg.sentAt, }); this._replaceMessage(conversationId, localMessageId, { ...msg, messageId: result.messageId || localMessageId, sentAt: result.sentAt || msg.sentAt, sendStatus: MESSAGE_SEND_STATUS.SENT, }); } catch (err) { this._patchMessage(conversationId, localMessageId, { sendStatus: MESSAGE_SEND_STATUS.FAILED, }); throw err; } } /** * 收到服务端推送的新消息(WebSocket 入口,AGENTS 第 9 条:DB 是 Source of Truth) * @param {import('../types').Message} msg */ onIncomingMessage(msg) { const cid = msg.conversationId; const list = this.state.messagesByConversation[cid]; if (list) { // 去重:messageId 已存在则不重复入队 if (!list.some((m) => m.messageId === msg.messageId)) { this._appendMessage(cid, { ...msg, sendStatus: MESSAGE_SEND_STATUS.SENT }); } } // 更新会话列表里 lastMessage + 未读 this._patchConversationLastMessage(cid, msg.content, msg.sentAt); if (cid !== this.state.activeConversationId) { this._bumpUnread(cid); } // 通知页面监听器 const listener = this._realtimeListeners.get(cid); if (listener) listener(msg); } /** 标记会话已读(清未读 + 同步服务端) */ async markConversationRead(conversationId) { const conv = this.state.conversations.find((c) => c.conversationId === conversationId); if (conv && conv.unreadCount > 0) { this.setState({ conversations: this.state.conversations.map((c) => c.conversationId === conversationId ? { ...c, unreadCount: 0 } : c, ), }); await this.refreshTotalUnread(); } } /** 刷新总未读数(Tab 徽标用,真值以服务端为准) */ async refreshTotalUnread() { try { const { unreadCount } = await messageService.getNotificationsUnreadCount(); this.setState({ totalUnreadCount: unreadCount }); this.persist(unreadCount, STORAGE_KEYS.TOTAL_UNREAD_COUNT); this._unreadListeners.forEach((fn) => fn(unreadCount)); } catch (err) { // 兜底:本地累加 const sum = this.state.conversations.reduce((s, c) => s + (c.unreadCount || 0), 0); this.setState({ totalUnreadCount: sum }); } } /** * Block/Unmatch 后由 matchStore 编排调用,标记会话不可发消息 * (AGENTS 第 4 条) */ markConversationBlocked(conversationId) { this.setState({ conversations: this.state.conversations.map((c) => c.conversationId === conversationId ? { ...c, canSendMessage: false, relationStatus: 'blocked' } : c, ), }); } /** * 注册实时消息监听(聊天页 onLoad 时调用) * @param {string} conversationId * @param {(msg: import('../types').Message) => void} listener * @returns {() => void} */ onMessage(conversationId, listener) { this._realtimeListeners.set(conversationId, listener); return () => this._realtimeListeners.delete(conversationId); } setConnectionState(state) { this.setState({ connectionState: state }); } // ============ 内部辅助 ============ _appendMessage(conversationId, msg) { const list = this.state.messagesByConversation[conversationId] || []; const next = [...list, msg].slice(-PER_CONVERSATION_CACHE * 2); this.setState({ messagesByConversation: { ...this.state.messagesByConversation, [conversationId]: next, }, }); } _patchMessage(conversationId, messageId, patch) { const list = this.state.messagesByConversation[conversationId] || []; const next = list.map((m) => m.messageId === messageId ? { ...m, ...patch } : m, ); this.setState({ messagesByConversation: { ...this.state.messagesByConversation, [conversationId]: next, }, }); } _replaceMessage(conversationId, oldId, newMsg) { const list = this.state.messagesByConversation[conversationId] || []; const idx = list.findIndex((m) => m.messageId === oldId); if (idx === -1) { this._appendMessage(conversationId, newMsg); return; } const next = [...list]; next[idx] = newMsg; this.setState({ messagesByConversation: { ...this.state.messagesByConversation, [conversationId]: next, }, }); } _patchConversationLastMessage(conversationId, content, sentAt) { this.setState({ conversations: this.state.conversations.map((c) => c.conversationId === conversationId ? { ...c, lastMessage: content, lastMessageAt: sentAt } : c, ), }); } _bumpUnread(conversationId) { this.setState({ conversations: this.state.conversations.map((c) => c.conversationId === conversationId ? { ...c, unreadCount: (c.unreadCount || 0) + 1 } : c, ), }); } } const messageStore = new MessageStore(); module.exports = messageStore;