✨ feat(im): 增加频道消息的已读状态

YunaiV
1 parent b18281fc
Showing 11 changed files with 185 additions and 9 deletions
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/manager/message/ImChannelMessageManagerController.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImChannelMessageController.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/channel/ImChannelMessagePullRespVO.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/ImChannelMessageReadRedisDAO.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageService.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImChannelMessageServiceImpl.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/ImChannelMessageDTO.java
yudao-module-im/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java
@@ -72,7 +72,8 @@ public class ImChannelMessageManagerController {
                 convertSet(pageResult.getList(), ImChannelMessageDO::getMaterialId));
         return success(BeanUtils.toBean(pageResult, ImChannelMessageRespVO.class, vo -> {
             MapUtils.findAndThen(channelMap, vo.getChannelId(), c -> vo.setChannelName(c.getName()));
-            MapUtils.findAndThen(materialMap, vo.getMaterialId(), m -> vo.setMaterialTitle(m.getTitle()));
+            MapUtils.findAndThen(materialMap, vo.getMaterialId(),
+                    material -> vo.setMaterialTitle(material.getTitle()).setMaterialCoverUrl(material.getCoverUrl()));
         }));
     }
 package cn.iocoder.yudao.module.im.controller.admin.message;
+import cn.hutool.core.collection.CollUtil;
 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.module.im.controller.admin.message.vo.channel.ImChannelMessagePullRespVO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImChannelMessageDO;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
 import cn.iocoder.yudao.module.im.service.message.ImChannelMessageService;
 import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.annotation.Resource;
 import jakarta.validation.constraints.Max;
@@ -13,13 +16,17 @@ import jakarta.validation.constraints.Min;
 import jakarta.validation.constraints.PositiveOrZero;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PutMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;
 import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.getLoginUserId;
 @Tag(name = "用户 APP - IM 频道消息")
@@ -38,8 +45,31 @@ public class ImChannelMessageController {
             @RequestParam(value = "size", defaultValue = "100")
             @Min(value = 1, message = "size 必须大于 0")
             @Max(value = 200, message = "size 一次最多 200 条") Integer size) {
-        List<ImChannelMessageDO> list = channelMessageService.getMessageListForPull(getLoginUserId(), minId, size);
-        return success(BeanUtils.toBean(list, ImChannelMessagePullRespVO.class));
+        // 1. 拉取消息列表
+        Long userId = getLoginUserId();
+        List<ImChannelMessageDO> list = channelMessageService.getMessageListForPull(userId, minId, size);
+        if (CollUtil.isEmpty(list)) {
+            return success(Collections.emptyList());
+        }
+        // 2. 按 Redis 已读游标补 status；device A 已读后 device B 拉到这条不再算入未读
+        Map<Long, Long> readMaxByChannel = channelMessageService.getChannelReadMaxMessageIdMap(
+                userId, convertSet(list, ImChannelMessageDO::getChannelId));
+        return success(BeanUtils.toBean(list, ImChannelMessagePullRespVO.class, vo -> {
+            Long readMax = readMaxByChannel.get(vo.getChannelId());
+            vo.setStatus(readMax != null && readMax >= vo.getId()
+                    ? ImMessageStatusEnum.READ.getStatus()
+                    : ImMessageStatusEnum.UNREAD.getStatus());
+        }));
+    }
+
+    @PutMapping("/read")
+    @Operation(summary = "标记频道消息已读")
+    @Parameter(name = "channelId", description = "频道编号", required = true, example = "1")
+    @Parameter(name = "messageId", description = "已读到的消息编号", required = true, example = "100")
+    public CommonResult<Boolean> readChannelMessages(@RequestParam("channelId") Long channelId,
+                                                     @RequestParam("messageId") Long messageId) {
+        channelMessageService.readChannelMessages(getLoginUserId(), channelId, messageId);
+        return success(true);
     }
 }
