BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 26413e09f3fff91d320fe8365e04f02a367a1818

Authored by YunaiV
1 parent 37cf16c1

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

Inline Side-by-side
Showing 15 changed files with 581 additions and 533 deletions
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImGroupMessageController.java
  View file @26413e0
1 1 package cn.iocoder.yudao.module.im.controller.admin.message;
2 2  
3 3 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  4 +import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
  5 +import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO;
4 6 import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageRespVO;
5 7 import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO;
6 8 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
... ... @@ -14,7 +16,6 @@ import org.springframework.validation.annotation.Validated;
14 16 import org.springframework.web.bind.annotation.*;
15 17  
16 18 import java.util.List;
17   -import java.util.stream.Collectors;
18 19  
19 20 import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
20 21 import static cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils.getLoginUserId;
... ... @@ -29,73 +30,55 @@ import static cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUti
29 30 public class ImGroupMessageController {
30 31  
31 32 @Resource
32   - private ImGroupMessageService imGroupMessageService;
  33 + private ImGroupMessageService groupMessageService;
33 34  
34 35 @PostMapping("/send")
35 36 @Operation(summary = "发送群聊消息")
36 37 public CommonResult<ImGroupMessageRespVO> sendGroupMessage(@Valid @RequestBody ImGroupMessageSendReqVO reqVO) {
37   - ImGroupMessageDO message = imGroupMessageService.sendGroupMessage(getLoginUserId(), reqVO);
38   - return success(toRespVO(message));
  38 + ImGroupMessageDO message = groupMessageService.sendGroupMessage(getLoginUserId(), reqVO);
  39 + return success(BeanUtils.toBean(message, ImGroupMessageRespVO.class));
39 40 }
40 41  
41 42 @GetMapping("/pull")
42 43 @Operation(summary = "拉取群聊消息（增量）")
43 44 @Parameter(name = "minId", description = "最小消息 id", required = true, example = "0")
44 45 @Parameter(name = "size", description = "拉取数量", required = true, example = "100")
45   - public CommonResult<List<ImGroupMessageRespVO>> pullGroupMessageList(
46   - @RequestParam("minId") Long minId,
47   - @RequestParam("size") Integer size) {
48   - List<ImGroupMessageDO> messages = imGroupMessageService.pullGroupMessageList(getLoginUserId(), minId, size);
49   - return success(messages.stream().map(this::toRespVO).collect(Collectors.toList()));
  46 + public CommonResult<List<ImGroupMessageRespVO>> pullGroupMessageList(@RequestParam("minId") Long minId,
  47 + @RequestParam("size") Integer size) {
  48 + List<ImGroupMessageDO> messages = groupMessageService.pullGroupMessageList(getLoginUserId(), minId, size);
  49 + return success(BeanUtils.toBean(messages, ImGroupMessageRespVO.class));
50 50 }
51 51  
52 52 @PutMapping("/read")
53 53 @Operation(summary = "标记群聊消息已读")
54 54 @Parameter(name = "groupId", description = "群编号", required = true, example = "1")
55 55 public CommonResult<Boolean> readGroupMessages(@RequestParam("groupId") Long groupId) {
56   - imGroupMessageService.readGroupMessages(getLoginUserId(), groupId);
  56 + groupMessageService.readGroupMessages(getLoginUserId(), groupId);
57 57 return success(true);
58 58 }
59 59  
60 60 @DeleteMapping("/recall")
61 61 @Operation(summary = "撤回群聊消息")
62 62 @Parameter(name = "id", description = "消息编号", required = true, example = "1")
63   - public CommonResult<Boolean> recallGroupMessage(@RequestParam("id") Long id) {
64   - imGroupMessageService.recallGroupMessage(getLoginUserId(), id);
65   - return success(true);
  63 + public CommonResult<ImGroupMessageRespVO> recallGroupMessage(@RequestParam("id") Long id) {
  64 + ImGroupMessageDO message = groupMessageService.recallGroupMessage(getLoginUserId(), id);
  65 + return success(BeanUtils.toBean(message, ImGroupMessageRespVO.class));
66 66 }
67 67  
68   - // DONE @AI：这个接口，确认需要！用于群回执人数展示
69 68 @GetMapping("/read-users")
70 69 @Operation(summary = "获取群消息已读用户列表")
71 70 @Parameter(name = "groupId", description = "群编号", required = true, example = "1")
72 71 @Parameter(name = "messageId", description = "消息编号", required = true, example = "1")
73   - public CommonResult<List<String>> getGroupReadUsers(
74   - @RequestParam("groupId") Long groupId,
75   - @RequestParam("messageId") Long messageId) {
76   - List<Long> readUserIds = imGroupMessageService.getGroupReadUsers(getLoginUserId(), groupId, messageId);
77   - // Long -> String 转换
78   - return success(readUserIds.stream().map(String::valueOf).collect(Collectors.toList()));
  72 + public CommonResult<List<Long>> getGroupReadUsers(@RequestParam("groupId") Long groupId,
  73 + @RequestParam("messageId") Long messageId) {
  74 + return success(groupMessageService.getGroupReadUsers(getLoginUserId(), groupId, messageId));
79 75 }
80 76  
81   - /**
82   - * DO 转换为 RespVO（Long id 统一转 String）
83   - */
84   - // DONE @AI：BeanUtils？结论：因 id/senderId/groupId 需 Long→String 手动转换，不适合 BeanUtils
85   - private ImGroupMessageRespVO toRespVO(ImGroupMessageDO message) {
86   - ImGroupMessageRespVO vo = new ImGroupMessageRespVO();
87   - vo.setId(message.getId().toString());
88   - vo.setClientMessageId(message.getClientMessageId());
89   - vo.setSenderId(message.getSenderId().toString());
90   - vo.setGroupId(message.getGroupId().toString());
91   - vo.setType(message.getType());
92   - vo.setContent(message.getContent());
93   - vo.setStatus(message.getStatus());
94   - vo.setSendTime(message.getSendTime());
95   - vo.setAtUserIds(message.getAtUserIds());
96   - vo.setReceiverUserIds(message.getReceiverUserIds());
97   - vo.setReceiptStatus(message.getReceiptStatus());
98   - return vo;
  77 + @GetMapping("/list")
  78 + @Operation(summary = "查询群聊历史消息")
  79 + public CommonResult<List<ImGroupMessageRespVO>> getGroupMessageList(@Valid ImGroupMessageListReqVO reqVO) {
  80 + List<ImGroupMessageDO> messages = groupMessageService.getGroupMessageList(getLoginUserId(), reqVO);
  81 + return success(BeanUtils.toBean(messages, ImGroupMessageRespVO.class));
99 82 }
100 83  
101 84 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageListReqVO.java 0 → 100644
  View file @26413e0
  1 +package cn.iocoder.yudao.module.im.controller.admin.message.vo.group;
  2 +
  3 +import io.swagger.v3.oas.annotations.media.Schema;
  4 +import jakarta.validation.constraints.Max;
  5 +import jakarta.validation.constraints.Min;
  6 +import jakarta.validation.constraints.NotNull;
  7 +import lombok.Data;
  8 +
  9 +@Schema(description = "管理后台 - 群聊历史消息列表 Request VO")
  10 +@Data
  11 +public class ImGroupMessageListReqVO {
  12 +
  13 + @Schema(description = "群编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "10")
  14 + @NotNull(message = "群编号不能为空")
  15 + private Long groupId;
  16 +
  17 + @Schema(description = "起始消息编号，从该 id 往前拉取（不含）。为空则从最新消息开始", example = "1024")
  18 + private Long maxId;
  19 +
  20 + @Schema(description = "拉取数量", requiredMode = Schema.RequiredMode.REQUIRED, example = "20")
  21 + @NotNull(message = "拉取数量不能为空")
  22 + @Min(value = 1, message = "拉取数量最小值为 1")
  23 + @Max(value = 200, message = "拉取数量最大值为 200")
  24 + private Integer limit;
  25 +
  26 +}
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageRespVO.java
  View file @26413e0
... ... @@ -14,16 +14,16 @@ import java.util.List;
14 14 public class ImGroupMessageRespVO {
15 15  
16 16 @Schema(description = "消息编号", example = "1")
17   - private String id;
  17 + private Long id;
18 18  
19 19 @Schema(description = "客户端消息编号", example = "uuid-xxx")
20 20 private String clientMessageId;
21 21  
22 22 @Schema(description = "发送人编号", example = "1")
23   - private String senderId;
  23 + private Long senderId;
24 24  
25 25 @Schema(description = "群编号", example = "1")
26   - private String groupId;
  26 + private Long groupId;
27 27  
28 28 @Schema(description = "消息类型", example = "0")
29 29 private Integer type;
... ... @@ -46,7 +46,4 @@ public class ImGroupMessageRespVO {
46 46 @Schema(description = "回执状态", example = "0")
47 47 private Integer receiptStatus;
48 48  
49   - @Schema(description = "已读人数（群回执使用）", example = "5")
50   - private Integer readCount;
51   -
52 49 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java
  View file @26413e0
... ... @@ -4,6 +4,8 @@ import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
4 4 import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
5 5 import cn.iocoder.yudao.framework.mybatis.core.query.QueryWrapperX;
6 6 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
  7 +import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
  8 +import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
7 9 import org.apache.ibatis.annotations.Mapper;
8 10  
9 11 import java.util.List;
... ... @@ -16,6 +18,14 @@ import java.util.List;
16 18 @Mapper
17 19 public interface ImGroupMessageMapper extends BaseMapperX<ImGroupMessageDO> {
18 20  
  21 + /**
  22 + * 根据 minId 增量拉取群聊消息
  23 + *
  24 + * @param groupIds 用户所在群编号列表
  25 + * @param minId 最小消息 id（不含）
  26 + * @param size 拉取数量
  27 + * @return 消息列表
  28 + */
19 29 default List<ImGroupMessageDO> selectListByMinId(List<Long> groupIds, Long minId, Integer size) {
20 30 QueryWrapperX<ImGroupMessageDO> wrapper = new QueryWrapperX<>();
21 31 wrapper.in("group_id", groupIds)
... ... @@ -25,33 +35,64 @@ public interface ImGroupMessageMapper extends BaseMapperX&lt;ImGroupMessageDO&gt; {
25 35 return selectList(wrapper);
26 36 }
27 37  
  38 + /**
  39 + * 查询群聊历史消息（游标拉取）
  40 + *
  41 + * @param groupId 群编号
  42 + * @param maxId 起始消息 id（不含），为空则从最新开始
  43 + * @param limit 拉取数量
  44 + * @return 消息列表（按 id 倒序）
  45 + */
  46 + default List<ImGroupMessageDO> selectHistoryList(Long groupId, Long maxId, Integer limit) {
  47 + QueryWrapperX<ImGroupMessageDO> wrapper = new QueryWrapperX<>();
  48 + wrapper.eq("group_id", groupId)
  49 + .ne("status", ImMessageStatusEnum.RECALL.getStatus())
  50 + .lt(maxId != null, "id", maxId)
  51 + .orderByDesc("id");
  52 + wrapper.limitN(limit);
  53 + return selectList(wrapper);
  54 + }
  55 +
28 56 default ImGroupMessageDO selectBySenderIdAndClientMessageId(Long senderId, String clientMessageId) {
29 57 return selectOne(new LambdaQueryWrapperX<ImGroupMessageDO>()
30 58 .eq(ImGroupMessageDO::getSenderId, senderId)
31 59 .eq(ImGroupMessageDO::getClientMessageId, clientMessageId));
32 60 }
33 61  
34   - // TODO @芋艿：暂时没 review
35 62 /**
36 63 * 获取群内最新消息 id
  64 + *
  65 + * @param groupId 群编号
  66 + * @return 最新消息 id，无消息时返回 null
37 67 */
38 68 default Long selectMaxIdByGroupId(Long groupId) {
39 69 QueryWrapperX<ImGroupMessageDO> wrapper = new QueryWrapperX<>();
40 70 wrapper.eq("group_id", groupId)
41   - .orderByDesc("id");
  71 + .orderByDesc("id")
  72 + .select("id");
42 73 wrapper.limitN(1);
43 74 ImGroupMessageDO msg = selectOne(wrapper);
44 75 return msg == null ? null : msg.getId();
45 76 }
46 77  
47   - // TODO @芋艿：暂时没 review
48 78 /**
49   - * 获取群内需要回执的消息（状态为 PENDING）
  79 + * 查询群内指定范围内待回执的消息
  80 + * <p>
  81 + * 仅在用户"已读位置前进"时调用，避免全量扫描：
  82 + * 只有位于 (minId, maxId] 范围内、且仍处于 PENDING 的回执消息可能因本次已读而状态变化。
  83 + *
  84 + * @param groupId 群编号
  85 + * @param minId 起始消息 id（不含，上一次已读位置）
  86 + * @param maxId 结束消息 id（含，本次已读位置）
  87 + * @return 待回执消息列表
50 88 */
51   - default List<ImGroupMessageDO> selectPendingReceiptMessages(Long groupId) {
  89 + default List<ImGroupMessageDO> selectPendingReceiptMessagesInRange(Long groupId, Long minId, Long maxId) {
52 90 return selectList(new LambdaQueryWrapperX<ImGroupMessageDO>()
53 91 .eq(ImGroupMessageDO::getGroupId, groupId)
54   - .eq(ImGroupMessageDO::getReceiptStatus, 1)); // PENDING
  92 + .eq(ImGroupMessageDO::getReceiptStatus, ImGroupMessageReceiptStatusEnum.PENDING.getStatus())
  93 + .gt(ImGroupMessageDO::getId, minId)
  94 + .le(ImGroupMessageDO::getId, maxId)
  95 + .ne(ImGroupMessageDO::getStatus, ImMessageStatusEnum.RECALL.getStatus()));
55 96 }
56 97  
57 98 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
  View file @26413e0
... ... @@ -8,20 +8,7 @@ package cn.iocoder.yudao.module.im.dal.redis;
8 8 */
9 9 public interface RedisKeyConstants {
10 10  
11   - /**
12   - * 收件箱序号生成器
13   - * KEY 格式： im:inbox:sequence:{userId}
14   - * VALUE 数据类型： String
15   - */
16   - String INBOX_SEQUENCE = "im_inbox_sequence:%s";
17   -
18   - /**
19   - * 收件箱的分布式锁
20   - * KEY 格式： im:inbox:lock:{userId}
21   - * VALUE 数据类型： String
22   - */
23   - String INBOX_LOCK = "im_inbox_lock:%s";
24   -
  11 + // TODO @AI：这个 key 可能要改下；
25 12 /**
26 13 * 群已读位置
27 14 * KEY 格式： im:group:read:{groupId}
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/group/GroupReadPositionRedisDAO.java renamed to yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupReadPositionRedisDAO.java
  View file @26413e0
1   -package cn.iocoder.yudao.module.im.dal.redis.group;
  1 +package cn.iocoder.yudao.module.im.dal.redis.message;
2 2  
3 3 import jakarta.annotation.Resource;
4 4 import org.springframework.data.redis.core.StringRedisTemplate;
... ... @@ -8,10 +8,11 @@ import java.util.Map;
8 8  
9 9 import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_READ_POSITION;
10 10  
  11 +// TODO @AI：需要改成，GroupMessageReadRedisDAO
11 12 /**
12 13 * 群已读位置 Redis DAO
13 14 * <p>
14   - * 采用 Redis Hash 存储：key=im:group:read:{groupId}, field=userId, value=maxReadMessageId
  15 + * 采用 Redis Hash 存储：key=im:group:read:{groupId}, field=userId, value=maxReadMessageId TODO @AI：这里不用注释；
15 16 *
16 17 * @author 芋道源码
17 18 */
... ... @@ -21,6 +22,7 @@ public class GroupReadPositionRedisDAO {
21 22 @Resource
22 23 private StringRedisTemplate stringRedisTemplate;
23 24  
  25 + // TODO @AI：updateReadMaxMessageId；
24 26 /**
25 27 * 更新用户在某群的已读位置
26 28 *
... ... @@ -29,14 +31,18 @@ public class GroupReadPositionRedisDAO {
29 31 * @param maxMessageId 最大已读消息编号
30 32 */
31 33 public void updateReadPosition(Long groupId, Long userId, Long maxMessageId) {
  34 + // TODO @AI：这里搞个 formatKey 方法，可以复用；
32 35 String key = String.format(GROUP_READ_POSITION, groupId);
33 36 // 只在新值大于旧值时更新
34 37 Object currentVal = stringRedisTemplate.opsForHash().get(key, userId.toString());
35 38 if (currentVal == null || Long.parseLong(currentVal.toString()) < maxMessageId) {
  39 + // TODO @AI：可以 put 的时候，就是 Long，Long 么？
36 40 stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString());
37 41 }
38 42 }
39 43  
  44 + // TODO @AI：不存在，则返回 null 更合理；
  45 + // TODO @AI：getReadMaxMessageId；
40 46 /**
41 47 * 获取用户在某群的已读位置
42 48 */
... ... @@ -46,28 +52,14 @@ public class GroupReadPositionRedisDAO {
46 52 return val == null ? 0L : Long.parseLong(val.toString());
47 53 }
48 54  
  55 + // TODO @AI：参考这个风格，优化下；
49 56 /**
50 57 * 获取某群所有用户的已读位置
51 58 */
52 59 public Map<Object, Object> getAllReadPositions(Long groupId) {
  60 + // TODO @AI：可以返回的时候，就是 Long，Long 么？
53 61 String key = String.format(GROUP_READ_POSITION, groupId);
54 62 return stringRedisTemplate.opsForHash().entries(key);
55 63 }
56 64  
57   - /**
58   - * 删除某群所有用户的已读位置（群解散时调用）
59   - */
60   - public void deleteGroupPositions(Long groupId) {
61   - String key = String.format(GROUP_READ_POSITION, groupId);
62   - stringRedisTemplate.delete(key);
63   - }
64   -
65   - /**
66   - * 删除用户在某群的已读位置（退群时调用）
67   - */
68   - public void deleteUserPosition(Long groupId, Long userId) {
69   - String key = String.format(GROUP_READ_POSITION, groupId);
70   - stringRedisTemplate.opsForHash().delete(key, userId.toString());
71   - }
72   -
73 65 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java
  View file @26413e0
1 1 package cn.iocoder.yudao.module.im.service.message;
2 2  
  3 +import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO;
3 4 import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO;
4 5 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
5 6  
... ... @@ -44,8 +45,9 @@ public interface ImGroupMessageService {
44 45 *
45 46 * @param userId 当前用户编号
46 47 * @param messageId 消息编号
  48 + * @return 撤回后的提示消息
47 49 */
48   - void recallGroupMessage(Long userId, Long messageId);
  50 + ImGroupMessageDO recallGroupMessage(Long userId, Long messageId);
49 51  
50 52 /**
51 53 * 获取群消息的已读用户列表
... ... @@ -57,4 +59,13 @@ public interface ImGroupMessageService {
57 59 */
58 60 List<Long> getGroupReadUsers(Long userId, Long groupId, Long messageId);
59 61  
  62 + /**
  63 + * 查询群聊历史消息（游标拉取）
  64 + *
  65 + * @param userId 当前用户编号
  66 + * @param reqVO 拉取请求
  67 + * @return 消息列表（按 id 倒序）
  68 + */
  69 + List<ImGroupMessageDO> getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO);
  70 +
60 71 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
  View file @26413e0
1 1 package cn.iocoder.yudao.module.im.service.message;
2 2  
3 3 import cn.hutool.core.collection.CollUtil;
  4 +import cn.hutool.core.util.IdUtil;
  5 +import cn.hutool.core.util.ObjUtil;
  6 +import cn.hutool.extra.spring.SpringUtil;
  7 +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
4 8 import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
  9 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  10 +import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
  11 +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
  12 +import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO;
5 13 import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO;
6   -import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
7 14 import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
8 15 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
  16 +import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
9 17 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
10   -import cn.iocoder.yudao.module.im.dal.redis.group.GroupReadPositionRedisDAO;
11   -import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
  18 +import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO;
12 19 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
13 20 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
14   -import cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum;
15 21 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
16   -import cn.iocoder.yudao.module.im.websocket.ImGroupMessageSendMessage;
17   -import cn.iocoder.yudao.module.im.websocket.ImMessageReadMessage;
18   -import cn.iocoder.yudao.module.im.websocket.ImMessageReceiptMessage;
19   -import cn.iocoder.yudao.module.im.websocket.ImMessageRecallMessage;
20   -import cn.iocoder.yudao.module.im.service.group.ImGroupService;
21 22 import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService;
  23 +import cn.iocoder.yudao.module.im.service.group.ImGroupService;
22 24 import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
23   -import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender;
  25 +import cn.iocoder.yudao.module.im.websocket.ImGroupMessageDTO;
24 26 import jakarta.annotation.Resource;
25 27 import lombok.extern.slf4j.Slf4j;
  28 +import org.springframework.scheduling.annotation.Async;
26 29 import org.springframework.stereotype.Service;
  30 +import org.springframework.transaction.annotation.Transactional;
27 31 import org.springframework.validation.annotation.Validated;
28 32  
29 33 import java.time.LocalDateTime;
30 34 import java.util.*;
31   -import java.util.stream.Collectors;
32 35  
33 36 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
  37 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
34 38 import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
35   -import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*;
  39 +import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_MAX_PULL_SIZE;
  40 +import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_RECALL_TIMEOUT_MINUTES;
36 41  
37 42 /**
38 43 * IM 群聊消息 Service 实现类
... ... @@ -45,60 +50,51 @@ import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*;
45 50 public class ImGroupMessageServiceImpl implements ImGroupMessageService {
46 51  
47 52 @Resource
48   - private ImGroupMessageMapper imGroupMessageMapper;
  53 + private ImGroupMessageMapper groupMessageMapper;
  54 + // TODO @AI：变量名，可能要改下；
49 55 @Resource
50   - private ImGroupService imGroupService;
  56 + private GroupReadPositionRedisDAO groupReadPositionRedisDAO;
  57 +
51 58 @Resource
52   - private ImGroupMemberService imGroupMemberService;
  59 + private ImGroupService groupService;
53 60 @Resource
54   - private ImSensitiveWordService imSensitiveWordService;
  61 + private ImGroupMemberService groupMemberService;
55 62 @Resource
56   - private GroupReadPositionRedisDAO groupReadPositionRedisDAO;
  63 + private ImSensitiveWordService sensitiveWordService;
  64 +
57 65 @Resource
58 66 private WebSocketMessageSender webSocketMessageSender;
59 67  
60 68 @Override
61 69 public ImGroupMessageDO sendGroupMessage(Long senderId, ImGroupMessageSendReqVO reqVO) {
62   - // 1. 幂等校验
63   - ImGroupMessageDO existing = imGroupMessageMapper.selectBySenderIdAndClientMessageId(
  70 + // 1.1 幂等校验：根据 senderId + clientMessageId 查重
  71 + ImGroupMessageDO existing = groupMessageMapper.selectBySenderIdAndClientMessageId(
64 72 senderId, reqVO.getClientMessageId());
65 73 if (existing != null) {
  74 + log.info("[sendGroupMessage][幂等命中 senderId({}) clientMessageId({}) 已存在消息({})]",
  75 + senderId, reqVO.getClientMessageId(), existing.getId());
66 76 return existing;
67 77 }
68   -
69   - // 2. 校验群存在且未解散
70   - ImGroupDO group = imGroupService.validateGroupExists(reqVO.getGroupId());
71   -
72   - // 3. 校验发送人在群中
73   - ImGroupMemberDO member = imGroupMemberService.getGroupMember(reqVO.getGroupId(), senderId);
74   - if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) {
75   - throw exception(GROUP_MEMBER_NOT_IN_GROUP);
76   - }
77   -
78   - // 4. 文本消息敏感词过滤
  78 + // 1.2 校验群存在、发送人仍在群中
  79 + groupService.validateGroupExists(reqVO.getGroupId());
  80 + validateMemberInGroup(reqVO.getGroupId(), senderId);
  81 + // 1.3 文本消息敏感词过滤
79 82 if (ImMessageTypeEnum.TEXT.getType().equals(reqVO.getType())) {
80   - imSensitiveWordService.validateText(reqVO.getContent());
  83 + sensitiveWordService.validateText(reqVO.getContent());
81 84 }
82 85  
83   - // 5. 构建消息
84   - ImGroupMessageDO message = ImGroupMessageDO.builder()
85   - .clientMessageId(reqVO.getClientMessageId())
86   - .senderId(senderId)
87   - .groupId(reqVO.getGroupId())
88   - .type(reqVO.getType())
89   - .content(reqVO.getContent())
90   - .status(ImMessageStatusEnum.UNREAD.getStatus()) // 群聊正常状态 = 0
91   - .sendTime(LocalDateTime.now())
92   - .atUserIds(reqVO.getAtUserIds())
93   - .receiverUserIds(reqVO.getReceiverUserIds())
94   - .receiptStatus(Boolean.TRUE.equals(reqVO.getNeedReceipt())
  86 + // 2. 构建并保存消息
  87 + // TODO @AI：content 要 json 下；
  88 + 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())
95 92 ? ImGroupMessageReceiptStatusEnum.PENDING.getStatus()
96   - : ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus())
97   - .build();
98   - imGroupMessageMapper.insert(message);
  93 + : ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus()));
  94 + groupMessageMapper.insert(message);
99 95  
100   - // 6. WebSocket 推送给群成员
101   - sendGroupMessageEvent(message, reqVO.getGroupId());
  96 + // 3. WebSocket 推送给群成员（异步）
  97 + getSelf().sendGroupMessageEvent(message);
102 98 return message;
103 99 }
104 100  
... ... @@ -107,272 +103,288 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
107 103 if (size > MESSAGE_MAX_PULL_SIZE) {
108 104 throw exception(MESSAGE_PULL_SIZE_EXCEEDED);
109 105 }
110   - // 获取用户所在的所有群
111   - List<ImGroupMemberDO> members = imGroupMemberService.getGroupMembersByUserId(userId);
  106 + // 1. 获取用户所在的所有群
  107 + List<ImGroupMemberDO> members = groupMemberService.getGroupMembersByUserId(userId);
112 108 if (CollUtil.isEmpty(members)) {
113 109 return Collections.emptyList();
114 110 }
115   - List<Long> groupIds = members.stream()
116   - .map(ImGroupMemberDO::getGroupId)
117   - .collect(Collectors.toList());
118   -
119   - // 拉取消息
120   - List<ImGroupMessageDO> messages = imGroupMessageMapper.selectListByMinId(groupIds, minId, size);
121   -
122   - // 过滤：入群前消息不可见、定向接收过滤、退群后消息边界
123   - Map<Long, ImGroupMemberDO> memberMap = members.stream()
124   - .collect(Collectors.toMap(ImGroupMemberDO::getGroupId, m -> m, (a, b) -> a));
125   -
126   - return messages.stream().filter(msg -> {
127   - ImGroupMemberDO m = memberMap.get(msg.getGroupId());
128   - if (m == null) {
129   - return false;
130   - }
131   - // 入群前消息不可见
132   - if (m.getJoinTime() != null && msg.getSendTime().isBefore(m.getJoinTime())) {
133   - return false;
134   - }
135   - // 退群后消息边界
136   - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus())
137   - && m.getQuitTime() != null && msg.getSendTime().isAfter(m.getQuitTime())) {
138   - return false;
139   - }
140   - // 定向接收过滤
141   - if (CollUtil.isNotEmpty(msg.getReceiverUserIds())) {
142   - if (!msg.getReceiverUserIds().contains(userId) && !msg.getSenderId().equals(userId)) {
143   - return false;
144   - }
145   - }
146   - return true;
147   - }).collect(Collectors.toList());
  111 + List<Long> groupIds = convertList(members, ImGroupMemberDO::getGroupId);
  112 +
  113 + // 2. 拉取消息
  114 + List<ImGroupMessageDO> messages = groupMessageMapper.selectListByMinId(groupIds, minId, size);
  115 +
  116 + // 3. 过滤：入群前不可见、退群后不可见、定向接收过滤
  117 + // TODO @AI：缺少各种逻辑；
  118 + Map<Long, ImGroupMemberDO> memberMap = CollUtil.toMap(members, new HashMap<>(), ImGroupMemberDO::getGroupId);
  119 + List<ImGroupMessageDO> result = messages.stream()
  120 + .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId))
  121 + .toList();
  122 + log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]",
  123 + userId, minId, size, result.size());
  124 + return result;
148 125 }
149 126  
  127 + // TODO @AI：增加 messageId 参数；允许空，则查询最大的一条；
150 128 @Override
151 129 public void readGroupMessages(Long userId, Long groupId) {
152   - // 1. 校验用户在群中（权限校验）
153   - ImGroupMemberDO member = imGroupMemberService.getGroupMember(groupId, userId);
154   - if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) {
155   - throw exception(GROUP_MEMBER_NOT_IN_GROUP);
  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;
156 136 }
157   -
158   - // 2. 高效获取群内最新消息 id（倒序 LIMIT 1，非全量扫描）
159   - Long maxMessageId = imGroupMessageMapper.selectMaxIdByGroupId(groupId);
160   - if (maxMessageId == null) {
  137 + // 1.3 已读位置未前进，直接返回
  138 + // TODO @AI：1.3 和 2 是不是可以合并？如果 update 失败，则直接回退；不然读取两次，貌似有 cas 问题；
  139 + Long prevMaxId = groupReadPositionRedisDAO.getReadPosition(groupId, userId);
  140 + if (newMaxId <= prevMaxId) {
161 141 return;
162 142 }
163 143  
164   - // 3. 更新 Redis 群已读位置
165   - groupReadPositionRedisDAO.updateReadPosition(groupId, userId, maxMessageId);
  144 + // 2. 更新 Redis 群已读位置
  145 + groupReadPositionRedisDAO.updateReadPosition(groupId, userId, newMaxId);
166 146  
167   - // 4. 发送 READ 事件给自己的其他终端
168   - ImMessageReadMessage readMessage = new ImMessageReadMessage()
169   - .setGroupId(groupId)
170   - .setScene(ImMessageSceneEnum.GROUP.getScene());
171   - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
172   - ImMessageReadMessage.TYPE, readMessage);
173   -
174   - // 5. 刷新群回执
175   - refreshGroupReceipts(groupId);
  147 + // 3. 异步发送 READ 事件 + 刷新范围内的群回执
  148 + getSelf().readGroupMessageEvent(userId, groupId, prevMaxId, newMaxId);
176 149 }
177 150  
178 151 @Override
179   - public void recallGroupMessage(Long userId, Long messageId) {
180   - // 1. 校验消息存在
181   - ImGroupMessageDO message = imGroupMessageMapper.selectById(messageId);
  152 + @Transactional(rollbackFor = Exception.class)
  153 + public ImGroupMessageDO recallGroupMessage(Long userId, Long messageId) {
  154 + // 1.1 校验消息存在
  155 + // TODO @AI：抽成一个 validateXXXExists 方法；
  156 + ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
182 157 if (message == null) {
183 158 throw exception(MESSAGE_NOT_EXISTS);
184 159 }
185   - // 2. 只能撤回自己发送的消息
186   - if (!message.getSenderId().equals(userId)) {
  160 + // 1.2 只能撤回自己发送的消息
  161 + if (ObjUtil.notEqual(message.getSenderId(), userId)) {
187 162 throw exception(MESSAGE_RECALL_DENIED);
188 163 }
189   - // 3. 不能重复撤回
  164 + // 1.3 不能重复撤回
190 165 if (ImMessageStatusEnum.RECALL.getStatus().equals(message.getStatus())) {
191 166 throw exception(MESSAGE_ALREADY_RECALLED);
192 167 }
193   - // 4. 更新消息状态
194   - ImGroupMessageDO updateObj = new ImGroupMessageDO();
195   - updateObj.setId(messageId);
196   - updateObj.setStatus(ImMessageStatusEnum.RECALL.getStatus());
197   - imGroupMessageMapper.updateById(updateObj);
198   -
199   - // 5. 发送 RECALL 事件给所有群成员
200   - List<ImGroupMemberDO> members = imGroupMemberService.selectByGroupId(message.getGroupId());
201   - ImMessageRecallMessage recallMessage = new ImMessageRecallMessage()
202   - .setMessageId(messageId)
203   - .setGroupId(message.getGroupId())
204   - .setSenderId(userId)
205   - .setScene(ImMessageSceneEnum.GROUP.getScene());
206   - for (ImGroupMemberDO m : members) {
207   - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus())) {
208   - continue;
209   - }
210   - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), m.getUserId(),
211   - ImMessageRecallMessage.TYPE, recallMessage);
  168 + // 1.4 只允许撤回限定时间内的消息
  169 + if (message.getSendTime().plusMinutes(MESSAGE_RECALL_TIMEOUT_MINUTES).isBefore(LocalDateTime.now())) {
  170 + throw exception(MESSAGE_RECALL_TIMEOUT, MESSAGE_RECALL_TIMEOUT_MINUTES);
212 171 }
  172 + // 1.5 校验撤回人仍在群中
  173 + validateMemberInGroup(message.getGroupId(), userId);
  174 +
  175 + // 2. 更新原消息状态为撤回
  176 + groupMessageMapper.updateById(new ImGroupMessageDO().setId(messageId)
  177 + .setStatus(ImMessageStatusEnum.RECALL.getStatus()));
  178 +
  179 + // 3. 插入一条 TIP_TEXT 消息作为撤回提示
  180 + RecallMessage recallContent = new RecallMessage().setMessageId(messageId);
  181 + ImGroupMessageDO recallMessage = new ImGroupMessageDO().setClientMessageId(IdUtil.fastSimpleUUID())
  182 + .setSenderId(userId).setGroupId(message.getGroupId())
  183 + .setType(ImMessageTypeEnum.TIP_TEXT.getType()).setContent(JsonUtils.toJsonString(recallContent))
  184 + .setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now())
  185 + .setReceiptStatus(ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus());
  186 + groupMessageMapper.insert(recallMessage);
  187 +
  188 + // 4. 异步推送撤回提示消息（前端据此更新原消息状态 + 插入撤回提示）
  189 + getSelf().sendGroupMessageEvent(recallMessage);
  190 + return recallMessage;
213 191 }
214 192  
215 193 @Override
216 194 public List<Long> getGroupReadUsers(Long userId, Long groupId, Long messageId) {
217   - // 1. 校验用户在群中（权限校验）
218   - ImGroupMemberDO currentMember = imGroupMemberService.getGroupMember(groupId, userId);
219   - if (currentMember == null || CommonStatusEnum.DISABLE.getStatus().equals(currentMember.getStatus())) {
220   - throw exception(GROUP_MEMBER_NOT_IN_GROUP);
221   - }
222   -
223   - // 2. 获取消息
224   - ImGroupMessageDO message = imGroupMessageMapper.selectById(messageId);
  195 + // 1.1 校验用户在群中（权限校验）
  196 + validateMemberInGroup(groupId, userId);
  197 + // 1.2 获取消息
  198 + ImGroupMessageDO message = groupMessageMapper.selectById(messageId);
225 199 if (message == null) {
226 200 return Collections.emptyList();
227 201 }
228 202  
229 203 // 2. 获取所有成员和已读位置
230   - List<ImGroupMemberDO> allMembers = imGroupMemberService.selectByGroupId(groupId);
  204 + List<ImGroupMemberDO> allMembers = groupMemberService.selectByGroupId(groupId);
231 205 Map<Object, Object> allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId);
232 206  
233   - // 3. 计算该消息的可见成员集合
  207 + // 3. 计算该消息的可见成员集合（排除发送者自己）
234 208 Set<Long> visibleUserIds = getVisibleUserIds(message, allMembers);
  209 + visibleUserIds.remove(message.getSenderId());
235 210  
236 211 // 4. 只返回在可见范围内且已读位置 >= messageId 的用户
237 212 List<Long> readUserIds = new ArrayList<>();
238   - for (Map.Entry<Object, Object> entry : allPositions.entrySet()) {
239   - Long uid = Long.parseLong(entry.getKey().toString());
240   - Long readMaxId = Long.parseLong(entry.getValue().toString());
241   - if (readMaxId >= messageId && visibleUserIds.contains(uid)
242   - && !uid.equals(message.getSenderId())) { // 排除发送者自己
  213 + allPositions.forEach((k, v) -> {
  214 + Long uid = Long.valueOf(k.toString());
  215 + if (visibleUserIds.contains(uid) && Long.parseLong(v.toString()) >= messageId) {
243 216 readUserIds.add(uid);
244 217 }
245   - }
  218 + });
246 219 return readUserIds;
247 220 }
248 221  
  222 + @Override
  223 + public List<ImGroupMessageDO> getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO) {
  224 + // 1. 校验用户在群中
  225 + ImGroupMemberDO member = validateMemberInGroup(reqVO.getGroupId(), userId);
  226 +
  227 + // 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();
  237 + }
  238 +
  239 + // ========== 异步 WebSocket 推送 ==========
  240 +
  241 + /**
  242 + * 发送群聊消息 WebSocket 事件
  243 + */
  244 + @Async
  245 + public void sendGroupMessageEvent(ImGroupMessageDO message) {
  246 + ImGroupMessageDTO dto = ImGroupMessageDTO.ofSend(message);
  247 + // 广播给群内可见成员（含发送方自己，支持多端同步）
  248 + // TODO @AI：getGroupMemberListByGroupId()；默认过滤有效的；
  249 + List<ImGroupMemberDO> members = groupMemberService.selectByGroupId(message.getGroupId());
  250 + List<Long> receiverList = message.getReceiverUserIds();
  251 + for (ImGroupMemberDO member : members) {
  252 + if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) {
  253 + continue;
  254 + }
  255 + // TODO @AI：这里是不是不用过滤啊？因为一般 send 的时候，肯定带了呀。按照群友就好了；
  256 + if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId())
  257 + && ObjUtil.notEqual(member.getUserId(), message.getSenderId())) {
  258 + continue;
  259 + }
  260 + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), member.getUserId(),
  261 + ImGroupMessageDTO.TYPE, dto);
  262 + }
  263 + }
  264 +
  265 + // TODO @AI：命名最后都是 pre 和 new 里面最好都有 messageId 的标识；包括上面的地方，更好理解吧！
249 266 /**
250   - * 刷新群回执（仅处理 PENDING 状态的消息，无需全量扫描）
  267 + * 发送已读 + 刷新群回执 WebSocket 事件
  268 + *
  269 + * @param userId 当前用户编号
  270 + * @param groupId 群编号
  271 + * @param prevMaxId 上次已读位置
  272 + * @param newMaxId 本次已读位置
251 273 */
252   - private void refreshGroupReceipts(Long groupId) {
253   - // 1. 只查 PENDING 状态的回执消息
254   - List<ImGroupMessageDO> pendingMessages = imGroupMessageMapper.selectPendingReceiptMessages(groupId);
  274 + @Async
  275 + public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxId, Long newMaxId) {
  276 + // 1. 发送 READ 事件给自己的其他终端（多端同步）
  277 + ImGroupMessageDTO readDto = ImGroupMessageDTO.ofRead(userId, groupId, newMaxId);
  278 + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
  279 + ImGroupMessageDTO.TYPE, readDto);
  280 +
  281 + // 2. 刷新 (prevMaxId, newMaxId] 区间内的待回执消息
  282 + // TODO @AI：selectListByXXX；注意，prevMaxId 可能为空的情况；
  283 + List<ImGroupMessageDO> pendingMessages = groupMessageMapper.selectPendingReceiptMessagesInRange(
  284 + groupId, prevMaxId, newMaxId);
255 285 if (CollUtil.isEmpty(pendingMessages)) {
256 286 return;
257 287 }
258   -
259   - // 2. 加载成员和已读位置
260   - List<ImGroupMemberDO> allMembers = imGroupMemberService.selectByGroupId(groupId);
  288 + // TODO @AI：是不是要过滤 status ？因为可能退出了。
  289 + List<ImGroupMemberDO> allMembers = groupMemberService.selectByGroupId(groupId);
  290 + // TODO @AI：变量也改下；
261 291 Map<Object, Object> allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId);
262   -
263   - for (ImGroupMessageDO msg : pendingMessages) {
264   - // 3. 计算该消息的可见成员（排除发送者自己）
265   - Set<Long> visibleUserIds = getVisibleUserIds(msg, allMembers);
266   - visibleUserIds.remove(msg.getSenderId()); // 发送者不计入回执统计
267   -
  292 + for (ImGroupMessageDO message : pendingMessages) {
  293 + // 2.1 计算该消息的可见成员（排除发送者自己）
  294 + Set<Long> visibleUserIds = getVisibleUserIds(message, allMembers);
  295 + visibleUserIds.remove(message.getSenderId());
  296 + // TODO @AI：CollUtil 来判断；
268 297 if (visibleUserIds.isEmpty()) {
269 298 continue;
270 299 }
271   -
272   - // 4. 统计可见成员中的已读人数
  300 + // 2.2 统计可见成员中的已读人数
273 301 int readCount = 0;
274 302 for (Long uid : visibleUserIds) {
  303 + // TODO @AI：根据返回结果，进行简化；应该都是 Long、Long；
275 304 Object val = allPositions.get(uid.toString());
276   - if (val != null && Long.parseLong(val.toString()) >= msg.getId()) {
  305 + if (val != null && Long.parseLong(val.toString()) >= message.getId()) {
277 306 readCount++;
278 307 }
279 308 }
280   -
281   - // 5. 判断是否全部已读
  309 + // 2.3 全部已读 → 标记回执完成
282 310 boolean allRead = readCount >= visibleUserIds.size();
  311 + Integer newReceiptStatus = allRead
  312 + ? ImGroupMessageReceiptStatusEnum.DONE.getStatus()
  313 + : ImGroupMessageReceiptStatusEnum.PENDING.getStatus();
283 314 if (allRead) {
284   - ImGroupMessageDO update = new ImGroupMessageDO();
285   - update.setId(msg.getId());
286   - update.setReceiptStatus(ImGroupMessageReceiptStatusEnum.DONE.getStatus());
287   - imGroupMessageMapper.updateById(update);
  315 + groupMessageMapper.updateById(new ImGroupMessageDO().setId(message.getId())
  316 + .setReceiptStatus(newReceiptStatus));
288 317 }
  318 + // 2.4 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度）
  319 + // TODO @AI：检查下别的 Dto = 的地方，都最好改成 DTO = ；仅限 im 模块；
  320 + ImGroupMessageDTO receiptDTO = ImGroupMessageDTO.ofReceipt(
  321 + message.getId(), groupId, readCount, newReceiptStatus);
  322 + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(),
  323 + ImGroupMessageDTO.TYPE, receiptDTO);
  324 + }
  325 + }
289 326  
290   - // 6. 广播 RECEIPT 事件给活跃成员
291   - ImMessageReceiptMessage receiptMessage = new ImMessageReceiptMessage()
292   - .setMessageId(msg.getId())
293   - .setGroupId(groupId)
294   - .setReadCount(readCount)
295   - .setReceiptStatus(allRead
296   - ? ImGroupMessageReceiptStatusEnum.DONE.getStatus()
297   - : ImGroupMessageReceiptStatusEnum.PENDING.getStatus())
298   - .setScene(ImMessageSceneEnum.GROUP.getScene());
299   -
300   - for (ImGroupMemberDO m : allMembers) {
301   - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus())) {
302   - continue;
303   - }
304   - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), m.getUserId(),
305   - ImMessageReceiptMessage.TYPE, receiptMessage);
306   - }
  327 + // ========== 私有工具方法 ==========
  328 +
  329 + // TODO @AI：抽到对应的 GroupMemberService 中，供其他地方复用
  330 + /**
  331 + * 校验用户是否为群的有效成员
  332 + *
  333 + * @param groupId 群编号
  334 + * @param userId 用户编号
  335 + * @return 群成员记录
  336 + */
  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);
