✨ feat(im): 优化私聊消息已读的处理逻辑

YunaiV
1 parent dc7be585
Showing 10 changed files with 134 additions and 64 deletions
yudao-module-im/yudao-module-im-api/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageSceneEnum.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImPrivateMessageController.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageService.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReadMessage.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageRecallMessage.java
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReceiptMessage.java
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImplTest.java
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImWebSocketEventTest.java
+package cn.iocoder.yudao.module.im.enums.message;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IM 消息场景枚举
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum ImMessageSceneEnum implements ArrayValuable<Integer> {
+
+    PRIVATE(1, "私聊"),
+    GROUP(2, "群聊");
+
+    public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImMessageSceneEnum::getScene).toArray(Integer[]::new);
+
+    /**
+     * 场景
+     */
+    private final Integer scene;
+    /**
+     * 名字
+     */
+    private final String name;
+
+    @Override
+    public Integer[] array() {
+        return ARRAYS;
+    }
+
+}
@@ -51,9 +51,9 @@ public class ImPrivateMessageController {
     @PutMapping("/read")
     @Operation(summary = "标记私聊消息已读")
-    @Parameter(name = "friendId", description = "好友用户编号", required = true, example = "2")
-    public CommonResult<Boolean> readPrivateMessages(@RequestParam("friendId") Long friendId) {
-        privateMessageService.readPrivateMessages(getLoginUserId(), friendId);
+    @Parameter(name = "receiverId", description = "接收方用户编号（对方）", required = true, example = "2")
+    public CommonResult<Boolean> readPrivateMessages(@RequestParam("receiverId") Long receiverId) {
+        privateMessageService.readPrivateMessages(getLoginUserId(), receiverId);
         return success(true);
     }
@@ -34,10 +34,10 @@ public interface ImPrivateMessageService {
     /**
      * 标记私聊消息已读
      *
-     * @param userId   当前用户编号
-     * @param friendId 好友用户编号
+     * @param userId     当前用户编号
+     * @param receiverId 接收方用户编号（对方）
      */
-    void readPrivateMessages(Long userId, Long friendId);
+    void readPrivateMessages(Long userId, Long receiverId);
     /**
      * 撤回私聊消息
 package cn.iocoder.yudao.module.im.service.message;
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
+import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageSendReqVO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImPrivateMessageMapper;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
 import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
 import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
 import cn.iocoder.yudao.module.im.websocket.ImMessageReadMessage;
-import cn.iocoder.yudao.module.im.websocket.ImMessageReceiptMessage;
 import cn.iocoder.yudao.module.im.websocket.ImMessageRecallMessage;
+import cn.iocoder.yudao.module.im.websocket.ImMessageReceiptMessage;
 import cn.iocoder.yudao.module.im.websocket.ImPrivateMessageSendMessage;
-import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Async;
@@ -25,6 +27,7 @@ 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.*;
 /**
@@ -91,31 +94,21 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
         return messages;
     }
-    // TODO @AI：应该是将 friendId 改成 receiverId，毕竟消息是发给对方的
     @Override
-    public void readPrivateMessages(Long userId, Long friendId) {
-        // 1. 批量更新消息状态为已读
-        // TODO @AI：先查询；然后在 update；避免锁的问题；
-        // TODO @AI：注意，update 的时候，也要 id in、status =，最后 updateObj 传入；避免并发问题；
-        int updated = privateMessageMapper.updateStatusToRead(userId, friendId);
-        if (updated == 0) {
-            return; // 没有需要更新的消息
+    public void readPrivateMessages(Long userId, Long receiverId) {
+        // 1.1 查询未读消息列表
+        List<ImPrivateMessageDO> unreadMessages = privateMessageMapper.selectListBySenderIdAndReceiverIdAndStatus(
+                receiverId, userId, ImMessageStatusEnum.UNREAD.getStatus());
+        if (CollUtil.isEmpty(unreadMessages)) {
+            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()));
-        // TODO @AI：继续 async ！
-        // 2.1 发送 READ 事件给自己的其他终端（多端同步）
-        ImMessageReadMessage readMessage = new ImMessageReadMessage()
-                .setFriendId(friendId.toString())
-                .setMessageScene("private");
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
-                ImMessageReadMessage.TYPE, readMessage);
-
-        // 2.2 发送 RECEIPT 事件给对方（已读回执）
-        ImMessageReceiptMessage receiptMessage = new ImMessageReceiptMessage()
-                .setUserId(userId.toString())
-                .setMessageScene("private");
-        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), friendId,
-                ImMessageReceiptMessage.TYPE, receiptMessage);
+        // 2. 异步发送 READ + RECEIPT 事件
+        getSelf().readPrivateMessageEvent(userId, receiverId);
     }
     // DONE @芋艿：已 review，逻辑正确
@@ -142,9 +135,7 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
         // 3. 发送 RECALL 事件
         ImMessageRecallMessage recallMessage = new ImMessageRecallMessage()
-                .setMessageId(messageId.toString())
-                .setSenderId(userId.toString())
-                .setMessageScene("private");
+                .setMessageId(messageId).setSenderId(userId).setScene(ImMessageSceneEnum.PRIVATE.getScene());
         // 3.1 通知接收方
         webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getReceiverId(),
                 ImMessageRecallMessage.TYPE, recallMessage);
@@ -162,11 +153,33 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
         // 推送给接收方
         webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getReceiverId(),
                 ImPrivateMessageSendMessage.TYPE, websocketMessage);
+
         // 推送给发送方自己的其他终端（多端同步）
         webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(),
                 ImPrivateMessageSendMessage.TYPE, websocketMessage);
     }
+    /**
+     * 发送已读 + 已读回执 WebSocket 事件
+     *
+     * @param userId     当前用户编号
+     * @param receiverId 对方用户编号
+     */
+    @Async
+    public void readPrivateMessageEvent(Long userId, Long receiverId) {
+        // 1. 发送 READ 事件给自己的其他终端（多端同步）
+        ImMessageReadMessage readMessage = new ImMessageReadMessage()
+                .setReceiverId(receiverId).setScene(ImMessageSceneEnum.PRIVATE.getScene());
+        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
+                ImMessageReadMessage.TYPE, readMessage);
+
+        // 2. 发送 RECEIPT 事件给对方（已读回执）
+        ImMessageReceiptMessage receiptMessage = new ImMessageReceiptMessage()
+                .setReceiverId(userId).setScene(ImMessageSceneEnum.PRIVATE.getScene());
+        webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), receiverId,
+                ImMessageReceiptMessage.TYPE, receiptMessage);
+    }
+
     private ImPrivateMessageServiceImpl getSelf() {
         return SpringUtil.getBean(getClass());
     }
