test(playback): 补充合并预写缓冲与迟到条目丢弃用例

fruitshorts-app-api/src/test/java/com/fruitshorts/playback/service/WatchProgressServiceTest.java

@@ -20,11 +20,14 @@ import com.fruitshorts.playback.vo.ProgressRespVO;
 import com.fruitshorts.playback.vo.ProgressSaveReqVO;
 import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.session.ExecutorType;
+import org.apache.ibatis.session.SqlSession;
 import org.apache.ibatis.session.SqlSessionFactory;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
@@ -38,6 +41,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -384,7 +388,58 @@ class WatchProgressServiceTest {
         verify(progressMapper, never()).selectBatchForPairs(any());
     }
 
+    @Test
+    void should_flush_device_buffer_before_merging_and_clear_it_after() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        given(buffer.getAll(any(InteractionIdentity.class))).willReturn(Map.of());
+        WatchProgressService service = serviceWithBuffer(buffer);
+        given(progressMapper.selectList(any())).willReturn(List.of());
+        given(historyMapper.selectList(any())).willReturn(List.of());
+
+        service.mergeDeviceIntoUser(9L, "device-a");
+
+        // 合并前先把缓冲落库，合并完再清掉设备侧缓冲
+        InOrder order = inOrder(buffer);
+        order.verify(buffer).getAll(any(InteractionIdentity.class));
+        order.verify(buffer).clear(any(InteractionIdentity.class));
+    }
+
+    @Test
+    void should_drop_late_buffered_ref_when_row_is_already_deleted() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        SqlSessionFactory sessionFactory = mock(SqlSessionFactory.class);
+        SqlSession session = mock(SqlSession.class);
+        WatchProgressMapper batchProgress = mock(WatchProgressMapper.class);
+        WatchHistoryMapper batchHistory = mock(WatchHistoryMapper.class);
+        given(sessionFactory.openSession(ExecutorType.BATCH, false)).willReturn(session);
+        given(session.getMapper(WatchProgressMapper.class)).willReturn(batchProgress);
+        given(session.getMapper(WatchHistoryMapper.class)).willReturn(batchHistory);
+        WatchProgressService service = serviceWithBuffer(buffer, sessionFactory);
+
+        given(buffer.drain(10)).willReturn(
+                List.of(new WatchProgressBuffer.IdentityRef("u:9", 9L, "", 5L)));
+        given(buffer.getAllBatch(List.of("u:9"))).willReturn(Map.of("u:9", Map.of(
+                5L, new WatchProgressBuffer.BufferedProgress(
+                        5L, 88L, 1_000L, 2_000L, false, "feed", "", 1_700_000_000_000L))));
+        // 库里是逻辑删除行，且删除时刻晚于该条缓冲的上报时刻
+        WatchProgress hidden = progress(5L, 88L);
+        hidden.setUserId(9L);
+        hidden.setDeleted(1);
+        hidden.setUpdatedAt(LocalDateTime.of(2026, 10, 28, 12, 0));
+        given(progressMapper.selectBatchForPairs(any())).willReturn(List.of(hidden));
+
+        service.flushBuffered(10);
+
+        verify(batchProgress, never()).updateRowAny(any());
+        verify(batchHistory, never()).insert(any(WatchHistory.class));
+    }
+
     private WatchProgressService serviceWithBuffer(WatchProgressBuffer buffer) {
+        return serviceWithBuffer(buffer, mock(SqlSessionFactory.class));
+    }
+
+    private WatchProgressService serviceWithBuffer(
+            WatchProgressBuffer buffer, SqlSessionFactory sessionFactory) {
         @SuppressWarnings("unchecked")
         ObjectProvider<WatchProgressBuffer> provider = mock(ObjectProvider.class);
         given(provider.getIfAvailable()).willReturn(buffer);
@@ -394,7 +449,7 @@ class WatchProgressServiceTest {
                 dramaService,
                 episodeService,
                 provider,
-                mock(SqlSessionFactory.class),
+                sessionFactory,
                 new TransactionTemplate(mock(PlatformTransactionManager.class)));
     }