BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 7057f4715b9ae7d609012a50dbf59b7be5bfec0d

Authored by YunaiV
1 parent 26413e09

✨ feat(message): 群聊消息的实现

Inline Side-by-side
Showing 6 changed files with 127 additions and 168 deletions
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java
  View file @7057f47
... ... @@ -8,6 +8,7 @@ import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
8 8 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
9 9 import org.apache.ibatis.annotations.Mapper;
10 10  
  11 +import java.time.LocalDateTime;
11 12 import java.util.List;
12 13  
13 14 /**
... ... @@ -30,6 +31,7 @@ public interface ImGroupMessageMapper extends BaseMapperX<ImGroupMessageDO> {
30 31 QueryWrapperX<ImGroupMessageDO> wrapper = new QueryWrapperX<>();
31 32 wrapper.in("group_id", groupIds)
32 33 .gt("id", minId)
  34 + .ne("status", ImMessageStatusEnum.RECALL.getStatus())
33 35 .orderByAsc("id");
34 36 wrapper.limitN(size);
35 37 return selectList(wrapper);
... ... @@ -38,16 +40,18 @@ public interface ImGroupMessageMapper extends BaseMapperX&lt;ImGroupMessageDO&gt; {
38 40 /**
39 41 * 查询群聊历史消息（游标拉取）
40 42 *
41   - * @param groupId 群编号
42   - * @param maxId 起始消息 id（不含），为空则从最新开始
43   - * @param limit 拉取数量
  43 + * @param groupId 群编号
  44 + * @param maxId 起始消息 id（不含），为空则从最新开始
  45 + * @param limit 拉取数量
  46 + * @param joinTime 入群时间，仅返回入群之后的消息
44 47 * @return 消息列表（按 id 倒序）
45 48 */
46   - default List<ImGroupMessageDO> selectHistoryList(Long groupId, Long maxId, Integer limit) {
  49 + default List<ImGroupMessageDO> selectHistoryList(Long groupId, Long maxId, Integer limit, LocalDateTime joinTime) {
47 50 QueryWrapperX<ImGroupMessageDO> wrapper = new QueryWrapperX<>();
48 51 wrapper.eq("group_id", groupId)
49 52 .ne("status", ImMessageStatusEnum.RECALL.getStatus())
50 53 .lt(maxId != null, "id", maxId)
  54 + .ge(joinTime != null, "send_time", joinTime)
51 55 .orderByDesc("id");
52 56 wrapper.limitN(limit);
53 57 return selectList(wrapper);
... ... @@ -60,22 +64,6 @@ public interface ImGroupMessageMapper extends BaseMapperX&lt;ImGroupMessageDO&gt; {
60 64 }
61 65  
62 66 /**
63   - * 获取群内最新消息 id
64   - *
65   - * @param groupId 群编号
66   - * @return 最新消息 id，无消息时返回 null
67   - */
68   - default Long selectMaxIdByGroupId(Long groupId) {
69   - QueryWrapperX<ImGroupMessageDO> wrapper = new QueryWrapperX<>();
70   - wrapper.eq("group_id", groupId)
71   - .orderByDesc("id")
72   - .select("id");
73   - wrapper.limitN(1);
74   - ImGroupMessageDO msg = selectOne(wrapper);
75   - return msg == null ? null : msg.getId();
76   - }
77   -
78   - /**
79 67 * 查询群内指定范围内待回执的消息
80 68 * <p>
81 69 * 仅在用户"已读位置前进"时调用，避免全量扫描：
... ... @@ -86,11 +74,11 @@ public interface ImGroupMessageMapper extends BaseMapperX&lt;ImGroupMessageDO&gt; {
86 74 * @param maxId 结束消息 id（含，本次已读位置）
87 75 * @return 待回执消息列表
88 76 */
89   - default List<ImGroupMessageDO> selectPendingReceiptMessagesInRange(Long groupId, Long minId, Long maxId) {
  77 + default List<ImGroupMessageDO> selectListByGroupIdAndPendingReceipt(Long groupId, Long minId, Long maxId) {
90 78 return selectList(new LambdaQueryWrapperX<ImGroupMessageDO>()
91 79 .eq(ImGroupMessageDO::getGroupId, groupId)
92 80 .eq(ImGroupMessageDO::getReceiptStatus, ImGroupMessageReceiptStatusEnum.PENDING.getStatus())
93   - .gt(ImGroupMessageDO::getId, minId)
  81 + .gt(minId != null, ImGroupMessageDO::getId, minId)
94 82 .le(ImGroupMessageDO::getId, maxId)
95 83 .ne(ImGroupMessageDO::getStatus, ImMessageStatusEnum.RECALL.getStatus()));
96 84 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
  View file @7057f47
... ... @@ -8,12 +8,11 @@ package cn.iocoder.yudao.module.im.dal.redis;
8 8 */
9 9 public interface RedisKeyConstants {
10 10  
11   - // TODO @AI：这个 key 可能要改下；
12 11 /**
13   - * 群已读位置
14   - * KEY 格式： im:group:read:{groupId}
  12 + * 群消息已读位置
  13 + * KEY 格式： im:group:message:read:{groupId}
15 14 * VALUE 数据类型： Hash (field: userId, value: maxReadMessageId)
16 15 */
17   - String GROUP_READ_POSITION = "im:group:read:%s";
  16 + String GROUP_MESSAGE_READ = "im:group:message:read:%s";
18 17  
19 18 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupReadPositionRedisDAO.java renamed to yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupMessageReadRedisDAO.java
  View file @7057f47
1 1 package cn.iocoder.yudao.module.im.dal.redis.message;
2 2  
  3 +import cn.hutool.core.convert.Convert;
3 4 import jakarta.annotation.Resource;
4 5 import org.springframework.data.redis.core.StringRedisTemplate;
5 6 import org.springframework.stereotype.Repository;
6 7  
  8 +import java.util.HashMap;
7 9 import java.util.Map;
8 10  
9   -import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_READ_POSITION;
  11 +import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_MESSAGE_READ;
10 12  
11   -// TODO @AI：需要改成，GroupMessageReadRedisDAO
12 13 /**
13   - * 群已读位置 Redis DAO
14   - * <p>
15   - * 采用 Redis Hash 存储：key=im:group:read:{groupId}, field=userId, value=maxReadMessageId TODO @AI：这里不用注释；
  14 + * 群消息已读位置 Redis DAO
16 15 *
17 16 * @author 芋道源码
18 17 */
19 18 @Repository
20   -public class GroupReadPositionRedisDAO {
  19 +public class GroupMessageReadRedisDAO {
21 20  
22 21 @Resource
23 22 private StringRedisTemplate stringRedisTemplate;
24 23  
25   - // TODO @AI：updateReadMaxMessageId；
26 24 /**
27   - * 更新用户在某群的已读位置
  25 + * 更新用户在某群的最大已读消息编号
28 26 *
29 27 * @param groupId 群编号
30 28 * @param userId 用户编号
31 29 * @param maxMessageId 最大已读消息编号
32 30 */
33   - public void updateReadPosition(Long groupId, Long userId, Long maxMessageId) {
34   - // TODO @AI：这里搞个 formatKey 方法，可以复用；
35   - String key = String.format(GROUP_READ_POSITION, groupId);
36   - // 只在新值大于旧值时更新
37   - Object currentVal = stringRedisTemplate.opsForHash().get(key, userId.toString());
38   - if (currentVal == null || Long.parseLong(currentVal.toString()) < maxMessageId) {
39   - // TODO @AI：可以 put 的时候，就是 Long，Long 么？
40   - stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString());
41   - }
  31 + public void updateReadMaxMessageId(Long groupId, Long userId, Long maxMessageId) {
  32 + String key = formatKey(groupId);
  33 + stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString());
42 34 }
43 35  
44   - // TODO @AI：不存在，则返回 null 更合理；
45   - // TODO @AI：getReadMaxMessageId；
46 36 /**
47   - * 获取用户在某群的已读位置
  37 + * 获取用户在某群的最大已读消息编号
  38 + *
  39 + * @param groupId 群编号
  40 + * @param userId 用户编号
  41 + * @return 最大已读消息编号，不存在则返回 null
48 42 */
49   - public Long getReadPosition(Long groupId, Long userId) {
50   - String key = String.format(GROUP_READ_POSITION, groupId);
  43 + public Long getReadMaxMessageId(Long groupId, Long userId) {
  44 + String key = formatKey(groupId);
51 45 Object val = stringRedisTemplate.opsForHash().get(key, userId.toString());
52   - return val == null ? 0L : Long.parseLong(val.toString());
  46 + return Convert.toLong(val);
53 47 }
54 48  
55   - // TODO @AI：参考这个风格，优化下；
56 49 /**
57 50 * 获取某群所有用户的已读位置
  51 + *
  52 + * @param groupId 群编号
  53 + * @return userId → maxReadMessageId 映射
58 54 */
59   - public Map<Object, Object> getAllReadPositions(Long groupId) {
60   - // TODO @AI：可以返回的时候，就是 Long，Long 么？
61   - String key = String.format(GROUP_READ_POSITION, groupId);
62   - return stringRedisTemplate.opsForHash().entries(key);
  55 + public Map<Long, Long> getReadMaxMessageIdMap(Long groupId) {
  56 + String key = formatKey(groupId);
  57 + Map<Object, Object> entries = stringRedisTemplate.opsForHash().entries(key);
  58 + // 转换为 Long → Long 的 Map
  59 + Map<Long, Long> result = new HashMap<>(entries.size());
  60 + entries.forEach((k, v) -> result.put(Long.parseLong(k.toString()), Long.parseLong(v.toString())));
  61 + return result;
  62 + }
  63 +
  64 + private static String formatKey(Long groupId) {
  65 + return String.format(GROUP_MESSAGE_READ, groupId);
63 66 }
64 67  
65 68 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
  View file @7057f47
... ... @@ -15,7 +15,7 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
15 15 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
16 16 import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
17 17 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
18   -import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO;
  18 +import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO;
19 19 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
20 20 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
21 21 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
... ... @@ -35,6 +35,7 @@ import java.util.*;
35 35  
36 36 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
37 37 import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
  38 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;
38 39 import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
39 40 import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_MAX_PULL_SIZE;
40 41 import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_RECALL_TIMEOUT_MINUTES;
... ... @@ -51,9 +52,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
51 52  
52 53 @Resource
53 54 private ImGroupMessageMapper groupMessageMapper;
54   - // TODO @AI：变量名，可能要改下；
55 55 @Resource
56   - private GroupReadPositionRedisDAO groupReadPositionRedisDAO;
  56 + private GroupMessageReadRedisDAO groupMessageReadRedisDAO;
57 57  
58 58 @Resource
59 59 private ImGroupService groupService;
... ... @@ -77,18 +77,16 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
77 77 }
78 78 // 1.2 校验群存在、发送人仍在群中
79 79 groupService.validateGroupExists(reqVO.getGroupId());
80   - validateMemberInGroup(reqVO.getGroupId(), senderId);
  80 + groupMemberService.validateMemberInGroup(reqVO.getGroupId(), senderId);
81 81 // 1.3 文本消息敏感词过滤
82 82 if (ImMessageTypeEnum.TEXT.getType().equals(reqVO.getType())) {
83 83 sensitiveWordService.validateText(reqVO.getContent());
84 84 }
85 85  
86 86 // 2. 构建并保存消息
87   - // TODO @AI：content 要 json 下；
88 87 ImGroupMessageDO message = BeanUtils.toBean(reqVO, ImGroupMessageDO.class, m -> m
89   - .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus())
90   - .setSendTime(LocalDateTime.now())
91   - .setReceiptStatus(Boolean.TRUE.equals(reqVO.getNeedReceipt())
  88 + .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now())
  89 + .setReceiptStatus(Boolean.TRUE.equals(reqVO.getReceipt())
92 90 ? ImGroupMessageReceiptStatusEnum.PENDING.getStatus()
93 91 : ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus()));
94 92 groupMessageMapper.insert(message);
... ... @@ -114,7 +112,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
114 112 List<ImGroupMessageDO> messages = groupMessageMapper.selectListByMinId(groupIds, minId, size);
115 113  
116 114 // 3. 过滤：入群前不可见、退群后不可见、定向接收过滤
117   - // TODO @AI：缺少各种逻辑；
  115 + // TODO @AI：Java 层过滤在 SQL limit 之后，可能导致返回数量不足 size（跨群场景 joinTime 各异，无法下沉 SQL）【后续优化】
118 116 Map<Long, ImGroupMemberDO> memberMap = CollUtil.toMap(members, new HashMap<>(), ImGroupMemberDO::getGroupId);
119 117 List<ImGroupMessageDO> result = messages.stream()
120 118 .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId))
... ... @@ -124,39 +122,29 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
124 122 return result;
125 123 }
126 124  
127   - // TODO @AI：增加 messageId 参数；允许空，则查询最大的一条；
128 125 @Override
129   - public void readGroupMessages(Long userId, Long groupId) {
130   - // 1.1 校验用户在群中（权限校验）
131   - validateMemberInGroup(groupId, userId);
132   - // 1.2 获取群内最新消息 id（倒序 LIMIT 1，避免全量扫描）
133   - Long newMaxId = groupMessageMapper.selectMaxIdByGroupId(groupId);
134   - if (newMaxId == null) {
135   - return;
136   - }
137   - // 1.3 已读位置未前进，直接返回
138   - // TODO @AI：1.3 和 2 是不是可以合并？如果 update 失败，则直接回退；不然读取两次，貌似有 cas 问题；
139   - Long prevMaxId = groupReadPositionRedisDAO.getReadPosition(groupId, userId);
140   - if (newMaxId <= prevMaxId) {
  126 + public void readGroupMessages(Long userId, Long groupId, Long messageId) {
  127 + Assert.notNull(messageId, "已读消息编号不能为空");
  128 + // 1. 校验用户在群中（权限校验）
  129 + groupMemberService.validateMemberInGroup(groupId, userId);
  130 +
  131 + // 2. 已读位置未前进，直接返回
  132 + Long prevMaxMessageId = groupMessageReadRedisDAO.getReadMaxMessageId(groupId, userId);
  133 + if (prevMaxMessageId != null && prevMaxMessageId >= messageId) {
141 134 return;
142 135 }
  136 + // 3. 更新 Redis 群已读位置
  137 + groupMessageReadRedisDAO.updateReadMaxMessageId(groupId, userId, messageId);
143 138  
144   - // 2. 更新 Redis 群已读位置
145   - groupReadPositionRedisDAO.updateReadPosition(groupId, userId, newMaxId);
146   -
147   - // 3. 异步发送 READ 事件 + 刷新范围内的群回执
148   - getSelf().readGroupMessageEvent(userId, groupId, prevMaxId, newMaxId);
  139 + // 4. 异步发送 READ 事件 + 刷新范围内的群回执
  140 + getSelf().readGroupMessageEvent(userId, groupId, prevMaxMessageId, messageId);
149 141 }
150 142  
151 143 @Override
152 144 @Transactional(rollbackFor = Exception.class)
153 145 public ImGroupMessageDO recallGroupMessage(Long userId, Long messageId) {
154 146 // 1.1 校验消息存在
155   - // TODO @AI：抽成一个 validateXXXExists 方法；
156   - ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
157   - if (message == null) {
158   - throw exception(MESSAGE_NOT_EXISTS);
159   - }
  147 + ImGroupMessageDO message = validateGroupMessageExists(messageId);
160 148 // 1.2 只能撤回自己发送的消息
161 149 if (ObjUtil.notEqual(message.getSenderId(), userId)) {
162 150 throw exception(MESSAGE_RECALL_DENIED);
... ... @@ -170,7 +158,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
170 158 throw exception(MESSAGE_RECALL_TIMEOUT, MESSAGE_RECALL_TIMEOUT_MINUTES);
171 159 }
172 160 // 1.5 校验撤回人仍在群中
173   - validateMemberInGroup(message.getGroupId(), userId);
  161 + groupMemberService.validateMemberInGroup(message.getGroupId(), userId);
174 162  
175 163 // 2. 更新原消息状态为撤回
176 164 groupMessageMapper.updateById(new ImGroupMessageDO().setId(messageId)
... ... @@ -193,7 +181,7 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
193 181 @Override
194 182 public List<Long> getGroupReadUsers(Long userId, Long groupId, Long messageId) {
195 183 // 1.1 校验用户在群中（权限校验）
196   - validateMemberInGroup(groupId, userId);
  184 + groupMemberService.validateMemberInGroup(groupId, userId);
197 185 // 1.2 获取消息
198 186 ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
199 187 if (message == null) {
... ... @@ -202,17 +190,17 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
202 190  
203 191 // 2. 获取所有成员和已读位置
204 192 List<ImGroupMemberDO> allMembers = groupMemberService.selectByGroupId(groupId);
205   - Map<Object, Object> allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId);
  193 + Map<Long, Long> allPositions = groupMessageReadRedisDAO.getReadMaxMessageIdMap(groupId);
206 194  
207 195 // 3. 计算该消息的可见成员集合（排除发送者自己）
208 196 Set<Long> visibleUserIds = getVisibleUserIds(message, allMembers);
209 197 visibleUserIds.remove(message.getSenderId());
210 198  
211 199 // 4. 只返回在可见范围内且已读位置 >= messageId 的用户
  200 + // DONE @AI：根据返回结果，进行简化；应该都是 Long、Long；——RedisDAO 已返回 Map<Long, Long>，直接使用
212 201 List<Long> readUserIds = new ArrayList<>();
213   - allPositions.forEach((k, v) -> {
214   - Long uid = Long.valueOf(k.toString());
215   - if (visibleUserIds.contains(uid) && Long.parseLong(v.toString()) >= messageId) {
  202 + allPositions.forEach((uid, readMaxMessageId) -> {
  203 + if (visibleUserIds.contains(uid) && readMaxMessageId >= messageId) {
216 204 readUserIds.add(uid);
217 205 }
218 206 });
... ... @@ -222,18 +210,11 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
222 210 @Override
223 211 public List<ImGroupMessageDO> getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO) {
224 212 // 1. 校验用户在群中
225   - ImGroupMemberDO member = validateMemberInGroup(reqVO.getGroupId(), userId);
  213 + ImGroupMemberDO member = groupMemberService.validateMemberInGroup(reqVO.getGroupId(), userId);
226 214  
227 215 // 2. 查询历史消息（仅入群之后）
228   - // TODO @AI：应该 mybatis 里就进行过滤！然后 joinTime 肯定不会是空的！！！
229   - List<ImGroupMessageDO> messages = groupMessageMapper.selectHistoryList(
230   - reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit());
231   - if (member.getJoinTime() == null) {
232   - return messages;
233   - }
234   - return messages.stream()
235   - .filter(msg -> !msg.getSendTime().isBefore(member.getJoinTime()))
236   - .toList();
  216 + return groupMessageMapper.selectHistoryList(
  217 + reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit(), member.getJoinTime());
237 218 }
238 219  
239 220 // ========== 异步 WebSocket 推送 ==========
... ... @@ -245,14 +226,13 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
245 226 public void sendGroupMessageEvent(ImGroupMessageDO message) {
246 227 ImGroupMessageDTO dto = ImGroupMessageDTO.ofSend(message);
247 228 // 广播给群内可见成员（含发送方自己，支持多端同步）
248   - // TODO @AI：getGroupMemberListByGroupId()；默认过滤有效的；
249 229 List<ImGroupMemberDO> members = groupMemberService.selectByGroupId(message.getGroupId());
250 230 List<Long> receiverList = message.getReceiverUserIds();
251 231 for (ImGroupMemberDO member : members) {
252 232 if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) {
253 233 continue;
254 234 }
255   - // TODO @AI：这里是不是不用过滤啊？因为一般 send 的时候，肯定带了呀。按照群友就好了；
  235 + // 特殊：仅在存在定向接收列表时，才进行过滤；如果没有定向列表，则默认所有成员可见
256 236 if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId())
257 237 && ObjUtil.notEqual(member.getUserId(), message.getSenderId())) {
258 238 continue;
... ... @@ -262,51 +242,45 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
262 242 }
263 243 }
264 244  
265   - // TODO @AI：命名最后都是 pre 和 new 里面最好都有 messageId 的标识；包括上面的地方，更好理解吧！
266 245 /**
267 246 * 发送已读 + 刷新群回执 WebSocket 事件
268 247 *
269   - * @param userId 当前用户编号
270   - * @param groupId 群编号
271   - * @param prevMaxId 上次已读位置
272   - * @param newMaxId 本次已读位置
  248 + * @param userId 当前用户编号
  249 + * @param groupId 群编号
  250 + * @param prevMaxMessageId 上次已读位置
  251 + * @param newMaxMessageId 本次已读位置
273 252 */
274 253 @Async
275   - public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxId, Long newMaxId) {
  254 + public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxMessageId, Long newMaxMessageId) {
276 255 // 1. 发送 READ 事件给自己的其他终端（多端同步）
277   - ImGroupMessageDTO readDto = ImGroupMessageDTO.ofRead(userId, groupId, newMaxId);
  256 + ImGroupMessageDTO readDTO = ImGroupMessageDTO.ofRead(userId, groupId, newMaxMessageId);
278 257 webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
279   - ImGroupMessageDTO.TYPE, readDto);
  258 + ImGroupMessageDTO.TYPE, readDTO);
280 259  
281   - // 2. 刷新 (prevMaxId, newMaxId] 区间内的待回执消息
282   - // TODO @AI：selectListByXXX；注意，prevMaxId 可能为空的情况；
283   - List<ImGroupMessageDO> pendingMessages = groupMessageMapper.selectPendingReceiptMessagesInRange(
284   - groupId, prevMaxId, newMaxId);
  260 + // 2. 刷新 (prevMaxMessageId, newMaxMessageId] 区间内的待回执消息
  261 + List<ImGroupMessageDO> pendingMessages = groupMessageMapper.selectListByGroupIdAndPendingReceipt(
  262 + groupId, prevMaxMessageId, newMaxMessageId);
285 263 if (CollUtil.isEmpty(pendingMessages)) {
286 264 return;
287 265 }
288   - // TODO @AI：是不是要过滤 status ？因为可能退出了。
289   - List<ImGroupMemberDO> allMembers = groupMemberService.selectByGroupId(groupId);
290   - // TODO @AI：变量也改下；
291   - Map<Object, Object> allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId);
  266 + // DONE @AI：收敛到 groupMemberService 方法里，会不会更通用；——已收敛为 getActiveMemberListByGroupId
  267 + List<ImGroupMemberDO> activeMembers = groupMemberService.getActiveMemberListByGroupId(groupId);
  268 + Map<Long, Long> allPositions = groupMessageReadRedisDAO.getReadMaxMessageIdMap(groupId);
292 269 for (ImGroupMessageDO message : pendingMessages) {
293   - // 2.1 计算该消息的可见成员（排除发送者自己）
294   - Set<Long> visibleUserIds = getVisibleUserIds(message, allMembers);
295   - visibleUserIds.remove(message.getSenderId());
296   - // TODO @AI：CollUtil 来判断；
297   - if (visibleUserIds.isEmpty()) {
  270 + // 2.1.1 统计可见成员中的已读人数
  271 + Set<Long> visibleUserIds = getVisibleUserIds(message, activeMembers);
  272 + visibleUserIds.remove(message.getSenderId()); // 发送者自己不算已读
  273 + if (CollUtil.isEmpty(visibleUserIds)) {
298 274 continue;
299 275 }
300   - // 2.2 统计可见成员中的已读人数
301 276 int readCount = 0;
302 277 for (Long uid : visibleUserIds) {
303   - // TODO @AI：根据返回结果，进行简化；应该都是 Long、Long；
304   - Object val = allPositions.get(uid.toString());
305   - if (val != null && Long.parseLong(val.toString()) >= message.getId()) {
  278 + Long readMaxMessageId = allPositions.get(uid);
  279 + if (readMaxMessageId != null && readMaxMessageId >= message.getId()) {
306 280 readCount++;
307 281 }
308 282 }
309   - // 2.3 全部已读 → 标记回执完成
  283 + // 2.1.2 全部已读 → 标记回执完成
310 284 boolean allRead = readCount >= visibleUserIds.size();
311 285 Integer newReceiptStatus = allRead
312 286 ? ImGroupMessageReceiptStatusEnum.DONE.getStatus()
... ... @@ -315,8 +289,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
315 289 groupMessageMapper.updateById(new ImGroupMessageDO().setId(message.getId())
316 290 .setReceiptStatus(newReceiptStatus));
317 291 }
318   - // 2.4 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度）
319   - // TODO @AI：检查下别的 Dto = 的地方，都最好改成 DTO = ；仅限 im 模块；
  292 +
  293 + // 224 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度）
320 294 ImGroupMessageDTO receiptDTO = ImGroupMessageDTO.ofReceipt(
321 295 message.getId(), groupId, readCount, newReceiptStatus);
322 296 webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(),
... ... @@ -326,20 +300,18 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
326 300  
327 301 // ========== 私有工具方法 ==========
328 302  
329   - // TODO @AI：抽到对应的 GroupMemberService 中，供其他地方复用
330 303 /**
331   - * 校验用户是否为群的有效成员
  304 + * 校验群聊消息存在
332 305 *
333   - * @param groupId 群编号
334   - * @param userId 用户编号
335   - * @return 群成员记录
  306 + * @param messageId 消息编号
  307 + * @return 群聊消息
336 308 */
337   - private ImGroupMemberDO validateMemberInGroup(Long groupId, Long userId) {
338   - ImGroupMemberDO member = groupMemberService.getGroupMember(groupId, userId);
339   - if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) {
340   - throw exception(GROUP_MEMBER_NOT_IN_GROUP);
  309 + private ImGroupMessageDO validateGroupMessageExists(Long messageId) {
  310 + ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
  311 + if (message == null) {
  312 + throw exception(MESSAGE_NOT_EXISTS);
341 313 }
342   - return member;
  314 + return message;
343 315 }
344 316  
345 317 /**
... ... @@ -373,14 +345,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
373 345 * 计算一条群消息的可见成员集合（含发送者）
374 346 */
375 347 private Set<Long> getVisibleUserIds(ImGroupMessageDO msg, List<ImGroupMemberDO> allMembers) {
376   - Set<Long> visible = new HashSet<>();
377   - // TODO @AI： CollectionUtils.converset
378   - for (ImGroupMemberDO member : allMembers) {
379   - if (isMessageVisible(msg, member, member.getUserId())) {
380   - visible.add(member.getUserId());
381   - }
382   - }
383   - return visible;
  348 + return convertSet(allMembers, ImGroupMemberDO::getUserId,
  349 + member -> isMessageVisible(msg, member, member.getUserId()));
384 350 }
385 351  
386 352 private ImGroupMessageServiceImpl getSelf() {
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java
  View file @7057f47
... ... @@ -175,14 +175,14 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
175 175 @Async
176 176 public void readPrivateMessageEvent(Long userId, Long receiverId, Long maxReadId) {
177 177 // 1. 发送 READ 事件给自己的其他终端（多端同步）
178   - ImPrivateMessageDTO readDto = ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId);
  178 + ImPrivateMessageDTO readDTO = ImPrivateMessageDTO.ofRead(userId, receiverId, maxReadId);
179 179 webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
180   - ImPrivateMessageDTO.TYPE, readDto);
  180 + ImPrivateMessageDTO.TYPE, readDTO);
181 181  
182 182 // 2. 发送 RECEIPT 事件给对方（已读回执）
183   - ImPrivateMessageDTO receiptDto = ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId);
  183 + ImPrivateMessageDTO receiptDTO = ImPrivateMessageDTO.ofReceipt(userId, receiverId, maxReadId);
184 184 webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), receiverId,
185   - ImPrivateMessageDTO.TYPE, receiptDto);
  185 + ImPrivateMessageDTO.TYPE, receiptDTO);
186 186 }
187 187  
188 188 @Override
... ...
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java
  View file @7057f47
... ... @@ -10,7 +10,7 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
10 10 import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
11 11 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
12 12 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
13   -import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO;
  13 +import cn.iocoder.yudao.module.im.dal.redis.message.GroupMessageReadRedisDAO;
14 14 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
15 15 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
16 16 import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService;
... ... @@ -50,7 +50,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
50 50 @Mock
51 51 private ImSensitiveWordService sensitiveWordService;
52 52 @Mock
53   - private GroupReadPositionRedisDAO groupReadPositionRedisDAO;
  53 + private GroupMessageReadRedisDAO groupMessageReadRedisDAO;
54 54 @Mock
55 55 private WebSocketMessageSender webSocketMessageSender;
56 56  
... ... @@ -82,7 +82,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
82 82  
83 83 ImGroupMemberDO member = ImGroupMemberDO.builder()
84 84 .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build();
85   - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(member);
  85 + when(groupMemberService.validateMemberInGroup(10L, 1L)).thenReturn(member);
86 86  
87 87 List<ImGroupMemberDO> allMembers = List.of(
88 88 member,
... ... @@ -141,7 +141,8 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
141 141 ImGroupDO group = new ImGroupDO();
142 142 group.setId(10L);
143 143 when(groupService.validateGroupExists(10L)).thenReturn(group);
144   - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null);
  144 + when(groupMemberService.validateMemberInGroup(10L, 1L))
  145 + .thenThrow(new ServiceException(GROUP_MEMBER_NOT_IN_GROUP.getCode(), GROUP_MEMBER_NOT_IN_GROUP.getMsg()));
145 146  
146 147 // 调用并断言
147 148 ServiceException exception = assertThrows(ServiceException.class,
... ... @@ -278,7 +279,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
278 279 // 撤回前需要校验用户仍在群中
279 280 ImGroupMemberDO selfMember = ImGroupMemberDO.builder()
280 281 .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build();
281   - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(selfMember);
  282 + when(groupMemberService.validateMemberInGroup(10L, 1L)).thenReturn(selfMember);
282 283 when(groupMessageMapper.updateById(any(ImGroupMessageDO.class))).thenReturn(1);
283 284 when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenReturn(1);
284 285  
... ... @@ -354,11 +355,12 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
354 355 @Test
355 356 public void testReadMessages_notInGroup() {
356 357 // 准备：当前用户不在群中
357   - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null);
  358 + when(groupMemberService.validateMemberInGroup(10L, 1L))
  359 + .thenThrow(new ServiceException(GROUP_MEMBER_NOT_IN_GROUP.getCode(), GROUP_MEMBER_NOT_IN_GROUP.getMsg()));
358 360  
359 361 // 调用并断言：越权校验
360 362 ServiceException exception = assertThrows(ServiceException.class,
361   - () -> groupMessageService.readGroupMessages(1L, 10L));
  363 + () -> groupMessageService.readGroupMessages(1L, 10L, 100L));
362 364 assertEquals(GROUP_MEMBER_NOT_IN_GROUP.getCode(), exception.getCode());
363 365 }
364 366  
... ... @@ -369,7 +371,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
369 371 .groupId(10L).userId(1L)
370 372 .status(CommonStatusEnum.ENABLE.getStatus())
371 373 .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build();
372   - when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(currentMember);
  374 + when(groupMemberService.validateMemberInGroup(10L, 1L)).thenReturn(currentMember);
373 375  
374 376 // 准备：消息由用户 5 发，发送时间在 2026-04-12 10:00
375 377 ImGroupMessageDO message = ImGroupMessageDO.builder()
... ... @@ -397,11 +399,11 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
397 399 when(groupMemberService.selectByGroupId(10L)).thenReturn(allMembers);
398 400  
399 401 // 准备：Redis 已读位置 — 用户 1 读到 100, 用户 2 读到 50, 用户 3 读到 200
400   - Map<Object, Object> positions = new HashMap<>();
401   - positions.put("1", "100");
402   - positions.put("2", "50");
403   - positions.put("3", "200");
404   - when(groupReadPositionRedisDAO.getAllReadPositions(10L)).thenReturn(positions);
  402 + Map<Long, Long> positions = new HashMap<>();
  403 + positions.put(1L, 100L);
  404 + positions.put(2L, 50L);
  405 + positions.put(3L, 200L);
  406 + when(groupMessageReadRedisDAO.getReadMaxMessageIdMap(10L)).thenReturn(positions);
405 407  
406 408 // 调用：查询 messageId=80 的已读用户
407 409 List<Long> readUsers = groupMessageService.getGroupReadUsers(1L, 10L, 80L);
... ... @@ -421,7 +423,8 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
421 423 @Test
422 424 public void testGetReadUsers_notInGroup() {
423 425 // 准备：当前用户不在群中
424   - when(groupMemberService.getGroupMember(10L, 99L)).thenReturn(null);
  426 + when(groupMemberService.validateMemberInGroup(10L, 99L))
  427 + .thenThrow(new ServiceException(GROUP_MEMBER_NOT_IN_GROUP.getCode(), GROUP_MEMBER_NOT_IN_GROUP.getMsg()));
425 428  
426 429 // 调用并断言：越权校验
427 430 ServiceException exception = assertThrows(ServiceException.class,
... ...