✨ feat(message): 优化群聊的 pullGroupMessageList 方法，解决空分页的风险(pro 增强）

YunaiV
1 parent 0ce9edde
Showing 2 changed files with 162 additions and 45 deletions
yudao-module-im/yudao-module-im-biz/src/main/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImpl.java
yudao-module-im/yudao-module-im-biz/src/test/java/cn/iocoder/yudao/module/im/service/message/ImGroupMessageServiceImplTest.java
@@ -51,8 +51,8 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
     /**
      * 仅用于规避群消息 pull 的"假空页"：
-     * 首批 raw 消息可能因入群时间或定向接收过滤后变成空列表，但后续更大的 id 仍然存在可见消息。
-     * 因此仅在过滤结果为空时，按本轮 raw 最大 id 向后再试几次。
+     * 首批消息可能因入群时间或定向接收过滤后变成空列表，但后续更大的 id 仍然存在可见消息。
+     * 因此仅在过滤结果为空时，按本轮消息最大 id 向后再试几次。
      */
     private static final int PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES = 3;
@@ -111,56 +111,71 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
         // 0. 拉取时间窗：超过窗口的老消息不再通过离线通道推送
         LocalDateTime minSendTime = LocalDateTime.now().minusDays(MESSAGE_GROUP_PULL_MAX_DAYS);
-        // 1. 拉取一批原始消息（包含当前在群、退群前补齐两部分），并过滤得到当前用户可见的结果
-        List<ImGroupMessageDO> result = Collections.emptyList();
-        Long currentMinId = minId; // 当前轮次的下限消息编号；初始为用户传入的 minId，后续会根据拉取结果更新
+        // 1.1 主查询：仅用"当前仍在群"的成员记录驱动；若首批消息过滤后为空，则允许内部重试
+        List<ImGroupMemberDO> activeMembers = groupMemberService.getActiveGroupMemberListByUserId(userId);
+        Map<Long, ImGroupMemberDO> memberMap = convertMap(activeMembers, ImGroupMemberDO::getGroupId);
+        List<ImGroupMessageDO> messages = new ArrayList<>();
+        if (CollUtil.isNotEmpty(activeMembers)) {
+            List<Long> groupIds = convertList(activeMembers, ImGroupMemberDO::getGroupId);
+            messages.addAll(pullActiveGroupMessageList(userId, groupIds, minId, size, minSendTime, memberMap));
+        }
+        // 1.2 补齐"退群前"的消息：
+        //    - 继续基于用户原始 minId 单次查询，不能被主查询内部探测游标带着前进；
+        //    - 若 minId > 0 且能查到对应消息，则进一步把下限抬到该消息的 sendTime，避免把客户端已拥有的老消息再次推送。
+        messages.addAll(pullQuitGroupMessageList(userId, minId, size, minSendTime, memberMap));
+
+        // 2. 合并后统一过滤，得到当前用户可见的结果
+        List<ImGroupMessageDO> result = filterGroupMessageList(messages, memberMap, userId, size);
+
+        // 3. 按当前用户补齐：消息已读态（相对 Redis 已读游标）、本人发送的回执消息的已读人数
+        appendMessageStatusAndReceipt(userId, result);
+        log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size());
+        return result;
+    }
+
+    /**
+     * 拉取当前仍在群主路径的消息。
+     *
+     * 仅当首批消息过滤后仍无可见消息时，才按消息最大 id 继续向后探测，
+     * 直到命中可见消息或确认该来源已耗尽。
+     */
+    private List<ImGroupMessageDO> pullActiveGroupMessageList(Long userId, List<Long> groupIds, Long minId,
+                                                              Integer size, LocalDateTime minSendTime,
+                                                              Map<Long, ImGroupMemberDO> memberMap) {
+        // 1. 主查询内部探测游标：仅用于向后探测，不代表客户端真实已送达边界
+        Long activeMinId = minId;
         for (int retryCount = 0; retryCount <= PULL_GROUP_MESSAGE_EMPTY_RETRY_TIMES; retryCount++) {
-            // 1.1 查询一批原始消息（包含当前在群、退群前补齐两部分）
-            Map<Long, ImGroupMemberDO> memberMap = new HashMap<>(); // 群编号 → 群成员记录（包含当前在群、退群前补齐两部分）
-            List<ImGroupMessageDO> messages = getGroupMessageList(userId, currentMinId, size, minSendTime, memberMap);
+            // 2. 查询本轮消息；若已无更多消息，则当前主路径直接结束
+            List<ImGroupMessageDO> messages = groupMessageMapper.selectListByMinId(groupIds, activeMinId,
+                    minSendTime, size);
             if (CollUtil.isEmpty(messages)) {
-                break;
+                return Collections.emptyList();
             }
-            // 1.2 过滤得到当前用户可见的结果（按 id 升序）
-            result = filterGroupMessageList(messages, memberMap, userId, size);
-            // 已拿到可见消息，或原始消息已不足一页（说明没有继续重试的必要）
-            if (CollUtil.isNotEmpty(result) || messages.size() < size) {
-                break;
+            boolean hasVisibleMessage = CollUtil.anyMatch(messages,
+                    message -> isMessageVisible(message, memberMap.get(message.getGroupId()), userId));
+            boolean sourceExhausted = messages.size() < size;
+            // 3. 本轮已命中可见消息，或主查询来源已耗尽，直接返回这一轮消息
+            if (hasVisibleMessage || sourceExhausted) {
+                return messages;
             }
-            // 1.3 更新 currentMinId，进行下一轮查询；如果本轮查询的最大消息编号没有前进，则说明后续没有更多消息了，直接 break
+            // 4. 按本轮消息最大 id 推进内部游标，跳过这段不可见区间；若游标未前进则直接停止
             Long maxMessageId = getMaxValue(messages, ImGroupMessageDO::getId);
-            if (maxMessageId == null || maxMessageId <= currentMinId) {
-                break;
+            if (maxMessageId <= activeMinId) {
+                return Collections.emptyList();
             }
-            currentMinId = maxMessageId;
+            activeMinId = maxMessageId;
         }
-
-        // 2. 按当前用户补齐：消息已读态（相对 Redis 已读游标）、本人发送的回执消息的已读人数
-        appendMessageStatusAndReceipt(userId, result);
-        log.info("[pullGroupMessageList][userId({}) minId({}) size({}) result({})]", userId, minId, size, result.size());
-        return result;
+        return Collections.emptyList();
     }
     /**
-     * 查询一批原始群消息（包含当前在群、退群前补齐两部分）
+     * 拉取已离开群路径的消息（退群前消息）
      */
-    private List<ImGroupMessageDO> getGroupMessageList(Long userId, Long minId, Integer size,
-                                                       LocalDateTime minSendTime,
-                                                       Map<Long, ImGroupMemberDO> memberMap) {
-        // 1. 主查询：仅用"当前仍在群"的成员记录驱动（已退群的群由步骤 2 补齐）
-        List<ImGroupMemberDO> activeMembers = groupMemberService.getActiveGroupMemberListByUserId(userId);
-        memberMap.putAll(convertMap(activeMembers, ImGroupMemberDO::getGroupId));
-        List<ImGroupMessageDO> messages = new ArrayList<>();
-        if (CollUtil.isNotEmpty(activeMembers)) {
-            List<Long> groupIds = convertList(activeMembers, ImGroupMemberDO::getGroupId);
-            messages.addAll(groupMessageMapper.selectListByMinId(groupIds, minId, minSendTime, size));
-        }
-
-        // 2. 补齐"退群前"的消息：
-        //    - 退群时间早于窗口起点的群直接忽略（消息已不在拉取范围）；
-        //    - 若 minId > 0 且能查到对应消息，则进一步把下限抬到该消息的 sendTime，
-        //      避免把客户端已拥有的老消息再次推送。
+    private List<ImGroupMessageDO> pullQuitGroupMessageList(Long userId, Long minId, Integer size,
+                                                            LocalDateTime minSendTime,
+                                                            Map<Long, ImGroupMemberDO> memberMap) {
+        // 1. 退群补齐始终基于用户原始 minId 计算时间边界，避免被主查询内部重试游标误伤
         LocalDateTime minQuitTime = minSendTime;
         if (minId != null && minId > 0) {
             ImGroupMessageDO minMessage = groupMessageMapper.selectById(minId);
@@ -169,8 +184,11 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
                 minQuitTime = minMessage.getSendTime();
             }
         }
+        // 2. 查询用户离开的群记录；若原始 minId 对应消息仍在窗口内，则用它的发送时间抬升退群筛选下限
+        List<ImGroupMessageDO> messages = new ArrayList<>();
         List<ImGroupMemberDO> quitMembers = groupMemberService.getQuitGroupMemberListByUserId(userId, minQuitTime);
         for (ImGroupMemberDO quitMember : quitMembers) {
+            // 3. 按原始 minId + 退群时间补齐该群退群前消息，并把成员记录写回 memberMap 供统一可见性过滤使用
             List<ImGroupMessageDO> quitGroupMessages = groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore(
                     quitMember.getGroupId(), minId, minSendTime, quitMember.getQuitTime(), size);
             if (CollUtil.isEmpty(quitGroupMessages)) {
@@ -185,6 +185,45 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
     }
     @Test
+    public void testPullMessages_activeQueryRetriesWhenFirstBatchInvisible() {
+        LocalDateTime now = LocalDateTime.now();
+        LocalDateTime joinTime = now.minusHours(1);
+        ImGroupMemberDO member = ImGroupMemberDO.builder()
+                .groupId(10L).userId(1L)
+                .status(CommonStatusEnum.ENABLE.getStatus())
+                .joinTime(joinTime).build();
+        when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(member));
+
+        ImGroupMessageDO beforeJoin = ImGroupMessageDO.builder()
+                .id(1L).groupId(10L).senderId(2L)
+                .sendTime(now.minusHours(3)).build();
+        ImGroupMessageDO directedMsg = ImGroupMessageDO.builder()
+                .id(2L).groupId(10L).senderId(2L)
+                .receiverUserIds(List.of(2L, 3L))
+                .sendTime(now.minusHours(2)).build();
+        ImGroupMessageDO visibleMsg = ImGroupMessageDO.builder()
+                .id(3L).groupId(10L).senderId(2L)
+                .sendTime(now.minusMinutes(30)).build();
+        when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(0L),
+                any(LocalDateTime.class), eq(2)))
+                .thenReturn(List.of(beforeJoin, directedMsg));
+        when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(2L),
+                any(LocalDateTime.class), eq(2)))
+                .thenReturn(List.of(visibleMsg));
+        when(groupMemberService.getQuitGroupMemberListByUserId(eq(1L), any(LocalDateTime.class)))
+                .thenReturn(List.of());
+
+        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessageList(1L, 0L, 2);
+
+        assertEquals(1, result.size());
+        assertEquals(3L, result.get(0).getId());
+        verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(0L),
+                any(LocalDateTime.class), eq(2));
+        verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(2L),
+                any(LocalDateTime.class), eq(2));
+    }
+
+    @Test
     public void testPullMessages_quitMemberLoadsPreQuitMessages() {
         // 准备：当前群（在群）+ 一个最近退群
         LocalDateTime now = LocalDateTime.now();
@@ -232,8 +271,6 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
         // 准备：minId > 0，边界消息 sendTime 晚于窗口起点，应当被用作 minQuitTime
         LocalDateTime now = LocalDateTime.now();
         when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of());
