fix(playback): 删除观看历史时同步清理进度缓冲

缓冲已是读接口的权威来源，只删库行会被读覆盖层读回、并被下一次落库把行复活。
缓冲补 remove/clear（HDEL/DEL），删除路径先清缓冲再删库行，读覆盖层按库行
deleted 过滤；落库时缓冲状态已失效的条目直接丢弃，不再整批退回队列。

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

@@ -93,6 +93,34 @@ public class WatchProgressBuffer {
         }
     }
 
+    /**
+     * 删除某部剧的缓冲状态，用于用户删除单条观看记录。
+     *
+     * <p>缓冲是读接口的权威来源，删库行必须同步删缓冲，否则会被读覆盖层读回；
+     * 同时缓冲状态消失后，dirty 队列里该条引用在落库时会被丢弃（见
+     * {@code WatchProgressService#flushRefs}），不会再被写回库。
+     */
+    public boolean remove(InteractionIdentity identity, long dramaId) {
+        try {
+            redis.opsForHash().delete(BUF_KEY_PREFIX + identityKey(identity), Long.toString(dramaId));
+            return true;
+        } catch (Exception e) {
+            log.warn("观看进度缓冲删除失败: {}", e.getMessage());
+            return false;
+        }
+    }
+
+    /** 清空某身份的全部缓冲状态，用于用户清空观看历史。 */
+    public boolean clear(InteractionIdentity identity) {
+        try {
+            redis.delete(BUF_KEY_PREFIX + identityKey(identity));
+            return true;
+        } catch (Exception e) {
+            log.warn("观看进度缓冲清空失败: {}", e.getMessage());
+            return false;
+        }
+    }
+
     /** 读单个剧的缓冲状态；无缓冲或解码失败返回 null。 */
     public BufferedProgress get(InteractionIdentity identity, long dramaId) {
         return get(identityKey(identity), dramaId);

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

@@ -279,14 +279,18 @@ public class WatchProgressService {
             List<Long> extraDramaIds = extra.stream()
                     .map(WatchProgressBuffer.BufferedProgress::dramaId)
                     .toList();
-            Set<Long> withRow = progressMapper.selectAnyForIdentityIn(
+            Map<Long, WatchProgress> rowByDrama = progressMapper.selectAnyForIdentityIn(
                             identity.userId(), identity.deviceId(), extraDramaIds)
                     .stream()
-                    .map(WatchProgress::getDramaId)
-                    .collect(Collectors.toSet());
+                    .collect(Collectors.toMap(WatchProgress::getDramaId, row -> row, (a, b) -> a));
             for (WatchProgressBuffer.BufferedProgress buffered : extra) {
+                WatchProgress existing = rowByDrama.get(buffered.dramaId());
+                // 用户已删除该剧（库里是逻辑删除行）：缓冲状态不能被读覆盖层再读回来
+                if (existing != null && existing.getDeleted() != null && existing.getDeleted() != 0) {
+                    continue;
+                }
                 merged.add(transientRow(buffered));
-                if (!withRow.contains(buffered.dramaId())) {
+                if (existing == null) {
                     total++;
                 }
             }
@@ -306,6 +310,12 @@ public class WatchProgressService {
 
     @Transactional(rollbackFor = Exception.class)
     public void remove(InteractionIdentity identity, Long dramaId) {
+        // 缓冲是读接口的权威来源：先删缓冲再删库行，否则删掉的记录会被读覆盖层读回、
+        // 并被下一次落库把行复活。
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer != null && dramaId != null) {
+            buffer.remove(identity, dramaId);
+        }
         progressMapper.delete(identityQuery(identity)
                 .eq(WatchProgress::getDramaId, dramaId));
     }
@@ -317,6 +327,10 @@ public class WatchProgressService {
 
     @Transactional(rollbackFor = Exception.class)
     public void clear(InteractionIdentity identity) {
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer != null) {
+            buffer.clear(identity);
+        }
         progressMapper.delete(identityQuery(identity));
     }
 
@@ -338,8 +352,7 @@ public class WatchProgressService {
             return 0;
         }
         try {
-            flushRefs(buffer, refs);
-            return refs.size();
+            return flushRefs(buffer, refs);
         } catch (Exception ex) {
             log.warn("观看进度缓冲合并失败，整批退回队列稍后重试: {}", ex.getMessage());
             buffer.requeue(refs);
@@ -347,7 +360,7 @@ public class WatchProgressService {
         }
     }
 
-    private void flushRefs(WatchProgressBuffer buffer, List<WatchProgressBuffer.IdentityRef> refs) {
+    private int flushRefs(WatchProgressBuffer buffer, List<WatchProgressBuffer.IdentityRef> refs) {
         // 1) pipeline 批量读缓冲状态，key -> (dramaId -> state)
         List<String> identityKeys = refs.stream()
                 .map(WatchProgressBuffer.IdentityRef::key)
@@ -372,7 +385,11 @@ public class WatchProgressService {
                     ref.userId(), ref.deviceId(), ref.dramaId()));
         }
         if (pendingRefs.isEmpty()) {
-            throw new IllegalStateException("缓冲状态读取为空，ref=" + refs.size());
+            // 缓冲状态已消失（用户清空/删除历史，或 HASH 12h TTL 到期）。此时 ref 已无
+            // 可落库的内容，必须丢弃：退回队列会在下个周期被原样弹出、再清空、再退回，
+            // 这批僵尸条目将永久占住 dirty 队列配额。
+            log.debug("观看进度缓冲条目已无对应状态，丢弃 {} 条", refs.size());
+            return 0;
         }
 
         // 3) 分块成对查询已有行（每块 100 组 OR 索引条件）
@@ -415,6 +432,7 @@ public class WatchProgressService {
             }
             session.commit();
         }
+        return pendingRefs.size();
     }
 
     /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */