danyang-match/miniapp/stores/messageStore.js

/**
 * 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 { formatConversationTime } = require('../utils/datetime');
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: this._decorateConversations(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, ...this._decorateConversations(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,
      },
    });
  }

  /**
   * 为会话补上模板所需的展示字段。
   *
   * components/conversation-item/index.wxml 读取 `lastMessageAtText`，
   * 该字段由服务端不产出（后端只给 epoch 毫秒的 lastMessageAt），
   * 故在进入 store 时统一格式化一次，避免每个页面各自实现。
   */
  _decorateConversations(items) {
    return (items || []).map((c) => ({
      ...c,
      lastMessageAtText: formatConversationTime(c.lastMessageAt),
    }));
  }

  _patchConversationLastMessage(conversationId, content, sentAt) {
    this.setState({
      conversations: this.state.conversations.map((c) =>
        c.conversationId === conversationId
          ? {
              ...c,
              lastMessage: content,
              lastMessageAt: sentAt,
              lastMessageAtText: formatConversationTime(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;