307 341 }
  342 + return member;
308 343 }
309 344  
310 345 /**
311   - * 计算一条群消息的可见成员集合
  346 + * 判断一条群消息对某个群成员是否可见
312 347 * <p>
313 348 * 规则：
314   - * 1. 成员必须是 NORMAL 状态，或虽已 QUIT 但消息发送时间在退群之前
315   - * 2. 成员的入群时间必须在消息发送之前（入群后才能看到消息）
316   - * 3. 如果消息有 receiverUserIds 定向接收，则只有列表中的人可见
  349 + * 1. 入群时间晚于消息发送时间 → 不可见
  350 + * 2. 已退群且退群时间早于消息发送时间 → 不可见
  351 + * 3. 存在定向接收列表且当前用户既不在列表中也不是发送者 → 不可见
317 352 */
318   - private Set<Long> getVisibleUserIds(ImGroupMessageDO msg, List<ImGroupMemberDO> allMembers) {
319   - // 定向接收集合
320   - List<Long> receiverList = msg.getReceiverUserIds();
321   -
322   - Set<Long> visible = new HashSet<>();
323   - for (ImGroupMemberDO m : allMembers) {
324   - // 入群时间在消息发送之后 → 不可见
325   - if (m.getJoinTime() != null && msg.getSendTime().isBefore(m.getJoinTime())) {
326   - continue;
327   - }
328   - // 已退群且退群时间在消息发送之前 → 不可见
329   - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus())
330   - && m.getQuitTime() != null && msg.getSendTime().isAfter(m.getQuitTime())) {
331   - continue;
332   - }
333   - // 定向接收过滤
334   - if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(m.getUserId())
335   - && !m.getUserId().equals(msg.getSenderId())) {
336   - continue;
337   - }
338   - visible.add(m.getUserId());
  353 + private boolean isMessageVisible(ImGroupMessageDO msg, ImGroupMemberDO member, Long userId) {
  354 + if (member == null) {
  355 + return false;
339 356 }
340   - return visible;
  357 + if (member.getJoinTime() != null && msg.getSendTime().isBefore(member.getJoinTime())) {
  358 + return false;
  359 + }
  360 + if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())
  361 + && member.getQuitTime() != null && msg.getSendTime().isAfter(member.getQuitTime())) {
  362 + return false;
  363 + }
  364 + if (CollUtil.isNotEmpty(msg.getReceiverUserIds())
  365 + && !msg.getReceiverUserIds().contains(userId)
  366 + && ObjUtil.notEqual(msg.getSenderId(), userId)) {
  367 + return false;
  368 + }
  369 + return true;
