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

- 新增 flushIdentity：把某身份缓冲里未落库的进度立即写进权威表；
  mergeDeviceIntoUser 先调用它再扫库，否则登录前 2 秒内的上报会被漏掉，
  稍后落库还会生成一条登录用户看不到的匿名孤儿行；合并完清掉设备侧缓冲。
- planOne 增加判定：逻辑删除会刷新 updated_at，上报时间不晚于它的条目直接丢弃，
  防止删除瞬间已 drain 的在途引用把 deleted 写回 0。
- 去掉 save(Long userId, ...) 上多余的 @Transactional（缓冲路径白开事务）。

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)));
     }
 

fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java

@@ -8,6 +8,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.data.redis.core.SessionCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.springframework.stereotype.Component;
 
 import java.time.Duration;
@@ -19,7 +20,7 @@ import java.util.Map;
 /**
  * 观看进度 Redis 缓冲（write-behind）。
  *
- * <p>高并发上报的落地形态：POST 只写 Redis（每次上报 1 次写），由
+ * <p>高并发上报的落地形态：POST 只写 Redis（Lua 脚本一次往返完成写状态与入队），由
  * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后批量 UPSERT
  * 进权威表 watch_progress。Redis 不可用时 put 返回 false，调用方回退同步写库。
  *
@@ -45,6 +46,22 @@ public class WatchProgressBuffer {
     private static final int MAX_DRAMAS_PER_IDENTITY = 30;
     private static final Duration TTL = Duration.ofHours(12);
 
+    /**
+     * 上报热路径的单次往返脚本：写状态、续 TTL、入待落库队列、截断队列一次做完。
+     * 拆成多条命令时进程若死在「写状态」与「入队」之间，该剧进度只留在 HASH 里、
+     * 永远不会落库，12 小时后被 TTL 清掉。返回 1 表示该身份的剧数超限，需要补充裁剪。
+     */
+    private static final DefaultRedisScript<Long> PUT_SCRIPT = new DefaultRedisScript<>("""
+            redis.call('HSET', KEYS[1], ARGV[1], ARGV[2])
+            redis.call('EXPIRE', KEYS[1], ARGV[3])
+            redis.call('LPUSH', KEYS[2], ARGV[4])
+            redis.call('LTRIM', KEYS[2], 0, ARGV[5])
+            if redis.call('HLEN', KEYS[1]) > tonumber(ARGV[6]) then
+              return 1
+            end
+            return 0
+            """, Long.class);
+
     private final StringRedisTemplate redis;
 
     /**
@@ -80,12 +97,18 @@ public class WatchProgressBuffer {
             String key = identityKey(identity);
             String hashKey = BUF_KEY_PREFIX + key;
             String field = Long.toString(dramaId);
-            redis.opsForHash().put(hashKey, field, encode(
-                    episodeId, positionMs, durationMs, finished, source, deviceId));
-            redis.expire(hashKey, TTL);
-            trimIdentity(hashKey);
-            redis.opsForList().leftPush(DIRTY_KEY, key + "|" + field);
-            redis.opsForList().trim(DIRTY_KEY, 0, MAX_DIRTY - 1);
+            Long overflow = redis.execute(PUT_SCRIPT,
+                    List.of(hashKey, DIRTY_KEY),
+                    field,
+                    encode(episodeId, positionMs, durationMs, finished, source, deviceId),
+                    Long.toString(TTL.toSeconds()),
+                    key + "|" + field,
+                    Long.toString(MAX_DIRTY - 1),
+                    Integer.toString(MAX_DRAMAS_PER_IDENTITY));
+            if (overflow != null && overflow > 0) {
+                // 超出 30 部才做一次 HGETALL + HDEL 裁剪，热路径保持单次往返
+                trimIdentity(hashKey);
+            }
             return true;
         } catch (Exception e) {
             log.warn("观看进度写入 Redis 缓冲失败，本次回退同步写库: {}", e.getMessage());

fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java

@@ -85,7 +85,6 @@ public class WatchProgressService {
             .expireAfterWrite(Duration.ofMinutes(5))
             .build();
 
-    @Transactional(rollbackFor = Exception.class)
     public ProgressRespVO save(Long userId, ProgressSaveReqVO req) {
         return save(new InteractionIdentity(userId, ""), req);
     }
@@ -460,6 +459,12 @@ public class WatchProgressService {
             return;
         }
         boolean hidden = row.getDeleted() != null && row.getDeleted() != 0;
+        long rowTs = toMillis(row.getUpdatedAt());
+        if (hidden && rowTs > 0 && state.ts() <= rowTs) {
+            // 逻辑删除会刷新 updated_at：上报时间早于它的条目是删除前排队的迟到数据，
+            // 丢弃即可，否则会把已删记录重新写活。
+            return;
+        }
         boolean sameEpisode = row.getEpisodeId() != null
                 && row.getEpisodeId().equals(state.episodeId());
         if (hidden && sameEpisode) {
@@ -661,6 +666,36 @@ public class WatchProgressService {
         historyMapper.insert(history);
     }
 
+    /**
+     * 把某个身份缓冲里尚未落库的进度立即写进权威表。
+     * 登录合并前必须先调用：合并只扫库表，缓冲里的设备进度会被漏掉，
+     * 随后定时落库又会生成一条登录用户看不到的匿名孤儿行。
+     *
+     * @return 本次合并的条目数
+     */
+    public int flushIdentity(InteractionIdentity identity) {
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer == null) {
+            return 0;
+        }
+        Map<Long, WatchProgressBuffer.BufferedProgress> states = buffer.getAll(identity);
+        if (states.isEmpty()) {
+            return 0;
+        }
+        String key = buffer.identityKey(identity);
+        List<WatchProgressBuffer.IdentityRef> refs = states.keySet().stream()
+                .map(dramaId -> new WatchProgressBuffer.IdentityRef(
+                        key, identity.userId(), identity.deviceId(), dramaId))
+                .toList();
+        try {
+            return flushRefs(buffer, refs);
+        } catch (Exception ex) {
+            // 预写失败不影响合并本身：dirty 队列里的引用还在，下个周期会再落库
+            log.warn("观看进度缓冲立即落库失败，跳过预写: {}", ex.getMessage());
+            return 0;
+        }
+    }
+
     /**
      * 登录时合并设备进度和观看流水。
      *
@@ -672,6 +707,9 @@ public class WatchProgressService {
         if (userId == null || !StringUtils.hasText(deviceId)) {
             return new MergeResult(0, 0, 0, 0);
         }
+        InteractionIdentity deviceIdentity = new InteractionIdentity(null, deviceId);
+        // 缓冲里可能还有没落库的设备进度，先落库再合并，否则这段进度会被漏掉
+        flushIdentity(deviceIdentity);
         List<WatchProgress> deviceProgress = progressMapper.selectList(
                 Wrappers.<WatchProgress>lambdaQuery()
                         .isNull(WatchProgress::getUserId)
@@ -731,6 +769,11 @@ public class WatchProgressService {
             }
             historyMapper.deleteById(device.getId());
         }
+        // 设备侧进度已迁走或淘汰，缓冲留着只会让读覆盖层把已删行读回来
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer != null) {
+            buffer.clear(deviceIdentity);
+        }
         return new MergeResult(
                 progressMigrated, progressIgnored, historyMigrated, historyIgnored);
     }