diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java
index 3dc36c2..3cf622b 100644
--- a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java
+++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java
@@ -93,6 +93,34 @@ public class WatchProgressBuffer {
}
}
+ /**
+ * 删除某部剧的缓冲状态,用于用户删除单条观看记录。
+ *
+ *
缓冲是读接口的权威来源,删库行必须同步删缓冲,否则会被读覆盖层读回;
+ * 同时缓冲状态消失后,dirty 队列里该条引用在落库时会被丢弃(见
+ * {@code WatchProgressService#flushRefs}),不会再被写回库。
+ */
+ public boolean remove(InteractionIdentity identity, long dramaId) {
+ try {
+ redis.opsForHash().delete(BUF_KEY_PREFIX + identityKey(identity), Long.toString(dramaId));
+ return true;
+ } catch (Exception e) {
+ log.warn("观看进度缓冲删除失败: {}", e.getMessage());
+ return false;
+ }
+ }
+
+ /** 清空某身份的全部缓冲状态,用于用户清空观看历史。 */
+ public boolean clear(InteractionIdentity identity) {
+ try {
+ redis.delete(BUF_KEY_PREFIX + identityKey(identity));
+ return true;
+ } catch (Exception e) {
+ log.warn("观看进度缓冲清空失败: {}", e.getMessage());
+ return false;
+ }
+ }
+
/** 读单个剧的缓冲状态;无缓冲或解码失败返回 null。 */
public BufferedProgress get(InteractionIdentity identity, long dramaId) {
return get(identityKey(identity), dramaId);
diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java
index 27f486c..cd4246b 100644
--- a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java
+++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java
@@ -279,14 +279,18 @@ public class WatchProgressService {
List extraDramaIds = extra.stream()
.map(WatchProgressBuffer.BufferedProgress::dramaId)
.toList();
- Set withRow = progressMapper.selectAnyForIdentityIn(
+ Map rowByDrama = progressMapper.selectAnyForIdentityIn(
identity.userId(), identity.deviceId(), extraDramaIds)
.stream()
- .map(WatchProgress::getDramaId)
- .collect(Collectors.toSet());
+ .collect(Collectors.toMap(WatchProgress::getDramaId, row -> row, (a, b) -> a));
for (WatchProgressBuffer.BufferedProgress buffered : extra) {
+ WatchProgress existing = rowByDrama.get(buffered.dramaId());
+ // 用户已删除该剧(库里是逻辑删除行):缓冲状态不能被读覆盖层再读回来
+ if (existing != null && existing.getDeleted() != null && existing.getDeleted() != 0) {
+ continue;
+ }
merged.add(transientRow(buffered));
- if (!withRow.contains(buffered.dramaId())) {
+ if (existing == null) {
total++;
}
}
@@ -306,6 +310,12 @@ public class WatchProgressService {
@Transactional(rollbackFor = Exception.class)
public void remove(InteractionIdentity identity, Long dramaId) {
+ // 缓冲是读接口的权威来源:先删缓冲再删库行,否则删掉的记录会被读覆盖层读回、
+ // 并被下一次落库把行复活。
+ WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+ if (buffer != null && dramaId != null) {
+ buffer.remove(identity, dramaId);
+ }
progressMapper.delete(identityQuery(identity)
.eq(WatchProgress::getDramaId, dramaId));
}
@@ -317,6 +327,10 @@ public class WatchProgressService {
@Transactional(rollbackFor = Exception.class)
public void clear(InteractionIdentity identity) {
+ WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
+ if (buffer != null) {
+ buffer.clear(identity);
+ }
progressMapper.delete(identityQuery(identity));
}
@@ -338,8 +352,7 @@ public class WatchProgressService {
return 0;
}
try {
- flushRefs(buffer, refs);
- return refs.size();
+ return flushRefs(buffer, refs);
} catch (Exception ex) {
log.warn("观看进度缓冲合并失败,整批退回队列稍后重试: {}", ex.getMessage());
buffer.requeue(refs);
@@ -347,7 +360,7 @@ public class WatchProgressService {
}
}
- private void flushRefs(WatchProgressBuffer buffer, List refs) {
+ private int flushRefs(WatchProgressBuffer buffer, List refs) {
// 1) pipeline 批量读缓冲状态,key -> (dramaId -> state)
List identityKeys = refs.stream()
.map(WatchProgressBuffer.IdentityRef::key)
@@ -372,7 +385,11 @@ public class WatchProgressService {
ref.userId(), ref.deviceId(), ref.dramaId()));
}
if (pendingRefs.isEmpty()) {
- throw new IllegalStateException("缓冲状态读取为空,ref=" + refs.size());
+ // 缓冲状态已消失(用户清空/删除历史,或 HASH 12h TTL 到期)。此时 ref 已无
+ // 可落库的内容,必须丢弃:退回队列会在下个周期被原样弹出、再清空、再退回,
+ // 这批僵尸条目将永久占住 dirty 队列配额。
+ log.debug("观看进度缓冲条目已无对应状态,丢弃 {} 条", refs.size());
+ return 0;
}
// 3) 分块成对查询已有行(每块 100 组 OR 索引条件)
@@ -415,6 +432,7 @@ public class WatchProgressService {
}
session.commit();
}
+ return pendingRefs.size();
}
/** 计算单条缓冲的合并动作:只改内存对象不触库,产出归入三组待执行列表。 */