341 370 }
342 371  
343 372 /**
344   - * 发送群聊消息 WebSocket 事件
  373 + * 计算一条群消息的可见成员集合（含发送者）
345 374 */
346   - private void sendGroupMessageEvent(ImGroupMessageDO message, Long groupId) {
347   - List<ImGroupMemberDO> members = imGroupMemberService.selectByGroupId(groupId);
348   - ImGroupMessageSendMessage sendMessage = new ImGroupMessageSendMessage()
349   - .setId(message.getId())
350   - .setClientMessageId(message.getClientMessageId())
351   - .setSenderId(message.getSenderId())
352   - .setGroupId(message.getGroupId())
353   - .setType(message.getType())
354   - .setContent(message.getContent())
355   - .setStatus(message.getStatus())
356   - .setSendTime(message.getSendTime())
357   - .setAtUserIds(message.getAtUserIds())
358   - .setReceiverUserIds(message.getReceiverUserIds())
359   - .setReceiptStatus(message.getReceiptStatus());
360   -
361   - // 确定推送目标
362   - List<Long> receiverList = message.getReceiverUserIds();
363   -
364   - for (ImGroupMemberDO member : members) {
365   - if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) {
366   - continue;
367   - }
368   - // 定向接收过滤
369   - if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId())
370   - && !member.getUserId().equals(message.getSenderId())) {
371   - continue;
  375 + 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());
