feat(watch): 支持设备观看与登录合并

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

@@ -10,13 +10,21 @@ import org.apache.ibatis.annotations.Update;
 
 public interface WatchProgressMapper extends BaseMapper<WatchProgress> {
 
-    /** 含逻辑删除行。唯一索引不区分 deleted，再看同一部剧必须能找回旧行。 */
+    /** 按登录用户或匿名设备读取同一部剧的记录，包含逻辑删除行。 */
     @Select("""
             SELECT * FROM watch_progress
-            WHERE user_id = #{userId} AND drama_id = #{dramaId}
+            WHERE (
+              (#{userId} IS NOT NULL AND user_id = #{userId})
+              OR
+              (#{userId} IS NULL AND user_id IS NULL AND device_id = #{deviceId})
+            )
+            AND drama_id = #{dramaId}
             LIMIT 1
             """)
-    WatchProgress selectAny(@Param("userId") Long userId, @Param("dramaId") Long dramaId);
+    WatchProgress selectAnyForIdentity(
+            @Param("userId") Long userId,
+            @Param("deviceId") String deviceId,
+            @Param("dramaId") Long dramaId);
 
     /** 恢复并更新。手写 SQL 不带 deleted=0 条件，逻辑删除行也能写回。 */
     @Update("""

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

@@ -152,14 +152,19 @@ public class FavoriteService {
 
     /** 分页查询收藏短剧，保持收藏时间与 dramaId 稳定排序。 */
     public PageResult<Drama> page(Long userId, int pageIndex, int pageSize) {
+        return page(new InteractionIdentity(userId, ""), pageIndex, pageSize);
+    }
+
+    /** 分页查询当前登录账号或匿名设备的收藏短剧。 */
+    public PageResult<Drama> page(
+            InteractionIdentity identity, int pageIndex, int pageSize) {
         int safeIndex = Math.max(pageIndex, 1);
         int safeSize = Math.min(Math.max(pageSize, 1), MAX_PAGE_SIZE);
         Page<UserFavorite> request = new Page<>(safeIndex, safeSize);
         String publishedDramaSql = "SELECT id FROM drama WHERE deleted = 0 AND status = '"
                 + ContentStatus.PUBLISHED.getCode() + "'";
         IPage<UserFavorite> result = favoriteMapper.selectPage(request,
-                Wrappers.<UserFavorite>lambdaQuery()
-                .eq(UserFavorite::getUserId, userId)
+                identityQuery(identity)
                         .inSql(UserFavorite::getDramaId, publishedDramaSql)
                         .orderByDesc(UserFavorite::getCreatedAt)
                         .orderByDesc(UserFavorite::getDramaId));

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

@@ -3,9 +3,11 @@
 package com.fruitshorts.playback.service;
 
 import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.fruitshorts.common.pojo.PageResult;
+import com.fruitshorts.common.interaction.InteractionIdentity;
 import com.fruitshorts.content.dal.entity.Drama;
 import com.fruitshorts.content.dal.entity.DramaEpisode;
 import com.fruitshorts.content.service.ContentLocaleTexts;
@@ -21,6 +23,7 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
 
 import java.time.LocalDateTime;
 import java.util.ArrayList;
@@ -33,7 +36,7 @@ import java.util.Objects;
  * 观看进度。
  *
  * <p>每次上报都写库在高并发下会成为瓶颈（1 万并发 × 每 10 秒一次 ≈ 1000 写/秒），
- * 所以生产形态是「先写 Redis，后台任务按 user+drama 合并后批量 UPSERT」。
+ * 所以生产形态是「先写 Redis，后台任务按登录账号或设备 + drama 合并后批量 UPSERT」。
  * 这里先落地权威表与合并规则，缓存层在接入 Redis 时补齐，接口契约不变。
  *
  * <p>合并规则：以客户端上报的 positionMs 为准，但只在「更靠后」时覆盖，
@@ -53,6 +56,11 @@ public class WatchProgressService {
 
     @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());
         long position = req.getPositionMs() == null ? 0L : Math.max(0L, req.getPositionMs());
@@ -60,11 +68,11 @@ public class WatchProgressService {
         boolean finished = Boolean.TRUE.equals(req.getFinished())
                 || (duration > 0 && position + FINISH_THRESHOLD_MS >= duration);
 
-        LoadedProgress loaded = loadForSave(userId, drama.getId(), episode, position, duration,
+        LoadedProgress loaded = loadForSave(identity, drama.getId(), episode, position, duration,
                 finished, req);
         WatchProgress existing = loaded.progress();
         if (loaded.created()) {
-            appendHistory(userId, drama.getId(), episode.getId(), position, duration,
+            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
                     existing.getSource(), existing.getDeviceId());
             return toResp(progressMapper.selectById(existing.getId()));
         }
@@ -74,7 +82,7 @@ public class WatchProgressService {
                 && existing.getEpisodeId().equals(episode.getId());
         if (hidden && sameEpisode) {
             existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
-            appendHistory(userId, drama.getId(), episode.getId(), position, duration,
+            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
                     req.getSource(), existing.getDeviceId());
         }
         if (!sameEpisode) {
@@ -84,7 +92,7 @@ public class WatchProgressService {
             existing.setDurationMs(duration);
             existing.setFinished(finished ? 1 : 0);
             existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1);
-            appendHistory(userId, drama.getId(), episode.getId(), position, duration,
+            appendHistory(identity, drama.getId(), episode.getId(), position, duration,
                     req.getSource(), existing.getDeviceId());
         } else {
             existing.setDurationMs(Math.max(existing.getDurationMs() == null ? 0L : existing.getDurationMs(),
@@ -96,7 +104,9 @@ public class WatchProgressService {
                 existing.setFinished(1);
             }
         }
-        existing.setDeviceId(req.getDeviceId() == null ? "" : req.getDeviceId());
+        existing.setDeviceId(identity.authenticated()
+                ? (req.getDeviceId() == null ? "" : req.getDeviceId())
+                : identity.deviceId());
         existing.setSource(req.getSource() == null ? "" : req.getSource());
         if (hidden) {
             existing.setDeleted(0);
@@ -107,18 +117,22 @@ public class WatchProgressService {
         return toResp(progressMapper.selectById(existing.getId()));
     }
 
-    private LoadedProgress loadForSave(Long userId, Long dramaId, DramaEpisode episode, long position,
+    private LoadedProgress loadForSave(InteractionIdentity identity, Long dramaId,
+                                       DramaEpisode episode, long position,
                                        long duration, boolean finished, ProgressSaveReqVO req) {
-        WatchProgress existing = progressMapper.selectAny(userId, dramaId);
+        WatchProgress existing = progressMapper.selectAnyForIdentity(
+                identity.userId(), identity.deviceId(), dramaId);
         if (existing != null) {
             return new LoadedProgress(existing, false);
         }
-        WatchProgress created = newProgress(userId, dramaId, episode, position, duration, finished, req);
+        WatchProgress created = newProgress(
+                identity, dramaId, episode, position, duration, finished, req);
         try {
             progressMapper.insert(created);
             return new LoadedProgress(created, true);
         } catch (DuplicateKeyException ex) {
-            WatchProgress raced = progressMapper.selectAny(userId, dramaId);
+            WatchProgress raced = progressMapper.selectAnyForIdentity(
+                    identity.userId(), identity.deviceId(), dramaId);
             if (raced == null) {
                 throw ex;
             }
@@ -126,17 +140,20 @@ public class WatchProgressService {
         }
     }
 
-    private WatchProgress newProgress(Long userId, Long dramaId, DramaEpisode episode, long position,
+    private WatchProgress newProgress(InteractionIdentity identity, Long dramaId,
+                                      DramaEpisode episode, long position,
                                       long duration, boolean finished, ProgressSaveReqVO req) {
         WatchProgress progress = new WatchProgress();
-        progress.setUserId(userId);
+        progress.setUserId(identity.userId());
         progress.setDramaId(dramaId);
         progress.setEpisodeId(episode.getId());
         progress.setPositionMs(position);
         progress.setDurationMs(duration);
         progress.setFinished(finished ? 1 : 0);
         progress.setPlayCount(1);
-        progress.setDeviceId(req.getDeviceId() == null ? "" : req.getDeviceId());
+        progress.setDeviceId(identity.authenticated()
+                ? (req.getDeviceId() == null ? "" : req.getDeviceId())
+                : identity.deviceId());
         progress.setSource(req.getSource() == null ? "" : req.getSource());
         return progress;
     }
@@ -145,8 +162,11 @@ public class WatchProgressService {
     }
 
     public ProgressRespVO ofDrama(Long userId, Long dramaId) {
-        WatchProgress progress = progressMapper.selectOne(Wrappers.<WatchProgress>lambdaQuery()
-                .eq(WatchProgress::getUserId, userId)
+        return ofDrama(new InteractionIdentity(userId, ""), dramaId);
+    }
+
+    public ProgressRespVO ofDrama(InteractionIdentity identity, Long dramaId) {
+        WatchProgress progress = progressMapper.selectOne(identityQuery(identity)
                 .eq(WatchProgress::getDramaId, dramaId)
                 .last("limit 1"));
         return progress == null ? null : toResp(progress);
@@ -154,12 +174,16 @@ public class WatchProgressService {
 
     /** 分页查询最近观看，按更新时间倒序并以 id 兜底保证稳定。 */
     public PageResult<ProgressRespVO> recentPage(Long userId, int pageIndex, int pageSize) {
+        return recentPage(new InteractionIdentity(userId, ""), pageIndex, pageSize);
+    }
+
+    public PageResult<ProgressRespVO> recentPage(
+            InteractionIdentity identity, int pageIndex, int pageSize) {
         int safeIndex = Math.max(pageIndex, 1);
         int safeSize = Math.min(Math.max(pageSize, 1), MAX_PAGE_SIZE);
         Page<WatchProgress> request = new Page<>(safeIndex, safeSize);
         IPage<WatchProgress> result = progressMapper.selectPage(request,
-                Wrappers.<WatchProgress>lambdaQuery()
-                        .eq(WatchProgress::getUserId, userId)
+                identityQuery(identity)
                         .orderByDesc(WatchProgress::getUpdatedAt)
                         .orderByDesc(WatchProgress::getId));
         return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords()));
@@ -167,15 +191,23 @@ public class WatchProgressService {
 
     @Transactional(rollbackFor = Exception.class)
     public void remove(Long userId, Long dramaId) {
-        progressMapper.delete(Wrappers.<WatchProgress>lambdaQuery()
-                .eq(WatchProgress::getUserId, userId)
+        remove(new InteractionIdentity(userId, ""), dramaId);
+    }
+
+    @Transactional(rollbackFor = Exception.class)
+    public void remove(InteractionIdentity identity, Long dramaId) {
+        progressMapper.delete(identityQuery(identity)
                 .eq(WatchProgress::getDramaId, dramaId));
     }
 
     @Transactional(rollbackFor = Exception.class)
     public void clear(Long userId) {
-        progressMapper.delete(Wrappers.<WatchProgress>lambdaQuery()
-                .eq(WatchProgress::getUserId, userId));
+        clear(new InteractionIdentity(userId, ""));
+    }
+
+    @Transactional(rollbackFor = Exception.class)
+    public void clear(InteractionIdentity identity) {
+        progressMapper.delete(identityQuery(identity));
     }
 
     private ProgressRespVO toResp(WatchProgress progress) {
@@ -248,19 +280,169 @@ public class WatchProgressService {
      * 追加一条观看流水。只在「首次观看一部剧」和「切换剧集」时调用，
      * 每 10 秒的进度上报不写这里 —— 否则单集会产出几十条噪音。
      */
-    private void appendHistory(Long userId, Long dramaId, Long episodeId, long positionMs,
+    private void appendHistory(InteractionIdentity identity, Long dramaId, Long episodeId,
+                               long positionMs,
                                long durationMs, String source, String deviceId) {
         WatchHistory history = new WatchHistory();
-        history.setUserId(userId);
+        history.setUserId(identity.userId());
         history.setDramaId(dramaId);
         history.setEpisodeId(episodeId);
         history.setPositionMs(positionMs);
         history.setDurationMs(durationMs);
         history.setSource(source == null ? "" : source);
-        history.setDeviceId(deviceId == null ? "" : deviceId);
+        history.setDeviceId(identity.authenticated()
+                ? (deviceId == null ? "" : deviceId)
+                : identity.deviceId());
         historyMapper.insert(history);
     }
 
+    /**
+     * 登录时合并设备进度和观看流水。
+     *
+     * <p>同一部剧只保留进度更靠前的账号记录：先比较集号，同集再比较播放位置，
+     * 最后用更新时间兜底。playCount 只取保留记录，不把两条记录的计数相加。
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public MergeResult mergeDeviceIntoUser(Long userId, String deviceId) {
+        if (userId == null || !StringUtils.hasText(deviceId)) {
+            return new MergeResult(0, 0, 0, 0);
+        }
+        List<WatchProgress> deviceProgress = progressMapper.selectList(
+                Wrappers.<WatchProgress>lambdaQuery()
+                        .isNull(WatchProgress::getUserId)
+                        .eq(WatchProgress::getDeviceId, deviceId));
+        Map<Long, Integer> episodeNumbers = episodeNumbers(deviceProgress);
+        int progressMigrated = 0;
+        int progressIgnored = 0;
+        for (WatchProgress device : deviceProgress) {
+            WatchProgress account = progressMapper.selectAnyForIdentity(
+                    userId, "", device.getDramaId());
+            if (account == null) {
+                device.setUserId(userId);
+                progressMapper.updateById(device);
+                progressMigrated++;
+                continue;
+            }
+            if (progressAhead(device, account, episodeNumbers)) {
+                copyProgress(device, account);
+                account.setDeleted(0);
+                progressMapper.updateAny(account);
+                progressMigrated++;
+            } else {
+                progressIgnored++;
+            }
+            progressMapper.deleteById(device.getId());
+        }
+
+        List<WatchHistory> deviceHistory = historyMapper.selectList(
+                Wrappers.<WatchHistory>lambdaQuery()
+                        .isNull(WatchHistory::getUserId)
+                        .eq(WatchHistory::getDeviceId, deviceId));
+        int historyMigrated = 0;
+        int historyIgnored = 0;
+        for (WatchHistory device : deviceHistory) {
+            WatchHistory account = historyMapper.selectOne(
+                    Wrappers.<WatchHistory>lambdaQuery()
+                            .eq(WatchHistory::getUserId, userId)
+                            .eq(WatchHistory::getDramaId, device.getDramaId())
+                            .eq(WatchHistory::getEpisodeId, device.getEpisodeId())
+                            .orderByDesc(WatchHistory::getWatchedAt)
+                            .last("limit 1"));
+            if (account == null) {
+                device.setUserId(userId);
+                historyMapper.updateById(device);
+                historyMigrated++;
+                continue;
+            }
+            if (historyAhead(device, account)) {
+                account.setPositionMs(device.getPositionMs());
+                account.setDurationMs(device.getDurationMs());
+                account.setSource(device.getSource());
+                account.setWatchedAt(device.getWatchedAt());
+                historyMapper.updateById(account);
+                historyMigrated++;
+            } else {
+                historyIgnored++;
+            }
+            historyMapper.deleteById(device.getId());
+        }
+        return new MergeResult(
+                progressMigrated, progressIgnored, historyMigrated, historyIgnored);
+    }
+
+    public record MergeResult(
+            int progressMigrated,
+            int progressIgnored,
+            int historyMigrated,
+            int historyIgnored) {
+    }
+
+    private Map<Long, Integer> episodeNumbers(List<WatchProgress> progress) {
+        List<Long> episodeIds = progress.stream()
+                .map(WatchProgress::getEpisodeId)
+                .filter(Objects::nonNull)
+                .distinct()
+                .toList();
+        Map<Long, Integer> result = new HashMap<>();
+        if (episodeIds.isEmpty()) {
+            return result;
+        }
+        for (DramaEpisode episode : episodeService.listByIds(episodeIds)) {
+            result.put(episode.getId(), episode.getEpNo());
+        }
+        return result;
+    }
+
+    private boolean progressAhead(
+            WatchProgress candidate,
+            WatchProgress current,
+            Map<Long, Integer> episodeNumbers) {
+        Integer candidateNo = episodeNumbers.get(candidate.getEpisodeId());
+        Integer currentNo = episodeNumbers.get(current.getEpisodeId());
+        if (candidateNo != null && currentNo != null && !candidateNo.equals(currentNo)) {
+            return candidateNo > currentNo;
+        }
+        long candidatePosition = candidate.getPositionMs() == null ? 0L : candidate.getPositionMs();
+        long currentPosition = current.getPositionMs() == null ? 0L : current.getPositionMs();
+        if (candidatePosition != currentPosition) {
+            return candidatePosition > currentPosition;
+        }
+        return candidate.getUpdatedAt() != null
+                && (current.getUpdatedAt() == null
+                || candidate.getUpdatedAt().isAfter(current.getUpdatedAt()));
+    }
+
+    private void copyProgress(WatchProgress source, WatchProgress target) {
+        target.setEpisodeId(source.getEpisodeId());
+        target.setPositionMs(source.getPositionMs());
+        target.setDurationMs(source.getDurationMs());
+        target.setFinished(source.getFinished());
+        target.setPlayCount(source.getPlayCount());
+        target.setDeviceId(source.getDeviceId());
+        target.setSource(source.getSource());
+    }
+
+    private boolean historyAhead(WatchHistory candidate, WatchHistory current) {
+        long candidatePosition = candidate.getPositionMs() == null ? 0L : candidate.getPositionMs();
+        long currentPosition = current.getPositionMs() == null ? 0L : current.getPositionMs();
+        if (candidatePosition != currentPosition) {
+            return candidatePosition > currentPosition;
+        }
+        return candidate.getWatchedAt() != null
+                && (current.getWatchedAt() == null
+                || candidate.getWatchedAt().isAfter(current.getWatchedAt()));
+    }
+
+    private LambdaQueryWrapper<WatchProgress> identityQuery(InteractionIdentity identity) {
+        if (identity.authenticated()) {
+            return Wrappers.<WatchProgress>lambdaQuery()
+                    .eq(WatchProgress::getUserId, identity.userId());
+        }
+        return Wrappers.<WatchProgress>lambdaQuery()
+                .isNull(WatchProgress::getUserId)
+                .eq(WatchProgress::getDeviceId, identity.deviceId());
+    }
+
     /** 供后续批量合并任务使用：只更新进度字段，避免整行覆盖。 */
     public void mergeInto(Long userId, Long dramaId, Long episodeId, long positionMs) {
         progressMapper.update(null, Wrappers.<WatchProgress>lambdaUpdate()