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

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

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

@@ -20,11 +20,14 @@ import com.fruitshorts.playback.vo.ProgressRespVO;
 import com.fruitshorts.playback.vo.ProgressSaveReqVO;
 import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.session.ExecutorType;
+import org.apache.ibatis.session.SqlSession;
 import org.apache.ibatis.session.SqlSessionFactory;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
@@ -38,6 +41,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -384,7 +388,58 @@ class WatchProgressServiceTest {
         verify(progressMapper, never()).selectBatchForPairs(any());
     }
 
+    @Test
+    void should_flush_device_buffer_before_merging_and_clear_it_after() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        given(buffer.getAll(any(InteractionIdentity.class))).willReturn(Map.of());
+        WatchProgressService service = serviceWithBuffer(buffer);
+        given(progressMapper.selectList(any())).willReturn(List.of());
+        given(historyMapper.selectList(any())).willReturn(List.of());
+
+        service.mergeDeviceIntoUser(9L, "device-a");
+
+        // 合并前先把缓冲落库，合并完再清掉设备侧缓冲
+        InOrder order = inOrder(buffer);
+        order.verify(buffer).getAll(any(InteractionIdentity.class));
+        order.verify(buffer).clear(any(InteractionIdentity.class));
+    }
+
+    @Test
+    void should_drop_late_buffered_ref_when_row_is_already_deleted() {
+        WatchProgressBuffer buffer = mock(WatchProgressBuffer.class);
+        SqlSessionFactory sessionFactory = mock(SqlSessionFactory.class);
+        SqlSession session = mock(SqlSession.class);
+        WatchProgressMapper batchProgress = mock(WatchProgressMapper.class);
+        WatchHistoryMapper batchHistory = mock(WatchHistoryMapper.class);
+        given(sessionFactory.openSession(ExecutorType.BATCH, false)).willReturn(session);
+        given(session.getMapper(WatchProgressMapper.class)).willReturn(batchProgress);
+        given(session.getMapper(WatchHistoryMapper.class)).willReturn(batchHistory);
+        WatchProgressService service = serviceWithBuffer(buffer, sessionFactory);
+
+        given(buffer.drain(10)).willReturn(
+                List.of(new WatchProgressBuffer.IdentityRef("u:9", 9L, "", 5L)));
+        given(buffer.getAllBatch(List.of("u:9"))).willReturn(Map.of("u:9", Map.of(
+                5L, new WatchProgressBuffer.BufferedProgress(
+                        5L, 88L, 1_000L, 2_000L, false, "feed", "", 1_700_000_000_000L))));
+        // 库里是逻辑删除行，且删除时刻晚于该条缓冲的上报时刻
+        WatchProgress hidden = progress(5L, 88L);
+        hidden.setUserId(9L);
+        hidden.setDeleted(1);
+        hidden.setUpdatedAt(LocalDateTime.of(2026, 10, 28, 12, 0));
+        given(progressMapper.selectBatchForPairs(any())).willReturn(List.of(hidden));
+
+        service.flushBuffered(10);
+
+        verify(batchProgress, never()).updateRowAny(any());
+        verify(batchHistory, never()).insert(any(WatchHistory.class));
+    }
+
     private WatchProgressService serviceWithBuffer(WatchProgressBuffer buffer) {
+        return serviceWithBuffer(buffer, mock(SqlSessionFactory.class));
+    }
+
+    private WatchProgressService serviceWithBuffer(
+            WatchProgressBuffer buffer, SqlSessionFactory sessionFactory) {
         @SuppressWarnings("unchecked")
         ObjectProvider<WatchProgressBuffer> provider = mock(ObjectProvider.class);
         given(provider.getIfAvailable()).willReturn(buffer);
@@ -394,7 +449,7 @@ class WatchProgressServiceTest {
                 dramaService,
                 episodeService,
                 provider,
-                mock(SqlSessionFactory.class),
+                sessionFactory,
                 new TransactionTemplate(mock(PlatformTransactionManager.class)));
     }
 

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

@@ -8,6 +8,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.data.redis.core.SessionCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.springframework.stereotype.Component;
 
 import java.time.Duration;