372 381 }
373   - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), member.getUserId(),
374   - ImGroupMessageSendMessage.TYPE, sendMessage);
375 382 }
  383 + return visible;
  384 + }
  385 +
  386 + private ImGroupMessageServiceImpl getSelf() {
  387 + return SpringUtil.getBean(getClass());
376 388 }
377 389  
378 390 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java
  View file @26413e0
... ... @@ -22,6 +22,7 @@ import jakarta.annotation.Resource;
22 22 import lombok.extern.slf4j.Slf4j;
23 23 import org.springframework.scheduling.annotation.Async;
24 24 import org.springframework.stereotype.Service;
  25 +import org.springframework.transaction.annotation.Transactional;
25 26 import org.springframework.validation.annotation.Validated;
26 27  
27 28 import java.time.LocalDateTime;
... ... @@ -92,6 +93,7 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
92 93 return messages;
93 94 }
94 95  
  96 + // TODO @AI：考虑到更稳妥，是不是前端传递 messageId？不然恰好发过来，然后读取了。【后续在优化】
95 97 @Override
96 98 public void readPrivateMessages(Long userId, Long receiverId) {
97 99 // 1.1 查询未读消息列表
... ... @@ -111,6 +113,7 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
111 113 }
112 114  
113 115 @Override
  116 + @Transactional(rollbackFor = Exception.class)
