397 lines
13 KiB
JavaScript
397 lines
13 KiB
JavaScript
/**
|
||
* 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<string, import('../types').Message[]>} 按 conversationId 索引的内存缓存 */
|
||
messagesByConversation: {},
|
||
/** @type {Record<string, string|null>} 游标 */
|
||
cursorsByConversation: {},
|
||
hasMoreByConversation: {},
|
||
|
||
totalUnreadCount: 0,
|
||
|
||
/** @type {CHAT_CONNECTION[keyof CHAT_CONNECTION]} WebSocket 连接状态 */
|
||
connectionState: CHAT_CONNECTION.DISCONNECTED,
|
||
|
||
lastError: null,
|
||
});
|
||
/** @type {Map<string, (msg: import('../types').Message) => 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<import('../types').Message>}
|
||
*/
|
||
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;
|