@@ -17,16 +17,18 @@ public class ImMessageReadMessage {
     public static final String TYPE = "im-message-read";
     /**
-     * 私聊场景：好友编号
+     * 私聊场景：接收方编号
      */
-    private String friendId;
+    private Long receiverId;
     /**
      * 群聊场景：群编号
      */
-    private String groupId;
+    private Long groupId;
     /**
-     * 消息场景：private / group
+     * 消息场景
+     *
+     * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum}
      */
-    private String messageScene;
+    private Integer scene;
 }
@@ -17,18 +17,20 @@ public class ImMessageRecallMessage {
     /**
      * 消息编号
      */
-    private String messageId;
+    private Long messageId;
     /**
      * 发送人编号
      */
-    private String senderId;
+    private Long senderId;
     /**
      * 群聊场景：群编号
      */
-    private String groupId;
+    private Long groupId;
     /**
-     * 消息场景：private / group
+     * 消息场景
+     *
+     * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum}
      */
-    private String messageScene;
+    private Integer scene;
 }
@@ -17,23 +17,30 @@ public class ImMessageReceiptMessage {
     public static final String TYPE = "im-message-receipt";
+    /**
+     * 消息场景
+     *
+     * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum}
+     */
+    private Integer scene;
+
     // ========== 私聊场景 ==========
     /**
      * 私聊场景：已读方的用户编号
      */
-    private String userId;
+    private Long receiverId;
     // ========== 群聊场景 ==========
     /**
      * 群聊场景：消息编号
      */
-    private String messageId;
+    private Long messageId;
     /**
      * 群聊场景：群编号
      */
-    private String groupId;
+    private Long groupId;
     /**
      * 群聊场景：已读人数
      */
@@ -43,11 +50,4 @@ public class ImMessageReceiptMessage {
      */
     private Integer receiptStatus;
-    // ========== 通用 ==========
-
-    /**
-     * 消息场景：private / group
-     */
-    private String messageScene;
-
 }
@@ -168,7 +168,7 @@ public class ImGroupMessageServiceImplTest {
                 .thenReturn(List.of(beforeJoin, afterJoin));
         // 调用
-        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessages(1L, 0L, 100);
+        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessageList(1L, 0L, 100);
         // 断言：入群前消息不可见
         assertEquals(1, result.size());
@@ -197,7 +197,7 @@ public class ImGroupMessageServiceImplTest {
                 .thenReturn(List.of(beforeQuit, afterQuit));
         // 调用
-        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessages(1L, 0L, 100);
+        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessageList(1L, 0L, 100);
         // 断言：退群后消息不可见
         assertEquals(1, result.size());
@@ -226,7 +226,7 @@ public class ImGroupMessageServiceImplTest {
                 .thenReturn(List.of(directedMsg, allMsg));
         // 调用
-        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessages(1L, 0L, 100);
+        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessageList(1L, 0L, 100);
         // 断言：定向接收的消息用户 1 看不到，只能看到全员消息
         assertEquals(1, result.size());
@@ -250,7 +250,7 @@ public class ImGroupMessageServiceImplTest {
                 .thenReturn(List.of(atMsg));
         // 调用
-        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessages(1L, 0L, 100);
+        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessageList(1L, 0L, 100);
         // 断言：@ 字段正确返回
         assertEquals(1, result.size());
@@ -137,7 +137,7 @@ public class ImPrivateMessageServiceImplTest {
         when(imPrivateMessageMapper.selectListByMinId(1L, 0L, 100)).thenReturn(mockMessages);
         // 调用
-        List<ImPrivateMessageDO> result = privateMessageService.pullPrivateMessages(1L, 0L, 100);
+        List<ImPrivateMessageDO> result = privateMessageService.pullPrivateMessageList(1L, 0L, 100);
         // 断言
         assertEquals(2, result.size());
@@ -147,7 +147,7 @@ public class ImPrivateMessageServiceImplTest {
     public void testPullMessages_sizeExceeded() {
         // 调用并断言
         ServiceException exception = assertThrows(ServiceException.class,
-                () -> privateMessageService.pullPrivateMessages(1L, 0L, 1001));
+                () -> privateMessageService.pullPrivateMessageList(1L, 0L, 1001));
         assertEquals(MESSAGE_PULL_SIZE_EXCEEDED.getCode(), exception.getCode());
     }
@@ -155,13 +155,21 @@ public class ImPrivateMessageServiceImplTest {
     @Test
     public void testReadMessages_success() {
-        // 准备
-        when(imPrivateMessageMapper.updateStatusToRead(1L, 2L)).thenReturn(3);
+        // 准备：mock 未读消息列表
+        List<ImPrivateMessageDO> unreadMessages = List.of(
+                ImPrivateMessageDO.builder().id(1L).senderId(2L).receiverId(1L)
+                        .status(ImMessageStatusEnum.UNREAD.getStatus()).build(),
+                ImPrivateMessageDO.builder().id(2L).senderId(2L).receiverId(1L)
+                        .status(ImMessageStatusEnum.UNREAD.getStatus()).build()
+        );
+        when(imPrivateMessageMapper.selectListBySenderIdAndReceiverIdAndStatus(2L, 1L,
+                ImMessageStatusEnum.UNREAD.getStatus())).thenReturn(unreadMessages);
         // 调用
         privateMessageService.readPrivateMessages(1L, 2L);
-        // 断言：发送了 READ 和 RECEIPT 事件
+        // 断言：更新了消息状态 + 发送了 READ 和 RECEIPT 事件
+        verify(imPrivateMessageMapper).update(any(ImPrivateMessageDO.class), any());
         verify(webSocketMessageSender, times(2)).sendObject(anyInt(), anyLong(), anyString(), any());
     }
@@ -3,6 +3,7 @@ package cn.iocoder.yudao.module.im.service.message;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageSendReqVO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImPrivateMessageMapper;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
 import cn.iocoder.yudao.module.im.websocket.ImMessageReadMessage;
 import cn.iocoder.yudao.module.im.websocket.ImMessageReceiptMessage;
 import cn.iocoder.yudao.module.im.websocket.ImMessageRecallMessage;
@@ -17,6 +18,8 @@ import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.List;
+
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
@@ -82,8 +85,13 @@ public class ImWebSocketEventTest {
     @Test
     public void testReadMessage_readAndReceiptEvents() {
-        // 准备
-        when(imPrivateMessageMapper.updateStatusToRead(1L, 2L)).thenReturn(1);
+        // 准备：mock 未读消息
+        List<ImPrivateMessageDO> unreadMessages = List.of(
+                ImPrivateMessageDO.builder().id(1L).senderId(2L).receiverId(1L)
+                        .status(ImMessageStatusEnum.UNREAD.getStatus()).build()
+        );
+        when(imPrivateMessageMapper.selectListBySenderIdAndReceiverIdAndStatus(2L, 1L,
+                ImMessageStatusEnum.UNREAD.getStatus())).thenReturn(unreadMessages);
         // 调用
         privateMessageService.readPrivateMessages(1L, 2L);