@@ -19,7 +20,7 @@ import java.util.Map;
 /**
  * 观看进度 Redis 缓冲（write-behind）。
  *
- * <p>高并发上报的落地形态：POST 只写 Redis（每次上报 1 次写），由
+ * <p>高并发上报的落地形态：POST 只写 Redis（Lua 脚本一次往返完成写状态与入队），由
  * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后批量 UPSERT
  * 进权威表 watch_progress。Redis 不可用时 put 返回 false，调用方回退同步写库。
  *
@@ -45,6 +46,22 @@ public class WatchProgressBuffer {
     private static final int MAX_DRAMAS_PER_IDENTITY = 30;
     private static final Duration TTL = Duration.ofHours(12);
 
+    /**
+     * 上报热路径的单次往返脚本：写状态、续 TTL、入待落库队列、截断队列一次做完。
+     * 拆成多条命令时进程若死在「写状态」与「入队」之间，该剧进度只留在 HASH 里、
+     * 永远不会落库，12 小时后被 TTL 清掉。返回 1 表示该身份的剧数超限，需要补充裁剪。
+     */
+    private static final DefaultRedisScript<Long> PUT_SCRIPT = new DefaultRedisScript<>("""
+            redis.call('HSET', KEYS[1], ARGV[1], ARGV[2])
+            redis.call('EXPIRE', KEYS[1], ARGV[3])
+            redis.call('LPUSH', KEYS[2], ARGV[4])
+            redis.call('LTRIM', KEYS[2], 0, ARGV[5])
+            if redis.call('HLEN', KEYS[1]) > tonumber(ARGV[6]) then
+              return 1
+            end
+            return 0
+            """, Long.class);
+
     private final StringRedisTemplate redis;
 
     /**
@@ -80,12 +97,18 @@ public class WatchProgressBuffer {
             String key = identityKey(identity);
             String hashKey = BUF_KEY_PREFIX + key;
             String field = Long.toString(dramaId);
-            redis.opsForHash().put(hashKey, field, encode(
-                    episodeId, positionMs, durationMs, finished, source, deviceId));
-            redis.expire(hashKey, TTL);
-            trimIdentity(hashKey);
-            redis.opsForList().leftPush(DIRTY_KEY, key + "|" + field);
-            redis.opsForList().trim(DIRTY_KEY, 0, MAX_DIRTY - 1);
+            Long overflow = redis.execute(PUT_SCRIPT,
+                    List.of(hashKey, DIRTY_KEY),
+                    field,
+                    encode(episodeId, positionMs, durationMs, finished, source, deviceId),
+                    Long.toString(TTL.toSeconds()),
+                    key + "|" + field,
+                    Long.toString(MAX_DIRTY - 1),
+                    Integer.toString(MAX_DRAMAS_PER_IDENTITY));
+            if (overflow != null && overflow > 0) {
+                // 超出 30 部才做一次 HGETALL + HDEL 裁剪，热路径保持单次往返
+                trimIdentity(hashKey);
+            }
             return true;
         } catch (Exception e) {
             log.warn("观看进度写入 Redis 缓冲失败，本次回退同步写库: {}", e.getMessage());

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

@@ -85,7 +85,6 @@ public class WatchProgressService {
             .expireAfterWrite(Duration.ofMinutes(5))
             .build();
 
-    @Transactional(rollbackFor = Exception.class)
     public ProgressRespVO save(Long userId, ProgressSaveReqVO req) {
         return save(new InteractionIdentity(userId, ""), req);
     }
@@ -460,6 +459,12 @@ public class WatchProgressService {
             return;
         }
         boolean hidden = row.getDeleted() != null && row.getDeleted() != 0;
+        long rowTs = toMillis(row.getUpdatedAt());
+        if (hidden && rowTs > 0 && state.ts() <= rowTs) {
+            // 逻辑删除会刷新 updated_at：上报时间早于它的条目是删除前排队的迟到数据，
+            // 丢弃即可，否则会把已删记录重新写活。
+            return;
+        }
         boolean sameEpisode = row.getEpisodeId() != null
                 && row.getEpisodeId().equals(state.episodeId());
         if (hidden && sameEpisode) {
@@ -661,6 +666,36 @@ public class WatchProgressService {
         historyMapper.insert(history);
     }
 
+    /**
+     * 把某个身份缓冲里尚未落库的进度立即写进权威表。
+     * 登录合并前必须先调用：合并只扫库表，缓冲里的设备进度会被漏掉，
+     * 随后定时落库又会生成一条登录用户看不到的匿名孤儿行。
+     *
+     * @return 本次合并的条目数
+     */
+    public int flushIdentity(InteractionIdentity identity) {
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer == null) {
+            return 0;
+        }
+        Map<Long, WatchProgressBuffer.BufferedProgress> states = buffer.getAll(identity);
+        if (states.isEmpty()) {
+            return 0;
+        }
+        String key = buffer.identityKey(identity);
+        List<WatchProgressBuffer.IdentityRef> refs = states.keySet().stream()
+                .map(dramaId -> new WatchProgressBuffer.IdentityRef(
+                        key, identity.userId(), identity.deviceId(), dramaId))
+                .toList();
+        try {
+            return flushRefs(buffer, refs);
+        } catch (Exception ex) {
+            // 预写失败不影响合并本身：dirty 队列里的引用还在，下个周期会再落库
+            log.warn("观看进度缓冲立即落库失败，跳过预写: {}", ex.getMessage());
+            return 0;
+        }
+    }
+
     /**
      * 登录时合并设备进度和观看流水。
      *
@@ -672,6 +707,9 @@ public class WatchProgressService {
         if (userId == null || !StringUtils.hasText(deviceId)) {
             return new MergeResult(0, 0, 0, 0);
         }
+        InteractionIdentity deviceIdentity = new InteractionIdentity(null, deviceId);
+        // 缓冲里可能还有没落库的设备进度，先落库再合并，否则这段进度会被漏掉
+        flushIdentity(deviceIdentity);
         List<WatchProgress> deviceProgress = progressMapper.selectList(
                 Wrappers.<WatchProgress>lambdaQuery()
                         .isNull(WatchProgress::getUserId)
@@ -731,6 +769,11 @@ public class WatchProgressService {
             }
             historyMapper.deleteById(device.getId());
         }
+        // 设备侧进度已迁走或淘汰，缓冲留着只会让读覆盖层把已删行读回来
+        WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+        if (buffer != null) {
+            buffer.clear(deviceIdentity);
+        }
         return new MergeResult(
                 progressMigrated, progressIgnored, historyMigrated, historyIgnored);
     }

sql/data-sensitive-words-id.sql

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