chore(merge): 合并 origin/develop 的看门狗周期巡检与 admin-api Redis 配置

watchdog.ps1 冲突取本地实现：计数落盘用 Import/Export 函数，另含 Webhook
三级兜底与告警节流，为远程内联版本的超集。

.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

@@ -32,9 +32,11 @@ 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.mock;
 import static org.mockito.Mockito.never;
@@ -330,6 +332,72 @@ 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());
+    }
+
+    private WatchProgressService serviceWithBuffer(WatchProgressBuffer buffer) {
+        @SuppressWarnings("unchecked")
+        ObjectProvider<WatchProgressBuffer> provider = mock(ObjectProvider.class);
+        given(provider.getIfAvailable()).willReturn(buffer);
+        return new WatchProgressService(
+                progressMapper,
+                historyMapper,
+                dramaService,
+                episodeService,
+                provider,
+                mock(SqlSessionFactory.class),
+                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

@@ -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();
     }
 
     /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */