🐛 fix(im): 私聊已读改用前端上报 messageId，修正 RECEIPT 语义

原 readPrivateMessages 走 select-then-update 两步：先按 status=UNREAD 查未读，再按 id in 翻转，maxReadId 取已 update 行的 max(id)。该实现有两个问题： 1. select 与 update 之间到达的消息会被一并标 READ，但 maxReadId 取的是 select 时刻的最大值，对方端拿到的「已读到哪条」语义偏弱，前端按 maxReadId 卡边界时也会与实际状态错位。 2. 接口与群聊 readGroupMessages(userId, groupId, messageId) 不对称。改造： - 接口签名增加 messageId（已读位置，含），与群聊接口对齐； - mapper 替换为单条 update SQL：id <= messageId AND status=UNREAD 直接翻转，消除两步竞态窗口； - READ / RECEIPT 推送的已读位置直接采用前端上报的 messageId； - 落地 ImPrivateMessageServiceImpl 原 TODO； - 同步更新 testReadMessages_success / testReadMessages_noUnread。

🐛 fix(im): 私聊已读改用前端上报 messageId，修正 RECEIPT 语义
原 readPrivateMessages 走 select-then-update 两步：先按 status=UNREAD 查未读，再按 id in 翻转，maxReadId 取已 update 行的 max(id)。该实现有两个问题： 1. select 与 update 之间到达的消息会被一并标 READ，但 maxReadId 取的是 select 时刻的最大值，对方端拿到的「已读到哪条」语义偏弱，前端按 maxReadId 卡边界时也会与实际状态错位。 2. 接口与群聊 readGroupMessages(userId, groupId, messageId) 不对称。改造： - 接口签名增加 messageId（已读位置，含），与群聊接口对齐； - mapper 替换为单条 update SQL：id <= messageId AND status=UNREAD 直接翻转，消除两步竞态窗口； - READ / RECEIPT 推送的已读位置直接采用前端上报的 messageId； - 落地 ImPrivateMessageServiceImpl 原 TODO； - 同步更新 testReadMessages_success / testReadMessages_noUnread。
YunaiV
1 parent ceb60978
Showing 5 changed files with 56 additions and 46 deletions
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImPrivateMessageMapper.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java
yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java
@@ -50,8 +50,10 @@ public class ImPrivateMessageController {
     @PutMapping("/read")
     @Operation(summary = "标记私聊消息已读")
     @Parameter(name = "receiverId", description = "接收方用户编号（对方）", required = true, example = "2")
-    public CommonResult<Boolean> readPrivateMessages(@RequestParam("receiverId") Long receiverId) {
-        privateMessageService.readPrivateMessages(getLoginUserId(), receiverId);
+    @Parameter(name = "messageId", description = "已读位置（含），通常是会话内最大消息编号", required = true, example = "100")
+    public CommonResult<Boolean> readPrivateMessages(@RequestParam("receiverId") Long receiverId,
+                                                     @RequestParam("messageId") Long messageId) {
+        privateMessageService.readPrivateMessages(getLoginUserId(), receiverId, messageId);
         return success(true);
     }
@@ -67,16 +67,25 @@ public interface ImPrivateMessageMapper extends BaseMapperX&lt;ImPrivateMessageDO&gt; 
                 .eq(ImPrivateMessageDO::getClientMessageId, clientMessageId));
     }
-    default List<ImPrivateMessageDO> selectListBySenderIdAndReceiverIdAndStatus(Long senderId, Long receiverId, Integer status) {
-        return selectList(new LambdaQueryWrapperX<ImPrivateMessageDO>()
+    /**
+     * 标记 (senderId → receiverId) 这条会话上、id <= maxMessageId 且 status = whereStatus 的消息为新状态
+     * <p>
+     * 用于「已读」语义：前端上报"我已读到 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<ImPrivateMessageDO>()
                 .eq(ImPrivateMessageDO::getSenderId, senderId)
                 .eq(ImPrivateMessageDO::getReceiverId, receiverId)
-                .eq(ImPrivateMessageDO::getStatus, status));
-    }
-
-    default void updateByIdsAndStatus(List<Long> ids, Integer whereStatus, ImPrivateMessageDO updateObj) {
-        update(updateObj, new LambdaQueryWrapperX<ImPrivateMessageDO>()
-                .in(ImPrivateMessageDO::getId, ids)
+                .le(ImPrivateMessageDO::getId, maxMessageId)
                 .eq(ImPrivateMessageDO::getStatus, whereStatus));
     }
@@ -34,11 +34,15 @@ public interface ImPrivateMessageService {
     /**
      * 标记私聊消息已读
+     * <p>
+     * 语义：将「对方发给当前用户、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);
     /**
      * 撤回私聊消息
 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<ImPrivateMessageDO> 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<Long> 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
@@ -153,21 +153,18 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
     @Test
     public void testReadMessages_success() {
-        // 准备：mock 未读消息（id=1 和 id=5，maxReadId 应为 5）
-        List<ImPrivateMessageDO> 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));
     }