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

Dockerfile

@@ -15,6 +15,20 @@
 FROM maven:3.9-eclipse-temurin-21 AS build
 WORKDIR /build
 
+# Maven Central 在国内网络下拉依赖不稳定（构建时曾出现 maven-filtering:3.3.1 解析失败），
+# 构建阶段固定走阿里云 central 镜像；其它仓库仍直连。
+RUN mkdir -p /root/.m2 \
+    && printf '%s\n' \
+       '<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0">' \
+       '  <mirrors>' \
+       '    <mirror>' \
+       '      <id>aliyun-central</id>' \
+       '      <mirrorOf>central</mirrorOf>' \
+       '      <url>https://maven.aliyun.com/repository/central</url>' \
+       '    </mirror>' \
+       '  </mirrors>' \
+       '</settings>' > /root/.m2/settings.xml
+
 # 先只拷各模块 pom 再拉依赖：依赖层可复用，改业务代码不会重新下载依赖
 COPY pom.xml .
 COPY fruitshorts-common/pom.xml fruitshorts-common/
@@ -30,7 +44,7 @@ COPY fruitshorts-module-creator/pom.xml fruitshorts-module-creator/
 COPY fruitshorts-admin-api/pom.xml fruitshorts-admin-api/
 COPY fruitshorts-app-api/pom.xml fruitshorts-app-api/
 COPY fruitshorts-transcode-worker/pom.xml fruitshorts-transcode-worker/
-RUN mvn -B -q -DskipTests dependency:go-offline || true
+RUN mvn -B -q -DskipTests dependency:go-offline
 
 COPY . .
 ARG MODULE=fruitshorts-app-api

deploy/backend-watchdog-task.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-16"?>
+<Task version="1.3" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
+  <RegistrationInfo>
+    <Description>FruitShorts backend watchdog: health check every 2 minutes and restart unhealthy services.</Description>
+  </RegistrationInfo>
+  <Principals>
+    <Principal id="Author">
+      <UserId>S-1-5-18</UserId>
+      <RunLevel>HighestAvailable</RunLevel>
+    </Principal>
+  </Principals>
+  <Settings>
+    <MultipleInstancesPolicy>IgnoreNew</MultipleInstancesPolicy>
+    <DisallowStartIfOnBatteries>false</DisallowStartIfOnBatteries>
+    <StopIfGoingOnBatteries>false</StopIfGoingOnBatteries>
+    <StartWhenAvailable>true</StartWhenAvailable>
+    <ExecutionTimeLimit>PT10M</ExecutionTimeLimit>
+  </Settings>
+  <Triggers>
+    <!-- BootTrigger 上挂 <Repetition> 在 Windows 计划任务里不生效（导入后 Next Run Time 为 N/A，
+         实际只在开机跑一次）。改成 TimeTrigger + 无限重复，配合 StartWhenAvailable
+         保证开机后若错过也会立即补跑；这样才真正做到每 2 分钟巡检一次。 -->
+    <TimeTrigger>
+      <StartBoundary>2026-01-01T00:00:00</StartBoundary>
+      <Repetition>
+        <Interval>PT2M</Interval>
+        <Duration>P3650D</Duration>
+        <StopAtDurationEnd>false</StopAtDurationEnd>
+      </Repetition>
+    </TimeTrigger>
+  </Triggers>
+  <Actions Context="Author">
+    <Exec>
+      <Command>powershell.exe</Command>
+      <Arguments>-NoProfile -ExecutionPolicy Bypass -File "C:\Users\Administrator\Desktop\ReelShorts\FruitServer\deploy\watchdog.ps1" -ComposeDir "C:\Users\Administrator\Desktop\ReelShorts\FruitServer" -LogDir "C:\Users\Administrator\Desktop\ReelShorts\FruitServer\logs"</Arguments>
+    </Exec>
+  </Actions>
+</Task>

deploy/watchdog.ps1

@@ -0,0 +1,215 @@
+# Author: fruit.com.cn
+#
+# 服务看门狗：每 2 分钟检查一次，挂了自动拉起，连续失败发飞书告警。
+# 注册为 Windows 计划任务后无人值守运行。
+
+param(
+    [string]$ComposeDir = "C:\FruitServer",
+    [string]$LogDir = "C:\FruitServer\logs",
+    [string]$FeishuWebhook = "",
+    [int]$MaxRestarts = 3,
+    [string]$FrontendTask = "FruitShorts-Web-3005"
+)
+
+$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"
+    Add-Content -Path $LogFile -Value $line -ErrorAction SilentlyContinue
+}
+
+function Send-Alert($title, $msg) {
+    if (-not $FeishuWebhook) { return }
+    $body = @{
+        msg_type = "text"
+        content  = @{ text = "[$title] $msg" }
+    } | ConvertTo-Json -Depth 3
+    try {
+        Invoke-RestMethod -Uri $FeishuWebhook -Method Post -Body $body -ContentType "application/json" -TimeoutSec 10
+    } catch {
+        Write-Log "飞书告警发送失败: $_"
+    }
+}
+
+# 连续失败计数的读写：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)
+}
+
+function Test-Http([string]$url) {
+    try {
+        $response = Invoke-WebRequest -Uri $url -TimeoutSec 10 -UseBasicParsing -ErrorAction Stop
+        return ($response.StatusCode -ge 200 -and $response.StatusCode -lt 400)
+    } catch {
+        return $false
+    }
+}
+
+function Test-Container([string]$container) {
+    $status = docker inspect -f "{{.State.Status}}" $container 2>$null
+    return ($status -eq "running")
+}
+
+function Invoke-Compose([string[]]$composeArgs) {
+    & docker compose -f $ComposeFile @composeArgs 2>&1 | Out-Null
+    return $LASTEXITCODE
+}
+
+# 确保日志目录存在
+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"
+    $dockerExe = "C:\Program Files\Docker\Docker\Docker Desktop.exe"
+    if (Test-Path $dockerExe) {
+        Start-Process -FilePath $dockerExe
+    }
+
+    for ($i = 0; $i -lt 18; $i++) {
+        Start-Sleep -Seconds 5
+        if (Test-Docker) { break }
+    }
+}
+
+if (-not (Test-Docker)) {
+    Write-Log "Docker 在 90 秒内未就绪，跳过本轮检查"
+    Send-Alert "服务异常" "Docker Desktop 无法启动，请人工检查桌面机。"
+    exit 1
+}
+
+# 健康检查定义：名称 → 检查方式
+$checks = @(
+    @{ Name = "app-api";    Type = "http";     Url = "http://127.0.0.1:8081/healthz"; Service = "app-api"; Container = "fruitshorts-app-api" },
+    @{ Name = "admin-api";  Type = "http";     Url = "http://127.0.0.1:8080/healthz"; Service = "admin-api"; Container = "fruitshorts-admin-api" },
+    @{ Name = "frontend";   Type = "frontend"; Url = "http://127.0.0.1:3005/dist-dev/" },
+    @{ Name = "Grafana";    Type = "http";     Url = "http://127.0.0.1:3000/api/health"; Service = "grafana"; Container = "fruitshorts-grafana" },
+    @{ Name = "PostgreSQL"; Type = "docker";   Service = "db"; Container = "fruitshorts-db" },
+    @{ Name = "Redis";      Type = "docker";   Service = "redis"; Container = "fruitshorts-redis" },
+    @{ Name = "Prometheus"; Type = "http";     Url = "http://127.0.0.1:9090/-/healthy"; Service = "prometheus"; Container = "fruitshorts-prometheus" }
+)
+
+# 跟踪每个服务的连续失败次数（跨进程从状态文件恢复）
+$failCounts = Import-FailCounts
+
+foreach ($check in $checks) {
+    $name = $check.Name
+    $healthy = $false
+
+    if ($check.Type -eq "http") {
+        $healthy = Test-Http $check.Url
+    } elseif ($check.Type -eq "frontend") {
+        $healthy = Test-Http $check.Url
+    } else {
+        $healthy = Test-Container $check.Container
+    }
+
+    if ($healthy) {
+        if ($failCounts[$name] -gt 0) {
+            Write-Log "$name 恢复正常（此前连续失败 $($failCounts[$name]) 次）"
+            Send-Alert "服务恢复" "$name 已恢复正常"
+        }
+        $failCounts[$name] = 0
+    } else {
+        if ($null -eq $failCounts[$name]) {
+            $failCounts[$name] = 0
+        }
+        $failCounts[$name] = $failCounts[$name] + 1
+        $count = $failCounts[$name]
+        Write-Log "$name 不健康（第 $count 次）"
+
+        if ($check.Type -eq "frontend") {
+            Write-Log "$name 重启计划任务 $FrontendTask"
+            Stop-ScheduledTask -TaskName $FrontendTask -ErrorAction SilentlyContinue
+            Start-Sleep -Seconds 2
+            Start-ScheduledTask -TaskName $FrontendTask -ErrorAction SilentlyContinue
+            Start-Sleep -Seconds 6
+            $healthy = Test-Http $check.Url
+        } elseif ($check.Type -eq "http") {
+            Write-Log "$name 重启 Compose 服务 $($check.Service)"
+            Invoke-Compose @("restart", $check.Service) | Out-Null
+            Start-Sleep -Seconds 12
+            $healthy = Test-Http $check.Url
+
+            if (-not $healthy) {
+                Write-Log "$name 重启后仍不健康，执行 compose up -d"
+                Invoke-Compose @("up", "-d", $check.Service) | Out-Null
+            }
+        } else {
+            Write-Log "$name 拉起容器 $($check.Container)"
+            docker start $check.Container 2>&1 | Out-Null
+            Start-Sleep -Seconds 5
+            $healthy = Test-Container $check.Container
+
+            if (-not $healthy) {
+                Write-Log "$name 启动失败，执行 compose up -d"
+                Invoke-Compose @("up", "-d", $check.Service) | Out-Null
+            }
+        }
+
+        # 连续失败达到上限 → 告警。计数会跨进程累积，所以只在首次达阈值和
+        # 之后每 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) {
+            Write-Log "$name 已自动恢复"
+        }
+    }
+}
+
+# 磁盘检查：超过 85% 告警
+$drive = (Get-PSDrive C -ErrorAction SilentlyContinue)
+if ($drive) {
+    $pct = [math]::Round(($drive.Used / ($drive.Used + $drive.Free)) * 100)
+    if ($pct -gt 85) {
+        Write-Log "磁盘使用率 ${pct}%（超过 85%）"
+        Send-Alert "磁盘告警" "C 盘使用率 ${pct}%，请清理日志或扩容"
+    }
+}
+
+# 本轮结果落盘，供下一轮（2 分钟后的新进程）继续累加
+Export-FailCounts $failCounts