-        when(groupMessageMapper.selectListByMinId(anyList(), anyLong(), any(LocalDateTime.class), anyInt()))
-                .thenReturn(List.of());
         LocalDateTime boundarySendTime = now.minusDays(2);
         ImGroupMessageDO boundary = ImGroupMessageDO.builder()
@@ -251,6 +288,70 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
     }
     @Test
+    public void testPullMessages_quitQueryKeepsOriginalMinIdWhenActiveQueryRetries() {
+        LocalDateTime now = LocalDateTime.now();
+        LocalDateTime joinTime = now.minusHours(1);
+        ImGroupMemberDO activeMember = ImGroupMemberDO.builder()
+                .groupId(10L).userId(1L)
+                .status(CommonStatusEnum.ENABLE.getStatus())
+                .joinTime(joinTime).build();
+        when(groupMemberService.getActiveGroupMemberListByUserId(1L)).thenReturn(List.of(activeMember));
+
+        ImGroupMessageDO beforeJoin = ImGroupMessageDO.builder()
+                .id(100L).groupId(10L).senderId(2L)
+                .sendTime(now.minusHours(3)).build();
+        ImGroupMessageDO directedMsg = ImGroupMessageDO.builder()
+                .id(101L).groupId(10L).senderId(2L)
+                .receiverUserIds(List.of(2L, 3L))
+                .sendTime(now.minusMinutes(50)).build();
+        ImGroupMessageDO visibleActiveMsg = ImGroupMessageDO.builder()
+                .id(102L).groupId(10L).senderId(2L)
+                .sendTime(now.minusMinutes(30)).build();
+        when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(8L),
+                any(LocalDateTime.class), eq(2)))
+                .thenReturn(List.of(beforeJoin, directedMsg));
+        when(groupMessageMapper.selectListByMinId(eq(List.of(10L)), eq(101L),
+                any(LocalDateTime.class), eq(2)))
+                .thenReturn(List.of(visibleActiveMsg));
+
+        LocalDateTime boundarySendTime = now.minusDays(2);
+        ImGroupMessageDO boundary = ImGroupMessageDO.builder()
+                .id(8L).groupId(99L).senderId(2L).sendTime(boundarySendTime).build();
+        when(groupMessageMapper.selectById(8L)).thenReturn(boundary);
+
+        LocalDateTime quitTime = now.minusHours(12);
+        ImGroupMemberDO quitMember = ImGroupMemberDO.builder()
+                .groupId(20L).userId(1L)
+                .status(CommonStatusEnum.DISABLE.getStatus())
+                .joinTime(now.minusDays(10))
+                .quitTime(quitTime).build();
+        when(groupMemberService.getQuitGroupMemberListByUserId(1L, boundarySendTime))
+                .thenReturn(List.of(quitMember));
+
+        ImGroupMessageDO quitGroupMsg = ImGroupMessageDO.builder()
+                .id(50L).groupId(20L).senderId(99L)
+                .sendTime(now.minusDays(1)).build();
+        when(groupMessageMapper.selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(8L),
+                any(LocalDateTime.class), eq(quitTime), eq(2)))
+                .thenReturn(List.of(quitGroupMsg));
+
+        List<ImGroupMessageDO> result = groupMessageService.pullGroupMessageList(1L, 8L, 2);
+
+        assertEquals(2, result.size());
+        assertEquals(50L, result.get(0).getId());
+        assertEquals(102L, result.get(1).getId());
+        verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(8L),
+                any(LocalDateTime.class), eq(2));
+        verify(groupMessageMapper).selectListByMinId(eq(List.of(10L)), eq(101L),
+                any(LocalDateTime.class), eq(2));
+        verify(groupMemberService).getQuitGroupMemberListByUserId(1L, boundarySendTime);
+        verify(groupMessageMapper).selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(8L),
+                any(LocalDateTime.class), eq(quitTime), eq(2));
+        verify(groupMessageMapper, never()).selectListByGroupIdAndMinIdAndQuitTimeBefore(eq(20L), eq(101L),
+                any(LocalDateTime.class), eq(quitTime), eq(2));
+    }
+
+    @Test
     public void testPullMessages_receiverUserIdsFilter() {
         // 准备
         LocalDateTime now = LocalDateTime.now();
@@ -452,9 +553,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
             List<ImGroupMemberDO> 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()
+                            .status(CommonStatusEnum.ENABLE.getStatus()).build()
             );
             when(groupMemberService.getActiveGroupMemberListByGroupId(10L)).thenReturn(members);