From a35698fc07863e95e6a8b1828da1820840ae3c81 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sun, 26 Apr 2026 00:28:43 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(im):=20=E7=BE=A4=E8=81=8A?= =?UTF-8?q?=E7=A6=BB=E7=BA=BF=E6=8B=89=E5=8F=96=E7=9C=8B=E4=B8=8D=E5=88=B0?= =?UTF-8?q?=E6=92=A4=E5=9B=9E=E6=8F=90=E7=A4=BA=EF=BC=8Cpull=20=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E6=8E=A5=E5=85=A5=20recallMessage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pullByType 之前对 RECALL 信号一律 skip、只靠原消息 status=RECALL 走 OR 兜底渲染。 当 pull 的 minId 卡在原消息处、回拉只返回信号时，本地缓存里的老消息没人翻成 RECALL，会一直停在原态——配合后端群聊 mapper 过滤掉 status=RECALL 的原消息，群聊 离线撤回完全不可见。 改成 pull / WS 走同一套 dispatch： - pullByType 信号转 conversationStore.recallMessage()，跟 WS 路径一致 - recallMessage 把 parseRecallMessageId 收敛进内部，第 3 个参数从 messageId: number 改成 recallSignalContent: string，4 个调用点都缩成一行 - MessageItem.isRecall 只判 type=RECALL，去掉 status=RECALL OR 分支 （conversationStore 里跳未读 / 跳已读那两处对 status 的判断是业务逻辑保留） --- .../im/home/composables/useMessagePuller.ts | 209 ++++++++++++++++++ src/views/im/home/store/conversationStore.ts | 12 +- src/views/im/home/store/friendStore.ts | 6 +- src/views/im/home/store/groupStore.ts | 10 +- src/views/im/home/store/websocketStore.ts | 124 ++++++----- src/views/im/home/types/index.ts | 4 +- 6 files changed, 295 insertions(+), 70 deletions(-) create mode 100644 src/views/im/home/composables/useMessagePuller.ts diff --git a/src/views/im/home/composables/useMessagePuller.ts b/src/views/im/home/composables/useMessagePuller.ts new file mode 100644 index 000000000..13372f7f5 --- /dev/null +++ b/src/views/im/home/composables/useMessagePuller.ts @@ -0,0 +1,209 @@ +import { watch } from 'vue' +import { useConversationStore } from '../store/conversationStore' +import { useImWebSocketStore } from '../store/websocketStore' +import { + pullPrivateMessages as apiPullPrivateMessages, + type ImPrivateMessageRespVO +} from '@/api/im/message/private' +import { + pullGroupMessages as apiPullGroupMessages, + type ImGroupMessageRespVO +} from '@/api/im/message/group' +import { + ImConversationType, + ImMessageType, + PRIVATE_MESSAGE_PULL_SIZE, + GROUP_MESSAGE_PULL_SIZE +} from '../../utils/constants' +import { useUserStore } from '@/store/modules/user' +import type { Message } from '../types' + +/** + * 消息增量拉取：登录后分页拉取离线期间的新消息 + * + * 设计要点： + * 1. 同时拉取私聊 + 群聊，使用各自的 `minId` 游标（privateMessageMaxId / groupMessageMaxId） + * 2. 后端一次最多返回 size 条；前端按 minId 持续翻页，直到接口返回空列表为止 + * 3. 拉取期间 conversationStore.loading=true： + * - conversationStore 跳过 localStorage 持久化，避免频繁写入卡顿 + * - websocketStore 把新来的 WS 普通消息丢进缓冲区，等循环结束后统一回放 + * 4. WebSocket 重连后会再触发一次拉取，补齐断网期间错过的消息 + */ +export const useMessagePuller = () => { + const conversationStore = useConversationStore() + const wsStore = useImWebSocketStore() + const userStore = useUserStore() + const currentUserId = Number(userStore.getUser?.id) || 0 + + /** 服务端私聊消息 -> 本地 Message */ + const convertPrivateMessage = (message: ImPrivateMessageRespVO): Message => { + return { + id: message.id, + clientMessageId: message.clientMessageId || '', + type: message.type, + content: message.content, + status: message.status, + sendTime: new Date(message.sendTime).getTime(), + senderId: message.senderId, + senderNickName: '', + targetId: message.receiverId, + selfSend: message.senderId === currentUserId + } + } + + /** 服务端群聊消息 -> 本地 Message */ + const convertGroupMessage = (message: ImGroupMessageRespVO): Message => { + return { + id: message.id, + clientMessageId: message.clientMessageId || '', + type: message.type, + content: message.content, + status: message.status, + sendTime: new Date(message.sendTime).getTime(), + senderId: message.senderId, + senderNickName: '', + targetId: message.groupId, + selfSend: message.senderId === currentUserId, + atUserIds: message.atUserIds || [], + receiverUserIds: message.receiverUserIds || [], + receiptStatus: message.receiptStatus, + readCount: message.readCount + } + } + + /** 私聊：会话归属到对端 userId */ + const convertPrivateConversation = (message: ImPrivateMessageRespVO) => { + const targetId = message.senderId === currentUserId ? message.receiverId : message.senderId + return { + type: ImConversationType.PRIVATE, + targetId, + name: String(targetId), + avatar: '' + } + } + + /** 群聊：会话归属到 groupId */ + const convertGroupConversation = (message: ImGroupMessageRespVO) => { + return { + type: ImConversationType.GROUP, + targetId: message.groupId, + name: String(message.groupId), + avatar: '' + } + } + + /** 循环拉取指定会话类型的消息：以列表最后一条 id 作为下次 minId，直到接口返回空列表 */ + const pullByType = async (conversationType: number, startMinId: number) => { + // 私聊 / 群聊各自一套接口和分页大小，按 isPrivate 在循环内分支调度 + let minId = startMinId || 0 + const isPrivate = conversationType === ImConversationType.PRIVATE + const size = isPrivate ? PRIVATE_MESSAGE_PULL_SIZE : GROUP_MESSAGE_PULL_SIZE + while (true) { + const list = isPrivate + ? await apiPullPrivateMessages({ minId, size }) + : await apiPullGroupMessages({ minId, size }) + if (!list || list.length === 0) { + break + } + + // 逐条 dispatch：原消息走 insertMessage；RECALL 信号走 recallMessage 把同批内已 insert 的原消息翻成撤回提示。 + // 后端按 id 升序返回，且信号 id 一定 > 原消息 id（先翻 status 再插信号），所以原消息一定先到、recallMessage 找得到 + for (const raw of list) { + if (isPrivate) { + const message = raw as ImPrivateMessageRespVO + if (message.type === ImMessageType.RECALL) { + conversationStore.recallMessage( + ImConversationType.PRIVATE, + message.senderId === currentUserId ? message.receiverId : message.senderId, + message.content, + '', + message.senderId === currentUserId + ) + continue + } + conversationStore.insertMessage( + convertPrivateConversation(message), + convertPrivateMessage(message) + ) + } else { + const message = raw as ImGroupMessageRespVO + if (message.type === ImMessageType.RECALL) { + conversationStore.recallMessage( + ImConversationType.GROUP, + message.groupId, + message.content, + '', + message.senderId === currentUserId + ) + continue + } + conversationStore.insertMessage( + convertGroupConversation(message), + convertGroupMessage(message) + ) + } + } + + // 游标推进到本批最后一条 id，下一轮从此处续翻 + const lastId = list[list.length - 1].id + if (lastId != null) { + minId = lastId + } + } + } + + /** 同一时刻只允许一次 pull：Index.vue 的手动调用与重连 watch 触发可能并发，共用同一个 promise 即可去重 */ + let pullPromise: Promise | null = null + + /** 执行一次全量增量拉取（重入安全：进行中再次调用复用同一个 promise） */ + const pullOnce = (): Promise => { + if (!currentUserId) { + return Promise.resolve() + } + if (pullPromise) { + return pullPromise + } + pullPromise = (async () => { + conversationStore.loading = true + try { + // 并发拉取私聊 + 群聊，降低初始加载耗时 + await Promise.all([ + pullByType(ImConversationType.PRIVATE, conversationStore.privateMessageMaxId), + pullByType(ImConversationType.GROUP, conversationStore.groupMessageMaxId) + ]) + + // 回放 WebSocket 在 loading 期间收到的缓冲消息 + const buffered = wsStore.flushBuffer() + for (const item of buffered) { + if (item.conversationType === ImConversationType.PRIVATE) { + wsStore.handlePrivateMessage(item.payload) + } else { + wsStore.handleGroupMessage(item.payload) + } + } + } catch (e) { + console.error('[IM] 拉取离线消息失败:', e) + } finally { + conversationStore.loading = false + conversationStore.sortConversations() + pullPromise = null + } + })() + return pullPromise + } + + /** + * 断网期间 WS 收不到推送，期间产生的消息只能靠拉取接口按 minId 游标补齐； + * 首次连接由 Index.vue 显式调 pullOnce，这里订阅 isConnected 的 false→true 转换，覆盖后续每次重连 + */ + watch( + () => wsStore.isConnected, + (isConnected) => { + if (isConnected) { + void pullOnce() + } + } + ) + + return { pullOnce } +} diff --git a/src/views/im/home/store/conversationStore.ts b/src/views/im/home/store/conversationStore.ts index ebbfdd1b8..2d172e292 100644 --- a/src/views/im/home/store/conversationStore.ts +++ b/src/views/im/home/store/conversationStore.ts @@ -13,6 +13,7 @@ import { buildRecallTip, generateClientMessageId, parseMessage, + parseRecallMessageId, type TextMessage } from '../../utils/message' import type { Conversation, ConversationStoreMeta, Message } from '../types' @@ -475,17 +476,18 @@ export const useConversationStore = defineStore('imConversationStore', { this.saveConversations(conversation) }, - /** - * 撤回消息：将原消息 type 改为 RECALL，并刷新会话摘要 - * 对应后端 RECALL 事件：按原 messageId 更新 - */ + /** 撤回消息：解析撤回信号 content（`{"messageId": xxx}`），找到原消息翻成 RECALL 态 + 刷新会话摘要 */ recallMessage( conversationType: number, targetId: number, - messageId: number, + recallSignalContent: string, senderNickName: string, selfSend: boolean ) { + const messageId = parseRecallMessageId(recallSignalContent) + if (messageId <= 0) { + return + } const conversation = this.getConversation(conversationType, targetId) if (!conversation) { return diff --git a/src/views/im/home/store/friendStore.ts b/src/views/im/home/store/friendStore.ts index 6b2962a9a..e5f4e8c1a 100644 --- a/src/views/im/home/store/friendStore.ts +++ b/src/views/im/home/store/friendStore.ts @@ -52,7 +52,7 @@ export const useFriendStore = defineStore('imFriendStore', { return } const list = await apiGetMyFriendList() - this.friends = (list || []).map(toFriend) + this.friends = (list || []).map(convertFriend) this.loaded = true // 同步 conversationStore 私聊会话的展示名 / 头像 / 免打扰 const conversationStore = useConversationStore() @@ -72,7 +72,7 @@ export const useFriendStore = defineStore('imFriendStore', { if (!data) { return } - this.upsertFriend(toFriend(data)) + this.upsertFriend(convertFriend(data)) } catch (e) { console.warn('[IM friendStore] loadFriendInfo 失败', e) } @@ -154,7 +154,7 @@ export const useFriendStore = defineStore('imFriendStore', { } }) -function toFriend(vo: ImFriendRespVO): Friend { +function convertFriend(vo: ImFriendRespVO): Friend { return { id: vo.id, friendUserId: vo.friendUserId, diff --git a/src/views/im/home/store/groupStore.ts b/src/views/im/home/store/groupStore.ts index 78e2262ca..064eddfd8 100644 --- a/src/views/im/home/store/groupStore.ts +++ b/src/views/im/home/store/groupStore.ts @@ -44,7 +44,7 @@ export const useGroupStore = defineStore('imGroupStore', { } // 拉取当前登录用户加入的所有群（不带成员；成员按需再走 loadGroupMembers） const list = await apiGetMyGroupList() - this.groups = (list || []).map(toGroup) + this.groups = (list || []).map(convertGroup) this.loaded = true const conversationStore = useConversationStore() for (const g of this.groups) { @@ -63,7 +63,7 @@ export const useGroupStore = defineStore('imGroupStore', { if (!data) { return } - this.upsertGroup(toGroup(data)) + this.upsertGroup(convertGroup(data)) } catch (e) { console.warn('[IM groupStore] loadGroupInfo 失败', e) } @@ -79,7 +79,7 @@ export const useGroupStore = defineStore('imGroupStore', { // 拉取该群所有成员（聚合自 AdminUser，含 nickname / avatar / displayUserName） const list = await apiGetGroupMemberList(groupId) - const members = (list || []).map((member) => toGroupMember(member, groupId)) + const members = (list || []).map((member) => convertGroupMember(member, groupId)) // 成员列表可能在群列表之前触发，此时需要占位一个 group if (!group) { this.upsertGroup({ @@ -138,7 +138,7 @@ export const useGroupStore = defineStore('imGroupStore', { } }) -function toGroup(vo: ImGroupRespVO): Group { +function convertGroup(vo: ImGroupRespVO): Group { return { id: vo.id, name: vo.name, @@ -148,7 +148,7 @@ function toGroup(vo: ImGroupRespVO): Group { } } -function toGroupMember(member: ImGroupMemberRespVO, groupId: number): GroupMember { +function convertGroupMember(member: ImGroupMemberRespVO, groupId: number): GroupMember { return { id: member.id, userId: member.userId, diff --git a/src/views/im/home/store/websocketStore.ts b/src/views/im/home/store/websocketStore.ts index bd60a0e17..1e6c4bc06 100644 --- a/src/views/im/home/store/websocketStore.ts +++ b/src/views/im/home/store/websocketStore.ts @@ -4,7 +4,7 @@ import { getRefreshToken } from '@/utils/auth' import { useUserStore } from '@/store/modules/user' import { ImWebSocketMessageType, ImMessageType, ImConversationType } from '../../utils/constants' -import { parseRecallMessageId, playAudioTip } from '../../utils/message' +import { playAudioTip } from '../../utils/message' import { useConversationStore } from './conversationStore' import { useFriendStore } from './friendStore' import { useGroupStore } from './groupStore' @@ -17,6 +17,44 @@ import type { Message } from '../types' +/** WebSocket 私聊 DTO -> 前端 Message：sendTime 转毫秒；senderNickName 由调用方按好友信息补 */ +const convertPrivateMessage = ( + websocketMessage: ImPrivateMessageDTO, + currentUserId: number, + senderNickName: string +): Message => ({ + id: websocketMessage.id, + clientMessageId: websocketMessage.clientMessageId, + type: websocketMessage.type, + content: websocketMessage.content, + status: websocketMessage.status, + sendTime: new Date(websocketMessage.sendTime).getTime(), + senderId: websocketMessage.senderId, + senderNickName, + targetId: websocketMessage.receiverId, + selfSend: websocketMessage.senderId === currentUserId +}) + +/** WebSocket 群聊 DTO -> 前端 Message：群消息额外带 atUserIds / receiverUserIds，给 @ 标记和回执用 */ +const convertGroupMessage = ( + websocketMessage: ImGroupMessageDTO, + currentUserId: number, + senderNickName: string +): Message => ({ + id: websocketMessage.id, + clientMessageId: websocketMessage.clientMessageId, + type: websocketMessage.type, + content: websocketMessage.content, + status: websocketMessage.status, + sendTime: new Date(websocketMessage.sendTime).getTime(), + senderId: websocketMessage.senderId, + senderNickName, + targetId: websocketMessage.groupId, + selfSend: websocketMessage.senderId === currentUserId, + atUserIds: websocketMessage.atUserIds || [], + receiverUserIds: websocketMessage.receiverUserIds || [] +}) + /** * IM WebSocket Store * @@ -38,8 +76,8 @@ export const useImWebSocketStore = defineStore('imWebSocketStore', { reconnectTimer: null as ReturnType | null, heartbeatTimer: null as ReturnType | null, messageBuffer: [] as Array< - | { kind: 'private'; payload: ImPrivateMessageDTO } - | { kind: 'group'; payload: ImGroupMessageDTO } + | { conversationType: typeof ImConversationType.PRIVATE; payload: ImPrivateMessageDTO } + | { conversationType: typeof ImConversationType.GROUP; payload: ImGroupMessageDTO } > // 初始化加载期内，先把普通消息丢进缓冲区，pull 完成后再一次性回放 }), @@ -228,7 +266,10 @@ export const useImWebSocketStore = defineStore('imWebSocketStore', { const conversationStore = useConversationStore() // 1. 离线加载期间先缓冲，等 pull 完成后再统一回放，避免重复或顺序错乱 if (conversationStore.loading) { - this.messageBuffer.push({ kind: 'private', payload: websocketMessage }) + this.messageBuffer.push({ + conversationType: ImConversationType.PRIVATE, + payload: websocketMessage + }) return } @@ -247,32 +288,18 @@ export const useImWebSocketStore = defineStore('imWebSocketStore', { // 3. 后端撤回：下发一条 RECALL 消息，content 为 `{"messageId": xxx}`（对齐 ImMessageTypeEnum.RECALL → RecallMessage） // 这里拦截下来改走 recallMessage（把原消息翻转为 RECALL 态），不让它作为新消息进列表 if (websocketMessage.type === ImMessageType.RECALL) { - const recallMessageId = parseRecallMessageId(websocketMessage.content) - if (recallMessageId) { - conversationStore.recallMessage( - ImConversationType.PRIVATE, - peerId, - recallMessageId, - friend?.nickname || '', - selfSend - ) - return - } + conversationStore.recallMessage( + ImConversationType.PRIVATE, + peerId, + websocketMessage.content, + friend?.nickname || '', + selfSend + ) + return } - // 4. 后端 DTO → 前端 Message：sendTime 转毫秒；selfSend / senderNickName 是前端补的 - const message: Message = { - id: websocketMessage.id, - clientMessageId: websocketMessage.clientMessageId, - type: websocketMessage.type, - content: websocketMessage.content, - status: websocketMessage.status, - sendTime: new Date(websocketMessage.sendTime).getTime(), - senderId: websocketMessage.senderId, - senderNickName: friend?.nickname || '', - targetId: websocketMessage.receiverId, - selfSend - } + // 4. 后端 DTO → 前端 Message + const message = convertPrivateMessage(websocketMessage, currentUserId, friend?.nickname || '') conversationStore.insertMessage( { type: ImConversationType.PRIVATE, @@ -339,7 +366,10 @@ export const useImWebSocketStore = defineStore('imWebSocketStore', { const conversationStore = useConversationStore() // 1. 离线加载期缓冲（与私聊对称） if (conversationStore.loading) { - this.messageBuffer.push({ kind: 'group', payload: websocketMessage }) + this.messageBuffer.push({ + conversationType: ImConversationType.GROUP, + payload: websocketMessage + }) return } const userStore = useUserStore() @@ -359,34 +389,18 @@ export const useImWebSocketStore = defineStore('imWebSocketStore', { // 3. 后端撤回：下发一条 RECALL 消息，content 为 `{"messageId": xxx}` // 这里拦截下来改走 recallMessage（把原消息翻转为 RECALL 态） if (websocketMessage.type === ImMessageType.RECALL) { - const recallMessageId = parseRecallMessageId(websocketMessage.content) - if (recallMessageId) { - conversationStore.recallMessage( - ImConversationType.GROUP, - websocketMessage.groupId, - recallMessageId, - senderNickName, - selfSend - ) - return - } + conversationStore.recallMessage( + ImConversationType.GROUP, + websocketMessage.groupId, + websocketMessage.content, + senderNickName, + selfSend + ) + return } - // 4. 后端 DTO → 前端 Message：群消息额外带 atUserIds / receiverUserIds，给 @ 标记和回执用 - const message: Message = { - id: websocketMessage.id, - clientMessageId: websocketMessage.clientMessageId, - type: websocketMessage.type, - content: websocketMessage.content, - status: websocketMessage.status, - sendTime: new Date(websocketMessage.sendTime).getTime(), - senderId: websocketMessage.senderId, - senderNickName, - targetId: websocketMessage.groupId, - selfSend, - atUserIds: websocketMessage.atUserIds || [], - receiverUserIds: websocketMessage.receiverUserIds || [] - } + // 4. 后端 DTO → 前端 Message + const message = convertGroupMessage(websocketMessage, currentUserId, senderNickName) conversationStore.insertMessage( { type: ImConversationType.GROUP, diff --git a/src/views/im/home/types/index.ts b/src/views/im/home/types/index.ts index ae59b9a10..d8fb63dc3 100644 --- a/src/views/im/home/types/index.ts +++ b/src/views/im/home/types/index.ts @@ -141,8 +141,8 @@ export interface Friend { avatar?: string // 好友头像 muted?: boolean // 是否免打扰（不展示未读徽标 + 不响提示音） status?: number // 好友状态，对齐 CommonStatusEnum（DISABLE = 已删除/墓碑） - addTime?: number // 添加好友时间（毫秒时间戳；后端为 LocalDateTime 字符串，在 toFriend 转换） - deleteTime?: number // 删除好友时间（毫秒时间戳；后端为 LocalDateTime 字符串，在 toFriend 转换） + addTime?: number // 添加好友时间（毫秒时间戳；后端为 LocalDateTime 字符串，在 convertFriend 转换） + deleteTime?: number // 删除好友时间（毫秒时间戳；后端为 LocalDateTime 字符串，在 convertFriend 转换） } // ==================== 用户名片 ====================