diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java index a65780f..f405316 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java @@ -8,6 +8,7 @@ import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum; import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum; import org.apache.ibatis.annotations.Mapper; +import java.time.LocalDateTime; import java.util.List; /** @@ -30,6 +31,7 @@ public interface ImGroupMessageMapper extends BaseMapperX { QueryWrapperX wrapper = new QueryWrapperX<>(); wrapper.in("group_id", groupIds) .gt("id", minId) + .ne("status", ImMessageStatusEnum.RECALL.getStatus()) .orderByAsc("id"); wrapper.limitN(size); return selectList(wrapper); @@ -38,16 +40,18 @@ public interface ImGroupMessageMapper extends BaseMapperX { /** * 查询群聊历史消息（游标拉取） * - * @param groupId 群编号 - * @param maxId 起始消息 id（不含），为空则从最新开始 - * @param limit 拉取数量 + * @param groupId 群编号 + * @param maxId 起始消息 id（不含），为空则从最新开始 + * @param limit 拉取数量 + * @param joinTime 入群时间，仅返回入群之后的消息 * @return 消息列表（按 id 倒序） */ - default List selectHistoryList(Long groupId, Long maxId, Integer limit) { + default List selectHistoryList(Long groupId, Long maxId, Integer limit, LocalDateTime joinTime) { QueryWrapperX wrapper = new QueryWrapperX<>(); wrapper.eq("group_id", groupId) .ne("status", ImMessageStatusEnum.RECALL.getStatus()) .lt(maxId != null, "id", maxId) + .ge(joinTime != null, "send_time", joinTime) .orderByDesc("id"); wrapper.limitN(limit); return selectList(wrapper); @@ -60,22 +64,6 @@ public interface ImGroupMessageMapper extends BaseMapperX { } /** - * 获取群内最新消息 id - * - * @param groupId 群编号 - * @return 最新消息 id，无消息时返回 null - */ - default Long selectMaxIdByGroupId(Long groupId) { - QueryWrapperX wrapper = new QueryWrapperX<>(); - wrapper.eq("group_id", groupId) - .orderByDesc("id") - .select("id"); - wrapper.limitN(1); - ImGroupMessageDO msg = selectOne(wrapper); - return msg == null ? null : msg.getId(); - } - - /** * 查询群内指定范围内待回执的消息 *

* 仅在用户"已读位置前进"时调用，避免全量扫描： @@ -86,11 +74,11 @@ public interface ImGroupMessageMapper extends BaseMapperX { * @param maxId 结束消息 id（含，本次已读位置） * @return 待回执消息列表 */ - default List selectPendingReceiptMessagesInRange(Long groupId, Long minId, Long maxId) { + default List selectListByGroupIdAndPendingReceipt(Long groupId, Long minId, Long maxId) { return selectList(new LambdaQueryWrapperX() .eq(ImGroupMessageDO::getGroupId, groupId) .eq(ImGroupMessageDO::getReceiptStatus, ImGroupMessageReceiptStatusEnum.PENDING.getStatus()) - .gt(ImGroupMessageDO::getId, minId) + .gt(minId != null, ImGroupMessageDO::getId, minId) .le(ImGroupMessageDO::getId, maxId) .ne(ImGroupMessageDO::getStatus, ImMessageStatusEnum.RECALL.getStatus())); } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java index 53e5498..9e07fbb 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java @@ -8,12 +8,11 @@ package cn.iocoder.yudao.module.im.dal.redis; */ public interface RedisKeyConstants { - // TODO @AI：这个 key 可能要改下； /** - * 群已读位置 - * KEY 格式： im:group:read:{groupId} + * 群消息已读位置 + * KEY 格式： im:group:message:read:{groupId} * VALUE 数据类型： Hash (field: userId, value: maxReadMessageId) */ - String GROUP_READ_POSITION = "im:group:read:%s"; + String GROUP_MESSAGE_READ = "im:group:message:read:%s"; } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupReadPositionRedisDAO.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupMessageReadRedisDAO.java index 8dd9dfb..3bc2159 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupReadPositionRedisDAO.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupMessageReadRedisDAO.java @@ -1,65 +1,68 @@ 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 java.util.HashMap; import java.util.Map; -import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_READ_POSITION; +import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_MESSAGE_READ; -// TODO @AI：需要改成，GroupMessageReadRedisDAO /** - * 群已读位置 Redis DAO - *

- * 采用 Redis Hash 存储：key=im:group:read:{groupId}, field=userId, value=maxReadMessageId TODO @AI：这里不用注释； + * 群消息已读位置 Redis DAO * * @author 芋道源码 */ @Repository -public class GroupReadPositionRedisDAO { +public class GroupMessageReadRedisDAO { @Resource private StringRedisTemplate stringRedisTemplate; - // TODO @AI：updateReadMaxMessageId； /** - * 更新用户在某群的已读位置 + * 更新用户在某群的最大已读消息编号 * * @param groupId 群编号 * @param userId 用户编号 * @param maxMessageId 最大已读消息编号 */ - public void updateReadPosition(Long groupId, Long userId, Long maxMessageId) { - // TODO @AI：这里搞个 formatKey 方法，可以复用； - String key = String.format(GROUP_READ_POSITION, groupId); - // 只在新值大于旧值时更新 - Object currentVal = stringRedisTemplate.opsForHash().get(key, userId.toString()); - if (currentVal == null || Long.parseLong(currentVal.toString()) < maxMessageId) { - // TODO @AI：可以 put 的时候，就是 Long，Long 么？ - stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString()); - } + public void updateReadMaxMessageId(Long groupId, Long userId, Long maxMessageId) { + String key = formatKey(groupId); + stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString()); } - // TODO @AI：不存在，则返回 null 更合理； - // TODO @AI：getReadMaxMessageId； /** - * 获取用户在某群的已读位置 + * 获取用户在某群的最大已读消息编号 + * + * @param groupId 群编号 + * @param userId 用户编号 + * @return 最大已读消息编号，不存在则返回 null */ - public Long getReadPosition(Long groupId, Long userId) { - String key = String.format(GROUP_READ_POSITION, groupId); + public Long getReadMaxMessageId(Long groupId, Long userId) { + String key = formatKey(groupId); Object val = stringRedisTemplate.opsForHash().get(key, userId.toString()); - return val == null ? 0L : Long.parseLong(val.toString()); + return Convert.toLong(val); } - // TODO @AI：参考这个风格，优化下； /** * 获取某群所有用户的已读位置 + * + * @param groupId 群编号 + * @return userId → maxReadMessageId 映射 */ - public Map getAllReadPositions(Long groupId) { - // TODO @AI：可以返回的时候，就是 Long，Long 么？ - String key = String.format(GROUP_READ_POSITION, groupId); - return stringRedisTemplate.opsForHash().entries(key); + public Map getReadMaxMessageIdMap(Long groupId) { + String key = formatKey(groupId); + Map entries = stringRedisTemplate.opsForHash().entries(key); + // 转换为 Long → Long 的 Map + Map result = new HashMap<>(entries.size()); + entries.forEach((k, v) -> result.put(Long.parseLong(k.toString()), Long.parseLong(v.toString()))); + return result; + } + + private static String formatKey(Long groupId) { + return String.format(GROUP_MESSAGE_READ, groupId); } } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java index e7e525f..286e37b 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java @@ -15,7 +15,7 @@ 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.dataobject.message.content.RecallMessage; import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper; -import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO; +import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO; 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; @@ -35,6 +35,7 @@ import java.util.*; 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.framework.common.util.collection.CollectionUtils.convertSet; import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*; import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_MAX_PULL_SIZE; import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_RECALL_TIMEOUT_MINUTES; @@ -51,9 +52,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { @Resource private ImGroupMessageMapper groupMessageMapper; - // TODO @AI：变量名，可能要改下； @Resource - private GroupReadPositionRedisDAO groupReadPositionRedisDAO; + private GroupMessageReadRedisDAO groupMessageReadRedisDAO; @Resource private ImGroupService groupService; @@ -77,18 +77,16 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { } // 1.2 校验群存在、发送人仍在群中 groupService.validateGroupExists(reqVO.getGroupId()); - validateMemberInGroup(reqVO.getGroupId(), senderId); + groupMemberService.validateMemberInGroup(reqVO.getGroupId(), senderId); // 1.3 文本消息敏感词过滤 if (ImMessageTypeEnum.TEXT.getType().equals(reqVO.getType())) { sensitiveWordService.validateText(reqVO.getContent()); } // 2. 构建并保存消息 - // TODO @AI：content 要 json 下； ImGroupMessageDO message = BeanUtils.toBean(reqVO, ImGroupMessageDO.class, m -> m - .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus()) - .setSendTime(LocalDateTime.now()) - .setReceiptStatus(Boolean.TRUE.equals(reqVO.getNeedReceipt()) + .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now()) + .setReceiptStatus(Boolean.TRUE.equals(reqVO.getReceipt()) ? ImGroupMessageReceiptStatusEnum.PENDING.getStatus() : ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus())); groupMessageMapper.insert(message); @@ -114,7 +112,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { List messages = groupMessageMapper.selectListByMinId(groupIds, minId, size); // 3. 过滤：入群前不可见、退群后不可见、定向接收过滤 - // TODO @AI：缺少各种逻辑； + // TODO @AI：Java 层过滤在 SQL limit 之后，可能导致返回数量不足 size（跨群场景 joinTime 各异，无法下沉 SQL）【后续优化】 Map memberMap = CollUtil.toMap(members, new HashMap<>(), ImGroupMemberDO::getGroupId); List result = messages.stream() .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId)) @@ -124,39 +122,29 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { return result; } - // TODO @AI：增加 messageId 参数；允许空，则查询最大的一条； @Override - public void readGroupMessages(Long userId, Long groupId) { - // 1.1 校验用户在群中（权限校验） - validateMemberInGroup(groupId, userId); - // 1.2 获取群内最新消息 id（倒序 LIMIT 1，避免全量扫描） - Long newMaxId = groupMessageMapper.selectMaxIdByGroupId(groupId); - if (newMaxId == null) { - return; - } - // 1.3 已读位置未前进，直接返回 - // TODO @AI：1.3 和 2 是不是可以合并？如果 update 失败，则直接回退；不然读取两次，貌似有 cas 问题； - Long prevMaxId = groupReadPositionRedisDAO.getReadPosition(groupId, userId); - if (newMaxId <= prevMaxId) { + public void readGroupMessages(Long userId, Long groupId, Long messageId) { + Assert.notNull(messageId, "已读消息编号不能为空"); + // 1. 校验用户在群中（权限校验） + groupMemberService.validateMemberInGroup(groupId, userId); + + // 2. 已读位置未前进，直接返回 + Long prevMaxMessageId = groupMessageReadRedisDAO.getReadMaxMessageId(groupId, userId); + if (prevMaxMessageId != null && prevMaxMessageId >= messageId) { return; } + // 3. 更新 Redis 群已读位置 + groupMessageReadRedisDAO.updateReadMaxMessageId(groupId, userId, messageId); - // 2. 更新 Redis 群已读位置 - groupReadPositionRedisDAO.updateReadPosition(groupId, userId, newMaxId); - - // 3. 异步发送 READ 事件 + 刷新范围内的群回执 - getSelf().readGroupMessageEvent(userId, groupId, prevMaxId, newMaxId); + // 4. 异步发送 READ 事件 + 刷新范围内的群回执 + getSelf().readGroupMessageEvent(userId, groupId, prevMaxMessageId, messageId); } @Override @Transactional(rollbackFor = Exception.class) public ImGroupMessageDO recallGroupMessage(Long userId, Long messageId) { // 1.1 校验消息存在 - // TODO @AI：抽成一个 validateXXXExists 方法； - ImGroupMessageDO message = groupMessageMapper.selectById(messageId); - if (message == null) { - throw exception(MESSAGE_NOT_EXISTS); - } + ImGroupMessageDO message = validateGroupMessageExists(messageId); // 1.2 只能撤回自己发送的消息 if (ObjUtil.notEqual(message.getSenderId(), userId)) { throw exception(MESSAGE_RECALL_DENIED); @@ -170,7 +158,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { throw exception(MESSAGE_RECALL_TIMEOUT, MESSAGE_RECALL_TIMEOUT_MINUTES); } // 1.5 校验撤回人仍在群中 - validateMemberInGroup(message.getGroupId(), userId); + groupMemberService.validateMemberInGroup(message.getGroupId(), userId); // 2. 更新原消息状态为撤回 groupMessageMapper.updateById(new ImGroupMessageDO().setId(messageId) @@ -193,7 +181,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { @Override public List getGroupReadUsers(Long userId, Long groupId, Long messageId) { // 1.1 校验用户在群中（权限校验） - validateMemberInGroup(groupId, userId); + groupMemberService.validateMemberInGroup(groupId, userId); // 1.2 获取消息 ImGroupMessageDO message = groupMessageMapper.selectById(messageId); if (message == null) { @@ -202,17 +190,17 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { // 2. 获取所有成员和已读位置 List allMembers = groupMemberService.selectByGroupId(groupId); - Map allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId); + Map allPositions = groupMessageReadRedisDAO.getReadMaxMessageIdMap(groupId); // 3. 计算该消息的可见成员集合（排除发送者自己） Set visibleUserIds = getVisibleUserIds(message, allMembers); visibleUserIds.remove(message.getSenderId()); // 4. 只返回在可见范围内且已读位置 >= messageId 的用户 + // DONE @AI：根据返回结果，进行简化；应该都是 Long、Long；——RedisDAO 已返回 Map，直接使用 List readUserIds = new ArrayList<>(); - allPositions.forEach((k, v) -> { - Long uid = Long.valueOf(k.toString()); - if (visibleUserIds.contains(uid) && Long.parseLong(v.toString()) >= messageId) { + allPositions.forEach((uid, readMaxMessageId) -> { + if (visibleUserIds.contains(uid) && readMaxMessageId >= messageId) { readUserIds.add(uid); } }); @@ -222,18 +210,11 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { @Override public List getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO) { // 1. 校验用户在群中 - ImGroupMemberDO member = validateMemberInGroup(reqVO.getGroupId(), userId); + ImGroupMemberDO member = groupMemberService.validateMemberInGroup(reqVO.getGroupId(), userId); // 2. 查询历史消息（仅入群之后） - // TODO @AI：应该 mybatis 里就进行过滤！然后 joinTime 肯定不会是空的！！！ - List messages = groupMessageMapper.selectHistoryList( - reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit()); - if (member.getJoinTime() == null) { - return messages; - } - return messages.stream() - .filter(msg -> !msg.getSendTime().isBefore(member.getJoinTime())) - .toList(); + return groupMessageMapper.selectHistoryList( + reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit(), member.getJoinTime()); } // ========== 异步 WebSocket 推送 ========== @@ -245,14 +226,13 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { public void sendGroupMessageEvent(ImGroupMessageDO message) { ImGroupMessageDTO dto = ImGroupMessageDTO.ofSend(message); // 广播给群内可见成员（含发送方自己，支持多端同步） - // TODO @AI：getGroupMemberListByGroupId()；默认过滤有效的； List members = groupMemberService.selectByGroupId(message.getGroupId()); List receiverList = message.getReceiverUserIds(); for (ImGroupMemberDO member : members) { if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { continue; } - // TODO @AI：这里是不是不用过滤啊？因为一般 send 的时候，肯定带了呀。按照群友就好了； + // 特殊：仅在存在定向接收列表时，才进行过滤；如果没有定向列表，则默认所有成员可见 if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId()) && ObjUtil.notEqual(member.getUserId(), message.getSenderId())) { continue; @@ -262,51 +242,45 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { } } - // TODO @AI：命名最后都是 pre 和 new 里面最好都有 messageId 的标识；包括上面的地方，更好理解吧！ /** * 发送已读 + 刷新群回执 WebSocket 事件 * - * @param userId 当前用户编号 - * @param groupId 群编号 - * @param prevMaxId 上次已读位置 - * @param newMaxId 本次已读位置 + * @param userId 当前用户编号 + * @param groupId 群编号 + * @param prevMaxMessageId 上次已读位置 + * @param newMaxMessageId 本次已读位置 */ @Async - public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxId, Long newMaxId) { + public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxMessageId, Long newMaxMessageId) { // 1. 发送 READ 事件给自己的其他终端（多端同步） - ImGroupMessageDTO readDto = ImGroupMessageDTO.ofRead(userId, groupId, newMaxId); + ImGroupMessageDTO readDTO = ImGroupMessageDTO.ofRead(userId, groupId, newMaxMessageId); webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId, - ImGroupMessageDTO.TYPE, readDto); + ImGroupMessageDTO.TYPE, readDTO); - // 2. 刷新 (prevMaxId, newMaxId] 区间内的待回执消息 - // TODO @AI：selectListByXXX；注意，prevMaxId 可能为空的情况； - List pendingMessages = groupMessageMapper.selectPendingReceiptMessagesInRange( - groupId, prevMaxId, newMaxId); + // 2. 刷新 (prevMaxMessageId, newMaxMessageId] 区间内的待回执消息 + List pendingMessages = groupMessageMapper.selectListByGroupIdAndPendingReceipt( + groupId, prevMaxMessageId, newMaxMessageId); if (CollUtil.isEmpty(pendingMessages)) { return; } - // TODO @AI：是不是要过滤 status ？因为可能退出了。 - List allMembers = groupMemberService.selectByGroupId(groupId); - // TODO @AI：变量也改下； - Map allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId); + // DONE @AI：收敛到 groupMemberService 方法里，会不会更通用；——已收敛为 getActiveMemberListByGroupId + List activeMembers = groupMemberService.getActiveMemberListByGroupId(groupId); + Map allPositions = groupMessageReadRedisDAO.getReadMaxMessageIdMap(groupId); for (ImGroupMessageDO message : pendingMessages) { - // 2.1 计算该消息的可见成员（排除发送者自己） - Set visibleUserIds = getVisibleUserIds(message, allMembers); - visibleUserIds.remove(message.getSenderId()); - // TODO @AI：CollUtil 来判断； - if (visibleUserIds.isEmpty()) { + // 2.1.1 统计可见成员中的已读人数 + Set visibleUserIds = getVisibleUserIds(message, activeMembers); + visibleUserIds.remove(message.getSenderId()); // 发送者自己不算已读 + if (CollUtil.isEmpty(visibleUserIds)) { continue; } - // 2.2 统计可见成员中的已读人数 int readCount = 0; for (Long uid : visibleUserIds) { - // TODO @AI：根据返回结果，进行简化；应该都是 Long、Long； - Object val = allPositions.get(uid.toString()); - if (val != null && Long.parseLong(val.toString()) >= message.getId()) { + Long readMaxMessageId = allPositions.get(uid); + if (readMaxMessageId != null && readMaxMessageId >= message.getId()) { readCount++; } } - // 2.3 全部已读 → 标记回执完成 + // 2.1.2 全部已读 → 标记回执完成 boolean allRead = readCount >= visibleUserIds.size(); Integer newReceiptStatus = allRead ? ImGroupMessageReceiptStatusEnum.DONE.getStatus() @@ -315,8 +289,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { groupMessageMapper.updateById(new ImGroupMessageDO().setId(message.getId()) .setReceiptStatus(newReceiptStatus)); } - // 2.4 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度） - // TODO @AI：检查下别的 Dto = 的地方，都最好改成 DTO = ；仅限 im 模块； + + // 224 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度） ImGroupMessageDTO receiptDTO = ImGroupMessageDTO.ofReceipt( message.getId(), groupId, readCount, newReceiptStatus); webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(), @@ -326,20 +300,18 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { // ========== 私有工具方法 ========== - // TODO @AI：抽到对应的 GroupMemberService 中，供其他地方复用 /** - * 校验用户是否为群的有效成员 + * 校验群聊消息存在 * - * @param groupId 群编号 - * @param userId 用户编号 - * @return 群成员记录 + * @param messageId 消息编号 + * @return 群聊消息 */ - private ImGroupMemberDO validateMemberInGroup(Long groupId, Long userId) { - ImGroupMemberDO member = groupMemberService.getGroupMember(groupId, userId); - if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { - throw exception(GROUP_MEMBER_NOT_IN_GROUP); + private ImGroupMessageDO validateGroupMessageExists(Long messageId) { + ImGroupMessageDO message = groupMessageMapper.selectById(messageId); + if (message == null) { + throw exception(MESSAGE_NOT_EXISTS); } - return member; + return message; } /** @@ -373,14 +345,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { * 计算一条群消息的可见成员集合（含发送者） */ private Set getVisibleUserIds(ImGroupMessageDO msg, List allMembers) { - Set visible = new HashSet<>(); - // TODO @AI： CollectionUtils.converset - for (ImGroupMemberDO member : allMembers) { - if (isMessageVisible(msg, member, member.getUserId())) { - visible.add(member.getUserId()); - } - } - return visible; + return convertSet(allMembers, ImGroupMemberDO::getUserId, + member -> isMessageVisible(msg, member, member.getUserId())); } private ImGroupMessageServiceImpl getSelf() { diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java index 06e8811..904c7b0 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java @@ -175,14 +175,14 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService { @Async public void readPrivateMessageEvent(Long userId, Long receiverId, Long maxReadId) { // 1. 发送 READ 事件给自己的其他终端（多端同步） - ImPrivateMessageDTO readDto = ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId); + ImPrivateMessageDTO readDTO = ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId); webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId, - ImPrivateMessageDTO.TYPE, readDto); + ImPrivateMessageDTO.TYPE, readDTO); // 2. 发送 RECEIPT 事件给对方（已读回执） - ImPrivateMessageDTO receiptDto = ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId); + ImPrivateMessageDTO receiptDTO = ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId); webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), receiverId, - ImPrivateMessageDTO.TYPE, receiptDto); + ImPrivateMessageDTO.TYPE, receiptDTO); } @Override diff --git a/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java b/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java index 9edb995..58abe5a 100644 --- a/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java +++ b/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java @@ -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.GroupReadPositionRedisDAO; +import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO; import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum; import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum; import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService; @@ -50,7 +50,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Mock private ImSensitiveWordService sensitiveWordService; @Mock - private GroupReadPositionRedisDAO groupReadPositionRedisDAO; + private GroupMessageReadRedisDAO groupMessageReadRedisDAO; @Mock private WebSocketMessageSender webSocketMessageSender; @@ -82,7 +82,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { ImGroupMemberDO member = ImGroupMemberDO.builder() .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build(); - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(member); + when(groupMemberService.validateMemberInGroup(10L, 1L)).thenReturn(member); List allMembers = List.of( member, @@ -141,7 +141,8 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { ImGroupDO group = new ImGroupDO(); group.setId(10L); when(groupService.validateGroupExists(10L)).thenReturn(group); - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null); + when(groupMemberService.validateMemberInGroup(10L, 1L)) + .thenThrow(new ServiceException(GROUP_MEMBER_NOT_IN_GROUP.getCode(), GROUP_MEMBER_NOT_IN_GROUP.getMsg())); // 调用并断言 ServiceException exception = assertThrows(ServiceException.class, @@ -278,7 +279,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { // 撤回前需要校验用户仍在群中 ImGroupMemberDO selfMember = ImGroupMemberDO.builder() .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build(); - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(selfMember); + when(groupMemberService.validateMemberInGroup(10L, 1L)).thenReturn(selfMember); when(groupMessageMapper.updateById(any(ImGroupMessageDO.class))).thenReturn(1); when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenReturn(1); @@ -354,11 +355,12 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testReadMessages_notInGroup() { // 准备：当前用户不在群中 - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null); + when(groupMemberService.validateMemberInGroup(10L, 1L)) + .thenThrow(new ServiceException(GROUP_MEMBER_NOT_IN_GROUP.getCode(), GROUP_MEMBER_NOT_IN_GROUP.getMsg())); // 调用并断言：越权校验 ServiceException exception = assertThrows(ServiceException.class, - () -> groupMessageService.readGroupMessages(1L, 10L)); + () -> groupMessageService.readGroupMessages(1L, 10L, 100L)); assertEquals(GROUP_MEMBER_NOT_IN_GROUP.getCode(), exception.getCode()); } @@ -369,7 +371,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .groupId(10L).userId(1L) .status(CommonStatusEnum.ENABLE.getStatus()) .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build(); - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(currentMember); + when(groupMemberService.validateMemberInGroup(10L, 1L)).thenReturn(currentMember); // 准备：消息由用户 5 发，发送时间在 2026-04-12 10:00 ImGroupMessageDO message = ImGroupMessageDO.builder() @@ -397,11 +399,11 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { when(groupMemberService.selectByGroupId(10L)).thenReturn(allMembers); // 准备：Redis 已读位置 — 用户 1 读到 100, 用户 2 读到 50, 用户 3 读到 200 - Map positions = new HashMap<>(); - positions.put("1", "100"); - positions.put("2", "50"); - positions.put("3", "200"); - when(groupReadPositionRedisDAO.getAllReadPositions(10L)).thenReturn(positions); + Map positions = new HashMap<>(); + positions.put(1L, 100L); + positions.put(2L, 50L); + positions.put(3L, 200L); + when(groupMessageReadRedisDAO.getReadMaxMessageIdMap(10L)).thenReturn(positions); // 调用：查询 messageId=80 的已读用户 List readUsers = groupMessageService.getGroupReadUsers(1L, 10L, 80L); @@ -421,7 +423,8 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testGetReadUsers_notInGroup() { // 准备：当前用户不在群中 - when(groupMemberService.getGroupMember(10L, 99L)).thenReturn(null); + when(groupMemberService.validateMemberInGroup(10L, 99L)) + .thenThrow(new ServiceException(GROUP_MEMBER_NOT_IN_GROUP.getCode(), GROUP_MEMBER_NOT_IN_GROUP.getMsg())); // 调用并断言：越权校验 ServiceException exception = assertThrows(ServiceException.class,