chore(merge): 合并 main 的印尼语敏感词库种子到 develop

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

.gitignore

@@ -20,3 +20,6 @@ data/uploads/*
 # 部署用的真实环境变量（含密钥），只提交 .env.example
 .env
 dump.rdb
+
+# 看门狗飞书告警 Webhook，只存在于部署机
+deploy/feishu-webhook.txt

deploy/watchdog.ps1

@@ -14,6 +14,9 @@ param(
 $ErrorActionPreference = "Continue"
 $ComposeFile = Join-Path $ComposeDir "docker-compose.yml"
 $LogFile = Join-Path $LogDir "watchdog-$(Get-Date -Format 'yyyy-MM-dd').log"
+# 连续失败计数落盘文件：计划任务每 2 分钟起一个新进程，计数只放进程内永远累积不到
+# $MaxRestarts，告警会静默哑火。
+$StateFile = Join-Path $LogDir "watchdog-state.json"
 
 function Write-Log($msg) {
     $line = "$(Get-Date -Format 'yyyy-MM-dd HH:mm:ss') $msg"
@@ -33,6 +36,29 @@ function Send-Alert($title, $msg) {
     }
 }
 
+# 连续失败计数的读写：JSON 落盘，跨进程保留
+function Import-FailCounts {
+    $counts = @{}
+    if (-not (Test-Path $StateFile)) { return $counts }
+    try {
+        $saved = Get-Content -Path $StateFile -Raw | ConvertFrom-Json
+        foreach ($prop in $saved.PSObject.Properties) {
+            $counts[$prop.Name] = [int]$prop.Value
+        }
+    } catch {
+        Write-Log "看门狗计数文件解析失败，各服务按 0 重新计数: $_"
+    }
+    return $counts
+}
+
+function Export-FailCounts($counts) {
+    try {
+        $counts | ConvertTo-Json | Set-Content -Path $StateFile -Encoding UTF8
+    } catch {
+        Write-Log "看门狗计数文件写入失败: $_"
+    }
+}
+
 function Test-Docker {
     docker info --format "{{.ServerVersion}}" 2>$null | Out-Null
     return ($LASTEXITCODE -eq 0)
@@ -60,6 +86,20 @@ function Invoke-Compose([string[]]$composeArgs) {
 # 确保日志目录存在
 New-Item -ItemType Directory -Force -Path $LogDir | Out-Null
 
+# 告警 Webhook：命令行参数 > 环境变量 > deploy 目录下的 feishu-webhook.txt。
+# 计划任务注册命令里不写密钥（密钥不该进版本库），由部署机本地放文件或配环境变量。
+if (-not $FeishuWebhook) { $FeishuWebhook = $env:FRUITSHORTS_FEISHU_WEBHOOK }
+if (-not $FeishuWebhook) {
+    $webhookFile = Join-Path $PSScriptRoot "feishu-webhook.txt"
+    if (Test-Path $webhookFile) {
+        $FeishuWebhook = (Get-Content -Path $webhookFile -Raw -ErrorAction SilentlyContinue)
+        if ($FeishuWebhook) { $FeishuWebhook = $FeishuWebhook.Trim() }
+    }
+}
+if (-not $FeishuWebhook) {
+    Write-Log "未配置飞书 Webhook（参数/环境变量/feishu-webhook.txt 均为空），告警只写本地日志"
+}
+
 # Docker Desktop 登录后偶尔尚未就绪，先等待并尝试拉起。
 if (-not (Test-Docker)) {
     Write-Log "Docker 未就绪，尝试启动 Docker Desktop"
@@ -91,18 +131,8 @@ $checks = @(
     @{ Name = "Prometheus"; Type = "http";     Url = "http://127.0.0.1:9090/-/healthy"; Service = "prometheus"; Container = "fruitshorts-prometheus" }
 )
 
-# 跟踪每个服务的连续失败次数。脚本由计划任务每 2 分钟拉起一次，
-# 进程内变量会跨轮丢失，因此计数持久化到状态文件，否则「连续 3 次告警」永远不触发。
-$StateFile = Join-Path $LogDir "watchdog-state.json"
-$failCounts = @{}
-if (Test-Path $StateFile) {
-    try {
-        (Get-Content $StateFile -Raw | ConvertFrom-Json).PSObject.Properties |
-            ForEach-Object { $failCounts[$_.Name] = [int]$_.Value }
-    } catch {
-        Write-Log "状态文件损坏，重置连续失败计数"
-    }
-}
+# 跟踪每个服务的连续失败次数（跨进程从状态文件恢复）
+$failCounts = Import-FailCounts
 
 foreach ($check in $checks) {
     $name = $check.Name
@@ -159,8 +189,10 @@ foreach ($check in $checks) {
             }
         }
 
-        # 连续失败达到上限 → 告警
-        if (-not $healthy -and $count -ge $MaxRestarts) {
+        # 连续失败达到上限 → 告警。计数会跨进程累积，所以只在首次达阈值和
+        # 之后每 30 轮（约 1 小时）提醒一次，避免每 2 分钟刷屏。
+        if (-not $healthy -and $count -ge $MaxRestarts -and
+                ($count -eq $MaxRestarts -or $count % 30 -eq 0)) {
             Send-Alert "服务异常" "$name 已连续 $count 次不健康，已尝试重启。请人工介入检查。"
             Write-Log "$name 连续 $count 次失败，已发告警"
         } elseif ($healthy) {
@@ -179,9 +211,5 @@ if ($drive) {
     }
 }
 
-# 落盘本轮结束时的连续失败计数，供下一轮接着累计
-try {
-    $failCounts | ConvertTo-Json | Set-Content -Path $StateFile -Encoding UTF8
-} catch {
-    Write-Log "状态文件写入失败: $_"
-}
+# 本轮结果落盘，供下一轮（2 分钟后的新进程）继续累加
+Export-FailCounts $failCounts

fruitshorts-app-api/src/main/resources/application.yml

@@ -20,6 +20,13 @@ spring:
   jackson:
     date-format: yyyy-MM-dd HH:mm:ss
     time-zone: Asia/Shanghai
+  task:
+    scheduling:
+      # 默认调度池只有 1 条线程：观看进度落库（每 2s）会和其他定时任务互相排队，
+      # 落库一旦拖慢就会连带推迟对时延敏感的任务，必须给池子多几条线程。
+      pool:
+        size: 4
+      thread-name-prefix: fruitshorts-sched-
   data:
     redis:
       host: ${REDIS_HOST:127.0.0.1}

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;
@@ -32,10 +35,13 @@ import org.springframework.transaction.support.TransactionTemplate;
 import java.time.LocalDateTime;
 import java.lang.reflect.Method;
 import java.util.List;
+import java.util.Map;
 
 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;
@@ -330,6 +336,123 @@ class WatchProgressServiceTest {
         verify(progressMapper).deleteById(1L);
     }
 
+    @Test
+    void should_delete_buffer_state_when_removing_or_clearing_history() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        WatchProgressService service = serviceWithBuffer(buffer);
+
+        service.remove(new InteractionIdentity(9L, ""), 5L);
+        service.clear(new InteractionIdentity(9L, ""));
+
+        verify(buffer).remove(any(InteractionIdentity.class), eq(5L));
+        verify(buffer).clear(any(InteractionIdentity.class));
+        verify(progressMapper, times(2)).delete(any());
+    }
+
+    @Test
+    void should_not_bring_back_deleted_drama_from_buffer_in_recent_page() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        WatchProgressService service = serviceWithBuffer(buffer);
+        Page<WatchProgress> emptyPage = new Page<>(1, 20, 0);
+        emptyPage.setRecords(List.of());
+        given(progressMapper.selectPage(any(Page.class), any())).willReturn(emptyPage);
+        given(buffer.getAll(any(InteractionIdentity.class))).willReturn(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.setDeleted(1);
+        given(progressMapper.selectAnyForIdentityIn(9L, "", List.of(5L)))
+                .willReturn(List.of(hidden));
+
+        PageResult<ProgressRespVO> result =
+                service.recentPage(new InteractionIdentity(9L, ""), 1, 20);
+
+        assertThat(result.getList()).isEmpty();
+        assertThat(result.getTotal()).isZero();
+    }
+
+    @Test
+    void should_drop_buffered_refs_whose_state_is_gone_instead_of_requeue() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        WatchProgressService service = serviceWithBuffer(buffer);
+        given(buffer.drain(10)).willReturn(
+                List.of(new WatchProgressBuffer.IdentityRef("u:9", 9L, "", 5L)));
+        given(buffer.getAllBatch(List.of("u:9"))).willReturn(Map.of());
+
+        int flushed = service.flushBuffered(10);
+
+        // 状态已消失（用户删除/清空历史或 TTL 过期）：必须丢弃，退回队列会变成永不消失的僵尸条目
+        assertThat(flushed).isZero();
+        verify(buffer, never()).requeue(any());
+        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);
+        return new WatchProgressService(
+                progressMapper,
+                historyMapper,
+                dramaService,
+                episodeService,
+                provider,
+                sessionFactory,
+                new TransactionTemplate(mock(PlatformTransactionManager.class)));
+    }
+
     private Drama drama(Long id, String title, int totalEpisodes) {
         Drama drama = new Drama();
         drama.setId(id);

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());
@@ -93,6 +116,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

@@ -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);
     }
@@ -279,14 +278,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 +309,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 +326,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 +351,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 +359,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 +384,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 +431,7 @@ public class WatchProgressService {
             }
             session.commit();
         }
+        return pendingRefs.size();
     }
 
     /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */
