From 711a692c3a490332fab5ef68dfd70837c2d299bb Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sun, 26 Apr 2026 09:41:44 +0800 Subject: [PATCH] 🐛 fix(im): 私聊已读改用前端上报 messageId，修正 RECEIPT 语义 --- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java | 6 ++++-- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java | 25 +++++++++++++++++-------- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java | 6 +++++- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java | 29 +++++++++++++---------------- yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java | 36 +++++++++++++++++------------------- 5 files changed, 56 insertions(+), 46 deletions(-) diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java index f6b25d6..1e51440 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java @@ -50,8 +50,10 @@ public class ImPrivateMessageController { @PutMapping("/read") @Operation(summary = "标记私聊消息已读") @Parameter(name = "receiverId", description = "接收方用户编号（对方）", required = true, example = "2") - public CommonResult readPrivateMessages(@RequestParam("receiverId") Long receiverId) { - privateMessageService.readPrivateMessages(getLoginUserId(), receiverId); + @Parameter(name = "messageId", description = "已读位置（含），通常是会话内最大消息编号", required = true, example = "100") + public CommonResult readPrivateMessages(@RequestParam("receiverId") Long receiverId, + @RequestParam("messageId") Long messageId) { + privateMessageService.readPrivateMessages(getLoginUserId(), receiverId, messageId); return success(true); } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java index 46c8630..d08b12f 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java @@ -67,16 +67,25 @@ public interface ImPrivateMessageMapper extends BaseMapperX .eq(ImPrivateMessageDO::getClientMessageId, clientMessageId)); } - default List selectListBySenderIdAndReceiverIdAndStatus(Long senderId, Long receiverId, Integer status) { - return selectList(new LambdaQueryWrapperX() + /** + * 标记 (senderId → receiverId) 这条会话上、id <= maxMessageId 且 status = whereStatus 的消息为新状态 + *

+ * 用于「已读」语义：前端上报"我已读到 maxMessageId"，由这一条 SQL 在 status=UNREAD 上幂等翻转， + * 避免"先 select 未读、再按 id in 更新"两步带来的竞态（select 后到达的消息被误标）。 + * + * @param senderId 发送方用户编号（对方） + * @param receiverId 接收方用户编号（当前用户） + * @param maxMessageId 已读位置（含） + * @param whereStatus 匹配的当前状态 + * @param updateObj 更新对象（仅设要变更的字段） + * @return 实际更新行数 + */ + default int updateBySenderIdAndReceiverIdAndIdLeAndStatus(Long senderId, Long receiverId, Long maxMessageId, + Integer whereStatus, ImPrivateMessageDO updateObj) { + return update(updateObj, new LambdaQueryWrapperX() .eq(ImPrivateMessageDO::getSenderId, senderId) .eq(ImPrivateMessageDO::getReceiverId, receiverId) - .eq(ImPrivateMessageDO::getStatus, status)); - } - - default void updateByIdsAndStatus(List ids, Integer whereStatus, ImPrivateMessageDO updateObj) { - update(updateObj, new LambdaQueryWrapperX() - .in(ImPrivateMessageDO::getId, ids) + .le(ImPrivateMessageDO::getId, maxMessageId) .eq(ImPrivateMessageDO::getStatus, whereStatus)); } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java index 0f21bd1..2f5eae2 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java @@ -34,11 +34,15 @@ public interface ImPrivateMessageService { /** * 标记私聊消息已读 + *

+ * 语义：将「对方发给当前用户、id <= messageId 的未读消息」一次性翻转为已读， + * 与群聊 readGroupMessages 对称，避免"select-then-update"两步式带来的竞态。 * * @param userId 当前用户编号 * @param receiverId 接收方用户编号（对方） + * @param messageId 已读位置（含），通常是前端会话内最大消息 id */ - void readPrivateMessages(Long userId, Long receiverId); + void readPrivateMessages(Long userId, Long receiverId, Long messageId); /** * 撤回私聊消息 diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java index ff6357a..6048620 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.module.im.service.message; -import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.lang.Assert; import cn.hutool.core.util.IdUtil; import cn.hutool.core.util.ObjUtil; import cn.iocoder.yudao.framework.common.util.json.JsonUtils; @@ -26,7 +26,6 @@ import java.time.LocalDateTime; import java.util.List; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; -import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList; import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*; import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*; @@ -97,26 +96,24 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService { return messages; } - // TODO @AI：考虑到更稳妥，是不是前端传递 messageId？不然恰好发过来，然后读取了。【后续在优化】 + // TODO DONE @AI：考虑到更稳妥，是不是前端传递 messageId？不然恰好发过来，然后读取了。【后续在优化】 @Override - public void readPrivateMessages(Long userId, Long receiverId) { - // 1.1 查询未读消息列表 - List unreadMessages = privateMessageMapper.selectListBySenderIdAndReceiverIdAndStatus( - receiverId, userId, ImMessageStatusEnum.UNREAD.getStatus()); - if (CollUtil.isEmpty(unreadMessages)) { + public void readPrivateMessages(Long userId, Long receiverId, Long messageId) { + Assert.notNull(messageId, "已读消息编号不能为空"); + // 1. 一步翻转：把 (receiverId → userId) 这条会话上、id <= messageId 的未读消息全部置为已读 + // 仅 UNREAD 行被命中，避免覆盖已撤回/已读的状态；select-then-update 拆成单条 SQL 后也消除了竞态窗口 + int updated = privateMessageMapper.updateBySenderIdAndReceiverIdAndIdLeAndStatus( + receiverId, userId, messageId, ImMessageStatusEnum.UNREAD.getStatus(), + new ImPrivateMessageDO().setStatus(ImMessageStatusEnum.READ.getStatus())); + if (updated == 0) { return; } - // 1.2 根据 id in + status 条件更新，避免并发问题 - List messageIds = convertList(unreadMessages, ImPrivateMessageDO::getId); - privateMessageMapper.updateByIdsAndStatus(messageIds, ImMessageStatusEnum.UNREAD.getStatus(), - new ImPrivateMessageDO().setStatus(ImMessageStatusEnum.READ.getStatus())); - // 2. 异步发送 READ + RECEIPT 事件（携带已读位置） - Long maxReadId = CollUtil.max(messageIds); + // 2. 异步发送 READ + RECEIPT 事件（已读位置以前端上报为准，与多端 / 对方 UI 显示一致） imWebSocketService.sendPrivateMessageAsync(userId, - ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId)); + ImPrivateMessageDTO.ofRead(userId, receiverId, messageId)); imWebSocketService.sendPrivateMessageAsync(receiverId, - ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId)); + ImPrivateMessageDTO.ofReceipt(userId, receiverId, messageId)); } @Override diff --git a/yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java b/yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java index ae40b55..a6eb305 100644 --- a/yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java +++ b/yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java @@ -153,21 +153,18 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testReadMessages_success() { - // 准备：mock 未读消息（id=1 和 id=5，maxReadId 应为 5） - List unreadMessages = List.of( - ImPrivateMessageDO.builder().id(1L).senderId(2L).receiverId(1L) - .status(ImMessageStatusEnum.UNREAD.getStatus()).build(), - ImPrivateMessageDO.builder().id(5L).senderId(2L).receiverId(1L) - .status(ImMessageStatusEnum.UNREAD.getStatus()).build() - ); - when(privateMessageMapper.selectListBySenderIdAndReceiverIdAndStatus(2L, 1L, - ImMessageStatusEnum.UNREAD.getStatus())).thenReturn(unreadMessages); + // 准备：前端上报已读到 messageId=5；mapper 返回更新行数 2 表示有未读被翻转 + when(privateMessageMapper.updateBySenderIdAndReceiverIdAndIdLeAndStatus( + eq(2L), eq(1L), eq(5L), + eq(ImMessageStatusEnum.UNREAD.getStatus()), any(ImPrivateMessageDO.class))) + .thenReturn(2); // 调用 - privateMessageService.readPrivateMessages(1L, 2L); + privateMessageService.readPrivateMessages(1L, 2L, 5L); // 断言：更新了消息状态 - verify(privateMessageMapper).updateByIdsAndStatus(eq(List.of(1L, 5L)), + verify(privateMessageMapper).updateBySenderIdAndReceiverIdAndIdLeAndStatus( + eq(2L), eq(1L), eq(5L), eq(ImMessageStatusEnum.UNREAD.getStatus()), any(ImPrivateMessageDO.class)); // 断言：发送了 READ + RECEIPT 事件，payload 字段正确 @@ -182,13 +179,13 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest { assertEquals(ImMessageTypeEnum.READ.getType(), readPayload.getType()); assertEquals(1L, readPayload.getSenderId()); assertEquals(2L, readPayload.getReceiverId()); - assertEquals(5L, readPayload.getId(), "READ id 应为 maxReadId"); + assertEquals(5L, readPayload.getId(), "READ id 应为前端上报的 messageId"); // 第二次：发给对方的 RECEIPT 事件 assertEquals(2L, userCaptor.getAllValues().get(1)); ImPrivateMessageDTO receiptPayload = contentCaptor.getAllValues().get(1); assertEquals(ImMessageTypeEnum.RECEIPT.getType(), receiptPayload.getType()); - assertEquals(5L, receiptPayload.getId(), "RECEIPT id 应为 maxReadId"); + assertEquals(5L, receiptPayload.getId(), "RECEIPT id 应为前端上报的 messageId"); } // ========== 撤回测试 ========== @@ -295,15 +292,16 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testReadMessages_noUnread() { - // 准备：没有未读消息 - when(privateMessageMapper.selectListBySenderIdAndReceiverIdAndStatus(2L, 1L, - ImMessageStatusEnum.UNREAD.getStatus())).thenReturn(List.of()); + // 准备：mapper 返回 0 表示 id <= messageId 范围内没有未读消息 + when(privateMessageMapper.updateBySenderIdAndReceiverIdAndIdLeAndStatus( + eq(2L), eq(1L), eq(5L), + eq(ImMessageStatusEnum.UNREAD.getStatus()), any(ImPrivateMessageDO.class))) + .thenReturn(0); // 调用 - privateMessageService.readPrivateMessages(1L, 2L); + privateMessageService.readPrivateMessages(1L, 2L, 5L); - // 断言：不更新、不推送 - verify(privateMessageMapper, never()).updateByIdsAndStatus(anyList(), anyInt(), any(ImPrivateMessageDO.class)); + // 断言：不推送 READ / RECEIPT verify(imWebSocketService, never()).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class)); } -- libgit2 0.21.4