diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImGroupMessageController.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImGroupMessageController.java index fe0d4a2..ee7be50 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImGroupMessageController.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/ImGroupMessageController.java @@ -1,6 +1,8 @@ package cn.iocoder.yudao.module.im.controller.admin.message; import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.framework.common.util.object.BeanUtils; +import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO; import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageRespVO; import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO; import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO; @@ -14,7 +16,6 @@ import org.springframework.validation.annotation.Validated; import org.springframework.web.bind.annotation.*; import java.util.List; -import java.util.stream.Collectors; import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success; 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 public class ImGroupMessageController { @Resource - private ImGroupMessageService imGroupMessageService; + private ImGroupMessageService groupMessageService; @PostMapping("/send") @Operation(summary = "发送群聊消息") public CommonResult sendGroupMessage(@Valid @RequestBody ImGroupMessageSendReqVO reqVO) { - ImGroupMessageDO message = imGroupMessageService.sendGroupMessage(getLoginUserId(), reqVO); - return success(toRespVO(message)); + ImGroupMessageDO message = groupMessageService.sendGroupMessage(getLoginUserId(), reqVO); + return success(BeanUtils.toBean(message, ImGroupMessageRespVO.class)); } @GetMapping("/pull") @Operation(summary = "拉取群聊消息（增量）") @Parameter(name = "minId", description = "最小消息 id", required = true, example = "0") @Parameter(name = "size", description = "拉取数量", required = true, example = "100") - public CommonResult> pullGroupMessageList( - @RequestParam("minId") Long minId, - @RequestParam("size") Integer size) { - List messages = imGroupMessageService.pullGroupMessageList(getLoginUserId(), minId, size); - return success(messages.stream().map(this::toRespVO).collect(Collectors.toList())); + public CommonResult> pullGroupMessageList(@RequestParam("minId") Long minId, + @RequestParam("size") Integer size) { + List messages = groupMessageService.pullGroupMessageList(getLoginUserId(), minId, size); + return success(BeanUtils.toBean(messages, ImGroupMessageRespVO.class)); } @PutMapping("/read") @Operation(summary = "标记群聊消息已读") @Parameter(name = "groupId", description = "群编号", required = true, example = "1") public CommonResult readGroupMessages(@RequestParam("groupId") Long groupId) { - imGroupMessageService.readGroupMessages(getLoginUserId(), groupId); + groupMessageService.readGroupMessages(getLoginUserId(), groupId); return success(true); } @DeleteMapping("/recall") @Operation(summary = "撤回群聊消息") @Parameter(name = "id", description = "消息编号", required = true, example = "1") - public CommonResult recallGroupMessage(@RequestParam("id") Long id) { - imGroupMessageService.recallGroupMessage(getLoginUserId(), id); - return success(true); + public CommonResult recallGroupMessage(@RequestParam("id") Long id) { + ImGroupMessageDO message = groupMessageService.recallGroupMessage(getLoginUserId(), id); + return success(BeanUtils.toBean(message, ImGroupMessageRespVO.class)); } - // DONE @AI：这个接口，确认需要！用于群回执人数展示 @GetMapping("/read-users") @Operation(summary = "获取群消息已读用户列表") @Parameter(name = "groupId", description = "群编号", required = true, example = "1") @Parameter(name = "messageId", description = "消息编号", required = true, example = "1") - public CommonResult> getGroupReadUsers( - @RequestParam("groupId") Long groupId, - @RequestParam("messageId") Long messageId) { - List readUserIds = imGroupMessageService.getGroupReadUsers(getLoginUserId(), groupId, messageId); - // Long -> String 转换 - return success(readUserIds.stream().map(String::valueOf).collect(Collectors.toList())); + public CommonResult> getGroupReadUsers(@RequestParam("groupId") Long groupId, + @RequestParam("messageId") Long messageId) { + return success(groupMessageService.getGroupReadUsers(getLoginUserId(), groupId, messageId)); } - /** - * DO 转换为 RespVO（Long id 统一转 String） - */ - // DONE @AI：BeanUtils？结论：因 id/senderId/groupId 需 Long→String 手动转换，不适合 BeanUtils - private ImGroupMessageRespVO toRespVO(ImGroupMessageDO message) { - ImGroupMessageRespVO vo = new ImGroupMessageRespVO(); - vo.setId(message.getId().toString()); - vo.setClientMessageId(message.getClientMessageId()); - vo.setSenderId(message.getSenderId().toString()); - vo.setGroupId(message.getGroupId().toString()); - vo.setType(message.getType()); - vo.setContent(message.getContent()); - vo.setStatus(message.getStatus()); - vo.setSendTime(message.getSendTime()); - vo.setAtUserIds(message.getAtUserIds()); - vo.setReceiverUserIds(message.getReceiverUserIds()); - vo.setReceiptStatus(message.getReceiptStatus()); - return vo; + @GetMapping("/list") + @Operation(summary = "查询群聊历史消息") + public CommonResult> getGroupMessageList(@Valid ImGroupMessageListReqVO reqVO) { + List messages = groupMessageService.getGroupMessageList(getLoginUserId(), reqVO); + return success(BeanUtils.toBean(messages, ImGroupMessageRespVO.class)); } } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageListReqVO.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageListReqVO.java new file mode 100644 index 0000000..76aa958 --- /dev/null +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageListReqVO.java @@ -0,0 +1,26 @@ +package cn.iocoder.yudao.module.im.controller.admin.message.vo.group; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.Max; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +@Schema(description = "管理后台 - 群聊历史消息列表 Request VO") +@Data +public class ImGroupMessageListReqVO { + + @Schema(description = "群编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "10") + @NotNull(message = "群编号不能为空") + private Long groupId; + + @Schema(description = "起始消息编号，从该 id 往前拉取（不含）。为空则从最新消息开始", example = "1024") + private Long maxId; + + @Schema(description = "拉取数量", requiredMode = Schema.RequiredMode.REQUIRED, example = "20") + @NotNull(message = "拉取数量不能为空") + @Min(value = 1, message = "拉取数量最小值为 1") + @Max(value = 200, message = "拉取数量最大值为 200") + private Integer limit; + +} diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageRespVO.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageRespVO.java index 2f5592f..428e1ca 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageRespVO.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/controller/admin/message/vo/group/ImGroupMessageRespVO.java @@ -14,16 +14,16 @@ import java.util.List; public class ImGroupMessageRespVO { @Schema(description = "消息编号", example = "1") - private String id; + private Long id; @Schema(description = "客户端消息编号", example = "uuid-xxx") private String clientMessageId; @Schema(description = "发送人编号", example = "1") - private String senderId; + private Long senderId; @Schema(description = "群编号", example = "1") - private String groupId; + private Long groupId; @Schema(description = "消息类型", example = "0") private Integer type; @@ -46,7 +46,4 @@ public class ImGroupMessageRespVO { @Schema(description = "回执状态", example = "0") private Integer receiptStatus; - @Schema(description = "已读人数（群回执使用）", example = "5") - private Integer readCount; - } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java index 3ab747f..a65780f 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/message/ImGroupMessageMapper.java @@ -4,6 +4,8 @@ import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX; import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX; import cn.iocoder.yudao.framework.mybatis.core.query.QueryWrapperX; import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO; +import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum; +import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum; import org.apache.ibatis.annotations.Mapper; import java.util.List; @@ -16,6 +18,14 @@ import java.util.List; @Mapper public interface ImGroupMessageMapper extends BaseMapperX { + /** + * 根据 minId 增量拉取群聊消息 + * + * @param groupIds 用户所在群编号列表 + * @param minId 最小消息 id（不含） + * @param size 拉取数量 + * @return 消息列表 + */ default List selectListByMinId(List groupIds, Long minId, Integer size) { QueryWrapperX wrapper = new QueryWrapperX<>(); wrapper.in("group_id", groupIds) @@ -25,33 +35,64 @@ public interface ImGroupMessageMapper extends BaseMapperX { return selectList(wrapper); } + /** + * 查询群聊历史消息（游标拉取） + * + * @param groupId 群编号 + * @param maxId 起始消息 id（不含），为空则从最新开始 + * @param limit 拉取数量 + * @return 消息列表（按 id 倒序） + */ + default List selectHistoryList(Long groupId, Long maxId, Integer limit) { + QueryWrapperX wrapper = new QueryWrapperX<>(); + wrapper.eq("group_id", groupId) + .ne("status", ImMessageStatusEnum.RECALL.getStatus()) + .lt(maxId != null, "id", maxId) + .orderByDesc("id"); + wrapper.limitN(limit); + return selectList(wrapper); + } + default ImGroupMessageDO selectBySenderIdAndClientMessageId(Long senderId, String clientMessageId) { return selectOne(new LambdaQueryWrapperX() .eq(ImGroupMessageDO::getSenderId, senderId) .eq(ImGroupMessageDO::getClientMessageId, clientMessageId)); } - // TODO @芋艿：暂时没 review /** * 获取群内最新消息 id + * + * @param groupId 群编号 + * @return 最新消息 id，无消息时返回 null */ default Long selectMaxIdByGroupId(Long groupId) { QueryWrapperX wrapper = new QueryWrapperX<>(); wrapper.eq("group_id", groupId) - .orderByDesc("id"); + .orderByDesc("id") + .select("id"); wrapper.limitN(1); ImGroupMessageDO msg = selectOne(wrapper); return msg == null ? null : msg.getId(); } - // TODO @芋艿：暂时没 review /** - * 获取群内需要回执的消息（状态为 PENDING） + * 查询群内指定范围内待回执的消息 + *

+ * 仅在用户"已读位置前进"时调用，避免全量扫描： + * 只有位于 (minId, maxId] 范围内、且仍处于 PENDING 的回执消息可能因本次已读而状态变化。 + * + * @param groupId 群编号 + * @param minId 起始消息 id（不含，上一次已读位置） + * @param maxId 结束消息 id（含，本次已读位置） + * @return 待回执消息列表 */ - default List selectPendingReceiptMessages(Long groupId) { + default List selectPendingReceiptMessagesInRange(Long groupId, Long minId, Long maxId) { return selectList(new LambdaQueryWrapperX() .eq(ImGroupMessageDO::getGroupId, groupId) - .eq(ImGroupMessageDO::getReceiptStatus, 1)); // PENDING + .eq(ImGroupMessageDO::getReceiptStatus, ImGroupMessageReceiptStatusEnum.PENDING.getStatus()) + .gt(ImGroupMessageDO::getId, minId) + .le(ImGroupMessageDO::getId, maxId) + .ne(ImGroupMessageDO::getStatus, ImMessageStatusEnum.RECALL.getStatus())); } } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java index f064fda..53e5498 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java @@ -8,20 +8,7 @@ package cn.iocoder.yudao.module.im.dal.redis; */ public interface RedisKeyConstants { - /** - * 收件箱序号生成器 - * KEY 格式： im:inbox:sequence:{userId} - * VALUE 数据类型： String - */ - String INBOX_SEQUENCE = "im_inbox_sequence:%s"; - - /** - * 收件箱的分布式锁 - * KEY 格式： im:inbox:lock:{userId} - * VALUE 数据类型： String - */ - String INBOX_LOCK = "im_inbox_lock:%s"; - + // TODO @AI：这个 key 可能要改下； /** * 群已读位置 * KEY 格式： im:group:read:{groupId} diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/group/GroupReadPositionRedisDAO.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupReadPositionRedisDAO.java index c21f332..8dd9dfb 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/group/GroupReadPositionRedisDAO.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/dal/redis/message/GroupReadPositionRedisDAO.java @@ -1,4 +1,4 @@ -package cn.iocoder.yudao.module.im.dal.redis.group; +package cn.iocoder.yudao.module.im.dal.redis.message; import jakarta.annotation.Resource; import org.springframework.data.redis.core.StringRedisTemplate; @@ -8,10 +8,11 @@ import java.util.Map; import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.GROUP_READ_POSITION; +// TODO @AI：需要改成，GroupMessageReadRedisDAO /** * 群已读位置 Redis DAO *

- * 采用 Redis Hash 存储：key=im:group:read:{groupId}, field=userId, value=maxReadMessageId + * 采用 Redis Hash 存储：key=im:group:read:{groupId}, field=userId, value=maxReadMessageId TODO @AI：这里不用注释； * * @author 芋道源码 */ @@ -21,6 +22,7 @@ public class GroupReadPositionRedisDAO { @Resource private StringRedisTemplate stringRedisTemplate; + // TODO @AI：updateReadMaxMessageId； /** * 更新用户在某群的已读位置 * @@ -29,14 +31,18 @@ public class GroupReadPositionRedisDAO { * @param maxMessageId 最大已读消息编号 */ public void updateReadPosition(Long groupId, Long userId, Long maxMessageId) { + // TODO @AI：这里搞个 formatKey 方法，可以复用； String key = String.format(GROUP_READ_POSITION, groupId); // 只在新值大于旧值时更新 Object currentVal = stringRedisTemplate.opsForHash().get(key, userId.toString()); if (currentVal == null || Long.parseLong(currentVal.toString()) < maxMessageId) { + // TODO @AI：可以 put 的时候，就是 Long，Long 么？ stringRedisTemplate.opsForHash().put(key, userId.toString(), maxMessageId.toString()); } } + // TODO @AI：不存在，则返回 null 更合理； + // TODO @AI：getReadMaxMessageId； /** * 获取用户在某群的已读位置 */ @@ -46,28 +52,14 @@ public class GroupReadPositionRedisDAO { return val == null ? 0L : Long.parseLong(val.toString()); } + // TODO @AI：参考这个风格，优化下； /** * 获取某群所有用户的已读位置 */ public Map getAllReadPositions(Long groupId) { + // TODO @AI：可以返回的时候，就是 Long，Long 么？ String key = String.format(GROUP_READ_POSITION, groupId); return stringRedisTemplate.opsForHash().entries(key); } - /** - * 删除某群所有用户的已读位置（群解散时调用） - */ - public void deleteGroupPositions(Long groupId) { - String key = String.format(GROUP_READ_POSITION, groupId); - stringRedisTemplate.delete(key); - } - - /** - * 删除用户在某群的已读位置（退群时调用） - */ - public void deleteUserPosition(Long groupId, Long userId) { - String key = String.format(GROUP_READ_POSITION, groupId); - stringRedisTemplate.opsForHash().delete(key, userId.toString()); - } - } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java index d5e4e42..371bc66 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageService.java @@ -1,5 +1,6 @@ package cn.iocoder.yudao.module.im.service.message; +import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO; import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO; import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO; @@ -44,8 +45,9 @@ public interface ImGroupMessageService { * * @param userId 当前用户编号 * @param messageId 消息编号 + * @return 撤回后的提示消息 */ - void recallGroupMessage(Long userId, Long messageId); + ImGroupMessageDO recallGroupMessage(Long userId, Long messageId); /** * 获取群消息的已读用户列表 @@ -57,4 +59,13 @@ public interface ImGroupMessageService { */ List getGroupReadUsers(Long userId, Long groupId, Long messageId); + /** + * 查询群聊历史消息（游标拉取） + * + * @param userId 当前用户编号 + * @param reqVO 拉取请求 + * @return 消息列表（按 id 倒序） + */ + List getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO); + } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java index 1fbccc9..e7e525f 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java @@ -1,38 +1,43 @@ package cn.iocoder.yudao.module.im.service.message; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.IdUtil; +import cn.hutool.core.util.ObjUtil; +import cn.hutool.extra.spring.SpringUtil; +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; import cn.iocoder.yudao.framework.common.enums.UserTypeEnum; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.framework.common.util.object.BeanUtils; +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageListReqVO; import cn.iocoder.yudao.module.im.controller.admin.message.vo.group.ImGroupMessageSendReqVO; -import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO; import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO; import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO; +import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage; import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper; -import cn.iocoder.yudao.module.im.dal.redis.group.GroupReadPositionRedisDAO; -import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; +import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO; import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum; import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum; -import cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum; import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum; -import cn.iocoder.yudao.module.im.websocket.ImGroupMessageSendMessage; -import cn.iocoder.yudao.module.im.websocket.ImMessageReadMessage; -import cn.iocoder.yudao.module.im.websocket.ImMessageReceiptMessage; -import cn.iocoder.yudao.module.im.websocket.ImMessageRecallMessage; -import cn.iocoder.yudao.module.im.service.group.ImGroupService; import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService; +import cn.iocoder.yudao.module.im.service.group.ImGroupService; import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService; -import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.module.im.websocket.ImGroupMessageDTO; import jakarta.annotation.Resource; import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; import org.springframework.validation.annotation.Validated; import java.time.LocalDateTime; import java.util.*; -import java.util.stream.Collectors; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList; import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*; -import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*; +import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_MAX_PULL_SIZE; +import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.MESSAGE_RECALL_TIMEOUT_MINUTES; /** * IM 群聊消息 Service 实现类 @@ -45,60 +50,51 @@ import static cn.iocoder.yudao.module.im.enums.ImCommonConstants.*; public class ImGroupMessageServiceImpl implements ImGroupMessageService { @Resource - private ImGroupMessageMapper imGroupMessageMapper; + private ImGroupMessageMapper groupMessageMapper; + // TODO @AI：变量名，可能要改下； @Resource - private ImGroupService imGroupService; + private GroupReadPositionRedisDAO groupReadPositionRedisDAO; + @Resource - private ImGroupMemberService imGroupMemberService; + private ImGroupService groupService; @Resource - private ImSensitiveWordService imSensitiveWordService; + private ImGroupMemberService groupMemberService; @Resource - private GroupReadPositionRedisDAO groupReadPositionRedisDAO; + private ImSensitiveWordService sensitiveWordService; + @Resource private WebSocketMessageSender webSocketMessageSender; @Override public ImGroupMessageDO sendGroupMessage(Long senderId, ImGroupMessageSendReqVO reqVO) { - // 1. 幂等校验 - ImGroupMessageDO existing = imGroupMessageMapper.selectBySenderIdAndClientMessageId( + // 1.1 幂等校验：根据 senderId + clientMessageId 查重 + ImGroupMessageDO existing = groupMessageMapper.selectBySenderIdAndClientMessageId( senderId, reqVO.getClientMessageId()); if (existing != null) { + log.info("[sendGroupMessage][幂等命中 senderId({}) clientMessageId({}) 已存在消息({})]", + senderId, reqVO.getClientMessageId(), existing.getId()); return existing; } - - // 2. 校验群存在且未解散 - ImGroupDO group = imGroupService.validateGroupExists(reqVO.getGroupId()); - - // 3. 校验发送人在群中 - ImGroupMemberDO member = imGroupMemberService.getGroupMember(reqVO.getGroupId(), senderId); - if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { - throw exception(GROUP_MEMBER_NOT_IN_GROUP); - } - - // 4. 文本消息敏感词过滤 + // 1.2 校验群存在、发送人仍在群中 + groupService.validateGroupExists(reqVO.getGroupId()); + validateMemberInGroup(reqVO.getGroupId(), senderId); + // 1.3 文本消息敏感词过滤 if (ImMessageTypeEnum.TEXT.getType().equals(reqVO.getType())) { - imSensitiveWordService.validateText(reqVO.getContent()); + sensitiveWordService.validateText(reqVO.getContent()); } - // 5. 构建消息 - ImGroupMessageDO message = ImGroupMessageDO.builder() - .clientMessageId(reqVO.getClientMessageId()) - .senderId(senderId) - .groupId(reqVO.getGroupId()) - .type(reqVO.getType()) - .content(reqVO.getContent()) - .status(ImMessageStatusEnum.UNREAD.getStatus()) // 群聊正常状态 = 0 - .sendTime(LocalDateTime.now()) - .atUserIds(reqVO.getAtUserIds()) - .receiverUserIds(reqVO.getReceiverUserIds()) - .receiptStatus(Boolean.TRUE.equals(reqVO.getNeedReceipt()) + // 2. 构建并保存消息 + // TODO @AI：content 要 json 下； + ImGroupMessageDO message = BeanUtils.toBean(reqVO, ImGroupMessageDO.class, m -> m + .setSenderId(senderId).setStatus(ImMessageStatusEnum.UNREAD.getStatus()) + .setSendTime(LocalDateTime.now()) + .setReceiptStatus(Boolean.TRUE.equals(reqVO.getNeedReceipt()) ? ImGroupMessageReceiptStatusEnum.PENDING.getStatus() - : ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus()) - .build(); - imGroupMessageMapper.insert(message); + : ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus())); + groupMessageMapper.insert(message); - // 6. WebSocket 推送给群成员 - sendGroupMessageEvent(message, reqVO.getGroupId()); + // 3. WebSocket 推送给群成员（异步） + getSelf().sendGroupMessageEvent(message); return message; } @@ -107,272 +103,288 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { if (size > MESSAGE_MAX_PULL_SIZE) { throw exception(MESSAGE_PULL_SIZE_EXCEEDED); } - // 获取用户所在的所有群 - List members = imGroupMemberService.getGroupMembersByUserId(userId); + // 1. 获取用户所在的所有群 + List members = groupMemberService.getGroupMembersByUserId(userId); if (CollUtil.isEmpty(members)) { return Collections.emptyList(); } - List groupIds = members.stream() - .map(ImGroupMemberDO::getGroupId) - .collect(Collectors.toList()); - - // 拉取消息 - List messages = imGroupMessageMapper.selectListByMinId(groupIds, minId, size); - - // 过滤：入群前消息不可见、定向接收过滤、退群后消息边界 - Map memberMap = members.stream() - .collect(Collectors.toMap(ImGroupMemberDO::getGroupId, m -> m, (a, b) -> a)); - - return messages.stream().filter(msg -> { - ImGroupMemberDO m = memberMap.get(msg.getGroupId()); - if (m == null) { - return false; - } - // 入群前消息不可见 - if (m.getJoinTime() != null && msg.getSendTime().isBefore(m.getJoinTime())) { - return false; - } - // 退群后消息边界 - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus()) - && m.getQuitTime() != null && msg.getSendTime().isAfter(m.getQuitTime())) { - return false; - } - // 定向接收过滤 - if (CollUtil.isNotEmpty(msg.getReceiverUserIds())) { - if (!msg.getReceiverUserIds().contains(userId) && !msg.getSenderId().equals(userId)) { - return false; - } - } - return true; - }).collect(Collectors.toList()); + List groupIds = convertList(members, ImGroupMemberDO::getGroupId); + + // 2. 拉取消息 + List messages = groupMessageMapper.selectListByMinId(groupIds, minId, size); + + // 3. 过滤：入群前不可见、退群后不可见、定向接收过滤 + // TODO @AI：缺少各种逻辑； + Map memberMap = CollUtil.toMap(members, new HashMap<>(), ImGroupMemberDO::getGroupId); + List result = messages.stream() + .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId)) + .toList(); + log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", + userId, minId, size, result.size()); + return result; } + // TODO @AI：增加 messageId 参数；允许空，则查询最大的一条； @Override public void readGroupMessages(Long userId, Long groupId) { - // 1. 校验用户在群中（权限校验） - ImGroupMemberDO member = imGroupMemberService.getGroupMember(groupId, userId); - if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { - throw exception(GROUP_MEMBER_NOT_IN_GROUP); + // 1.1 校验用户在群中（权限校验） + validateMemberInGroup(groupId, userId); + // 1.2 获取群内最新消息 id（倒序 LIMIT 1，避免全量扫描） + Long newMaxId = groupMessageMapper.selectMaxIdByGroupId(groupId); + if (newMaxId == null) { + return; } - - // 2. 高效获取群内最新消息 id（倒序 LIMIT 1，非全量扫描） - Long maxMessageId = imGroupMessageMapper.selectMaxIdByGroupId(groupId); - if (maxMessageId == null) { + // 1.3 已读位置未前进，直接返回 + // TODO @AI：1.3 和 2 是不是可以合并？如果 update 失败，则直接回退；不然读取两次，貌似有 cas 问题； + Long prevMaxId = groupReadPositionRedisDAO.getReadPosition(groupId, userId); + if (newMaxId <= prevMaxId) { return; } - // 3. 更新 Redis 群已读位置 - groupReadPositionRedisDAO.updateReadPosition(groupId, userId, maxMessageId); + // 2. 更新 Redis 群已读位置 + groupReadPositionRedisDAO.updateReadPosition(groupId, userId, newMaxId); - // 4. 发送 READ 事件给自己的其他终端 - ImMessageReadMessage readMessage = new ImMessageReadMessage() - .setGroupId(groupId) - .setScene(ImMessageSceneEnum.GROUP.getScene()); - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId, - ImMessageReadMessage.TYPE, readMessage); - - // 5. 刷新群回执 - refreshGroupReceipts(groupId); + // 3. 异步发送 READ 事件 + 刷新范围内的群回执 + getSelf().readGroupMessageEvent(userId, groupId, prevMaxId, newMaxId); } @Override - public void recallGroupMessage(Long userId, Long messageId) { - // 1. 校验消息存在 - ImGroupMessageDO message = imGroupMessageMapper.selectById(messageId); + @Transactional(rollbackFor = Exception.class) + public ImGroupMessageDO recallGroupMessage(Long userId, Long messageId) { + // 1.1 校验消息存在 + // TODO @AI：抽成一个 validateXXXExists 方法； + ImGroupMessageDO message = groupMessageMapper.selectById(messageId); if (message == null) { throw exception(MESSAGE_NOT_EXISTS); } - // 2. 只能撤回自己发送的消息 - if (!message.getSenderId().equals(userId)) { + // 1.2 只能撤回自己发送的消息 + if (ObjUtil.notEqual(message.getSenderId(), userId)) { throw exception(MESSAGE_RECALL_DENIED); } - // 3. 不能重复撤回 + // 1.3 不能重复撤回 if (ImMessageStatusEnum.RECALL.getStatus().equals(message.getStatus())) { throw exception(MESSAGE_ALREADY_RECALLED); } - // 4. 更新消息状态 - ImGroupMessageDO updateObj = new ImGroupMessageDO(); - updateObj.setId(messageId); - updateObj.setStatus(ImMessageStatusEnum.RECALL.getStatus()); - imGroupMessageMapper.updateById(updateObj); - - // 5. 发送 RECALL 事件给所有群成员 - List members = imGroupMemberService.selectByGroupId(message.getGroupId()); - ImMessageRecallMessage recallMessage = new ImMessageRecallMessage() - .setMessageId(messageId) - .setGroupId(message.getGroupId()) - .setSenderId(userId) - .setScene(ImMessageSceneEnum.GROUP.getScene()); - for (ImGroupMemberDO m : members) { - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus())) { - continue; - } - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), m.getUserId(), - ImMessageRecallMessage.TYPE, recallMessage); + // 1.4 只允许撤回限定时间内的消息 + if (message.getSendTime().plusMinutes(MESSAGE_RECALL_TIMEOUT_MINUTES).isBefore(LocalDateTime.now())) { + throw exception(MESSAGE_RECALL_TIMEOUT, MESSAGE_RECALL_TIMEOUT_MINUTES); } + // 1.5 校验撤回人仍在群中 + validateMemberInGroup(message.getGroupId(), userId); + + // 2. 更新原消息状态为撤回 + groupMessageMapper.updateById(new ImGroupMessageDO().setId(messageId) + .setStatus(ImMessageStatusEnum.RECALL.getStatus())); + + // 3. 插入一条 TIP_TEXT 消息作为撤回提示 + RecallMessage recallContent = new RecallMessage().setMessageId(messageId); + ImGroupMessageDO recallMessage = new ImGroupMessageDO().setClientMessageId(IdUtil.fastSimpleUUID()) + .setSenderId(userId).setGroupId(message.getGroupId()) + .setType(ImMessageTypeEnum.TIP_TEXT.getType()).setContent(JsonUtils.toJsonString(recallContent)) + .setStatus(ImMessageStatusEnum.UNREAD.getStatus()).setSendTime(LocalDateTime.now()) + .setReceiptStatus(ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus()); + groupMessageMapper.insert(recallMessage); + + // 4. 异步推送撤回提示消息（前端据此更新原消息状态 + 插入撤回提示） + getSelf().sendGroupMessageEvent(recallMessage); + return recallMessage; } @Override public List getGroupReadUsers(Long userId, Long groupId, Long messageId) { - // 1. 校验用户在群中（权限校验） - ImGroupMemberDO currentMember = imGroupMemberService.getGroupMember(groupId, userId); - if (currentMember == null || CommonStatusEnum.DISABLE.getStatus().equals(currentMember.getStatus())) { - throw exception(GROUP_MEMBER_NOT_IN_GROUP); - } - - // 2. 获取消息 - ImGroupMessageDO message = imGroupMessageMapper.selectById(messageId); + // 1.1 校验用户在群中（权限校验） + validateMemberInGroup(groupId, userId); + // 1.2 获取消息 + ImGroupMessageDO message = groupMessageMapper.selectById(messageId); if (message == null) { return Collections.emptyList(); } // 2. 获取所有成员和已读位置 - List allMembers = imGroupMemberService.selectByGroupId(groupId); + List allMembers = groupMemberService.selectByGroupId(groupId); Map allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId); - // 3. 计算该消息的可见成员集合 + // 3. 计算该消息的可见成员集合（排除发送者自己） Set visibleUserIds = getVisibleUserIds(message, allMembers); + visibleUserIds.remove(message.getSenderId()); // 4. 只返回在可见范围内且已读位置 >= messageId 的用户 List readUserIds = new ArrayList<>(); - for (Map.Entry entry : allPositions.entrySet()) { - Long uid = Long.parseLong(entry.getKey().toString()); - Long readMaxId = Long.parseLong(entry.getValue().toString()); - if (readMaxId >= messageId && visibleUserIds.contains(uid) - && !uid.equals(message.getSenderId())) { // 排除发送者自己 + allPositions.forEach((k, v) -> { + Long uid = Long.valueOf(k.toString()); + if (visibleUserIds.contains(uid) && Long.parseLong(v.toString()) >= messageId) { readUserIds.add(uid); } - } + }); return readUserIds; } + @Override + public List getGroupMessageList(Long userId, ImGroupMessageListReqVO reqVO) { + // 1. 校验用户在群中 + ImGroupMemberDO member = validateMemberInGroup(reqVO.getGroupId(), userId); + + // 2. 查询历史消息（仅入群之后） + // TODO @AI：应该 mybatis 里就进行过滤！然后 joinTime 肯定不会是空的！！！ + List messages = groupMessageMapper.selectHistoryList( + reqVO.getGroupId(), reqVO.getMaxId(), reqVO.getLimit()); + if (member.getJoinTime() == null) { + return messages; + } + return messages.stream() + .filter(msg -> !msg.getSendTime().isBefore(member.getJoinTime())) + .toList(); + } + + // ========== 异步 WebSocket 推送 ========== + + /** + * 发送群聊消息 WebSocket 事件 + */ + @Async + public void sendGroupMessageEvent(ImGroupMessageDO message) { + ImGroupMessageDTO dto = ImGroupMessageDTO.ofSend(message); + // 广播给群内可见成员（含发送方自己，支持多端同步） + // TODO @AI：getGroupMemberListByGroupId()；默认过滤有效的； + List members = groupMemberService.selectByGroupId(message.getGroupId()); + List receiverList = message.getReceiverUserIds(); + for (ImGroupMemberDO member : members) { + if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { + continue; + } + // TODO @AI：这里是不是不用过滤啊？因为一般 send 的时候，肯定带了呀。按照群友就好了； + if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId()) + && ObjUtil.notEqual(member.getUserId(), message.getSenderId())) { + continue; + } + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), member.getUserId(), + ImGroupMessageDTO.TYPE, dto); + } + } + + // TODO @AI：命名最后都是 pre 和 new 里面最好都有 messageId 的标识；包括上面的地方，更好理解吧！ /** - * 刷新群回执（仅处理 PENDING 状态的消息，无需全量扫描） + * 发送已读 + 刷新群回执 WebSocket 事件 + * + * @param userId 当前用户编号 + * @param groupId 群编号 + * @param prevMaxId 上次已读位置 + * @param newMaxId 本次已读位置 */ - private void refreshGroupReceipts(Long groupId) { - // 1. 只查 PENDING 状态的回执消息 - List pendingMessages = imGroupMessageMapper.selectPendingReceiptMessages(groupId); + @Async + public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxId, Long newMaxId) { + // 1. 发送 READ 事件给自己的其他终端（多端同步） + ImGroupMessageDTO readDto = ImGroupMessageDTO.ofRead(userId, groupId, newMaxId); + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId, + ImGroupMessageDTO.TYPE, readDto); + + // 2. 刷新 (prevMaxId, newMaxId] 区间内的待回执消息 + // TODO @AI：selectListByXXX；注意，prevMaxId 可能为空的情况； + List pendingMessages = groupMessageMapper.selectPendingReceiptMessagesInRange( + groupId, prevMaxId, newMaxId); if (CollUtil.isEmpty(pendingMessages)) { return; } - - // 2. 加载成员和已读位置 - List allMembers = imGroupMemberService.selectByGroupId(groupId); + // TODO @AI：是不是要过滤 status ？因为可能退出了。 + List allMembers = groupMemberService.selectByGroupId(groupId); + // TODO @AI：变量也改下； Map allPositions = groupReadPositionRedisDAO.getAllReadPositions(groupId); - - for (ImGroupMessageDO msg : pendingMessages) { - // 3. 计算该消息的可见成员（排除发送者自己） - Set visibleUserIds = getVisibleUserIds(msg, allMembers); - visibleUserIds.remove(msg.getSenderId()); // 发送者不计入回执统计 - + for (ImGroupMessageDO message : pendingMessages) { + // 2.1 计算该消息的可见成员（排除发送者自己） + Set visibleUserIds = getVisibleUserIds(message, allMembers); + visibleUserIds.remove(message.getSenderId()); + // TODO @AI：CollUtil 来判断； if (visibleUserIds.isEmpty()) { continue; } - - // 4. 统计可见成员中的已读人数 + // 2.2 统计可见成员中的已读人数 int readCount = 0; for (Long uid : visibleUserIds) { + // TODO @AI：根据返回结果，进行简化；应该都是 Long、Long； Object val = allPositions.get(uid.toString()); - if (val != null && Long.parseLong(val.toString()) >= msg.getId()) { + if (val != null && Long.parseLong(val.toString()) >= message.getId()) { readCount++; } } - - // 5. 判断是否全部已读 + // 2.3 全部已读 → 标记回执完成 boolean allRead = readCount >= visibleUserIds.size(); + Integer newReceiptStatus = allRead + ? ImGroupMessageReceiptStatusEnum.DONE.getStatus() + : ImGroupMessageReceiptStatusEnum.PENDING.getStatus(); if (allRead) { - ImGroupMessageDO update = new ImGroupMessageDO(); - update.setId(msg.getId()); - update.setReceiptStatus(ImGroupMessageReceiptStatusEnum.DONE.getStatus()); - imGroupMessageMapper.updateById(update); + groupMessageMapper.updateById(new ImGroupMessageDO().setId(message.getId()) + .setReceiptStatus(newReceiptStatus)); } + // 2.4 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度） + // TODO @AI：检查下别的 Dto = 的地方，都最好改成 DTO = ；仅限 im 模块； + ImGroupMessageDTO receiptDTO = ImGroupMessageDTO.ofReceipt( + message.getId(), groupId, readCount, newReceiptStatus); + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), message.getSenderId(), + ImGroupMessageDTO.TYPE, receiptDTO); + } + } - // 6. 广播 RECEIPT 事件给活跃成员 - ImMessageReceiptMessage receiptMessage = new ImMessageReceiptMessage() - .setMessageId(msg.getId()) - .setGroupId(groupId) - .setReadCount(readCount) - .setReceiptStatus(allRead - ? ImGroupMessageReceiptStatusEnum.DONE.getStatus() - : ImGroupMessageReceiptStatusEnum.PENDING.getStatus()) - .setScene(ImMessageSceneEnum.GROUP.getScene()); - - for (ImGroupMemberDO m : allMembers) { - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus())) { - continue; - } - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), m.getUserId(), - ImMessageReceiptMessage.TYPE, receiptMessage); - } + // ========== 私有工具方法 ========== + + // TODO @AI：抽到对应的 GroupMemberService 中，供其他地方复用 + /** + * 校验用户是否为群的有效成员 + * + * @param groupId 群编号 + * @param userId 用户编号 + * @return 群成员记录 + */ + private ImGroupMemberDO validateMemberInGroup(Long groupId, Long userId) { + ImGroupMemberDO member = groupMemberService.getGroupMember(groupId, userId); + if (member == null || CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { + throw exception(GROUP_MEMBER_NOT_IN_GROUP); } + return member; } /** - * 计算一条群消息的可见成员集合 + * 判断一条群消息对某个群成员是否可见 *

* 规则： - * 1. 成员必须是 NORMAL 状态，或虽已 QUIT 但消息发送时间在退群之前 - * 2. 成员的入群时间必须在消息发送之前（入群后才能看到消息） - * 3. 如果消息有 receiverUserIds 定向接收，则只有列表中的人可见 + * 1. 入群时间晚于消息发送时间 → 不可见 + * 2. 已退群且退群时间早于消息发送时间 → 不可见 + * 3. 存在定向接收列表且当前用户既不在列表中也不是发送者 → 不可见 */ - private Set getVisibleUserIds(ImGroupMessageDO msg, List allMembers) { - // 定向接收集合 - List receiverList = msg.getReceiverUserIds(); - - Set visible = new HashSet<>(); - for (ImGroupMemberDO m : allMembers) { - // 入群时间在消息发送之后 → 不可见 - if (m.getJoinTime() != null && msg.getSendTime().isBefore(m.getJoinTime())) { - continue; - } - // 已退群且退群时间在消息发送之前 → 不可见 - if (CommonStatusEnum.DISABLE.getStatus().equals(m.getStatus()) - && m.getQuitTime() != null && msg.getSendTime().isAfter(m.getQuitTime())) { - continue; - } - // 定向接收过滤 - if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(m.getUserId()) - && !m.getUserId().equals(msg.getSenderId())) { - continue; - } - visible.add(m.getUserId()); + private boolean isMessageVisible(ImGroupMessageDO msg, ImGroupMemberDO member, Long userId) { + if (member == null) { + return false; } - return visible; + if (member.getJoinTime() != null && msg.getSendTime().isBefore(member.getJoinTime())) { + return false; + } + if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus()) + && member.getQuitTime() != null && msg.getSendTime().isAfter(member.getQuitTime())) { + return false; + } + if (CollUtil.isNotEmpty(msg.getReceiverUserIds()) + && !msg.getReceiverUserIds().contains(userId) + && ObjUtil.notEqual(msg.getSenderId(), userId)) { + return false; + } + return true; } /** - * 发送群聊消息 WebSocket 事件 + * 计算一条群消息的可见成员集合（含发送者） */ - private void sendGroupMessageEvent(ImGroupMessageDO message, Long groupId) { - List members = imGroupMemberService.selectByGroupId(groupId); - ImGroupMessageSendMessage sendMessage = new ImGroupMessageSendMessage() - .setId(message.getId()) - .setClientMessageId(message.getClientMessageId()) - .setSenderId(message.getSenderId()) - .setGroupId(message.getGroupId()) - .setType(message.getType()) - .setContent(message.getContent()) - .setStatus(message.getStatus()) - .setSendTime(message.getSendTime()) - .setAtUserIds(message.getAtUserIds()) - .setReceiverUserIds(message.getReceiverUserIds()) - .setReceiptStatus(message.getReceiptStatus()); - - // 确定推送目标 - List receiverList = message.getReceiverUserIds(); - - for (ImGroupMemberDO member : members) { - if (CommonStatusEnum.DISABLE.getStatus().equals(member.getStatus())) { - continue; - } - // 定向接收过滤 - if (CollUtil.isNotEmpty(receiverList) && !receiverList.contains(member.getUserId()) - && !member.getUserId().equals(message.getSenderId())) { - continue; + private Set getVisibleUserIds(ImGroupMessageDO msg, List allMembers) { + Set visible = new HashSet<>(); + // TODO @AI： CollectionUtils.converset + for (ImGroupMemberDO member : allMembers) { + if (isMessageVisible(msg, member, member.getUserId())) { + visible.add(member.getUserId()); } - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), member.getUserId(), - ImGroupMessageSendMessage.TYPE, sendMessage); } + return visible; + } + + private ImGroupMessageServiceImpl getSelf() { + return SpringUtil.getBean(getClass()); } } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java index e0fed38..06e8811 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImPrivateMessageServiceImpl.java @@ -22,6 +22,7 @@ import jakarta.annotation.Resource; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; import org.springframework.validation.annotation.Validated; import java.time.LocalDateTime; @@ -92,6 +93,7 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService { return messages; } + // TODO @AI：考虑到更稳妥，是不是前端传递 messageId？不然恰好发过来，然后读取了。【后续在优化】 @Override public void readPrivateMessages(Long userId, Long receiverId) { // 1.1 查询未读消息列表 @@ -111,6 +113,7 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService { } @Override + @Transactional(rollbackFor = Exception.class) public ImPrivateMessageDO recallPrivateMessage(Long userId, Long messageId) { // 1.1 校验消息存在 ImPrivateMessageDO message = privateMessageMapper.selectById(messageId); diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImGroupMessageSendMessage.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImGroupMessageDTO.java index ae1653a..bc9fd2a 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImGroupMessageSendMessage.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImGroupMessageDTO.java @@ -1,24 +1,32 @@ package cn.iocoder.yudao.module.im.websocket; +import cn.iocoder.yudao.framework.common.util.object.BeanUtils; +import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO; +import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum; import lombok.Data; -import lombok.experimental.Accessors; import java.time.LocalDateTime; import java.util.List; /** - * IM 群聊消息 WebSocket 推送消息 + * IM 群聊消息 WebSocket 统一推送 DTO + *

+ * 发送、已读、回执都通过 {@link #type} 字段区分， + * 并统一复用 {@link #id} 字段表达目标消息或已读位置。 * * @author 芋道源码 */ @Data -@Accessors(chain = true) -public class ImGroupMessageSendMessage { +public class ImGroupMessageDTO { - public static final String TYPE = "im-group-message-send"; + public static final String TYPE = "im-group-message"; /** * 消息编号 + *

+ * 普通消息：当前消息 id + * READ：无（群已读不需要） + * RECEIPT：需要回执的目标消息 id */ private Long id; /** @@ -50,7 +58,7 @@ public class ImGroupMessageSendMessage { */ private LocalDateTime sendTime; /** - * @ 的用户编号列表 + * @ 目标用户编号列表 */ private List atUserIds; /** @@ -61,5 +69,51 @@ public class ImGroupMessageSendMessage { * 群回执状态 */ private Integer receiptStatus; + /** + * 群回执已读人数 + */ + private Integer readCount; + + // ========== 静态工厂方法 ========== + + /** + * 构建发送消息 DTO + * + * @param message 群聊消息 DO + * @return 群聊 DTO + */ + public static ImGroupMessageDTO ofSend(ImGroupMessageDO message) { + return BeanUtils.toBean(message, ImGroupMessageDTO.class); + } + + /** + * 构建已读同步 DTO（多端同步：通知自己的其他终端"我已经读了某个群"） + * + * @param senderId 当前用户编号 + * @param groupId 群编号 + * @param readId 已读位置（最大已读消息编号） + * @return 群聊 DTO + */ + public static ImGroupMessageDTO ofRead(Long senderId, Long groupId, Long readId) { + return new ImGroupMessageDTO() + .setId(readId).setType(ImMessageTypeEnum.READ.getType()) + .setSenderId(senderId).setGroupId(groupId); + } + + /** + * 构建群回执 DTO（广播回执状态和已读人数） + * + * @param messageId 消息编号 + * @param groupId 群编号 + * @param readCount 已读人数 + * @param receiptStatus 回执状态 + * @return 群聊 DTO + */ + public static ImGroupMessageDTO ofReceipt(Long messageId, Long groupId, + Integer readCount, Integer receiptStatus) { + return new ImGroupMessageDTO() + .setId(messageId).setType(ImMessageTypeEnum.RECEIPT.getType()) + .setGroupId(groupId).setReadCount(readCount).setReceiptStatus(receiptStatus); + } } diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReadMessage.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReadMessage.java deleted file mode 100644 index 2f96101..0000000 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReadMessage.java +++ /dev/null @@ -1,34 +0,0 @@ -package cn.iocoder.yudao.module.im.websocket; - -import lombok.Data; -import lombok.experimental.Accessors; - -/** - * IM 消息已读 WebSocket 推送消息 - *

- * 用于多端同步：通知自己的其他终端"我已经读了某个会话" - * - * @author 芋道源码 - */ -@Data -@Accessors(chain = true) -public class ImMessageReadMessage { - - public static final String TYPE = "im-message-read"; - - /** - * 私聊场景：接收方编号 - */ - private Long receiverId; - /** - * 群聊场景：群编号 - */ - private Long groupId; - /** - * 消息场景 - * - * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum} - */ - private Integer scene; - -} diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageRecallMessage.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageRecallMessage.java deleted file mode 100644 index d3e3c36..0000000 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageRecallMessage.java +++ /dev/null @@ -1,36 +0,0 @@ -package cn.iocoder.yudao.module.im.websocket; - -import lombok.Data; -import lombok.experimental.Accessors; - -/** - * IM 消息撤回 WebSocket 推送消息 - * - * @author 芋道源码 - */ -@Data -@Accessors(chain = true) -public class ImMessageRecallMessage { - - public static final String TYPE = "im-message-recall"; - - /** - * 消息编号 - */ - private Long messageId; - /** - * 发送人编号 - */ - private Long senderId; - /** - * 群聊场景：群编号 - */ - private Long groupId; - /** - * 消息场景 - * - * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum} - */ - private Integer scene; - -} diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReceiptMessage.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReceiptMessage.java deleted file mode 100644 index 205efc3..0000000 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImMessageReceiptMessage.java +++ /dev/null @@ -1,53 +0,0 @@ -package cn.iocoder.yudao.module.im.websocket; - -import lombok.Data; -import lombok.experimental.Accessors; - -/** - * IM 已读回执 WebSocket 推送消息 - *

- * 私聊：通知对方"我已读了你的消息" - * 群聊：广播回执状态和已读人数 - * - * @author 芋道源码 - */ -@Data -@Accessors(chain = true) -public class ImMessageReceiptMessage { - - public static final String TYPE = "im-message-receipt"; - - /** - * 消息场景 - * - * 关联 {@link cn.iocoder.yudao.module.im.enums.message.ImMessageSceneEnum} - */ - private Integer scene; - - // ========== 私聊场景 ========== - - /** - * 私聊场景：已读方的用户编号 - */ - private Long receiverId; - - // ========== 群聊场景 ========== - - /** - * 群聊场景：消息编号 - */ - private Long messageId; - /** - * 群聊场景：群编号 - */ - private Long groupId; - /** - * 群聊场景：已读人数 - */ - private Integer readCount; - /** - * 群聊场景：回执状态 - */ - private Integer receiptStatus; - -} diff --git a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImPrivateMessageDTO.java b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImPrivateMessageDTO.java index 4779e81..471c75f 100644 --- a/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImPrivateMessageDTO.java +++ b/yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/websocket/ImPrivateMessageDTO.java @@ -45,8 +45,6 @@ public class ImPrivateMessageDTO { private Long receiverId; /** * 消息类型 - *

- * TEXT / IMAGE / FILE / VOICE / VIDEO / READ / RECEIPT / RECALL / TIP_TEXT */ private Integer type; /** diff --git a/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java b/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java index 5f96582..9edb995 100644 --- a/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java +++ b/yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java @@ -1,5 +1,6 @@ package cn.iocoder.yudao.module.im.service.message; +import cn.hutool.extra.spring.SpringUtil; import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; import cn.iocoder.yudao.framework.common.exception.ServiceException; import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest; @@ -9,7 +10,7 @@ import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO; import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO; import cn.iocoder.yudao.module.im.dal.dataobject.message.ImGroupMessageDO; import cn.iocoder.yudao.module.im.dal.mysql.message.ImGroupMessageMapper; -import cn.iocoder.yudao.module.im.dal.redis.group.GroupReadPositionRedisDAO; +import cn.iocoder.yudao.module.im.dal.redis.message.GroupReadPositionRedisDAO; import cn.iocoder.yudao.module.im.enums.message.ImGroupMessageReceiptStatusEnum; import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum; import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService; @@ -18,9 +19,12 @@ import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; import org.mockito.Mock; +import org.mockito.MockedStatic; import java.time.LocalDateTime; -import java.util.*; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*; import static org.junit.jupiter.api.Assertions.*; @@ -38,13 +42,13 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { private ImGroupMessageServiceImpl groupMessageService; @Mock - private ImGroupMessageMapper imGroupMessageMapper; + private ImGroupMessageMapper groupMessageMapper; @Mock - private ImGroupService imGroupService; + private ImGroupService groupService; @Mock - private ImGroupMemberService imGroupMemberService; + private ImGroupMemberService groupMemberService; @Mock - private ImSensitiveWordService imSensitiveWordService; + private ImSensitiveWordService sensitiveWordService; @Mock private GroupReadPositionRedisDAO groupReadPositionRedisDAO; @Mock @@ -63,47 +67,50 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testSendMessage_success() { - // 准备 - ImGroupMessageSendReqVO reqVO = buildSendReqVO(); - when(imGroupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001")) - .thenReturn(null); - ImGroupDO group = new ImGroupDO(); - group.setId(10L); - group.setName("测试群"); - when(imGroupService.validateGroupExists(10L)).thenReturn(group); - - ImGroupMemberDO member = new ImGroupMemberDO(); - member.setGroupId(10L); - member.setUserId(1L); - member.setStatus(CommonStatusEnum.ENABLE.getStatus()); - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(member); - - List allMembers = List.of( - member, - ImGroupMemberDO.builder().groupId(10L).userId(2L) - .status(CommonStatusEnum.ENABLE.getStatus()).build(), - ImGroupMemberDO.builder().groupId(10L).userId(3L) - .status(CommonStatusEnum.ENABLE.getStatus()).build() - ); - when(imGroupMemberService.selectByGroupId(10L)).thenReturn(allMembers); - when(imGroupMessageMapper.insert(any(ImGroupMessageDO.class))).thenAnswer(invocation -> { - ImGroupMessageDO msg = invocation.getArgument(0); - msg.setId(99L); - return 1; - }); - - // 调用 - ImGroupMessageDO result = groupMessageService.sendGroupMessage(1L, reqVO); - - // 断言 - assertNotNull(result); - assertEquals(1L, result.getSenderId()); - assertEquals(10L, result.getGroupId()); - assertEquals(ImMessageStatusEnum.UNREAD.getStatus(), result.getStatus()); - assertEquals(ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus(), result.getReceiptStatus()); - - // 验证推送给 3 个群成员 - verify(webSocketMessageSender, times(3)).sendObject(anyInt(), anyLong(), anyString(), any()); + try (MockedStatic springUtilMockedStatic = mockStatic(SpringUtil.class)) { + springUtilMockedStatic.when(() -> SpringUtil.getBean(eq(ImGroupMessageServiceImpl.class))) + .thenReturn(groupMessageService); + + // 准备 + ImGroupMessageSendReqVO reqVO = buildSendReqVO(); + when(groupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001")) + .thenReturn(null); + ImGroupDO group = new ImGroupDO(); + group.setId(10L); + group.setName("测试群"); + when(groupService.validateGroupExists(10L)).thenReturn(group); + + ImGroupMemberDO member = ImGroupMemberDO.builder() + .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build(); + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(member); + + List allMembers = List.of( + member, + ImGroupMemberDO.builder().groupId(10L).userId(2L) + .status(CommonStatusEnum.ENABLE.getStatus()).build(), + ImGroupMemberDO.builder().groupId(10L).userId(3L) + .status(CommonStatusEnum.ENABLE.getStatus()).build() + ); + when(groupMemberService.selectByGroupId(10L)).thenReturn(allMembers); + when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenAnswer(invocation -> { + ImGroupMessageDO msg = invocation.getArgument(0); + msg.setId(99L); + return 1; + }); + + // 调用 + ImGroupMessageDO result = groupMessageService.sendGroupMessage(1L, reqVO); + + // 断言 + assertNotNull(result); + assertEquals(1L, result.getSenderId()); + assertEquals(10L, result.getGroupId()); + assertEquals(ImMessageStatusEnum.UNREAD.getStatus(), result.getStatus()); + assertEquals(ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus(), result.getReceiptStatus()); + + // 验证推送给 3 个群成员（含发送者自己，用于多端同步） + verify(webSocketMessageSender, times(3)).sendObject(anyInt(), anyLong(), anyString(), any()); + } } @Test @@ -114,7 +121,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .id(100L).clientMessageId("test-uuid-group-001").senderId(1L).groupId(10L) .type(0).content("{\"content\":\"群聊你好\"}").status(0) .sendTime(LocalDateTime.now()).build(); - when(imGroupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001")) + when(groupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001")) .thenReturn(existing); // 调用 @@ -122,19 +129,19 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { // 断言 assertEquals(100L, result.getId()); - verify(imGroupMessageMapper, never()).insert(any(ImGroupMessageDO.class)); + verify(groupMessageMapper, never()).insert(any(ImGroupMessageDO.class)); } @Test public void testSendMessage_notInGroup() { // 准备 ImGroupMessageSendReqVO reqVO = buildSendReqVO(); - when(imGroupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001")) + when(groupMessageMapper.selectBySenderIdAndClientMessageId(1L, "test-uuid-group-001")) .thenReturn(null); ImGroupDO group = new ImGroupDO(); group.setId(10L); - when(imGroupService.validateGroupExists(10L)).thenReturn(group); - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(null); + when(groupService.validateGroupExists(10L)).thenReturn(group); + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null); // 调用并断言 ServiceException exception = assertThrows(ServiceException.class, @@ -152,7 +159,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .groupId(10L).userId(1L) .status(CommonStatusEnum.ENABLE.getStatus()) .joinTime(joinTime).build(); - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); // 模拟一条入群前的消息和一条入群后的消息 ImGroupMessageDO beforeJoin = ImGroupMessageDO.builder() @@ -161,7 +168,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { ImGroupMessageDO afterJoin = ImGroupMessageDO.builder() .id(2L).groupId(10L).senderId(2L) .sendTime(LocalDateTime.of(2026, 4, 12, 11, 0, 0)).build(); - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) .thenReturn(List.of(beforeJoin, afterJoin)); // 调用 @@ -181,7 +188,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .groupId(10L).userId(1L) .status(CommonStatusEnum.DISABLE.getStatus()) .joinTime(joinTime).quitTime(quitTime).build(); - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); // 退群前后各一条消息 ImGroupMessageDO beforeQuit = ImGroupMessageDO.builder() @@ -190,7 +197,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { ImGroupMessageDO afterQuit = ImGroupMessageDO.builder() .id(2L).groupId(10L).senderId(2L) .sendTime(LocalDateTime.of(2026, 4, 13, 10, 0, 0)).build(); - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) .thenReturn(List.of(beforeQuit, afterQuit)); // 调用 @@ -208,7 +215,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .groupId(10L).userId(1L) .status(CommonStatusEnum.ENABLE.getStatus()) .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build(); - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); // 定向接收消息：只给用户 2 和 3 ImGroupMessageDO directedMsg = ImGroupMessageDO.builder() @@ -219,7 +226,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { ImGroupMessageDO allMsg = ImGroupMessageDO.builder() .id(2L).groupId(10L).senderId(5L) .sendTime(LocalDateTime.of(2026, 4, 12, 10, 0, 0)).build(); - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) .thenReturn(List.of(directedMsg, allMsg)); // 调用 @@ -237,13 +244,13 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .groupId(10L).userId(1L) .status(CommonStatusEnum.ENABLE.getStatus()) .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build(); - when(imGroupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); + when(groupMemberService.getGroupMembersByUserId(1L)).thenReturn(List.of(member)); ImGroupMessageDO atMsg = ImGroupMessageDO.builder() .id(1L).groupId(10L).senderId(2L) .atUserIds(List.of(1L, 3L)) .sendTime(LocalDateTime.of(2026, 4, 12, 10, 0, 0)).build(); - when(imGroupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) + when(groupMessageMapper.selectListByMinId(List.of(10L), 0L, 100)) .thenReturn(List.of(atMsg)); // 调用 @@ -258,28 +265,88 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testRecallMessage_success() { + try (MockedStatic springUtilMockedStatic = mockStatic(SpringUtil.class)) { + springUtilMockedStatic.when(() -> SpringUtil.getBean(eq(ImGroupMessageServiceImpl.class))) + .thenReturn(groupMessageService); + + // 准备：消息由用户 1 发送，刚发送 5 分钟内 + ImGroupMessageDO message = ImGroupMessageDO.builder() + .id(50L).senderId(1L).groupId(10L) + .status(ImMessageStatusEnum.UNREAD.getStatus()) + .sendTime(LocalDateTime.now()).build(); + when(groupMessageMapper.selectById(50L)).thenReturn(message); + // 撤回前需要校验用户仍在群中 + ImGroupMemberDO selfMember = ImGroupMemberDO.builder() + .groupId(10L).userId(1L).status(CommonStatusEnum.ENABLE.getStatus()).build(); + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(selfMember); + when(groupMessageMapper.updateById(any(ImGroupMessageDO.class))).thenReturn(1); + when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenReturn(1); + + List members = List.of( + selfMember, + ImGroupMemberDO.builder().groupId(10L).userId(2L) + .status(CommonStatusEnum.ENABLE.getStatus()).build(), + ImGroupMemberDO.builder().groupId(10L).userId(3L) + .status(CommonStatusEnum.DISABLE.getStatus()).build() + ); + when(groupMemberService.selectByGroupId(10L)).thenReturn(members); + + // 调用 + ImGroupMessageDO result = groupMessageService.recallGroupMessage(1L, 50L); + + // 断言：返回 TIP_TEXT 消息 + assertNotNull(result); + // 验证：更新原消息状态 + 插入 tipMessage + verify(groupMessageMapper).updateById(any(ImGroupMessageDO.class)); + verify(groupMessageMapper).insert(any(ImGroupMessageDO.class)); + // 验证：只给 2 个活跃成员推送撤回提示（userId=3 已退群） + verify(webSocketMessageSender, times(2)).sendObject(anyInt(), anyLong(), anyString(), any()); + } + } + + @Test + public void testRecallMessage_notOwn() { + // 准备 + ImGroupMessageDO message = ImGroupMessageDO.builder() + .id(50L).senderId(2L).groupId(10L) + .status(ImMessageStatusEnum.UNREAD.getStatus()) + .sendTime(LocalDateTime.now()).build(); + when(groupMessageMapper.selectById(50L)).thenReturn(message); + + // 调用并断言 + ServiceException exception = assertThrows(ServiceException.class, + () -> groupMessageService.recallGroupMessage(1L, 50L)); + assertEquals(MESSAGE_RECALL_DENIED.getCode(), exception.getCode()); + } + + @Test + public void testRecallMessage_alreadyRecalled() { // 准备 ImGroupMessageDO message = ImGroupMessageDO.builder() .id(50L).senderId(1L).groupId(10L) - .status(ImMessageStatusEnum.UNREAD.getStatus()).build(); - when(imGroupMessageMapper.selectById(50L)).thenReturn(message); - when(imGroupMessageMapper.updateById(any(ImGroupMessageDO.class))).thenReturn(1); + .status(ImMessageStatusEnum.RECALL.getStatus()) + .sendTime(LocalDateTime.now()).build(); + when(groupMessageMapper.selectById(50L)).thenReturn(message); - List members = List.of( - ImGroupMemberDO.builder().groupId(10L).userId(1L) - .status(CommonStatusEnum.ENABLE.getStatus()).build(), - ImGroupMemberDO.builder().groupId(10L).userId(2L) - .status(CommonStatusEnum.ENABLE.getStatus()).build(), - ImGroupMemberDO.builder().groupId(10L).userId(3L) - .status(CommonStatusEnum.DISABLE.getStatus()).build() - ); - when(imGroupMemberService.selectByGroupId(10L)).thenReturn(members); + // 调用并断言 + ServiceException exception = assertThrows(ServiceException.class, + () -> groupMessageService.recallGroupMessage(1L, 50L)); + assertEquals(MESSAGE_ALREADY_RECALLED.getCode(), exception.getCode()); + } - // 调用 - groupMessageService.recallGroupMessage(1L, 50L); + @Test + public void testRecallMessage_timeout() { + // 准备：消息发送于 10 分钟前（超过 5 分钟窗口） + ImGroupMessageDO message = ImGroupMessageDO.builder() + .id(50L).senderId(1L).groupId(10L) + .status(ImMessageStatusEnum.UNREAD.getStatus()) + .sendTime(LocalDateTime.now().minusMinutes(10)).build(); + when(groupMessageMapper.selectById(50L)).thenReturn(message); - // 断言：只给 2 个活跃成员发送 RECALL 事件（userId=3 已退群） - verify(webSocketMessageSender, times(2)).sendObject(anyInt(), anyLong(), anyString(), any()); + // 调用并断言 + ServiceException exception = assertThrows(ServiceException.class, + () -> groupMessageService.recallGroupMessage(1L, 50L)); + assertEquals(MESSAGE_RECALL_TIMEOUT.getCode(), exception.getCode()); } // ========== 群已读测试 ========== @@ -287,7 +354,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testReadMessages_notInGroup() { // 准备：当前用户不在群中 - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(null); + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(null); // 调用并断言：越权校验 ServiceException exception = assertThrows(ServiceException.class, @@ -302,13 +369,13 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .groupId(10L).userId(1L) .status(CommonStatusEnum.ENABLE.getStatus()) .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build(); - when(imGroupMemberService.getGroupMember(10L, 1L)).thenReturn(currentMember); + when(groupMemberService.getGroupMember(10L, 1L)).thenReturn(currentMember); // 准备：消息由用户 5 发，发送时间在 2026-04-12 10:00 ImGroupMessageDO message = ImGroupMessageDO.builder() .id(80L).groupId(10L).senderId(5L) .sendTime(LocalDateTime.of(2026, 4, 12, 10, 0, 0)).build(); - when(imGroupMessageMapper.selectById(80L)).thenReturn(message); + when(groupMessageMapper.selectById(80L)).thenReturn(message); // 准备：群成员一览 // 用户 1: 正常，入群在消息之前 @@ -327,7 +394,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { .status(CommonStatusEnum.ENABLE.getStatus()) .joinTime(LocalDateTime.of(2026, 1, 1, 0, 0, 0)).build() // 发送者 ); - when(imGroupMemberService.selectByGroupId(10L)).thenReturn(allMembers); + when(groupMemberService.selectByGroupId(10L)).thenReturn(allMembers); // 准备：Redis 已读位置 — 用户 1 读到 100, 用户 2 读到 50, 用户 3 读到 200 Map positions = new HashMap<>(); @@ -340,7 +407,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { List readUsers = groupMessageService.getGroupReadUsers(1L, 10L, 80L); // 断言： - // 用户 1: 可见 + readMaxId=100>=80 → 已读 ✓（但排除发送者？不是发送者，所以算入） + // 用户 1: 可见 + readMaxId=100>=80 → 已读 ✓ // 用户 2: 可见 + readMaxId=50<80 → 未读 ✗ // 用户 3: 入群在消息之后 → 不可见 → 不算 // 用户 5: 发送者 → 排除 @@ -354,7 +421,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { @Test public void testGetReadUsers_notInGroup() { // 准备：当前用户不在群中 - when(imGroupMemberService.getGroupMember(10L, 99L)).thenReturn(null); + when(groupMemberService.getGroupMember(10L, 99L)).thenReturn(null); // 调用并断言：越权校验 ServiceException exception = assertThrows(ServiceException.class,