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 72c193c..7577e4b 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 @@ -138,16 +138,71 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService { } } - // 3. 按可见性过滤(入群前不可见、定向消息排除),并按 id 升序重排 + // 3.1 按可见性过滤(入群前不可见、定向消息排除),按 id 升序后仅取本页 size 条, + // 避免「在群 + 退群前」多路合并时一次响应跨度过大、游标直接跳到全局最大 id 而漏拉中间消息 List result = messages.stream() .filter(msg -> isMessageVisible(msg, memberMap.get(msg.getGroupId()), userId)) .sorted(Comparator.comparing(ImGroupMessageDO::getId)) + .limit(size) .toList(); + // 3.2 按当前用户补齐离线拉取结果:消息已读态(相对 Redis 已读游标)、本人发送的回执消息的已读人数 + enrichPullResultForViewer(userId, result); + log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size()); return result; } + /** + * 离线拉取结果按当前用户补齐:消息已读态(相对 Redis 已读游标)、本人发送的回执消息的已读人数 + */ + @SuppressWarnings("StatementWithEmptyBody") + private void enrichPullResultForViewer(Long userId, List messages) { + if (CollUtil.isEmpty(messages)) { + return; + } + Map viewerReadCursorByGroup = new HashMap<>(); // 群 → 已读位置 + Map> readPositionsByGroup = new HashMap<>(); // 群 → (用户 → 已读位置) + Map> membersByGroup = new HashMap<>(); // 群 → 全部成员列表 + for (ImGroupMessageDO message : messages) { + // 消息已读态(status):撤回 > 已读 > 未读 + Long groupId = message.getGroupId(); + long readCursor = viewerReadCursorByGroup.computeIfAbsent(groupId, gid -> { + Long readMax = groupMessageReadRedisDAO.getReadMaxMessageId(gid, userId); + return readMax != null ? readMax : -1L; + }); + if (ImMessageStatusEnum.RECALL.getStatus().equals(message.getStatus())) { + // 保持撤回态 + } else if (readCursor >= message.getId()) { + message.setStatus(ImMessageStatusEnum.READ.getStatus()); + } else { + message.setStatus(ImMessageStatusEnum.UNREAD.getStatus()); + } + + // 回执消息的已读人数(readCount):仅补齐本人发送的,其他消息不处理(回执消息才关心已读人数,且只对发送者可见) + if (ObjUtil.notEqual(message.getSenderId(), userId) + || ImGroupMessageReceiptStatusEnum.NO_RECEIPT.getStatus().equals(message.getReceiptStatus())) { + continue; + } + Map positions = readPositionsByGroup.computeIfAbsent(groupId, + gid -> groupMessageReadRedisDAO.getReadMaxMessageIdMap(gid)); + List allMembers = membersByGroup.computeIfAbsent(groupId, + groupMemberService::getGroupMemberListByGroupId); + Set visibleUserIds = getVisibleUserIds(message, allMembers); + visibleUserIds.remove(message.getSenderId()); + int readCount = 0; + if (CollUtil.isNotEmpty(visibleUserIds)) { + for (Long uid : visibleUserIds) { + Long pos = positions.get(uid); + if (pos != null && pos >= message.getId()) { + readCount++; + } + } + } + message.setReadCount(readCount); + } + } + @Override public void readGroupMessages(Long userId, Long groupId, Long messageId) { Assert.notNull(messageId, "已读消息编号不能为空"); 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 e3b2f9f..355dad7 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 @@ -311,6 +311,123 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest { assertEquals(List.of(1L, 3L), result.get(0).getAtUserIds()); } + @Test + public void testPullMessages_pageTrimmedToSize() { + LocalDateTime now = LocalDateTime.now(); + ImGroupMemberDO activeMember = ImGroupMemberDO.builder() + .groupId(10L).userId(1L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(now.minusDays(10)).build(); + when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(activeMember)); + + ImGroupMessageDO activeMsg1 = ImGroupMessageDO.builder() + .id(1000L).groupId(10L).senderId(2L).sendTime(now.minusHours(2)).build(); + ImGroupMessageDO activeMsg2 = ImGroupMessageDO.builder() + .id(2000L).groupId(10L).senderId(2L).sendTime(now.minusHours(1)).build(); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(0L), + any(LocalDateTime.class), eq(2))) + .thenReturn(List.of(activeMsg1, activeMsg2)); + + LocalDateTime quitTime = now.minusDays(1); + ImGroupMemberDO quitMember = ImGroupMemberDO.builder() + .groupId(20L).userId(1L) + .status(CommonStatusEnum.DISABLE.getStatus()) + .joinTime(now.minusDays(20)) + .quitTime(quitTime).build(); + when(groupMemberService.getQuitGroupMemberListByUserId(eq(1L), any(LocalDateTime.class))) + .thenReturn(List.of(quitMember)); + + ImGroupMessageDO quit1 = ImGroupMessageDO.builder() + .id(150L).groupId(20L).senderId(99L).sendTime(now.minusDays(5)).build(); + ImGroupMessageDO quit2 = ImGroupMessageDO.builder() + .id(160L).groupId(20L).senderId(99L).sendTime(now.minusDays(4)).build(); + when(groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(0L), + any(LocalDateTime.class), eq(quitTime), eq(2))) + .thenReturn(List.of(quit1, quit2)); + + when(groupMessageReadRedisDAO.getReadMaxMessageId(anyLong(), eq(1L))).thenReturn(null); + + List result = groupMessageService.pullGroupMessageList(1L, 0L, 2); + + assertEquals(2, result.size()); + assertEquals(150L, result.get(0).getId()); + assertEquals(160L, result.get(1).getId()); + } + + @Test + public void testPullMessages_pullSetsReadFromRedisCursor() { + LocalDateTime now = LocalDateTime.now(); + ImGroupMemberDO member = ImGroupMemberDO.builder() + .groupId(10L).userId(1L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(now.minusDays(10)).build(); + when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(member)); + + ImGroupMessageDO low = ImGroupMessageDO.builder() + .id(5L).groupId(10L).senderId(2L).status(ImMessageStatusEnum.UNREAD.getStatus()) + .sendTime(now.minusHours(2)).build(); + ImGroupMessageDO high = ImGroupMessageDO.builder() + .id(10L).groupId(10L).senderId(2L).status(ImMessageStatusEnum.UNREAD.getStatus()) + .sendTime(now.minusHours(1)).build(); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(0L), + any(LocalDateTime.class), eq(100))) + .thenReturn(List.of(low, high)); + when(groupMemberService.getQuitGroupMemberListByUserId(eq(1L), any(LocalDateTime.class))) + .thenReturn(List.of()); + + when(groupMessageReadRedisDAO.getReadMaxMessageId(10L, 1L)).thenReturn(7L); + + List result = groupMessageService.pullGroupMessageList(1L, 0L, 100); + + assertEquals(2, result.size()); + assertEquals(ImMessageStatusEnum.READ.getStatus(), result.get(0).getStatus()); + assertEquals(ImMessageStatusEnum.UNREAD.getStatus(), result.get(1).getStatus()); + } + + @Test + public void testPullMessages_receiptReadCountForSender() { + LocalDateTime now = LocalDateTime.now(); + ImGroupMemberDO member = ImGroupMemberDO.builder() + .groupId(10L).userId(1L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(now.minusDays(10)).build(); + when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(member)); + + ImGroupMessageDO receiptMsg = ImGroupMessageDO.builder() + .id(100L).groupId(10L).senderId(1L) + .status(ImMessageStatusEnum.UNREAD.getStatus()) + .receiptStatus(ImGroupMessageReceiptStatusEnum.PENDING.getStatus()) + .sendTime(now.minusHours(1)).build(); + when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(0L), + any(LocalDateTime.class), eq(100))) + .thenReturn(List.of(receiptMsg)); + when(groupMemberService.getQuitGroupMemberListByUserId(eq(1L), any(LocalDateTime.class))) + .thenReturn(List.of()); + + when(groupMessageReadRedisDAO.getReadMaxMessageId(10L, 1L)).thenReturn(100L); + Map positions = new HashMap<>(); + positions.put(1L, 100L); + positions.put(2L, 100L); + positions.put(3L, 50L); + when(groupMessageReadRedisDAO.getReadMaxMessageIdMap(10L)).thenReturn(positions); + + List allMembers = List.of( + member, + ImGroupMemberDO.builder().groupId(10L).userId(2L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(now.minusDays(20)).build(), + ImGroupMemberDO.builder().groupId(10L).userId(3L) + .status(CommonStatusEnum.ENABLE.getStatus()) + .joinTime(now.minusDays(20)).build() + ); + when(groupMemberService.getGroupMemberListByGroupId(10L)).thenReturn(allMembers); + + List result = groupMessageService.pullGroupMessageList(1L, 0L, 100); + + assertEquals(1, result.size()); + assertEquals(1, result.get(0).getReadCount()); + } + // ========== 撤回测试 ========== @Test