Merge remote-tracking branch 'origin/dev_pro2.0' into dev_pro2.0

gaoluyang

昨天 b64a0deae5b5d33f9e20671a68936b27f0b9b00b

 src/views/im/home/composables/useMessagePuller.ts

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,464 @@
import type { Message } from '../types'

import type { ImChannelMessageApi } from '#/api/im/message/channel'
import type { ImGroupMessageApi } from '#/api/im/message/group'
import type { ImPrivateMessageApi } from '#/api/im/message/private'

import { watch } from 'vue'

import { pullChannelMessageList as apiPullChannelMessageList } from '#/api/im/message/channel'
import { pullGroupMessageList as apiPullGroupMessageList } from '#/api/im/message/group'
import { getPrivateMaxReadMessageId as apiGetPrivateMaxReadMessageId, pullPrivateMessageList as apiPullPrivateMessageList } from '#/api/im/message/private'
import { getCurrentUserId } from '#/views/im/utils/auth'

import { buildChannelConversationStub } from '../../utils/channel'
import {
  MESSAGE_GROUP_PULL_SIZE,
  MESSAGE_PRIVATE_PULL_SIZE,
  MESSAGE_PRIVATE_READ_ENABLED
} from '../../utils/config'
import {
  ImContentType,
  ImConversationType,
  ImMessageStatus,
  isFriendChatTip,
  isFriendNotification
} from '../../utils/constants'
import { generateClientMessageId, getPrivateMessagePeerId } from '../../utils/message'
import { runMinIdPull } from '../../utils/pull'
import { getFriendDisplayName, getGroupDisplayName } from '../../utils/user'
import { useConversationStore } from '../store/conversationStore'
import { useFriendStore } from '../store/friendStore'
import { useGroupRequestStore } from '../store/groupRequestStore'
import { useGroupStore } from '../store/groupStore'
import { type PulledMessage, useMessageStore } from '../store/messageStore'
import { useRtcStore } from '../store/rtcStore'
import { useImWebSocketStore } from '../store/websocketStore'

/** ä¸‰ç±»æ¶ˆæ¯ pull æŽ¥å£è¿”回的原始 VO è”合类型；runMinIdPull åªéœ€ id æŽ¨è¿›æ¸¸æ ‡ï¼Œå…·ä½“分发在 applyPage å†…按类型 cast */
type PulledRawMessage = ImChannelMessageApi.ChannelMessageRespVO | ImGroupMessageApi.GroupMessageRespVO | ImPrivateMessageApi.PrivateMessageRespVO

/**
 * æ¶ˆæ¯å¢žé‡æ‹‰å–：登录后分页拉取离线期间的新消息
 *
 * è®¾è®¡è¦ç‚¹ï¼š
 * 1. åŒæ—¶æ‹‰å–私聊 + ç¾¤èŠï¼Œä½¿ç”¨å„自的 `minId` æ¸¸æ ‡ï¼ˆprivateMessageMaxId / groupMessageMaxId）
 * 2. åŽç«¯ä¸€æ¬¡æœ€å¤šè¿”回 size æ¡ï¼›å‰ç«¯æŒ‰ minId æŒç»­ç¿»é¡µï¼Œç›´åˆ°æŽ¥å£è¿”回空列表为止
 * 3. æ‹‰å–期间 conversationStore.loading=true：
 *    - conversationStore è·³è¿‡æ‰¹é‡æŒä¹…化，避免频繁写入卡顿
 *    - websocketStore æŠŠæ–°æ¥çš„ WS æ™®é€šæ¶ˆæ¯ä¸¢è¿›ç¼“冲区，等循环结束后统一回放
 * 4. WebSocket é‡è¿žåŽä¼šå†è§¦å‘一次拉取，补齐断网期间错过的消息
 */