docker-compose.yml

@@ -71,8 +71,10 @@ services:
     env_file: .env
     environment:
       SERVER_PORT: "8081"
-      # 默认连编排里的 db；要连外部库就在 .env 里覆盖 DB_URL
+      # 默认连编排里的 db；要连外部库就在 .env 里覆盖 DB_URL。
-      DB_URL: ${DB_URL:-jdbc:postgresql://db:5432/${DB_NAME:-fruitshorts}}
+      # reWriteBatchedInserts=true：pgjdbc 把 JDBC 批内多条 INSERT 重写成多值
+      # INSERT，观看进度缓冲落库（watch:pg:dirty 消费任务）依赖它跑满吞吐。
+      DB_URL: ${DB_URL:-jdbc:postgresql://db:5432/${DB_NAME:-fruitshorts}?reWriteBatchedInserts=true}
       # 限流/缓存用 Redis；容器内 127.0.0.1 是容器自身，必须指向编排里的 redis
       REDIS_HOST: ${REDIS_HOST:-redis}
       REDIS_PORT: ${REDIS_PORT:-6379}
@@ -197,9 +199,40 @@ services:
       GF_SECURITY_ADMIN_USER: ${GRAFANA_ADMIN_USER:-admin}
       GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_ADMIN_PASSWORD:?Grafana 管理员密码必填，见 .env.example}
       GF_USERS_ALLOW_SIGN_UP: "false"
+      # 新用户默认界面语言：简体中文
+      GF_USERS_LANGUAGE: "zh-Hans"
       TZ: Asia/Shanghai
     depends_on:
       - prometheus
+      - loki
+
+  # ─────────── 日志（Loki + Promtail）───────────
+  # Loki：日志存储与查询引擎；Promtail：采集容器 stdout/stderr 推给 Loki。
+  # Grafana 里通过 Loki 数据源可直接搜索/过滤各容器的请求日志。
+  loki:
+    image: grafana/loki:latest
+    container_name: fruitshorts-loki
+    restart: unless-stopped
+    command: -config.file=/etc/loki/local-config.yaml
+    volumes:
+      - ./loki-config.yml:/etc/loki/local-config.yaml:ro
+      - loki-data:/loki
+    expose:
+      - "3100"
+    # 镜像为 distroless（无 shell/wget），无法在容器内做 HTTP 探活；
+    # 靠 restart: unless-stopped 自动拉起，Grafana 数据源状态可反映可用性。
+
+  promtail:
+    image: grafana/promtail:latest
+    container_name: fruitshorts-promtail
+    restart: unless-stopped
+    command: -config.file=/etc/promtail/config.yml
+    volumes:
+      - ./promtail-config.yml:/etc/promtail/config.yml:ro
+      # 读取容器日志需要 Docker socket
+      - /var/run/docker.sock:/var/run/docker.sock:ro
+    depends_on:
+      - loki
 
   # 黑盒探活：从外部探测接口可用性与耗时
   blackbox:
@@ -227,3 +260,4 @@ volumes:
   redis-data:
   prometheus-data:
   grafana-data:
+  loki-data:

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

@@ -14,6 +14,12 @@ spring:
     url: ${DB_URL:jdbc:postgresql://127.0.0.1:5432/fruitshorts}
     username: ${DB_USER:fruitshorts}
     password: ${DB_PASSWORD:fruitshorts}
+  data:
+    redis:
+      # 与 app-api 共用同一个 Redis 实例；容器内必须显式指定 host，
+      # 否则 Spring Boot 默认连 127.0.0.1（容器内即自身）导致健康检查报 DOWN。
+      host: ${REDIS_HOST:127.0.0.1}
+      port: ${REDIS_PORT:6379}
   jackson:
     date-format: yyyy-MM-dd HH:mm:ss
     time-zone: Asia/Shanghai

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

@@ -23,6 +23,13 @@ fruitshorts:
   cors:
     # 生产只允许显式配置的来源；没配就等于空，浏览器侧会被拦住（比"*"安全）
     allowed-origins: ${CORS_ALLOWED_ORIGINS:}
+  watch-progress:
+    # 观看进度写缓冲：POST 先写 Redis 立即返回，后台任务批量合并进 watch_progress。
+    # 消化能力 = batch-size / interval ≈ 5000 条/秒，覆盖 30 万 DAU 峰值 ~3700 条/秒上报。
+    # 关掉（false）即回退同步写库，接口契约不变。
+    buffer: true
+    flush-interval-ms: 2000
+    flush-batch-size: 10000
 
 springdoc:
   # 生产只留机器可读契约，关掉 UI（doc.html 不对外开放）

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

@@ -11,13 +11,22 @@ spring:
   datasource:
     driver-class-name: org.postgresql.Driver
     # 地址必须外置：不同环境（本机 / 测试 / 生产）连的库不一样，
-    # 生产请用 DB_URL 注入（prod profile 会校验它非空）
+    # 生产请用 DB_URL 注入（prod profile 会校验它非空）。
-    url: ${DB_URL:jdbc:postgresql://127.0.0.1:5432/fruitshorts}
+    # reWriteBatchedInserts：pgjdbc 把批内多条 INSERT 重写成多值 INSERT，
+    # 观看进度缓冲落库依赖它才能跑满批量吞吐。
+    url: ${DB_URL:jdbc:postgresql://127.0.0.1:5432/fruitshorts?reWriteBatchedInserts=true}
     username: ${DB_USER:fruitshorts}
     password: ${DB_PASSWORD:fruitshorts}
   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,18 +20,28 @@ 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;
 
 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;
@@ -58,11 +68,18 @@ class WatchProgressServiceTest {
         historyMapper = mock(WatchHistoryMapper.class);
         dramaService = mock(DramaService.class);
         episodeService = mock(EpisodeService.class);
+        // 测试统一走缓冲关闭（同步写库）路径：getIfAvailable 返回 null
+        @SuppressWarnings("unchecked")
+        ObjectProvider<WatchProgressBuffer> bufferProvider = mock(ObjectProvider.class);
+        given(bufferProvider.getIfAvailable()).willReturn(null);
         progressService = new WatchProgressService(
                 progressMapper,
                 historyMapper,
                 dramaService,
-                episodeService);
+                episodeService,
+                bufferProvider,
+                mock(SqlSessionFactory.class),
+                new TransactionTemplate(mock(PlatformTransactionManager.class)));
     }
 
     @Test
@@ -319,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/pom.xml

@@ -33,5 +33,10 @@
             <groupId>com.fruitshorts</groupId>
             <artifactId>fruitshorts-module-user</artifactId>
         </dependency>
+        <!-- 进度上报路径的剧集/短剧进程内缓存（版本由 Spring Boot BOM 管理） -->
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+        </dependency>
     </dependencies>
 </project>

fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/dal/mapper/WatchProgressMapper.java

@@ -8,6 +8,8 @@ import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Select;
 import org.apache.ibatis.annotations.Update;
 
+import java.util.List;
+
 public interface WatchProgressMapper extends BaseMapper<WatchProgress> {
 
     /** 按登录用户或匿名设备读取同一部剧的记录，包含逻辑删除行。 */
@@ -40,4 +42,64 @@ public interface WatchProgressMapper extends BaseMapper<WatchProgress> {
             WHERE id = #{id}
             """)
     int updateAny(WatchProgress progress);
+
+    /** 固定形状的全字段按 id 更新（绕过逻辑删除过滤）。SQL 不随字段空值变化，
+     * 缓冲落库时连续调用可被 MyBatis BATCH 执行器合成一个 JDBC batch。 */
+    @Update("""
+            UPDATE watch_progress SET
+              episode_id = #{et.episodeId},
+              position_ms = #{et.positionMs},
+              duration_ms = #{et.durationMs},
+              finished = #{et.finished},
+              play_count = #{et.playCount},
+              device_id = #{et.deviceId},
+              source = #{et.source},
+              deleted = #{et.deleted}
+            WHERE id = #{et.id}
+            """)
+    int updateRowAny(@Param("et") WatchProgress et);
+
+    /** 批量读取同一身份下多部剧的记录（含逻辑删除行），供缓冲落库合并使用。 */
+    @Select("""
+            <script>
+            SELECT * FROM watch_progress
+            WHERE
+              <choose>
+                <when test="userId != null">user_id = #{userId,jdbcType=BIGINT}</when>
+                <otherwise>user_id IS NULL AND device_id = #{deviceId}</otherwise>
+              </choose>
+              AND drama_id IN
+              <foreach collection="dramaIds" item="d" open="(" separator="," close=")">#{d}</foreach>
+            </script>
+            """)
+    List<WatchProgress> selectAnyForIdentityIn(
+            @Param("userId") Long userId,
+            @Param("deviceId") String deviceId,
+            @Param("dramaIds") List<Long> dramaIds);
+
+    /** 落库合并的查询参数：身份（登录用户或匿名设备）+ drama。 */
+    @lombok.Data
+    @lombok.NoArgsConstructor
+    @lombok.AllArgsConstructor
+    class IdentityDramaPair {
+        private Long userId;
+        private String deviceId;
+        private long dramaId;
+    }
+
+    /** 按多组「身份 + drama」批量读取记录（含逻辑删除行），一组一条索引条件，单次往返。 */
+    @Select("""
+            <script>
+            SELECT * FROM watch_progress WHERE
+            <foreach collection="pairs" item="p" separator=" OR ">
+              (
+                (#{p.userId,jdbcType=BIGINT} IS NOT NULL AND user_id = #{p.userId,jdbcType=BIGINT})
+                OR
+                (#{p.userId,jdbcType=BIGINT} IS NULL AND user_id IS NULL AND device_id = #{p.deviceId})
+              )
+              AND drama_id = #{p.dramaId}
+            </foreach>
+            </script>
+            """)
+    List<WatchProgress> selectBatchForPairs(@Param("pairs") List<IdentityDramaPair> pairs);
 }

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

@@ -0,0 +1,355 @@
+// Author: fruit.com.cn
+
+package com.fruitshorts.playback.service;
+
+import com.fruitshorts.common.interaction.InteractionIdentity;
+import lombok.RequiredArgsConstructor;
+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;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 观看进度 Redis 缓冲（write-behind）。
+ *
+ * <p>高并发上报的落地形态：POST 只写 Redis（Lua 脚本一次往返完成写状态与入队），由
+ * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后批量 UPSERT
+ * 进权威表 watch_progress。Redis 不可用时 put 返回 false，调用方回退同步写库。
+ *
+ * <p>Redis 结构：
+ * <ul>
+ *   <li>{@code watch:pg:buf:u:{userId}} / {@code watch:pg:buf:d:{deviceId}}：
+ *       HASH，field=dramaId，value=进度状态，作为读接口的「读己之写」覆盖层，TTL 12h；</li>
+ *   <li>{@code watch:pg:dirty}：LIST，value={identityKey}|{dramaId}，待落库队列，
+ *       定时任务 RPOP 消费；重复条目靠幂等合并消化。</li>
+ * </ul>
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty(name = "fruitshorts.watch-progress.buffer", havingValue = "true")
+public class WatchProgressBuffer {
+
+    private static final String BUF_KEY_PREFIX = "watch:pg:buf:";
+    private static final String DIRTY_KEY = "watch:pg:dirty";
+    /** 待落库队列长度上限，防止 Redis 在落库停滞时无限膨胀。 */
+    private static final long MAX_DIRTY = 200_000;
+    /** 每个身份最多缓冲 30 部剧（续看列表第一页用不完，同时限制 HGETALL 大小）。 */
+    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;
+
+    /**
+     * 缓冲中的一条进度状态。ts 为客户端上报时刻（服务端毫秒），用于判断
+     * 缓冲值是否比数据库行新，以及合并时丢弃乱序旧值。
+     */
+    public record BufferedProgress(
+            long dramaId,
+            long episodeId,
+            long positionMs,
+            long durationMs,
+            boolean finished,
+            String source,
+            String deviceId,
+            long ts) {
+    }
+
+    /** dirty 队列弹出的一条待落库引用。 */
+    public record IdentityRef(String key, Long userId, String deviceId, long dramaId) {
+
+        public InteractionIdentity toIdentity() {
+            return userId != null
+                    ? new InteractionIdentity(userId, "")
+                    : new InteractionIdentity(null, deviceId);
+        }
+    }
+
+    /** 上报进度入缓冲。成功返回 true；Redis 异常返回 false，调用方回退同步写库。 */
+    public boolean put(InteractionIdentity identity, long dramaId, long episodeId,
+                       long positionMs, long durationMs, boolean finished,
+                       String source, String deviceId) {
+        try {
+            String key = identityKey(identity);
+            String hashKey = BUF_KEY_PREFIX + key;
+            String field = Long.toString(dramaId);
+            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());
+            return false;
+        }
+    }
+
+    /**
+     * 删除某部剧的缓冲状态，用于用户删除单条观看记录。
+     *
+     * <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);
+    }
+
+    public BufferedProgress get(String identityKey, long dramaId) {
+        try {
+            Object value = redis.opsForHash().get(BUF_KEY_PREFIX + identityKey, Long.toString(dramaId));
+            return value == null ? null : decode(dramaId, value.toString());
+        } catch (Exception e) {
+            log.warn("观看进度读取 Redis 缓冲失败: {}", e.getMessage());
+            return null;
+        }
+    }
+
+    /** 读某身份全部缓冲状态（≤30 部剧），读接口覆盖层用。 */
+    public Map<Long, BufferedProgress> getAll(InteractionIdentity identity) {
+        return getAll(identityKey(identity));
+    }
+
+    public Map<Long, BufferedProgress> getAll(String identityKey) {
+        try {
+            return decodeAll(redis.opsForHash().entries(BUF_KEY_PREFIX + identityKey));
+        } catch (Exception e) {
+            log.warn("观看进度读取 Redis 缓冲失败: {}", e.getMessage());
+            return Map.of();
+        }
+    }
+
+    /** 批量读取多个身份的缓冲状态（pipeline 单次往返），落库任务用。 */
+    @SuppressWarnings("unchecked")
+    public Map<String, Map<Long, BufferedProgress>> getAllBatch(List<String> identityKeys) {
+        if (identityKeys.isEmpty()) {
+            return Map.of();
+        }
+        try {
+            List<Object> results = redis.executePipelined(new SessionCallback<>() {
+                @Override
+                public Object execute(org.springframework.data.redis.core.RedisOperations operations) {
+                    for (String key : identityKeys) {
+                        operations.opsForHash().entries(BUF_KEY_PREFIX + key);
+                    }
+                    return null;
+                }
+            });
+            Map<String, Map<Long, BufferedProgress>> result = new LinkedHashMap<>();
+            for (int i = 0; i < identityKeys.size(); i++) {
+                Object raw = results.get(i);
+                result.put(identityKeys.get(i),
+                        raw instanceof Map<?, ?> map ? decodeAll((Map<Object, Object>) map) : Map.of());
+            }
+            return result;
+        } catch (Exception e) {
+            log.warn("观看进度批量读取 Redis 缓冲失败: {}", e.getMessage());
+            throw new IllegalStateException("Redis 缓冲批量读取失败", e);
+        }
+    }
+
+    /** 落库任务从 dirty 队头弹出最多 max 条（按 identity+drama 去重，保留顺序）。 */
+    public List<IdentityRef> drain(int max) {
+        List<String> raw;
+        try {
+            raw = redis.opsForList().leftPop(DIRTY_KEY, Math.max(1, max));
+        } catch (Exception e) {
+            log.warn("观看进度 dirty 队列弹出失败: {}", e.getMessage());
+            return List.of();
+        }
+        if (raw == null || raw.isEmpty()) {
+            return List.of();
+        }
+        LinkedHashMap<String, IdentityRef> dedup = new LinkedHashMap<>();
+        for (String value : raw) {
+            int split = value.lastIndexOf('|');
+            if (split <= 0) {
+                continue;
+            }
+            String key = value.substring(0, split);
+            long dramaId;
+            try {
+                dramaId = Long.parseLong(value.substring(split + 1));
+            } catch (NumberFormatException ignore) {
+                continue;
+            }
+            dedup.putIfAbsent(key + "|" + dramaId, toRef(key, dramaId));
+        }
+        return new ArrayList<>(dedup.values());
+    }
+
+    /** 落库失败的条目退回队列，等待下个周期重试。 */
+    public void requeue(List<IdentityRef> refs) {
+        try {
+            List<String> raw = refs.stream()
+                    .map(r -> r.key() + "|" + r.dramaId())
+                    .toList();
+            redis.opsForList().rightPushAll(DIRTY_KEY, raw);
+            redis.opsForList().trim(DIRTY_KEY, 0, MAX_DIRTY - 1);
+        } catch (Exception e) {
+            log.error("观看进度缓冲条目退回队列失败，这批进度将丢失: {}", e.getMessage());
+        }
+    }
+
+    /** 身份的 Redis 键：登录用 u:{userId}，匿名用 d:{deviceId}。 */
+    public String identityKey(InteractionIdentity identity) {
+        return identity.authenticated()
+                ? "u:" + identity.userId()
+                : "d:" + cleanToken(identity.deviceId());
+    }
+
+    private IdentityRef toRef(String key, long dramaId) {
+        int split = key.indexOf(':');
+        if (split <= 0) {
+            return new IdentityRef(key, null, key, dramaId);
+        }
+        String scope = key.substring(0, split);
+        String id = key.substring(split + 1);
+        if ("u".equals(scope)) {
+            try {
+                return new IdentityRef(key, Long.valueOf(id), "", dramaId);
+            } catch (NumberFormatException ignore) {
+                // 落到匿名分支
+            }
+        }
+        return new IdentityRef(key, null, id, dramaId);
+    }
+
+    private String encode(long episodeId, long positionMs, long durationMs,
+                          boolean finished, String source, String deviceId) {
+        return episodeId + "|" + positionMs + "|" + durationMs + "|" + (finished ? 1 : 0)
+                + "|" + cleanToken(source) + "|" + cleanToken(deviceId) + "|" + System.currentTimeMillis();
+    }
+
+    private BufferedProgress decode(long dramaId, String value) {
+        try {
+            String[] parts = value.split("\\|", -1);
+            if (parts.length != 7) {
+                return null;
+            }
+            return new BufferedProgress(
+                    dramaId,
+                    Long.parseLong(parts[0]),
+                    Long.parseLong(parts[1]),
+                    Long.parseLong(parts[2]),
+                    "1".equals(parts[3]),
+                    parts[4],
+                    parts[5],
+                    Long.parseLong(parts[6]));
+        } catch (NumberFormatException e) {
+            log.debug("观看进度缓冲值解码失败，忽略: {}", value);
+            return null;
+        }
+    }
+
+    private Map<Long, BufferedProgress> decodeAll(Map<Object, Object> raw) {
+        if (raw == null || raw.isEmpty()) {
+            return Map.of();
+        }
+        Map<Long, BufferedProgress> result = new LinkedHashMap<>();
+        for (Map.Entry<Object, Object> e : raw.entrySet()) {
+            long dramaId;
+            try {
+                dramaId = Long.parseLong(e.getKey().toString());
+            } catch (NumberFormatException ignore) {
+                continue;
+            }
+            BufferedProgress state = decode(dramaId, e.getValue().toString());
+            if (state != null) {
+                result.put(dramaId, state);
+            }
+        }
+        return result;
+    }
+
+    /** 限制单身份缓冲的剧数：超出时淘汰最旧条目，保证 HGETALL 与读覆盖层开销有界。 */
+    private void trimIdentity(String hashKey) {
+        Long size = redis.opsForHash().size(hashKey);
+        if (size == null || size <= MAX_DRAMAS_PER_IDENTITY) {
+            return;
+        }
+        Map<Object, Object> all = redis.opsForHash().entries(hashKey);
+        List<Map.Entry<Object, Long>> byTs = new ArrayList<>(all.size());
+        for (Map.Entry<Object, Object> e : all.entrySet()) {
+            long ts = 0;
+            if (e.getValue() != null) {
+                String[] parts = e.getValue().toString().split("\\|", -1);
+                try {
+                    ts = Long.parseLong(parts[parts.length - 1]);
+                } catch (NumberFormatException ignore) {
+                    // ts 取不到当最旧，优先淘汰
+                }
+            }
+            byTs.add(Map.entry(e.getKey(), ts));
+        }
+        byTs.sort(Map.Entry.comparingByValue());
+        int toRemove = Math.max(0, byTs.size() - MAX_DRAMAS_PER_IDENTITY);
+        if (toRemove > 0) {
+            redis.opsForHash().delete(hashKey,
+                    byTs.subList(0, toRemove).stream().map(Map.Entry::getKey).toArray());
+        }
+    }
+
+    /** value 中的 source/deviceId 不允许出现分隔符 | 与换行，统一清洗；长度截到 128。 */
+    private String cleanToken(String token) {
+        if (token == null) {
+            return "";
+        }
+        String cleaned = token.replace("|", "").replace("\n", "").replace("\r", "").trim();
+        return cleaned.length() > 128 ? cleaned.substring(0, 128) : cleaned;
+    }
+}

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

@@ -0,0 +1,43 @@
+// Author: fruit.com.cn
+
+package com.fruitshorts.playback.service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 观看进度缓冲落库任务：定时把 Redis 缓冲里的进度按身份分组合并进 watch_progress。
+ *
+ * <p>只在开启缓冲（fruitshorts.watch-progress.buffer=true）的应用内生效，
+ * app-api 的 prod profile 开启，admin-api 与测试环境不受影响。
+ * fixedDelay 串行执行，同一时刻只有一个落库批次在跑。
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty(name = "fruitshorts.watch-progress.buffer", havingValue = "true")
+public class WatchProgressFlusher {
+
+    private final WatchProgressService watchProgressService;
+
+    @Value("${fruitshorts.watch-progress.flush-batch-size:10000}")
+    private int batchSize;
+
+    @Scheduled(
+            fixedDelayString = "${fruitshorts.watch-progress.flush-interval-ms:2000}",
+            initialDelayString = "${fruitshorts.watch-progress.flush-initial-delay-ms:15000}")
+    public void flush() {
+        try {
+            int flushed = watchProgressService.flushBuffered(batchSize);
+            if (flushed > 0) {
+                log.debug("观看进度缓冲落库完成，本批次 {} 条", flushed);
+            }
+        } catch (Exception e) {
+            log.warn("观看进度缓冲落库失败，等待下个周期重试: {}", e.getMessage());
+        }
+    }
+}

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

@@ -19,29 +19,47 @@ import com.fruitshorts.playback.dal.mapper.WatchHistoryMapper;
 import com.fruitshorts.playback.dal.mapper.WatchProgressMapper;
 import com.fruitshorts.playback.vo.ProgressRespVO;
 import com.fruitshorts.playback.vo.ProgressSaveReqVO;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.ibatis.session.ExecutorType;
+import org.apache.ibatis.session.SqlSession;
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.StringUtils;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * 观看进度。
  *
- * <p>每次上报都写库在高并发下会成为瓶颈（1 万并发 × 每 10 秒一次 ≈ 1000 写/秒），
+ * <p>生产形态（fruitshorts.watch-progress.buffer=true，prod 已开启）：
- * 所以生产形态是「先写 Redis，后台任务按登录账号或设备 + drama 合并后批量 UPSERT」。
+ * POST 只写 Redis 缓冲（{@link WatchProgressBuffer}）立即返回，
- * 这里先落地权威表与合并规则，缓存层在接入 Redis 时补齐，接口契约不变。
+ * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后
+ * 批量 UPSERT 进权威表，读接口用缓冲覆盖保证读己之写。
+ * 缓冲关闭或 Redis 异常时自动回退同步写库，接口契约不变。
  *
  * <p>合并规则：以客户端上报的 positionMs 为准，但只在「更靠后」时覆盖，
  * 避免用户换设备后进度倒退。
  */
+@Slf4j
 @Service
 @RequiredArgsConstructor
 public class WatchProgressService {
@@ -53,68 +71,101 @@ public class WatchProgressService {
     private final WatchHistoryMapper historyMapper;
     private final DramaService dramaService;
     private final EpisodeService episodeService;
+    private final ObjectProvider<WatchProgressBuffer> bufferProvider;
+    private final SqlSessionFactory sqlSessionFactory;
+    private final TransactionTemplate transactionTemplate;
+
+    /** 已发布剧集/短剧的短 TTL 进程内缓存：上报路径的热点校验，免去每次两条 DB 查询。 */
+    private final Cache<Long, DramaEpisode> episodeCache = Caffeine.newBuilder()
+            .maximumSize(50_000)
+            .expireAfterWrite(Duration.ofMinutes(5))
+            .build();
+    private final Cache<Long, Drama> dramaCache = Caffeine.newBuilder()
+            .maximumSize(50_000)
+            .expireAfterWrite(Duration.ofMinutes(5))
+            .build();
 
-    @Transactional(rollbackFor = Exception.class)
     public ProgressRespVO save(Long userId, ProgressSaveReqVO req) {
         return save(new InteractionIdentity(userId, ""), req);
     }
 
-    @Transactional(rollbackFor = Exception.class)
     public ProgressRespVO save(InteractionIdentity identity, ProgressSaveReqVO req) {
-        DramaEpisode episode = episodeService.getPublished(req.getEpisodeId());
+        DramaEpisode episode = publishedEpisode(req.getEpisodeId());
-        Drama drama = dramaService.getPublished(episode.getDramaId());
+        Drama drama = publishedDrama(episode.getDramaId());
         long position = req.getPositionMs() == null ? 0L : Math.max(0L, req.getPositionMs());
         long duration = req.getDurationMs() == null ? 0L : Math.max(0L, req.getDurationMs());
         boolean finished = Boolean.TRUE.equals(req.getFinished())
                 || (duration > 0 && position + FINISH_THRESHOLD_MS >= duration);
 
-        LoadedProgress loaded = loadForSave(identity, drama.getId(), episode, position, duration,
+        // 高并发形态：写 Redis 缓冲立即返回，落库任务稍后合并进权威表
-                finished, req);
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
-        WatchProgress existing = loaded.progress();
+        if (buffer != null && buffer.put(identity, drama.getId(), episode.getId(),
-        if (loaded.created()) {
+                position, duration, finished, req.getSource(), req.getDeviceId())) {
-            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
+            ProgressRespVO vo = new ProgressRespVO();
-                    existing.getSource(), existing.getDeviceId());
+            vo.setDramaId(drama.getId());
-            return toResp(progressMapper.selectById(existing.getId()));
+            vo.setEpisodeId(episode.getId());
+            vo.setPositionMs(position);
+            vo.setDurationMs(duration);
+            vo.setFinished(finished);
+            vo.setUpdatedAt(LocalDateTime.now());
+            return vo;
         }
+        return saveToDb(identity, drama, episode, position, duration, finished, req);
+    }
 
-        boolean hidden = existing.getDeleted() != null && existing.getDeleted() != 0;
+    /** 同步写库路径：缓冲关闭或 Redis 异常时的回退，合并规则与缓冲落库一致。 */
-        boolean sameEpisode = existing.getEpisodeId() != null
+    private ProgressRespVO saveToDb(InteractionIdentity identity, Drama drama, DramaEpisode episode,
-                && existing.getEpisodeId().equals(episode.getId());
+                                    long position, long duration, boolean finished,
-        if (hidden && sameEpisode) {
+                                    ProgressSaveReqVO req) {
-            existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
+        return transactionTemplate.execute(status -> {
-            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
+            LoadedProgress loaded = loadForSave(identity, drama.getId(), episode, position, duration,
-                    req.getSource(), existing.getDeviceId());
+                    finished, req);
-        }
+            WatchProgress existing = loaded.progress();
-        if (!sameEpisode) {
+            if (loaded.created()) {
-            // 换集：直接采纳新集与新的播放位置
+                appendHistory(identity, drama.getId(), episode.getId(), position, duration,
-            existing.setEpisodeId(episode.getId());
+                        existing.getSource(), existing.getDeviceId());
-            existing.setPositionMs(position);
+                return toResp(progressMapper.selectById(existing.getId()));
-            existing.setDurationMs(duration);
+            }
-            existing.setFinished(finished ? 1 : 0);
+
-            existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
+            boolean hidden = existing.getDeleted() != null && existing.getDeleted() != 0;
-            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
+            boolean sameEpisode = existing.getEpisodeId() != null
-                    req.getSource(), existing.getDeviceId());
+                    && existing.getEpisodeId().equals(episode.getId());
-        } else {
+            if (hidden && sameEpisode) {
-            existing.setDurationMs(Math.max(existing.getDurationMs() == null ? 0L : existing.getDurationMs(),
+                existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
-                    duration));
+                appendHistory(identity, drama.getId(), episode.getId(), position, duration,
-            if (position >= (existing.getPositionMs() == null ? 0L : existing.getPositionMs())) {
+                        req.getSource(), existing.getDeviceId());
+            }
+            if (!sameEpisode) {
+                // 换集：直接采纳新集与新的播放位置
+                existing.setEpisodeId(episode.getId());
                 existing.setPositionMs(position);
+                existing.setDurationMs(duration);
+                existing.setFinished(finished ? 1 : 0);
+                existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
+                appendHistory(identity, drama.getId(), episode.getId(), position, duration,
+                        req.getSource(), existing.getDeviceId());
+            } else {
+                existing.setDurationMs(Math.max(existing.getDurationMs() == null ? 0L : existing.getDurationMs(),
+                        duration));
+                if (position >= (existing.getPositionMs() == null ? 0L : existing.getPositionMs())) {
+                    existing.setPositionMs(position);
+                }
+                if (finished) {
+                    existing.setFinished(1);
+                }
             }
-            if (finished) {
+            existing.setDeviceId(identity.authenticated()
-                existing.setFinished(1);
+                    ? (req.getDeviceId() == null ? "" : req.getDeviceId())
+                    : identity.deviceId());
+            existing.setSource(req.getSource() == null ? "" : req.getSource());
+            if (hidden) {
+                existing.setDeleted(0);
+                progressMapper.updateAny(existing);
+            } else {
+                progressMapper.updateById(existing);
             }
-        }
+            return toResp(progressMapper.selectById(existing.getId()));
-        existing.setDeviceId(identity.authenticated()
+        });
-                ? (req.getDeviceId() == null ? "" : req.getDeviceId())
-                : identity.deviceId());
-        existing.setSource(req.getSource() == null ? "" : req.getSource());
-        if (hidden) {
-            existing.setDeleted(0);
-            progressMapper.updateAny(existing);
-        } else {
-            progressMapper.updateById(existing);
-        }
-        return toResp(progressMapper.selectById(existing.getId()));
     }
 
     private LoadedProgress loadForSave(InteractionIdentity identity, Long dramaId,
@@ -169,6 +220,16 @@ public class WatchProgressService {
         WatchProgress progress = progressMapper.selectOne(identityQuery(identity)
                 .eq(WatchProgress::getDramaId, dramaId)
                 .last("limit 1"));
+        // 读己之写：缓冲里有比库行更新的状态时以缓冲为准
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer != null) {
+            WatchProgressBuffer.BufferedProgress buffered = buffer.get(identity, dramaId);
+            if (buffered != null
+                    && (progress == null
+                    || buffered.ts() > toMillis(progress.getUpdatedAt()))) {
+                return toResp(buffered, progress);
+            }
+        }
         return progress == null ? null : toResp(progress);
     }
 
@@ -186,7 +247,59 @@ public class WatchProgressService {
                 identityQuery(identity)
                         .orderByDesc(WatchProgress::getUpdatedAt)
                         .orderByDesc(WatchProgress::getId));
-        return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords()));
+
+        // 缓冲的条目都是最近上报的，只会出现在第一页；缓冲关闭时走原逻辑
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer == null || safeIndex > 1) {
+            return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords()));
+        }
+        Map<Long, WatchProgressBuffer.BufferedProgress> states = buffer.getAll(identity);
+        if (states.isEmpty()) {
+            return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords()));
+        }
+
+        List<WatchProgress> merged = new ArrayList<>();
+        Set<Long> pageDramas = new HashSet<>();
+        for (WatchProgress row : result.getRecords()) {
+            pageDramas.add(row.getDramaId());
+            WatchProgressBuffer.BufferedProgress buffered = row.getDramaId() == null
+                    ? null
+                    : states.get(row.getDramaId());
+            merged.add(overlayRow(row, buffered));
+        }
+        long total = result.getTotal();
+        // 缓冲里有、页里没有的剧补进第一页（刚上报首集、库里还没落的场景）
+        List<WatchProgressBuffer.BufferedProgress> extra = states.entrySet().stream()
+                .filter(e -> !pageDramas.contains(e.getKey()))
+                .map(Map.Entry::getValue)
+                .sorted(Comparator.comparingLong(WatchProgressBuffer.BufferedProgress::ts).reversed())
+                .toList();
+        if (!extra.isEmpty()) {
+            List<Long> extraDramaIds = extra.stream()
+                    .map(WatchProgressBuffer.BufferedProgress::dramaId)
+                    .toList();
+            Map<Long, WatchProgress> rowByDrama = progressMapper.selectAnyForIdentityIn(
+                            identity.userId(), identity.deviceId(), extraDramaIds)
+                    .stream()
+                    .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 (existing == null) {
+                    total++;
+                }
+            }
+        }
+        merged.sort(Comparator.comparing(WatchProgress::getUpdatedAt,
+                Comparator.nullsFirst(Comparator.naturalOrder())).reversed());
+        List<WatchProgress> page = merged.size() > safeSize
+                ? merged.subList(0, safeSize)
+                : merged;
+        return new PageResult<>(total, enrichDisplay(page));
     }
 
     @Transactional(rollbackFor = Exception.class)
@@ -196,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));
     }
@@ -207,9 +326,198 @@ 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));
     }
 
+    /**
+     * 缓冲落库：由 {@link WatchProgressFlusher} 定时调用。
+     * 消费 dirty 队列，pipeline 读状态、分块成对查行，先在内存里合并出
+     * 「新增进度 / 新增流水 / 更新行」三组，再在一个 JDBC batch 会话内
+     * 按类型连续攒批执行、一次提交（交替提交会退化成逐条往返，吞吐只有约 200 行/秒）。
+     *
+     * @return 本批次合并的进度条数
+     */
+    public int flushBuffered(int batchSize) {
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer == null) {
+            return 0;
+        }
+        List<WatchProgressBuffer.IdentityRef> refs = buffer.drain(batchSize);
+        if (refs.isEmpty()) {
+            return 0;
+        }
+        try {
+            return flushRefs(buffer, refs);
+        } catch (Exception ex) {
+            log.warn("观看进度缓冲合并失败，整批退回队列稍后重试: {}", ex.getMessage());
+            buffer.requeue(refs);
+            return 0;
+        }
+    }
+
+    private int flushRefs(WatchProgressBuffer buffer, List<WatchProgressBuffer.IdentityRef> refs) {
+        // 1) pipeline 批量读缓冲状态，key -> (dramaId -> state)
+        List<String> identityKeys = refs.stream()
+                .map(WatchProgressBuffer.IdentityRef::key)
+                .distinct()
+                .toList();
+        Map<String, Map<Long, WatchProgressBuffer.BufferedProgress>> statesByIdentity =
+                buffer.getAllBatch(identityKeys);
+
+        // 2) 组装待合并条目；状态读不到视为异常（ref 必然来自 put），整批退回
+        List<WatchProgressBuffer.IdentityRef> pendingRefs = new ArrayList<>(refs.size());
+        List<WatchProgressBuffer.BufferedProgress> pendingStates = new ArrayList<>(refs.size());
+        List<WatchProgressMapper.IdentityDramaPair> pairs = new ArrayList<>(refs.size());
+        for (WatchProgressBuffer.IdentityRef ref : refs) {
+            WatchProgressBuffer.BufferedProgress state =
+                    statesByIdentity.getOrDefault(ref.key(), Map.of()).get(ref.dramaId());
+            if (state == null) {
+                continue;
+            }
+            pendingRefs.add(ref);
+            pendingStates.add(state);
+            pairs.add(new WatchProgressMapper.IdentityDramaPair(
+                    ref.userId(), ref.deviceId(), ref.dramaId()));
+        }
+        if (pendingRefs.isEmpty()) {
+            // 缓冲状态已消失（用户清空/删除历史，或 HASH 12h TTL 到期）。此时 ref 已无
+            // 可落库的内容，必须丢弃：退回队列会在下个周期被原样弹出、再清空、再退回，
+            // 这批僵尸条目将永久占住 dirty 队列配额。
+            log.debug("观看进度缓冲条目已无对应状态，丢弃 {} 条", refs.size());
+            return 0;
+        }
+
+        // 3) 分块成对查询已有行（每块 100 组 OR 索引条件）
+        Map<String, WatchProgress> rowByPair = new HashMap<>(pendingRefs.size());
+        for (int i = 0; i < pairs.size(); i += 100) {
+            List<WatchProgressMapper.IdentityDramaPair> chunk =
+                    pairs.subList(i, Math.min(i + 100, pairs.size()));
+            for (WatchProgress row : progressMapper.selectBatchForPairs(chunk)) {
+                rowByPair.put(rowKey(row.getUserId(), row.getDeviceId(), row.getDramaId()), row);
+            }
+        }
+
+        // 4) 计算阶段：纯内存合并出「新增进度 / 新增流水 / 更新行」三组
+        List<WatchProgress> insertProgress = new ArrayList<>(pendingRefs.size());
+        List<WatchHistory> insertHistory = new ArrayList<>(pendingRefs.size());
+        List<WatchProgress> updateRows = new ArrayList<>(pendingRefs.size());
+        for (int i = 0; i < pendingRefs.size(); i++) {
+            WatchProgress row = rowByPair.get(rowKey(
+                    pendingRefs.get(i).userId(),
+                    pendingRefs.get(i).deviceId(),
+                    pendingStates.get(i).dramaId()));
+            planOne(pendingRefs.get(i).toIdentity(), pendingStates.get(i), row,
+                    insertProgress, insertHistory, updateRows);
+        }
+
+        // 5) 执行阶段：同一个 batch 会话按类型连续执行，最后一次提交。
+        //    MyBatis BATCH 执行器只合并「连续同形状」的语句，进度/流水交替提交会
+        //    退化成逐条网络往返（实测只有约 200 行/秒），必须分组攒批。
+        try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
+            WatchProgressMapper batchProgress = session.getMapper(WatchProgressMapper.class);
+            WatchHistoryMapper batchHistory = session.getMapper(WatchHistoryMapper.class);
+            for (WatchProgress created : insertProgress) {
+                batchProgress.insert(created);
+            }
+            for (WatchHistory history : insertHistory) {
+                batchHistory.insert(history);
+            }
+            for (WatchProgress row : updateRows) {
+                batchProgress.updateRowAny(row);
+            }
+            session.commit();
+        }
+        return pendingRefs.size();
+    }
+
+    /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */
+    private void planOne(InteractionIdentity identity,
+                         WatchProgressBuffer.BufferedProgress state,
+                         WatchProgress row,
+                         List<WatchProgress> insertProgress,
+                         List<WatchHistory> insertHistory,
+                         List<WatchProgress> updateRows) {
+        if (row == null) {
+            WatchProgress created = new WatchProgress();
+            created.setUserId(identity.userId());
+            created.setDramaId(state.dramaId());
+            created.setEpisodeId(state.episodeId());
+            created.setPositionMs(state.positionMs());
+            created.setDurationMs(state.durationMs());
+            created.setFinished(state.finished() ? 1 : 0);
+            created.setPlayCount(1);
+            created.setDeviceId(identity.authenticated()
+                    ? state.deviceId()
+                    : identity.deviceId());
+            created.setSource(state.source());
+            insertProgress.add(created);
+            insertHistory.add(historyOf(identity, state, created.getDeviceId()));
+            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) {
+            row.setPlayCount((row.getPlayCount() == null ? 0 : row.getPlayCount()) + 1);
+            insertHistory.add(historyOf(identity, state, row.getDeviceId()));
+        } else if (!sameEpisode) {
+            // 换集：直接采纳新集与新的播放位置
+            row.setEpisodeId(state.episodeId());
+            row.setPositionMs(state.positionMs());
+            row.setDurationMs(state.durationMs());
+            row.setFinished(state.finished() ? 1 : 0);
+            row.setPlayCount((row.getPlayCount() == null ? 0 : row.getPlayCount()) + 1);
+            insertHistory.add(historyOf(identity, state, row.getDeviceId()));
+        } else {
+            row.setDurationMs(Math.max(row.getDurationMs() == null ? 0L : row.getDurationMs(),
+                    state.durationMs()));
+            if (state.positionMs() >= (row.getPositionMs() == null ? 0L : row.getPositionMs())) {
+                row.setPositionMs(state.positionMs());
+            }
+            if (state.finished()) {
+                row.setFinished(1);
+            }
+        }
+        row.setDeviceId(identity.authenticated() ? state.deviceId() : identity.deviceId());
+        row.setSource(state.source());
+        if (hidden) {
+            row.setDeleted(0);
+        }
+        updateRows.add(row);
+    }
+
+    /** 组装一条观看流水（首次观剧/换集各记一条）。 */
+    private WatchHistory historyOf(InteractionIdentity identity,
+                                   WatchProgressBuffer.BufferedProgress state,
+                                   String deviceId) {
+        WatchHistory history = new WatchHistory();
+        history.setUserId(identity.userId());
+        history.setDramaId(state.dramaId());
+        history.setEpisodeId(state.episodeId());
+        history.setPositionMs(state.positionMs());
+        history.setDurationMs(state.durationMs());
+        history.setSource(state.source() == null ? "" : state.source());
+        history.setDeviceId(identity.authenticated()
+                ? (deviceId == null ? "" : deviceId)
+                : identity.deviceId());
+        return history;
+    }
+
+    /** 行定位键：登录身份用 userId，匿名用 deviceId。 */
+    private String rowKey(Long userId, String deviceId, Long dramaId) {
+        return (userId != null ? "u:" + userId : "d:" + deviceId) + ":" + dramaId;
+    }
+
     private ProgressRespVO toResp(WatchProgress progress) {
         ProgressRespVO vo = new ProgressRespVO();
         vo.setDramaId(progress.getDramaId());
@@ -222,6 +530,68 @@ public class WatchProgressService {
         return vo;
     }
 
+    /** 以缓冲状态为主、库行 playCount 兜底构建续看响应。 */
+    private ProgressRespVO toResp(WatchProgressBuffer.BufferedProgress buffered, WatchProgress row) {
+        ProgressRespVO vo = new ProgressRespVO();
+        vo.setDramaId(buffered.dramaId());
+        vo.setEpisodeId(buffered.episodeId());
+        vo.setPositionMs(buffered.positionMs());
+        vo.setDurationMs(buffered.durationMs());
+        vo.setFinished(buffered.finished());
+        vo.setPlayCount(row == null || row.getPlayCount() == null ? 1 : row.getPlayCount());
+        vo.setUpdatedAt(fromMillis(buffered.ts()));
+        return vo;
+    }
+
+    /** 库行被缓冲覆盖时返回覆盖后的瞬态行；缓冲不比库行新时原样返回。 */
+    private WatchProgress overlayRow(WatchProgress row, WatchProgressBuffer.BufferedProgress buffered) {
+        if (buffered == null || row.getUpdatedAt() == null
+                || buffered.ts() <= toMillis(row.getUpdatedAt())) {
+            return row;
+        }
+        WatchProgress overlaid = transientRow(buffered);
+        overlaid.setPlayCount(row.getPlayCount());
+        return overlaid;
+    }
+
+    private WatchProgress transientRow(WatchProgressBuffer.BufferedProgress buffered) {
+        WatchProgress row = new WatchProgress();
+        row.setDramaId(buffered.dramaId());
+        row.setEpisodeId(buffered.episodeId());
+        row.setPositionMs(buffered.positionMs());
+        row.setDurationMs(buffered.durationMs());
+        row.setFinished(buffered.finished() ? 1 : 0);
+        row.setPlayCount(1);
+        row.setUpdatedAt(fromMillis(buffered.ts()));
+        return row;
+    }
+
+    private long toMillis(LocalDateTime time) {
+        return time == null ? 0L : time.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+    }
+
+    private LocalDateTime fromMillis(long epochMillis) {
+        return LocalDateTime.ofInstant(Instant.ofEpochMilli(epochMillis), ZoneId.systemDefault());
+    }
+
+    private DramaEpisode publishedEpisode(Long episodeId) {
+        DramaEpisode episode = episodeCache.getIfPresent(episodeId);
+        if (episode == null) {
+            episode = episodeService.getPublished(episodeId);
+            episodeCache.put(episodeId, episode);
+        }
+        return episode;
+    }
+
+    private Drama publishedDrama(Long dramaId) {
+        Drama drama = dramaCache.getIfPresent(dramaId);
+        if (drama == null) {
+            drama = dramaService.getPublished(dramaId);
+            dramaCache.put(dramaId, drama);
+        }
+        return drama;
+    }
+
     private List<ProgressRespVO> enrichDisplay(List<WatchProgress> records) {
         if (records.isEmpty()) {
             return List.of();
@@ -296,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;
+        }
+    }
+
     /**
      * 登录时合并设备进度和观看流水。
      *
@@ -307,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)
@@ -366,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);
     }

grafana/provisioning/datasources/loki.yml

@@ -0,0 +1,14 @@
+# Author: fruit.com.cn
+# Grafana 数据源：Loki 日志
+apiVersion: 1
+
+datasources:
+  - name: Loki
+    uid: loki
+    type: loki
+    access: proxy
+    url: http://loki:3100
+    isDefault: false
+    editable: false
+    jsonData:
+      maxLines: 1000

loki-config.yml

@@ -0,0 +1,55 @@
+# Author: fruit.com.cn
+# Loki 单机配置：文件系统存储，7 天保留，供 Grafana 查询容器日志。
+auth_enabled: false
+
+server:
+  http_listen_port: 3100
+  grpc_listen_port: 9096
+
+common:
+  instance_addr: 127.0.0.1
+  path_prefix: /loki
+  storage:
+    filesystem:
+      chunks_directory: /loki/chunks
+      rules_directory: /loki/rules
+  replication_factor: 1
+  ring:
+    kvstore:
+      store: inmemory
+
+query_range:
+  results_cache:
+    cache:
+      embedded_cache:
+        enabled: true
+        max_size_mb: 100
+
+schema_config:
+  configs:
+    - from: 2024-01-01
+      store: tsdb
+      object_store: filesystem
+      schema: v13
+      index:
+        prefix: index_
+        period: 24h
+
+# 7 天保留，避免磁盘无限增长
+limits_config:
+  retention_period: 168h
+  max_query_series: 1000
+  # 允许采集历史日志（容器启动时的旧日志也能入库），靠 retention_period 控制总量
+  reject_old_samples: false
+  ingestion_rate_mb: 20
+  ingestion_burst_size_mb: 40
+  per_stream_rate_limit: 10MB
+  per_stream_rate_limit_burst: 40MB
+
+compactor:
+  working_directory: /loki/compactor
+  compaction_interval: 10m
+  retention_enabled: true
+  retention_delete_delay: 2h
+  retention_delete_worker_count: 150
+  delete_request_store: filesystem

promtail-config.yml

@@ -0,0 +1,34 @@
+# Author: fruit.com.cn
+# Promtail：通过 Docker 服务发现自动采集所有容器的 stdout/stderr 日志，推送到 Loki。
+server:
+  http_listen_port: 9080
+  grpc_listen_port: 0
+
+positions:
+  filename: /tmp/positions.yaml
+
+clients:
+  - url: http://loki:3100/loki/api/v1/push
+    batchwait: 1s
+    batchsize: 1048576
+
+scrape_configs:
+  - job_name: docker-containers
+    docker_sd_configs:
+      - host: unix:///var/run/docker.sock
+        refresh_interval: 5s
+    relabel_configs:
+      # 容器名（去掉开头的 /）
+      - source_labels: ['__meta_docker_container_name']
+        regex: '/(.*)'
+        target_label: container
+      # 日志流：stdout / stderr
+      - source_labels: ['__meta_docker_container_log_stream']
+        target_label: stream
+      # 容器 ID 前 12 位
+      - source_labels: ['__meta_docker_container_id']
+        target_label: container_id
+        regex: '(.{12}).*'
+      # 镜像名
+      - source_labels: ['__meta_docker_container_image']
+        target_label: image

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;

sql/migrate-k114-thumbnail-partial-index.sql

@@ -0,0 +1,7 @@
+-- 缩略图时间轴唯一约束改为部分索引：
+-- media_thumbnail.deleted 为 @TableLogic 逻辑删除，软删行仍占用 episode_id 会导致
+-- ThumbnailRecordService.upsert 的 delete+insert 触发 uk_media_thumbnail_episode 冲突。
+-- 唯一性只约束未删除行（deleted = 0）。
+DROP INDEX IF EXISTS uk_media_thumbnail_episode;
+CREATE UNIQUE INDEX IF NOT EXISTS uk_media_thumbnail_episode
+  ON media_thumbnail (episode_id) WHERE deleted = 0;

sql/schema-phase1.sql

@@ -806,7 +806,7 @@ CREATE TABLE IF NOT EXISTS media_thumbnail (
   deleted        SMALLINT     NOT NULL DEFAULT 0
 );
 CREATE UNIQUE INDEX IF NOT EXISTS uk_media_thumbnail_episode
-  ON media_thumbnail (episode_id);
+  ON media_thumbnail (episode_id) WHERE deleted = 0;
 CREATE INDEX IF NOT EXISTS idx_media_thumbnail_drama
   ON media_thumbnail (drama_id);
 COMMENT ON TABLE media_thumbnail IS '单集缩略图雪碧图与 VTT 时间轴';