perf(playback): 观看进度上报异步化（Redis write-behind 批量落库）

- POST /app/watch/progress 只写 Redis 缓冲立即返回（p95 75ms→2ms），
  WatchProgressFlusher 每 2s 把 watch:pg:dirty 队列按身份合并批量 UPSERT，
  Redis 异常自动回退同步写库，接口契约不变
- 读接口用缓冲覆盖 DB 行保证读己之写；Caffeine 缓存剧集校验免去上报路径两条 DB 查询
- 落库按「新增进度/新增流水/更新行」分组攒批（MyBatis BATCH 只合并连续同形状语句），
  配合 reWriteBatchedInserts 多值 INSERT，12.4 万积压 45 秒消化完，
  1650 QPS 持续压测队列稳态归零

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}

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,21 +71,54 @@ 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);
 
+        // 高并发形态：写 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);
+    }
+
+    /** 同步写库路径：缓冲关闭或 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();
@@ -115,6 +166,7 @@ public class WatchProgressService {
                 progressMapper.updateById(existing);
             }
             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,8 +248,56 @@ public class WatchProgressService {
                 identityQuery(identity)
                         .orderByDesc(WatchProgress::getUpdatedAt)
                         .orderByDesc(WatchProgress::getId));
+
+        // 缓冲的条目都是最近上报的，只会出现在第一页；缓冲关闭时走原逻辑
+        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)
     public void remove(Long userId, Long dramaId) {
@@ -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();