fix(ops): 修复管理端 Redis 未配置与看门狗未真正周期巡检

- admin-api 的 application.yml 缺少 spring.data.redis 配置，容器内默认连
  127.0.0.1 导致 /actuator/health 返回 DOWN（redis 组件不可用）；
  补齐后与 app-api 共用 REDIS_HOST/REDIS_PORT，健康检查恢复 UP
- 计划任务原先用 BootTrigger 挂 Repetition，Windows 不生效（Next Run Time
  为 N/A，9 小时只跑过一次），看门狗形同虚设；改为 TimeTrigger + PT2M
  无限重复 + StartWhenAvailable，实测每 2 分钟稳定巡检
- watchdog.ps1 的连续失败计数改为落盘 watchdog-state.json，避免每轮进程
  重启后清零导致「连续 3 次失败告警」永不触发

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,187 @@
+# 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"
+
+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 "飞书告警发送失败: $_"
+    }
+}
+
+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
+
+# 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" }
+)
+
+# 跟踪每个服务的连续失败次数。脚本由计划任务每 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 "状态文件损坏，重置连续失败计数"
+    }
+}
+
+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
+            }
+        }
+
+        # 连续失败达到上限 → 告警
+        if (-not $healthy -and $count -ge $MaxRestarts) {
+            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}%，请清理日志或扩容"
+    }
+}
+
+# 落盘本轮结束时的连续失败计数，供下一轮接着累计
+try {
+    $failCounts | ConvertTo-Json | Set-Content -Path $StateFile -Encoding UTF8
+} catch {
+    Write-Log "状态文件写入失败: $_"
+}

docker-compose.yml

@@ -71,8 +71,10 @@ services:
     env_file: .env
     environment:
       SERVER_PORT: "8081"