@@ -24,6 +24,9 @@ public class ImChannelMessagePullRespVO {
     @Schema(description = "消息内容；payload JSON 快照", requiredMode = Schema.RequiredMode.REQUIRED)
     private String content;
+    @Schema(description = "当前用户的已读态；按 Redis 游标计算填充", requiredMode = Schema.RequiredMode.REQUIRED, example = "0")
+    private Integer status; // 参见 ImMessageStatusEnum 枚举类
+
     @Schema(description = "发送时间", requiredMode = Schema.RequiredMode.REQUIRED)
     private LocalDateTime sendTime;
@@ -16,6 +16,13 @@ public interface RedisKeyConstants {
     String GROUP_MESSAGE_READ = "im:group:message:read:%s";
     /**
+     * 频道消息已读位置
+     * KEY 格式：  im:channel:message:read:{channelId}
+     * VALUE 数据类型： Hash (field: userId, value: maxReadMessageId)
+     */
+    String CHANNEL_MESSAGE_READ = "im:channel:message:read:%s";
+
+    /**
      * 好友关系状态缓存（合并「是否好友」+「是否拉黑」两态）
      * <p>
      * KEY 格式：friend_state:{userId}_{friendUserId}
+package cn.iocoder.yudao.module.im.dal.redis.message;
+
+import cn.hutool.core.convert.Convert;
+import jakarta.annotation.Resource;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Repository;
+
+import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.CHANNEL_MESSAGE_READ;
+
+/**
+ * IM 频道消息已读位置 Redis DAO
+ *
+ * @author 芋道源码
+ */
+@Repository
+public class ImChannelMessageReadRedisDAO {
+
+    @Resource
+    private StringRedisTemplate stringRedisTemplate;
+
+    /**
+     * 更新用户在某频道的最大已读消息编号
+     *
+     * @param channelId    频道编号
+     * @param userId       用户编号
+     * @param maxMessageId 最大已读消息编号
+     */
+    public void updateReadMaxMessageId(Long channelId, Long userId, Long maxMessageId) {
+        String key = formatKey(channelId);
+        stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString());
+    }
+
+    /**
+     * 获取用户在某频道的最大已读消息编号
+     *
+     * @param channelId 频道编号
+     * @param userId    用户编号
+     * @return 最大已读消息编号；不存在则返回 null
+     */
+    public Long getReadMaxMessageId(Long channelId, Long userId) {
+        String key = formatKey(channelId);
+        Object val = stringRedisTemplate.opsForHash().get(key, userId.toString());
+        return Convert.toLong(val);
+    }
+
+    private static String formatKey(Long channelId) {
+        return String.format(CHANNEL_MESSAGE_READ, channelId);
+    }
+
+}
@@ -12,12 +12,12 @@ import java.util.Map;
 import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_MESSAGE_READ;
 /**
- * 群消息已读位置 Redis DAO
+ * IM 群消息已读位置 Redis DAO
  *
  * @author 芋道源码
  */
 @Repository
-public class GroupMessageReadRedisDAO {
+public class ImGroupMessageReadRedisDAO {
     @Resource
     private StringRedisTemplate stringRedisTemplate;
@@ -6,7 +6,9 @@ import cn.iocoder.yudao.module.im.controller.admin.manager.message.vo.channel.Im
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImChannelMessageDO;
 import jakarta.validation.Valid;
+import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 /**
  * IM 频道消息 Service 接口
@@ -27,6 +29,23 @@ public interface ImChannelMessageService {
      */
     List<ImChannelMessageDO> getMessageListForPull(Long userId, Long minId, Integer size);
+    /**
+     * 上报频道消息已读位置；同步推 READ 事件给自己多端
+     *
+     * @param userId    当前用户编号
+     * @param channelId 频道编号
+     * @param messageId 已读到的最大消息编号
+     */
+    void readChannelMessages(Long userId, Long channelId, Long messageId);
+
+    /**
+     * 批量查询用户在多个频道下的已读游标
+     *
+     * @param userId     当前用户编号
+     * @param channelIds 频道编号集合
+     * @return channelId → 已读到的最大消息编号（未读到的频道不出现在返回 map 中）
+     */
+    Map<Long, Long> getChannelReadMaxMessageIdMap(Long userId, Collection<Long> channelIds);
     // ==================== 管理后台 ====================
 package cn.iocoder.yudao.module.im.service.message;
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.lang.Assert;
+import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
@@ -9,18 +11,23 @@ import cn.iocoder.yudao.module.im.controller.admin.manager.message.vo.channel.Im
 import cn.iocoder.yudao.module.im.dal.dataobject.channel.ImChannelMaterialDO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImChannelMessageDO;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImChannelMessageMapper;
+import cn.iocoder.yudao.module.im.dal.redis.message.ImChannelMessageReadRedisDAO;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
 import cn.iocoder.yudao.module.im.service.channel.ImChannelMaterialService;
 import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
 import cn.iocoder.yudao.module.im.service.websocket.dto.ImChannelMessageDTO;
 import cn.iocoder.yudao.module.im.service.websocket.dto.message.MaterialMessage;
 import jakarta.annotation.Resource;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.validation.annotation.Validated;
 import java.time.LocalDateTime;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
 import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.IM_CHANNEL_MESSAGE_NOT_EXISTS;
@@ -36,11 +43,15 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
     @Resource
     private ImChannelMessageMapper channelMessageMapper;
+
     @Resource
     private ImChannelMaterialService channelMaterialService;
     @Resource
     private ImWebSocketService webSocketService;
+    @Resource
+    private ImChannelMessageReadRedisDAO channelMessageReadRedisDAO;
+
     // ==================== 用户端 ====================
     @Override
@@ -48,6 +59,47 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
         return channelMessageMapper.selectListByUserAndMinId(userId, minId, size);
     }
+    @Override
+    public Map<Long, Long> getChannelReadMaxMessageIdMap(Long userId, Collection<Long> channelIds) {
+        Map<Long, Long> result = new HashMap<>(channelIds.size());
+        for (Long channelId : channelIds) {
+            Long max = channelMessageReadRedisDAO.getReadMaxMessageId(channelId, userId);
+            if (max != null) {
+                result.put(channelId, max);
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public void readChannelMessages(Long userId, Long channelId, Long messageId) {
+        Assert.notNull(channelId, "频道编号不能为空");
+        Assert.notNull(messageId, "已读消息编号不能为空");
+        // 1. 已读位置未前进，直接返回
+        Long prevMaxMessageId = channelMessageReadRedisDAO.getReadMaxMessageId(channelId, userId);
+        if (prevMaxMessageId != null && prevMaxMessageId >= messageId) {
+            return;
+        }
+
+        // 2. 更新 Redis 频道已读位置
+        channelMessageReadRedisDAO.updateReadMaxMessageId(channelId, userId, messageId);
+
+        // 3. 异步推 READ 事件给自己多端同步
+        getSelf().readChannelMessageEvent(userId, channelId, messageId);
+    }
+
+    /**
+     * 发送频道已读 READ 事件给自己其他终端；频道无「给发送方刷回执」概念，不广播
+     */
+    @Async
+    public void readChannelMessageEvent(Long userId, Long channelId, Long readId) {
+        webSocketService.sendChannelMessageAsync(userId, ImChannelMessageDTO.ofRead(channelId, readId));
+    }
+
+    private ImChannelMessageServiceImpl getSelf() {
+        return SpringUtil.getBean(getClass());
+    }
+
     // ==================== 管理后台 ====================
     @Override
@@ -17,7 +17,7 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
 import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
-import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO;
+import cn.iocoder.yudao.module.im.dal.redis.message.ImGroupMessageReadRedisDAO;
 import cn.iocoder.yudao.module.im.enums.group.ImGroupMemberRoleEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
@@ -66,7 +66,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
     @Resource
     private ImGroupMessageMapper groupMessageMapper;
     @Resource
-    private GroupMessageReadRedisDAO groupMessageReadRedisDAO;
+    private ImGroupMessageReadRedisDAO groupMessageReadRedisDAO;
     @Resource
     private ImGroupService groupService;
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.im.service.websocket.dto;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImChannelMessageDO;
+import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
 import lombok.Data;
 import lombok.experimental.Accessors;
@@ -53,4 +54,17 @@ public class ImChannelMessageDTO {
         return BeanUtils.toBean(message, ImChannelMessageDTO.class);
     }
+    /**
+     * 构建已读同步 DTO；多端同步：通知自己的其他终端「我已经读了某频道」
+     *
+     * @param channelId 频道编号
+     * @param readId    已读位置（最大已读消息编号）
+     * @return 频道 DTO
+     */
+    public static ImChannelMessageDTO ofRead(Long channelId, Long readId) {
+        return new ImChannelMessageDTO()
+                .setId(readId).setType(ImMessageTypeEnum.READ.getType())
+                .setChannelId(channelId);
+    }
+
 }
@@ -10,7 +10,7 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
 import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
-import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO;
+import cn.iocoder.yudao.module.im.dal.redis.message.ImGroupMessageReadRedisDAO;
 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
@@ -56,7 +56,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
     @Mock
     private ImSensitiveWordService sensitiveWordService;
     @Mock
-    private GroupMessageReadRedisDAO groupMessageReadRedisDAO;
+    private ImGroupMessageReadRedisDAO groupMessageReadRedisDAO;
     @Mock
     private ImWebSocketService imWebSocketService;