export const useMessagePuller = () => {
  const conversationStore = useConversationStore()
  const messageStore = useMessageStore()
  const wsStore = useImWebSocketStore()
  const friendStore = useFriendStore()
  const groupStore = useGroupStore()
  const groupRequestStore = useGroupRequestStore()
  const rtcStore = useRtcStore()
  const currentUserId = getCurrentUserId()

  /** åˆ¤æ–­è¯·æ±‚是否被主动取消 */
  const isAbortError = (e: unknown): boolean => {
    const error = e as { code?: string; message?: string; name?: string; }
    return (
      error?.name === 'CanceledError' ||
      error?.code === 'ERR_CANCELED' ||
      error?.message === 'canceled'
    )
  }

  /** ç§èŠä¼šè¯å½’属：自己发的算"发给 receiverId çš„会话"，否则算"发送方的会话"；curry currentUserId è¿›é—­åŒ…减少 3 å¤„调用方的样板 */
  const getPrivatePeerId = (message: ImPrivateMessageApi.PrivateMessageRespVO) =>
    getPrivateMessagePeerId(message, currentUserId)

  /** æœåŠ¡ç«¯ç§èŠæ¶ˆæ¯ -> æœ¬åœ° Message：targetId æ˜¯ä¼šè¯ä¸»é”®ï¼ˆå¯¹ç«¯ userId） */
  const convertPrivateMessage = (message: ImPrivateMessageApi.PrivateMessageRespVO): Message => {
    return {
      id: message.id,
      clientMessageId: message.clientMessageId || generateClientMessageId(),
      type: message.type,
      content: message.content,
      status: message.status,
      receiptStatus: message.receiptStatus,
      sendTime: new Date(message.sendTime).getTime(),
      senderId: message.senderId,
      targetId: getPrivatePeerId(message),
      selfSend: message.senderId === currentUserId
    }
  }

  /** æœåŠ¡ç«¯ç¾¤èŠæ¶ˆæ¯ -> æœ¬åœ° Message */
  const convertGroupMessage = (message: ImGroupMessageApi.GroupMessageRespVO): Message => {
    return {
      id: message.id,
      clientMessageId: message.clientMessageId || generateClientMessageId(),
      type: message.type,
      content: message.content,
      status: message.status,
      sendTime: new Date(message.sendTime).getTime(),
      senderId: message.senderId,
      targetId: message.groupId,
      selfSend: message.senderId === currentUserId,
      atUserIds: message.atUserIds || [],
      receiverUserIds: message.receiverUserIds || [],
      receiptStatus: message.receiptStatus,
      readCount: message.readCount
    }
  }

  /** æœåŠ¡ç«¯é¢‘道消息 -> æœ¬åœ° Message */
  const convertChannelMessage = (message: ImChannelMessageApi.ChannelMessageRespVO): Message => {
    return {
      id: message.id,
      clientMessageId: message.clientMessageId || generateClientMessageId(),
      type: message.type,
      content: message.content,
      status: ImMessageStatus.NORMAL, // é¢‘道无撤回，恒为正常
      receiptStatus: message.receiptStatus, // é¢‘道已读态：DONE å·²è¯» / PENDING æœªè¯»
      sendTime: new Date(message.sendTime).getTime(),
      senderId: 0, // ç³»ç»Ÿä¸‹å‘，无发送人
      targetId: message.channelId, // ä¼šè¯å½’属到频道编号
      selfSend: false,
      materialId: message.materialId // è¯¦æƒ…页拉富文本用
    }
  }

  /** é¢‘道：会话归属到 channelId；name / avatar æš‚用占位，将来接入 channelStore åŽå†å¡«çœŸå€¼ */
  const convertChannelConversation = (message: ImChannelMessageApi.ChannelMessageRespVO) =>
    buildChannelConversationStub(message.channelId)

  /** ç§èŠï¼šä¼šè¯å½’属到对端 userId */
  const convertPrivateConversation = (message: ImPrivateMessageApi.PrivateMessageRespVO) => {
    const targetId = getPrivatePeerId(message)
    const friend = friendStore.getFriend(targetId)
    return {
      type: ImConversationType.PRIVATE,
      targetId,
      name: friend ? getFriendDisplayName(friend) : String(targetId), // ä¼šè¯åˆ—表 / é¡¶éƒ¨æ ‡é¢˜å±•ç¤ºï¼šå¥½å‹å¤‡æ³¨ > çœŸå®žæ˜µç§°
      avatar: friend?.avatar || '',
      silent: friend?.silent
    }
  }

  /** ç¾¤èŠï¼šä¼šè¯å½’属到 groupId */
  const convertGroupConversation = (message: ImGroupMessageApi.GroupMessageRespVO) => {
    const group = groupStore.getGroup(message.groupId)
    return {
      type: ImConversationType.GROUP,
      targetId: message.groupId,
      name: group ? getGroupDisplayName(group) : String(message.groupId),
      avatar: group?.avatar || '',
      silent: group?.silent
    }
  }

  /**
   * åˆ†ç±»åž‹æ‹‰å–离线消息：翻页 / minId æ¸¸æ ‡æŽ¨è¿› / ç©ºé¡µåœç”± runMinIdPull è´Ÿè´£ï¼Œè¿™é‡ŒåªåšæŽ¥å£åˆ†æ”¯ + é€æ¡ä¸šåŠ¡åˆ†å‘
   * ï¼ˆæ’¤å›ž / å¥½å‹é€šçŸ¥ / æ™®é€šæ¶ˆæ¯ï¼‰+ å…¥åº“。
   *
   * å–消语义两层守卫，经 isActive ä¼ å…¥ runMinIdPull，任一不等即丢弃本批不入库、停止翻页，避免旧 session å“åº”落到新 store：
   * 1. startEpoch：cancelPull() é€’增 pullEpoch；离开 IM / åˆ‡è´¦å·æ—¶è·³å‡º
   * 2. startUserId：每批 await åŽæ¯”对当前登录 userId；防御 logout / å¤š tab ä¸‹ç”¨æˆ·å·²åˆ‡ä½† cancelPull æœªè§¦å‘
   */
  const pullByType = async (
    conversationType: number,
    startMinId: number,
    startEpoch: number,
    startUserId: number,
    signal: AbortSignal
  ) => {
    // ç§èŠ / ç¾¤èŠ / é¢‘道各自一套接口；按 conversationType åˆ†æ”¯è°ƒåº¦ã€‚翻页机制（minId æ¸¸æ ‡ / ç©ºé¡µåˆ¤æ–­ / é˜²æ­»ç¿»ï¼‰äº¤ç»™ runMinIdPull
    const isPrivate = conversationType === ImConversationType.PRIVATE
    const isChannel = conversationType === ImConversationType.CHANNEL
    const size = isPrivate ? MESSAGE_PRIVATE_PULL_SIZE : MESSAGE_GROUP_PULL_SIZE
    const isStillValid = () =>
      !signal.aborted && pullEpoch === startEpoch && getCurrentUserId() === startUserId
    await runMinIdPull<PulledRawMessage>({
      initialMinId: startMinId,
      pageSize: size,
      isActive: isStillValid,
      fetchPage: ({ minId, size }) => {
        if (isPrivate) {
          return apiPullPrivateMessageList({ minId, size }, signal)
        }
        if (isChannel) {
          return apiPullChannelMessageList({ minId, size }, signal)
        }
        return apiPullGroupMessageList({ minId, size }, signal)
      },
      applyPage: async (list, nextMinId) => {
        const pulledMessages: PulledMessage[] = []
        // é€æ¡ dispatch：原消息走批量 insert；RECALL ä¿¡å·èµ°æ‰¹é‡ recall æŠŠåŒæ‰¹å†…å·² insert çš„原消息更新为撤回提示。
        // åŽç«¯æŒ‰ id å‡åºè¿”回，且信号 id ä¸€å®š > åŽŸæ¶ˆæ¯ id（先更新 status å†æ’信号），所以原消息一定先到、recallMessage æ‰¾å¾—到
        for (const raw of list) {
          if (isChannel) {
            const message = raw as ImChannelMessageApi.ChannelMessageRespVO
            pulledMessages.push({
              kind: 'insert',
              conversationInfo: convertChannelConversation(message),
              message: convertChannelMessage(message)
            })
            continue
          }
          if (isPrivate) {
            const message = raw as ImPrivateMessageApi.PrivateMessageRespVO
            // ç‰¹æ®Šï¼šæ’¤å›žæ¶ˆæ¯çš„处理
            if (message.type === ImContentType.RECALL) {
              pulledMessages.push({
                kind: 'recall',
                conversationType: ImConversationType.PRIVATE,
                targetId: getPrivatePeerId(message),
                recallSignalContent: message.content
              })
              continue
            }
            // ç‰¹æ®Šï¼šåŽ†å²å¥½å‹äº‹ä»¶åªè¿˜åŽŸèŠå¤©æ°”泡；好友主数据由好友增量补偿同步
            // ä»… FRIEND_ADD / FRIEND_DELETE æ‰ä½œä¸ºä¼šè¯æ°”泡入消息列表
            if (isFriendNotification(message.type) && !isFriendChatTip(message.type)) {
              continue
            }
            // å…¶å®ƒæ¶ˆæ¯æ­£å¸¸å…¥ä¼šè¯æ¶ˆæ¯åˆ—表
            pulledMessages.push({
              kind: 'insert',
              conversationInfo: convertPrivateConversation(message),
              message: convertPrivateMessage(message)
            })
          } else {
            const message = raw as ImGroupMessageApi.GroupMessageRespVO
            // ç‰¹æ®Šï¼šæ’¤å›žæ¶ˆæ¯çš„处理
            if (message.type === ImContentType.RECALL) {
              pulledMessages.push({
                kind: 'recall',
                conversationType: ImConversationType.GROUP,
                targetId: message.groupId,
                recallSignalContent: message.content
              })
              continue
            }
            pulledMessages.push({
              kind: 'insert',
              conversationInfo: convertGroupConversation(message),
              message: convertGroupMessage(message)
            })
          }
        }
        // å…¥åº“ + æŽ¨è¿› messageMaxId；nextMinId ä¸ºç©ºï¼ˆæœ¬æ‰¹æ— æœ‰æ•ˆ id）时不推进游标，与旧逻辑一致
        await messageStore.applyPulledMessageList(pulledMessages, conversationType, nextMinId)
      }
    })
  }

  /** åŒä¸€æ—¶åˆ»åªå…è®¸ä¸€æ¬¡ pull：index.vue çš„手动调用与重连 watch è§¦å‘可能并发，共用同一个 promise å³å¯åŽ»é‡ */
  let pullPromise: null | Promise<void> = null
  let pullAbortController: AbortController | null = null

  /**
   * é¦–次 pull æ˜¯å¦å·²å®Œæˆã€‚仅在置 true åŽï¼ŒisConnected watch æ‰ä¼šè§¦å‘ pull。
   * é˜²æ­¢ socket onopen æ¯” friendStore/groupStore é¢„拉先到达时，watcher æŠ¢è·‘造成消息插入早于会话元数据可见
   */
  let initialPulled = false

  /**
   * pull è½®æ¬¡è®¡æ•°ï¼›åˆ‡è´¦å· / ç¦»å¼€ IM æ—¶ cancelPull() é€’增，旧 pullByType å¾ªçŽ¯æŒ‰ epoch è‡ªæ£€åŽè·³å‡º
   * é¿å…æ—§ session çš„接口响应在新 session è½åœ°ï¼Œé€ æˆè·¨è´¦å·æ¶ˆæ¯æ³„漏
   *
   * æ³¨æ„ï¼šæ™®é€šæ–­è¿žï¼ˆWS çŸ­æ–­ï¼‰ä¸å–消 pull——网络抖动 / æœåŠ¡ç«¯é‡å¯éƒ½å±žäºŽæœ¬è´¦å·æ­£å¸¸ç”Ÿå‘½å‘¨æœŸï¼Œ
   * å–消会导致首拉被中断后 initialPulled æ°¸è¿œåœåœ¨ false，后续重连 watcher ä¸å†è¡¥æ‹‰
   */
  let pullEpoch = 0

  /** æ˜¾å¼å–消：仅由 index.vue onUnmounted（离开 IM / åˆ‡è´¦å· / è·¯ç”±è·³å‡ºï¼‰è°ƒç”¨ */
  const cancelPull = () => {
    pullEpoch++
    pullAbortController?.abort()
    pullAbortController = null
    // æ—§ promise ä»åœ¨ finally é˜¶æ®µè·‘，但 epoch å®ˆå«å·²é˜»æ–­åŽç»­å‰¯ä½œç”¨ï¼›è¿™é‡Œç«‹åˆ»è®© pullPromise = null è®©æ–°ä¸€è½®å¯é‡å…¥
    pullPromise = null
    // åŒæ­¥ä¸¢å¼ƒ WS ç¼“冲帧；旧 pull å·²ä¸ä¼š flushBuffer，若不清下次进 IM ç¬¬ä¸€æ¬¡ pullOnce ä¼šæŠŠæ—§ session çš„帧回放进新 store
    wsStore.discardBuffer()
  }

  /**
   * çŠ¶æ€äº‹ä»¶è¡¥å¿ï¼šå¥½å‹ / å¥½å‹ç”³è¯·èµ°å¢žé‡ï¼›ç¾¤åˆ—表和群申请红点走快照刷新
   *
   * é¦–登主数据由 index.vue é©±åŠ¨ï¼Œé‡è¿žæ—¶å„ store å·²å°±ä½ï¼Œå¤šè·¯ allSettled å¹¶å‘互不影响，单路失败仅记日志。
   * ç¾¤æˆå‘˜ä¸åšå…¨å±€å¢žé‡åŒæ­¥ï¼Œé‡è¿žåªæ ‡è®°æœ¬åœ°ç¾¤æˆå‘˜ cache è¿‡æœŸï¼Œè¿›å…¥ç¾¤ä¼šè¯æˆ–成员列表时再按 groupId åˆ·æ–°ã€‚
   */
  const pullStateEvents = async (): Promise<void> => {
    // 1. æ¸…理连接级缓存
    messageStore.clearPrivateReadMaxIdCache()
    rtcStore.clearGroupCallCache()
    groupStore.markAllGroupActiveCallsExpired()
    groupStore.markAllGroupInfoExpired()
    groupStore.markAllGroupMembersExpired()
    // 2. å¹¶å‘补偿远端状态
    const results = await Promise.allSettled([
      friendStore.pullFriends(),
      friendStore.pullFriendRequests(),
      conversationStore.pullConversationReads(),
      groupStore.fetchGroupList(true),
      groupRequestStore.pullGroupRequests(),
      groupRequestStore.fetchUnhandledGroupRequestList()
    ])
    for (const result of results) {
      if (result.status === 'rejected') {
        console.warn('[IM] çŠ¶æ€äº‹ä»¶å¢žé‡è¡¥å¿å¤±è´¥', result.reason)
      }
    }
  }

  /** æ‰§è¡Œä¸€æ¬¡å…¨é‡å¢žé‡æ‹‰å–（重入安全：进行中再次调用复用同一个 promise） */
  const pullOnce = (): Promise<void> => {
    if (!currentUserId) {
      return Promise.resolve()
    }
    if (pullPromise) {
      return pullPromise
    }
    const startEpoch = pullEpoch
    // å¯åŠ¨æ—¶çš„用户快照；pullByType æ¯æ‰¹ await åŽæ¯”对当前登录用户，账号变了立刻丢弃
    const startUserId = currentUserId
    const abortController = new AbortController()
    pullAbortController = abortController
    // æœ¬è½® pull ä»å±žäºŽå½“前 session：epoch æœªæ¼‚ + ç”¨æˆ·æœªåˆ‡ï¼›ä»»ä½•åŠ¨æ–° store çŠ¶æ€çš„副作用都要先过这道关
    const isCurrentPull = () =>
      !abortController.signal.aborted &&
      pullEpoch === startEpoch &&
      getCurrentUserId() === startUserId
    pullPromise = (async () => {
      try {
        // æ—§ puller åœ¨ cancelPull æœªè§¦å‘的异常路径上再进来时，先于任何副作用退出，避免污染新 session çš„ loading
        if (!isCurrentPull()) {
          return
        }
        conversationStore.loading = true
        let messagePullSucceeded = false
        try {
          // å¹¶å‘拉取私聊 + ç¾¤èŠ + é¢‘道消息，降低初始加载耗时
          await Promise.all([
            pullByType(
              ImConversationType.PRIVATE,
              messageStore.privateMessageMaxId,
              startEpoch,
              startUserId,
              abortController.signal
            ),
            pullByType(
              ImConversationType.GROUP,
              messageStore.groupMessageMaxId,
              startEpoch,
              startUserId,
              abortController.signal
            ),
            pullByType(
              ImConversationType.CHANNEL,
              messageStore.channelMessageMaxId,
              startEpoch,
              startUserId,
              abortController.signal
            )
          ])
          messagePullSucceeded = true
        } catch (error) {
          if (isAbortError(error)) {
            return
          }
          console.error('[IM] æ‹‰å–离线消息失败:', error)
        } finally {
          // ä»å±žæœ¬è½®æ‰å¤ä½ loading；旧轮被 cancel / åˆ‡è´¦å·æ—¶ç”±æ–°ä¸€è½®è‡ªç®¡ï¼Œé¿å…è¦†ç›–æ–° session çš„ true
          if (isCurrentPull()) {
            conversationStore.loading = false
          }
        }

        // å–消 / åˆ‡è´¦å·åŽè·³è¿‡ flushBuffer / æŽ’序 / å·²è¯»ä½ç½®è¡¥é½
        if (!isCurrentPull()) {
          return
        }
        if (!messagePullSucceeded) {
          return
        }

        // å›žæ”¾ WebSocket åœ¨ loading æœŸé—´æ”¶åˆ°çš„缓冲消息
        const buffered = wsStore.flushBuffer()
        const replayPersistPromises: Promise<void>[] = []
        for (const item of buffered) {
          if (item.conversationType === ImConversationType.PRIVATE) {
            replayPersistPromises.push(wsStore.handlePrivateMessage(item.payload))
          } else if (item.conversationType === ImConversationType.CHANNEL) {
            replayPersistPromises.push(wsStore.handleChannelMessage(item.payload))
          } else {
            replayPersistPromises.push(wsStore.handleGroupMessage(item.payload))
          }
        }
        await Promise.all(replayPersistPromises)

        // pull + replay éƒ½å®ŒæˆåŽå†æŽ’序，避免回放消息打乱顺序
        conversationStore.sortConversationList()

        // é‡è¿ž / å†·å¯åŠ¨åŽè¡¥é½å½“前激活私聊会话的「对方已读位置」
        // ç¦»çº¿æœŸé—´é”™è¿‡çš„ RECEIPT æŽ¨é€ä¼šè¢«è¿™é‡Œè¡¥å›žï¼›å…¶ä»–私聊会话等用户点开时由 index.vue çš„ watch è§¦å‘
        // ç§èŠå·²è¯»å…³é—­æ—¶è·³è¿‡ï¼Œé¿å…æ‰“到已禁用接口触发错误日志
        const active = conversationStore.activeConversation
        if (MESSAGE_PRIVATE_READ_ENABLED && active && active.type === ImConversationType.PRIVATE) {
          try {
            const maxReadId = await apiGetPrivateMaxReadMessageId(
              active.targetId,
              abortController.signal
            )
            if (!isCurrentPull()) {
              return
            }
            messageStore.updatePrivateReadMaxId(active.targetId, maxReadId)
            if (maxReadId) {
              messageStore.applyMessageReadReceipt({
                conversationType: ImConversationType.PRIVATE,
                targetId: active.targetId,
                privateReadMaxId: maxReadId
              })
            }
          } catch (error) {
            if (isAbortError(error)) {
              return
            }
            console.warn('[IM] æ‹‰å–对方已读位置失败', error)
          }
        }
      } finally {
        // ä»å±žæœ¬è½®ï¼šæ­£å¸¸å®Œæˆé¦–拉；epoch ç­‰ä½† userId åˆ‡äº†ï¼šæ¸… pullPromise é˜²å¡æ­»ã€ä¸æ ‡é¦–拉；epoch æ¼‚：cancelPull å·²æ¸…，no-op
        if (isCurrentPull()) {
          pullPromise = null
          initialPulled = true
          if (pullAbortController === abortController) {
            pullAbortController = null
          }
        } else if (pullEpoch === startEpoch) {
          pullPromise = null
          if (pullAbortController === abortController) {
            pullAbortController = null
          }
        }
      }
    })()
    return pullPromise
  }

  /**
   * æ–­ç½‘期间 WS æ”¶ä¸åˆ°æŽ¨é€ï¼šé‡è¿žåŽæ—¢è¦æŒ‰ minId è¡¥é½æ¶ˆæ¯ï¼Œä¹Ÿè¦æŒ‰ update_time + id è¡¥é½å¥½å‹ / ç¾¤ / ç¾¤ç”³è¯·çŠ¶æ€ã€‚
   * é¦–次连接由 index.vue æ˜¾å¼é©±åŠ¨ï¼ˆpullOnce æ‹‰æ¶ˆæ¯ + å„ store é¦–拉），这里仅覆盖之后的重连。
   * é‡è¿žæ—¶ store å·²å°±ä½ï¼ŒpullStateEvents ä¸Ž pullOnce å¹¶å‘即可，无需「先就位再拉消息」的首登顺序约束。
   */
  watch(
    () => wsStore.isConnected,
    (isConnected) => {
      if (isConnected && initialPulled) {
        void pullOnce()
        void pullStateEvents()
      }
    }
  )

  return { pullOnce, cancelPull, convertPrivateMessage, convertGroupMessage }
}