-      # 默认连编排里的 db；要连外部库就在 .env 里覆盖 DB_URL
-      DB_URL: ${DB_URL:-jdbc:postgresql://db:5432/${DB_NAME:-fruitshorts}}
+      # 默认连编排里的 db；要连外部库就在 .env 里覆盖 DB_URL。
+      # 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,8 +11,10 @@ spring:
   datasource:
     driver-class-name: org.postgresql.Driver
     # 地址必须外置：不同环境（本机 / 测试 / 生产）连的库不一样，
-    # 生产请用 DB_URL 注入（prod profile 会校验它非空）
-    url: ${DB_URL:jdbc:postgresql://127.0.0.1:5432/fruitshorts}
+    # 生产请用 DB_URL 注入（prod profile 会校验它非空）。
+    # 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:

fruitshorts-app-api/src/test/java/com/fruitshorts/playback/service/WatchProgressServiceTest.java

@@ -20,10 +20,14 @@ import com.fruitshorts.playback.vo.ProgressRespVO;
 import com.fruitshorts.playback.vo.ProgressSaveReqVO;
 import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.session.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.springframework.beans.factory.ObjectProvider;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
 
 import java.time.LocalDateTime;
 import java.lang.reflect.Method;
@@ -58,11 +62,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

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,304 @@
+// 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.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（每次上报 1 次写），由
+ * {@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);
+
+    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);
+            redis.opsForHash().put(hashKey, field, encode(
+                    episodeId, positionMs, durationMs, finished, source, deviceId));
+            redis.expire(hashKey, TTL);
+            trimIdentity(hashKey);
+            redis.opsForList().leftPush(DIRTY_KEY, key + "|" + field);
+            redis.opsForList().trim(DIRTY_KEY, 0, MAX_DIRTY - 1);
+            return true;
+        } catch (Exception e) {
+            log.warn("观看进度写入 Redis 缓冲失败，本次回退同步写库: {}", 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 写/秒），
- * 所以生产形态是「先写 Redis，后台任务按登录账号或设备 + drama 合并后批量 UPSERT」。
- * 这里先落地权威表与合并规则，缓存层在接入 Redis 时补齐，接口契约不变。
+ * <p>生产形态（fruitshorts.watch-progress.buffer=true，prod 已开启）：
+ * POST 只写 Redis 缓冲（{@link WatchProgressBuffer}）立即返回，
+ * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后
+ * 批量 UPSERT 进权威表，读接口用缓冲覆盖保证读己之写。
+ * 缓冲关闭或 Redis 异常时自动回退同步写库，接口契约不变。
  *
  * <p>合并规则：以客户端上报的 positionMs 为准，但只在「更靠后」时覆盖，
  * 避免用户换设备后进度倒退。
  */
+@Slf4j
 @Service
 @RequiredArgsConstructor
 public class WatchProgressService {
@@ -53,68 +71,102 @@ 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());
-        Drama drama = dramaService.getPublished(episode.getDramaId());
+        DramaEpisode episode = publishedEpisode(req.getEpisodeId());
+        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,
-                finished, req);
-        WatchProgress existing = loaded.progress();
-        if (loaded.created()) {
-            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
-                    existing.getSource(), existing.getDeviceId());
-            return toResp(progressMapper.selectById(existing.getId()));
+        // 高并发形态：写 Redis 缓冲立即返回，落库任务稍后合并进权威表
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer != null && buffer.put(identity, drama.getId(), episode.getId(),
+                position, duration, finished, req.getSource(), req.getDeviceId())) {
+            ProgressRespVO vo = new ProgressRespVO();
+            vo.setDramaId(drama.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;
-        boolean sameEpisode = existing.getEpisodeId() != null
-                && existing.getEpisodeId().equals(episode.getId());
-        if (hidden && sameEpisode) {
-            existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
-            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
-                    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())) {
+    /** 同步写库路径：缓冲关闭或 Redis 异常时的回退，合并规则与缓冲落库一致。 */
+    private ProgressRespVO saveToDb(InteractionIdentity identity, Drama drama, DramaEpisode episode,
+                                    long position, long duration, boolean finished,
+                                    ProgressSaveReqVO req) {
+        return transactionTemplate.execute(status -> {
+            LoadedProgress loaded = loadForSave(identity, drama.getId(), episode, position, duration,
+                    finished, req);
+            WatchProgress existing = loaded.progress();
+            if (loaded.created()) {
+                appendHistory(identity, drama.getId(), episode.getId(), position, duration,
+                        existing.getSource(), existing.getDeviceId());
+                return toResp(progressMapper.selectById(existing.getId()));
+            }
+
+            boolean hidden = existing.getDeleted() != null && existing.getDeleted() != 0;
+            boolean sameEpisode = existing.getEpisodeId() != null
+                    && existing.getEpisodeId().equals(episode.getId());
+            if (hidden && sameEpisode) {
+                existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
+                appendHistory(identity, drama.getId(), episode.getId(), position, duration,
+                        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.setFinished(1);
+            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);
             }
-        }
-        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()));
+            return toResp(progressMapper.selectById(existing.getId()));
+        });
     }
 
     private LoadedProgress loadForSave(InteractionIdentity identity, Long dramaId,
@@ -169,6 +221,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 +248,55 @@ 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();
+            Set<Long> withRow = progressMapper.selectAnyForIdentityIn(
+                            identity.userId(), identity.deviceId(), extraDramaIds)
+                    .stream()
+                    .map(WatchProgress::getDramaId)
+                    .collect(Collectors.toSet());
+            for (WatchProgressBuffer.BufferedProgress buffered : extra) {
+                merged.add(transientRow(buffered));
+                if (!withRow.contains(buffered.dramaId())) {
+                    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)
@@ -210,6 +320,181 @@ public class WatchProgressService {
         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 {
+            flushRefs(buffer, refs);
+            return refs.size();
+        } catch (Exception ex) {
+            log.warn("观看进度缓冲合并失败，整批退回队列稍后重试: {}", ex.getMessage());
+            buffer.requeue(refs);
+            return 0;
+        }
+    }
+
+    private void 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()) {
+            throw new IllegalStateException("缓冲状态读取为空，ref=" + refs.size());
+        }
+
+        // 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();
+        }
+    }
+
+    /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */
+    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;
+        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 +507,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();

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/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 时间轴';