@@ -442,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) {
@@ -643,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;
+        }
+    }
+
     /**
      * 登录时合并设备进度和观看流水。
      *
@@ -654,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)
@@ -713,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);
     }

sql/data-sensitive-words-id.sql

@@ -0,0 +1,113 @@
+-- Author: fruit.com.cn
+
+-- FruitShorts 短剧服务 · 印尼语（id-ID）敏感词库增量补全（PostgreSQL 18）
+--
+-- 续接基线：sensitive_word 已有 id 9000001~9000053（zh-CN/en-US/id-ID 三语初版）。
+-- 本文件只补 id-ID 缺失/不足的分类，id 从 9000054 续接，避免与现有雪花 ID 冲突。
+--
+-- 分类（category）：POLITICS 政治 / PORN 色情 / ABUSE 辱骂 / AD 广告导流 / RELIGION 宗教
+-- 级别（level）：
+--   BLOCK   直接拒绝发布
+--   REVIEW  命中转人工审核（歧义词、可能误伤的业务词放这里）
+--   REPLACE 替换后放行（轻度脏话，用 replace_text 打码）
+--
+-- 幂等：唯一索引 uk_sensitive_word(language, word)，重复执行靠 ON CONFLICT DO NOTHING 跳过。
+-- 说明：词库内容需印尼运营/合规方复核后启用，本文件仅提供开发联调可用的初始基线。
+
+-- ============ 辱骂 / 脏话（ABUSE）· 补全 ============
+INSERT INTO sensitive_word (id, word, category, level, replace_text, language, status) VALUES
+  (9000054, 'babi',      'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000055, 'brengsek',  'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000056, 'kampret',   'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000057, 'keparat',   'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000058, 'sundal',    'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000059, 'lonte',     'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000060, 'pelacur',   'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000061, 'jancok',    'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000062, 'asu',       'ABUSE', 'BLOCK',   '*',   'id-ID', 1),
+  (9000063, 'sialan',    'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000064, 'tolol',     'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000065, 'goblok',    'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000066, 'bodoh',     'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000067, 'bego',      'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000068, 'kirik',     'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000069, 'tai',       'ABUSE', 'REPLACE', '***', 'id-ID', 1),
+  (9000070, 'bejat',     'ABUSE', 'REPLACE', '***', 'id-ID', 1)
+ON CONFLICT (language, word) DO NOTHING;
+
+-- ============ 色情（PORN）· 少量补全 ============
+INSERT INTO sensitive_word (id, word, category, level, replace_text, language, status) VALUES
+  (9000071, 'ngewe',        'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000072, 'toket',        'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000073, 'onani',        'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000074, 'perkosa',      'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000075, 'pemerkosaan',  'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000076, 'jembut',       'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000077, 'pepek',        'PORN', 'BLOCK',  '*', 'id-ID', 1),
+  (9000078, 'selingkuh',    'PORN', 'REVIEW', '*', 'id-ID', 1),
+  (9000079, 'gairah',       'PORN', 'REVIEW', '*', 'id-ID', 1)
+ON CONFLICT (language, word) DO NOTHING;
+
+-- ============ 政治（POLITICS）· 新增（基线缺失） ============
+INSERT INTO sensitive_word (id, word, category, level, replace_text, language, status) VALUES
+  (9000080, 'PKI',              'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000081, 'komunis',          'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000082, 'Papua Merdeka',    'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000083, 'OPM',              'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000084, 'GAM',              'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000085, 'separatis',        'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000086, 'separatisme',      'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000087, 'teroris',          'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000088, 'terorisme',        'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000089, 'ISIS',             'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000090, 'jihadis',          'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000091, 'radikalisme',      'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000092, 'kudeta',           'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000093, 'makar',            'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000094, 'pemberontak',      'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000095, 'pemberontakan',    'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000096, 'kerusuhan',        'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000097, 'penjarahan',       'POLITICS', 'BLOCK',  '*', 'id-ID', 1),
+  (9000098, 'radikal',          'POLITICS', 'REVIEW', '*', 'id-ID', 1),
+  (9000099, 'jihad',            'POLITICS', 'REVIEW', '*', 'id-ID', 1),
+  (9000100, 'anti-pemerintah',  'POLITICS', 'REVIEW', '*', 'id-ID', 1),
+  (9000101, 'bubarkan',         'POLITICS', 'REVIEW', '*', 'id-ID', 1)
+ON CONFLICT (language, word) DO NOTHING;
+
+-- ============ 宗教（RELIGION）· 补全 ============
+INSERT INTO sensitive_word (id, word, category, level, replace_text, language, status) VALUES
+  (9000102, 'penistaan agama', 'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000103, 'menghina agama',  'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000104, 'menghina Nabi',   'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000105, 'menghina Allah',  'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000106, 'menghina Kristen','RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000107, 'menghina Katolik','RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000108, 'menghina Hindu',  'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000109, 'menghina Buddha', 'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000110, 'kafir',           'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000111, 'murtad',          'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000112, 'sara',            'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000113, 'provokasi agama', 'RELIGION', 'BLOCK',  '*', 'id-ID', 1),
+  (9000114, 'konflik agama',   'RELIGION', 'REVIEW', '*', 'id-ID', 1)
+ON CONFLICT (language, word) DO NOTHING;
+
+-- ============ 广告导流 / 非法（AD）· 补全 ============
+INSERT INTO sensitive_word (id, word, category, level, replace_text, language, status) VALUES
+  (9000115, 'judi online',      'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000116, 'slot online',      'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000117, 'casino',           'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000118, 'poker online',     'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000119, 'situs judi',       'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000120, 'deposit slot',     'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000121, 'link alternatif',  'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000122, 'pinjol',           'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000123, 'rentenir',         'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000124, 'scam',             'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000125, 'penipuan',         'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000126, 'investasi bodong', 'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000127, 'arisan berantai',  'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000128, 'money game',       'AD', 'BLOCK',  '*', 'id-ID', 1),
+  (9000129, 'pinjaman online',  'AD', 'REVIEW', '*', 'id-ID', 1),
+  (9000130, 'hack',             'AD', 'REVIEW', '*', 'id-ID', 1),
+  (9000131, 'jual akun',        'AD', 'REVIEW', '*', 'id-ID', 1)
+ON CONFLICT (language, word) DO NOTHING;