114 117 public ImPrivateMessageDO recallPrivateMessage(Long userId, Long messageId) {
115 118 // 1.1 校验消息存在
116 119 ImPrivateMessageDO message = privateMessageMapper.selectById(messageId);
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImGroupMessageSendMessage.java renamed to yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImGroupMessageDTO.java
  View file @26413e0
1 1 package cn.iocoder.yudao.module.im.websocket;
2 2  
  3 +import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
  4 +import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
  5 +import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
3 6 import lombok.Data;
4   -import lombok.experimental.Accessors;
5 7  
6 8 import java.time.LocalDateTime;
7 9 import java.util.List;
8 10  
9 11 /**
10   - * IM 群聊消息 WebSocket 推送消息
  12 + * IM 群聊消息 WebSocket 统一推送 DTO
  13 + * <p>
  14 + * 发送、已读、回执都通过 {@link #type} 字段区分，
  15 + * 并统一复用 {@link #id} 字段表达目标消息或已读位置。
11 16 *
12 17 * @author 芋道源码
13 18 */
14 19 @Data
15   -@Accessors(chain = true)
16   -public class ImGroupMessageSendMessage {
  20 +public class ImGroupMessageDTO {
17 21  
18   - public static final String TYPE = "im-group-message-send";
  22 + public static final String TYPE = "im-group-message";
19 23  
20 24 /**
21 25 * 消息编号
  26 + * <p>
  27 + * 普通消息：当前消息 id
  28 + * READ：无（群已读不需要）
  29 + * RECEIPT：需要回执的目标消息 id
22 30 */
23 31 private Long id;
24 32 /**
... ... @@ -50,7 +58,7 @@ public class ImGroupMessageSendMessage {
50 58 */
51 59 private LocalDateTime sendTime;
52 60 /**
53   - * @ 的用户编号列表
  61 + * @ 目标用户编号列表
54 62 */
55 63 private List<Long> atUserIds;
56 64 /**
... ... @@ -61,5 +69,51 @@ public class ImGroupMessageSendMessage {
61 69 * 群回执状态
62 70 */
63 71 private Integer receiptStatus;
  72 + /**
  73 + * 群回执已读人数
  74 + */
  75 + private Integer readCount;
  76 +
  77 + // ========== 静态工厂方法 ==========
  78 +
  79 + /**
  80 + * 构建发送消息 DTO
  81 + *
  82 + * @param message 群聊消息 DO
  83 + * @return 群聊 DTO
  84 + */
  85 + public static ImGroupMessageDTO ofSend(ImGroupMessageDO message) {
  86 + return BeanUtils.toBean(message, ImGroupMessageDTO.class);
  87 + }
  88 +
  89 + /**
  90 + * 构建已读同步 DTO（多端同步：通知自己的其他终端"我已经读了某个群"）
  91 + *
  92 + * @param senderId 当前用户编号
  93 + * @param groupId 群编号
  94 + * @param readId 已读位置（最大已读消息编号）
  95 + * @return 群聊 DTO
  96 + */
  97 + public static ImGroupMessageDTO ofRead(Long senderId, Long groupId, Long readId) {
  98 + return new ImGroupMessageDTO()
  99 + .setId(readId).setType(ImMessageTypeEnum.READ.getType())
  100 + .setSenderId(senderId).setGroupId(groupId);
  101 + }
  102 +
  103 + /**
  104 + * 构建群回执 DTO（广播回执状态和已读人数）
  105 + *
  106 + * @param messageId 消息编号
  107 + * @param groupId 群编号
  108 + * @param readCount 已读人数
  109 + * @param receiptStatus 回执状态
  110 + * @return 群聊 DTO
  111 + */
  112 + public static ImGroupMessageDTO ofReceipt(Long messageId, Long groupId,
  113 + Integer readCount, Integer receiptStatus) {
  114 + return new ImGroupMessageDTO()
  115 + .setId(messageId).setType(ImMessageTypeEnum.RECEIPT.getType())
  116 + .setGroupId(groupId).setReadCount(readCount).setReceiptStatus(receiptStatus);
  117 + }
64 118  
65 119 }
... ...
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReadMessage.java deleted
View file @37cf16c
1   -package cn.iocoder.yudao.module.im.websocket;
2   -
3   -import lombok.Data;
4   -import lombok.experimental.Accessors;
5   -
6   -/**
7   - * IM 消息已读 WebSocket 推送消息
8   - * <p>
9   - * 用于多端同步：通知自己的其他终端"我已经读了某个会话"
10   - *
11   - * @author 芋道源码
12   - */
13   -@Data
14   -@Accessors(chain = true)
15   -public class ImMessageReadMessage {
16   -
17   - public static final String TYPE = "im-message-read";
18   -
19   - /**
20   - * 私聊场景：接收方编号
21   - */
22   - private Long receiverId;
23   - /**
24   - * 群聊场景：群编号
25   - */
26   - private Long groupId;
27   - /**
28   - * 消息场景
29   - *
30   - * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum}
31   - */
32   - private Integer scene;
33   -
34   -}
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageRecallMessage.java deleted
View file @37cf16c
1   -package cn.iocoder.yudao.module.im.websocket;
2   -
3   -import lombok.Data;
4   -import lombok.experimental.Accessors;
5   -
6   -/**
7   - * IM 消息撤回 WebSocket 推送消息
8   - *
9   - * @author 芋道源码
10   - */
11   -@Data
12   -@Accessors(chain = true)
13   -public class ImMessageRecallMessage {
14   -
15   - public static final String TYPE = "im-message-recall";
16   -
17   - /**
18   - * 消息编号
19   - */
20   - private Long messageId;
21   - /**
22   - * 发送人编号
23   - */
24   - private Long senderId;
25   - /**
26   - * 群聊场景：群编号
27   - */
28   - private Long groupId;
29   - /**
30   - * 消息场景
31   - *
32   - * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum}
33   - */
34   - private Integer scene;
35   -
36   -}
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReceiptMessage.java deleted
View file @37cf16c
1   -package cn.iocoder.yudao.module.im.websocket;
2   -
3   -import lombok.Data;
4   -import lombok.experimental.Accessors;
5   -
6   -/**
7   - * IM 已读回执 WebSocket 推送消息
8   - * <p>
9   - * 私聊：通知对方"我已读了你的消息"
10   - * 群聊：广播回执状态和已读人数
11   - *
12   - * @author 芋道源码
13   - */
14   -@Data
15   -@Accessors(chain = true)
16   -public class ImMessageReceiptMessage {
17   -
18   - public static final String TYPE = "im-message-receipt";
19   -
20   - /**
21   - * 消息场景
22   - *
23   - * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum}
24   - */
25   - private Integer scene;
26   -
27   - // ========== 私聊场景 ==========
28   -
29   - /**
30   - * 私聊场景：已读方的用户编号
31   - */
32   - private Long receiverId;
33   -
34   - // ========== 群聊场景 ==========
35   -
36   - /**
37   - * 群聊场景：消息编号
38   - */
39   - private Long messageId;
40   - /**
41   - * 群聊场景：群编号
42   - */
43   - private Long groupId;
44   - /**
45   - * 群聊场景：已读人数
46   - */
47   - private Integer readCount;
48   - /**
49   - * 群聊场景：回执状态
50   - */
51   - private Integer receiptStatus;
52   -
53   -}
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImPrivateMessageDTO.java
  View file @26413e0
... ... @@ -45,8 +45,6 @@ public class ImPrivateMessageDTO {
45 45 private Long receiverId;
46 46 /**
47 47 * 消息类型
48   - * <p>
49   - * TEXT / IMAGE / FILE / VOICE / VIDEO / READ / RECEIPT / RECALL / TIP_TEXT
50 48 */
51 49 private Integer type;
52 50 /**
... ...
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java
  View file @26413e0
1 1 package cn.iocoder.yudao.module.im.service.message;
2 2  
  3 +import cn.hutool.extra.spring.SpringUtil;
3 4 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
4 5 import cn.iocoder.yudao.framework.common.exception.ServiceException;
5 6 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
... ... @@ -9,7 +10,7 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
9 10 import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
10 11 import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO;
11 12 import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper;
12   -import cn.iocoder.yudao.module.im.dal.redis.group.GroupReadPositionRedisDAO;
  13 +import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO;
13 14 import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum;
14 15 import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
15 16 import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService;
... ... @@ -18,9 +19,12 @@ import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
18 19 import org.junit.jupiter.api.Test;
19 20 import org.mockito.InjectMocks;
20 21 import org.mockito.Mock;
  22 +import org.mockito.MockedStatic;
21 23  
22 24 import java.time.LocalDateTime;
23   -import java.util.*;
  25 +import java.util.HashMap;
  26 +import java.util.List;
  27 +import java.util.Map;
24 28  
25 29 import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
26 30 import static org.junit.jupiter.api.Assertions.*;
... ... @@ -38,13 +42,13 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
38 42 private ImGroupMessageServiceImpl groupMessageService;
39 43  
40 44 @Mock
41   - private ImGroupMessageMapper imGroupMessageMapper;
  45 + private ImGroupMessageMapper groupMessageMapper;
42 46 @Mock
43   - private ImGroupService imGroupService;
  47 + private ImGroupService groupService;
44 48 @Mock
45   - private ImGroupMemberService imGroupMemberService;
  49 + private ImGroupMemberService groupMemberService;
46 50 @Mock
47   - private ImSensitiveWordService imSensitiveWordService;
  51 + private ImSensitiveWordService sensitiveWordService;
48 52 @Mock
49 53 private GroupReadPositionRedisDAO groupReadPositionRedisDAO;
50 54 @Mock
... ... @@ -63,47 +67,50 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
63 67  
64 68 @Test
65 69 public void testSendMessage_success() {
66   - // 准备
67   - ImGroupMessageSendReqVO reqVO = buildSendReqVO();
68   - when(imGroupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001"))
69   - .thenReturn(null);
70   - ImGroupDO group = new ImGroupDO();
71   - group.setId(10L);
72   - group.setName("测试群");
73   - when(imGroupService.validateGroupExists(10L)).thenReturn(group);
74   -
75   - ImGroupMemberDO member = new ImGroupMemberDO();
76   - member.setGroupId(10L);
77   - member.setUserId(1L);
78   - member.setStatus(CommonStatusEnum.ENABLE.getStatus());
79   - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(member);
80   -
81   - List<ImGroupMemberDO> allMembers = List.of(
82   - member,
83   - ImGroupMemberDO.builder().groupId(10L).userId(2L)
84   - .status(CommonStatusEnum.ENABLE.getStatus()).build(),
85   - ImGroupMemberDO.builder().groupId(10L).userId(3L)
86   - .status(CommonStatusEnum.ENABLE.getStatus()).build()
87   - );
88   - when(imGroupMemberService.selectByGroupId(10L)).thenReturn(allMembers);
89   - when(imGroupMessageMapper.insert(any(ImGroupMessageDO.class))).thenAnswer(invocation -> {
90   - ImGroupMessageDO msg = invocation.getArgument(0);
91   - msg.setId(99L);
92   - return 1;
93   - });
94   -
95   - // 调用
96   - ImGroupMessageDO result = groupMessageService.sendGroupMessage(1L, reqVO);
97   -
98   - // 断言
99   - assertNotNull(result);
100   - assertEquals(1L, result.getSenderId());
101   - assertEquals(10L, result.getGroupId());
102   - assertEquals(ImMessageStatusEnum.UNREAD.getStatus(), result.getStatus());
103   - assertEquals(ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus(), result.getReceiptStatus());
104   -
105   - // 验证推送给 3 个群成员
106   - verify(webSocketMessageSender, times(3)).sendObject(anyInt(), anyLong(), anyString(), any());
  70 + try (MockedStatic<SpringUtil> springUtilMockedStatic = mockStatic(SpringUtil.class)) {
  71 + springUtilMockedStatic.when(() -> SpringUtil.getBean(eq(ImGroupMessageServiceImpl.class)))
  72 + .thenReturn(groupMessageService);
  73 +
  74 + // 准备
  75 + ImGroupMessageSendReqVO reqVO = buildSendReqVO();
  76 + when(groupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001"))
  77 + .thenReturn(null);
  78 + ImGroupDO group = new ImGroupDO();
  79 + group.setId(10L);
  80 + group.setName("测试群");
  81 + when(groupService.validateGroupExists(10L)).thenReturn(group);
  82 +
  83 + ImGroupMemberDO member = ImGroupMemberDO.builder()
  84 + .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build();
  85 + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(member);
  86 +
  87 + List<ImGroupMemberDO> allMembers = List.of(
  88 + member,
  89 + ImGroupMemberDO.builder().groupId(10L).userId(2L)
  90 + .status(CommonStatusEnum.ENABLE.getStatus()).build(),
  91 + ImGroupMemberDO.builder().groupId(10L).userId(3L)
  92 + .status(CommonStatusEnum.ENABLE.getStatus()).build()
  93 + );
  94 + when(groupMemberService.selectByGroupId(10L)).thenReturn(allMembers);
  95 + when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenAnswer(invocation -> {
  96 + ImGroupMessageDO msg = invocation.getArgument(0);
  97 + msg.setId(99L);
  98 + return 1;
  99 + });
  100 +
  101 + // 调用
  102 + ImGroupMessageDO result = groupMessageService.sendGroupMessage(1L, reqVO);
  103 +
  104 + // 断言
  105 + assertNotNull(result);
  106 + assertEquals(1L, result.getSenderId());
  107 + assertEquals(10L, result.getGroupId());
  108 + assertEquals(ImMessageStatusEnum.UNREAD.getStatus(), result.getStatus());
  109 + assertEquals(ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus(), result.getReceiptStatus());
  110 +
  111 + // 验证推送给 3 个群成员（含发送者自己，用于多端同步）
  112 + verify(webSocketMessageSender, times(3)).sendObject(anyInt(), anyLong(), anyString(), any());
  113 + }
107 114 }
108 115  
109 116 @Test
... ... @@ -114,7 +121,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
114 121 .id(100L).clientMessageId("test-uuid-group-001").senderId(1L).groupId(10L)
115 122 .type(0).content("{\"content\":\"群聊你好\"}").status(0)
116 123 .sendTime(LocalDateTime.now()).build();
117   - when(imGroupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001"))
  124 + when(groupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001"))
118 125 .thenReturn(existing);
119 126  
120 127 // 调用
... ... @@ -122,19 +129,19 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
122 129  
123 130 // 断言
124 131 assertEquals(100L, result.getId());
125   - verify(imGroupMessageMapper, never()).insert(any(ImGroupMessageDO.class));
  132 + verify(groupMessageMapper, never()).insert(any(ImGroupMessageDO.class));
126 133 }
127 134  
128 135 @Test
129 136 public void testSendMessage_notInGroup() {
130 137 // 准备
131 138 ImGroupMessageSendReqVO reqVO = buildSendReqVO();
132   - when(imGroupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001"))
  139 + when(groupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001"))
133 140 .thenReturn(null);
134 141 ImGroupDO group = new ImGroupDO();
135 142 group.setId(10L);
136   - when(imGroupService.validateGroupExists(10L)).thenReturn(group);
137   - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(null);
  143 + when(groupService.validateGroupExists(10L)).thenReturn(group);
  144 + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null);
138 145  
139 146 // 调用并断言
140 147 ServiceException exception = assertThrows(ServiceException.class,
... ... @@ -152,7 +159,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
152 159 .groupId(10L).userId(1L)
153 160 .status(CommonStatusEnum.ENABLE.getStatus())
154 161 .joinTime(joinTime).build();
155   - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
  162 + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
156 163  
157 164 // 模拟一条入群前的消息和一条入群后的消息
158 165 ImGroupMessageDO beforeJoin = ImGroupMessageDO.builder()
... ... @@ -161,7 +168,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
161 168 ImGroupMessageDO afterJoin = ImGroupMessageDO.builder()
162 169 .id(2L).groupId(10L).senderId(2L)
163 170 .sendTime(LocalDateTime.of(2026, 4, 12, 11, 0, 0)).build();
164   - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
  171 + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
165 172 .thenReturn(List.of(beforeJoin, afterJoin));
166 173  
167 174 // 调用
... ... @@ -181,7 +188,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
181 188 .groupId(10L).userId(1L)
182 189 .status(CommonStatusEnum.DISABLE.getStatus())
183 190 .joinTime(joinTime).quitTime(quitTime).build();
184   - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
  191 + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
185 192  
186 193 // 退群前后各一条消息
187 194 ImGroupMessageDO beforeQuit = ImGroupMessageDO.builder()
... ... @@ -190,7 +197,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
190 197 ImGroupMessageDO afterQuit = ImGroupMessageDO.builder()
191 198 .id(2L).groupId(10L).senderId(2L)
192 199 .sendTime(LocalDateTime.of(2026, 4, 13, 10, 0, 0)).build();
193   - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
  200 + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
194 201 .thenReturn(List.of(beforeQuit, afterQuit));
195 202  
196 203 // 调用
... ... @@ -208,7 +215,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
208 215 .groupId(10L).userId(1L)
209 216 .status(CommonStatusEnum.ENABLE.getStatus())
210 217 .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build();
211   - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
  218 + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
212 219  
213 220 // 定向接收消息：只给用户 2 和 3
214 221 ImGroupMessageDO directedMsg = ImGroupMessageDO.builder()
... ... @@ -219,7 +226,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
219 226 ImGroupMessageDO allMsg = ImGroupMessageDO.builder()
220 227 .id(2L).groupId(10L).senderId(5L)
221 228 .sendTime(LocalDateTime.of(2026, 4, 12, 10, 0, 0)).build();
222   - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
  229 + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
223 230 .thenReturn(List.of(directedMsg, allMsg));
224 231  
225 232 // 调用
... ... @@ -237,13 +244,13 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
237 244 .groupId(10L).userId(1L)
238 245 .status(CommonStatusEnum.ENABLE.getStatus())
239 246 .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build();
240   - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
  247 + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member));
241 248  
242 249 ImGroupMessageDO atMsg = ImGroupMessageDO.builder()
243 250 .id(1L).groupId(10L).senderId(2L)
244 251 .atUserIds(List.of(1L, 3L))
245 252 .sendTime(LocalDateTime.of(2026, 4, 12, 10, 0, 0)).build();
246   - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
  253 + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100))
247 254 .thenReturn(List.of(atMsg));
248 255  
249 256 // 调用
... ... @@ -258,28 +265,88 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
258 265  
259 266 @Test
260 267 public void testRecallMessage_success() {
  268 + try (MockedStatic<SpringUtil> springUtilMockedStatic = mockStatic(SpringUtil.class)) {
  269 + springUtilMockedStatic.when(() -> SpringUtil.getBean(eq(ImGroupMessageServiceImpl.class)))
  270 + .thenReturn(groupMessageService);
  271 +
  272 + // 准备：消息由用户 1 发送，刚发送 5 分钟内
  273 + ImGroupMessageDO message = ImGroupMessageDO.builder()
  274 + .id(50L).senderId(1L).groupId(10L)
  275 + .status(ImMessageStatusEnum.UNREAD.getStatus())
  276 + .sendTime(LocalDateTime.now()).build();
  277 + when(groupMessageMapper.selectById(50L)).thenReturn(message);
  278 + // 撤回前需要校验用户仍在群中
  279 + ImGroupMemberDO selfMember = ImGroupMemberDO.builder()
  280 + .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build();
  281 + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(selfMember);
  282 + when(groupMessageMapper.updateById(any(ImGroupMessageDO.class))).thenReturn(1);
  283 + when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenReturn(1);
  284 +
  285 + List<ImGroupMemberDO> members = List.of(
  286 + selfMember,
  287 + ImGroupMemberDO.builder().groupId(10L).userId(2L)
  288 + .status(CommonStatusEnum.ENABLE.getStatus()).build(),
  289 + ImGroupMemberDO.builder().groupId(10L).userId(3L)
  290 + .status(CommonStatusEnum.DISABLE.getStatus()).build()
  291 + );
  292 + when(groupMemberService.selectByGroupId(10L)).thenReturn(members);
  293 +
  294 + // 调用
  295 + ImGroupMessageDO result = groupMessageService.recallGroupMessage(1L, 50L);
  296 +
  297 + // 断言：返回 TIP_TEXT 消息
  298 + assertNotNull(result);
  299 + // 验证：更新原消息状态 + 插入 tipMessage
  300 + verify(groupMessageMapper).updateById(any(ImGroupMessageDO.class));
  301 + verify(groupMessageMapper).insert(any(ImGroupMessageDO.class));
  302 + // 验证：只给 2 个活跃成员推送撤回提示（userId=3 已退群）
  303 + verify(webSocketMessageSender, times(2)).sendObject(anyInt(), anyLong(), anyString(), any());
  304 + }
  305 + }
  306 +
  307 + @Test
  308 + public void testRecallMessage_notOwn() {
  309 + // 准备
  310 + ImGroupMessageDO message = ImGroupMessageDO.builder()
  311 + .id(50L).senderId(2L).groupId(10L)
  312 + .status(ImMessageStatusEnum.UNREAD.getStatus())
  313 + .sendTime(LocalDateTime.now()).build();
  314 + when(groupMessageMapper.selectById(50L)).thenReturn(message);
  315 +
  316 + // 调用并断言
  317 + ServiceException exception = assertThrows(ServiceException.class,
  318 + () -> groupMessageService.recallGroupMessage(1L, 50L));
  319 + assertEquals(MESSAGE_RECALL_DENIED.getCode(), exception.getCode());
  320 + }
  321 +
  322 + @Test
  323 + public void testRecallMessage_alreadyRecalled() {
261 324 // 准备
262 325 ImGroupMessageDO message = ImGroupMessageDO.builder()
263 326 .id(50L).senderId(1L).groupId(10L)
264   - .status(ImMessageStatusEnum.UNREAD.getStatus()).build();
265   - when(imGroupMessageMapper.selectById(50L)).thenReturn(message);
266   - when(imGroupMessageMapper.updateById(any(ImGroupMessageDO.class))).thenReturn(1);
  327 + .status(ImMessageStatusEnum.RECALL.getStatus())
  328 + .sendTime(LocalDateTime.now()).build();
  329 + when(groupMessageMapper.selectById(50L)).thenReturn(message);
267 330  
268   - List<ImGroupMemberDO> members = List.of(
269   - ImGroupMemberDO.builder().groupId(10L).userId(1L)
270   - .status(CommonStatusEnum.ENABLE.getStatus()).build(),
271   - ImGroupMemberDO.builder().groupId(10L).userId(2L)
272   - .status(CommonStatusEnum.ENABLE.getStatus()).build(),
273   - ImGroupMemberDO.builder().groupId(10L).userId(3L)
274   - .status(CommonStatusEnum.DISABLE.getStatus()).build()
275   - );
276   - when(imGroupMemberService.selectByGroupId(10L)).thenReturn(members);
  331 + // 调用并断言
  332 + ServiceException exception = assertThrows(ServiceException.class,
  333 + () -> groupMessageService.recallGroupMessage(1L, 50L));
  334 + assertEquals(MESSAGE_ALREADY_RECALLED.getCode(), exception.getCode());
  335 + }
277 336  
278   - // 调用
279   - groupMessageService.recallGroupMessage(1L, 50L);
  337 + @Test
  338 + public void testRecallMessage_timeout() {
  339 + // 准备：消息发送于 10 分钟前（超过 5 分钟窗口）
  340 + ImGroupMessageDO message = ImGroupMessageDO.builder()
  341 + .id(50L).senderId(1L).groupId(10L)
  342 + .status(ImMessageStatusEnum.UNREAD.getStatus())
  343 + .sendTime(LocalDateTime.now().minusMinutes(10)).build();
  344 + when(groupMessageMapper.selectById(50L)).thenReturn(message);
280 345  
281   - // 断言：只给 2 个活跃成员发送 RECALL 事件（userId=3 已退群）
282   - verify(webSocketMessageSender, times(2)).sendObject(anyInt(), anyLong(), anyString(), any());
  346 + // 调用并断言
  347 + ServiceException exception = assertThrows(ServiceException.class,
  348 + () -> groupMessageService.recallGroupMessage(1L, 50L));
  349 + assertEquals(MESSAGE_RECALL_TIMEOUT.getCode(), exception.getCode());
283 350 }
284 351  
285 352 // ========== 群已读测试 ==========
... ... @@ -287,7 +354,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
287 354 @Test
288 355 public void testReadMessages_notInGroup() {
289 356 // 准备：当前用户不在群中
290   - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(null);
  357 + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null);
291 358  
292 359 // 调用并断言：越权校验
293 360 ServiceException exception = assertThrows(ServiceException.class,
... ... @@ -302,13 +369,13 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
302 369 .groupId(10L).userId(1L)
303 370 .status(CommonStatusEnum.ENABLE.getStatus())
304 371 .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build();
305   - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(currentMember);
  372 + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(currentMember);
306 373  
307 374 // 准备：消息由用户 5 发，发送时间在 2026-04-12 10:00
308 375 ImGroupMessageDO message = ImGroupMessageDO.builder()
309 376 .id(80L).groupId(10L).senderId(5L)
310 377 .sendTime(LocalDateTime.of(2026, 4, 12, 10, 0, 0)).build();
311   - when(imGroupMessageMapper.selectById(80L)).thenReturn(message);
  378 + when(groupMessageMapper.selectById(80L)).thenReturn(message);
312 379  
313 380 // 准备：群成员一览
314 381 // 用户 1: 正常，入群在消息之前
... ... @@ -327,7 +394,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
327 394 .status(CommonStatusEnum.ENABLE.getStatus())
328 395 .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build() // 发送者
329 396 );
330   - when(imGroupMemberService.selectByGroupId(10L)).thenReturn(allMembers);
  397 + when(groupMemberService.selectByGroupId(10L)).thenReturn(allMembers);
331 398  
332 399 // 准备：Redis 已读位置 — 用户 1 读到 100, 用户 2 读到 50, 用户 3 读到 200
333 400 Map<Object, Object> positions = new HashMap<>();
... ... @@ -340,7 +407,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
340 407 List<Long> readUsers = groupMessageService.getGroupReadUsers(1L, 10L, 80L);
341 408  
342 409 // 断言：
343   - // 用户 1: 可见 + readMaxId=100>=80 → 已读 ✓（但排除发送者？不是发送者，所以算入）
  410 + // 用户 1: 可见 + readMaxId=100>=80 → 已读 ✓
344 411 // 用户 2: 可见 + readMaxId=50<80 → 未读 ✗
345 412 // 用户 3: 入群在消息之后 → 不可见 → 不算
346 413 // 用户 5: 发送者 → 排除
... ... @@ -354,7 +421,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
354 421 @Test
355 422 public void testGetReadUsers_notInGroup() {
356 423 // 准备：当前用户不在群中
357   - when(imGroupMemberService.getGroupMember(10L, 99L)).thenReturn(null);
  424 + when(groupMemberService.getGroupMember(10L, 99L)).thenReturn(null);
358 425  
359 426 // 调用并断言：越权校验
360 427 ServiceException exception = assertThrows(ServiceException.class,
... ...