fix(live): index mic rooms for reconciliation

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java

@@ -16,10 +16,9 @@ import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.UUID;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -31,11 +30,20 @@ import org.springframework.stereotype.Component;
 @RequiredArgsConstructor
 public class AgoraMicStateReconcileTask {
 
-  private static final String LOCK_KEY = "agora:mic:reconcile:lock";
-  private static final String SEAT_KEY_PREFIX = "LiveMick:SEAT:";
+  /**
+   * v2 锁与旧版整数值/无 token 锁隔离。滚动发布期间旧实例仍会无条件 DEL 旧 key，
+   * 若复用同名锁就可能误删新实例的 UUID 锁，破坏 compare-delete 的安全保证。
+   */
+  private static final String LOCK_KEY = "agora:mic:reconcile:lock:v2";
+  private static final long LOCK_TTL_MINUTES = 5L;
+  private static final String RELEASE_LOCK_SCRIPT = """
+      if redis.call('get', KEYS[1]) == ARGV[1] then
+        return redis.call('del', KEYS[1])
+      end
+      return 0
+      """;
 
   private final RedisService redisService;
-  private final RedisTemplate<String, Object> redisTemplate;
   private final LiveMicCacheService liveMicCacheService;
   private final AgoraRoomStateService agoraRoomStateService;
   private final AgoraMicMissingConfirmService agoraMicMissingConfirmService;
@@ -43,15 +51,19 @@ public class AgoraMicStateReconcileTask {
 
   @Scheduled(fixedDelayString = "${red-circle.live.agora-mic-reconcile-delay-ms:30000}")
   public void reconcileAgoraMicState() {
-    if (!redisService.setIfAbsent(LOCK_KEY, 1, 25, TimeUnit.SECONDS)) {
+    String lockToken = UUID.randomUUID().toString();
+    // 两台 live 共用同一轮对账。TTL 覆盖正常有界房间集合的完整处理窗口；异常退出仍可自动恢复。
+    if (!redisService.setIfAbsent(LOCK_KEY, lockToken, LOCK_TTL_MINUTES, TimeUnit.MINUTES)) {
       return;
     }
     try {
-      for (Long roomId : scanSeatRoomIds()) {
+      // 房间候选只来自麦位写链路维护的 Set。禁止为滚动升级或历史补数重新引入全库 SCAN；
+      // 老版本活跃麦位会在下一次心跳 goUp 时自动 SADD，未再心跳的旧 Hash 已有 120 秒 TTL 自然消失。
+      for (Long roomId : liveMicCacheService.listReconcileRoomIds()) {
         reconcileRoom(roomId);
       }
     } finally {
-      redisService.delete(LOCK_KEY);
+      releaseLock(lockToken);
     }
   }
 
@@ -59,6 +71,13 @@ public class AgoraMicStateReconcileTask {
     if (Objects.isNull(roomId)) {
       return;
     }
+
+    List<LiveMicrophone> microphones = liveMicCacheService.listLiveMicrophone(roomId);
+    if (CollectionUtils.isEmpty(microphones)) {
+      // Hash TTL 到期和异常中断都可能留下 Set 成员；Lua 会在删除前再次 HLEN，避免误删并发 goUp。
+      liveMicCacheService.pruneReconcileRoomIndexIfEmpty(roomId);
+      return;
+    }
     if (!shouldReconcileAgoraRoom(roomId)) {
       log.debug("声网麦位对账跳过非 Agora 房间 roomId={}", roomId);
       return;
@@ -70,11 +89,6 @@ public class AgoraMicStateReconcileTask {
       return;
     }
 
-    List<LiveMicrophone> microphones = liveMicCacheService.listLiveMicrophone(roomId);
-    if (CollectionUtils.isEmpty(microphones)) {
-      return;
-    }
-
     if (!state.channelExist()) {
       for (LiveMicrophone microphone : microphones) {
         cleanupMissingMicrophoneUser(roomId, microphone, state, "reconcile agora channel not exist");
@@ -130,27 +144,21 @@ public class AgoraMicStateReconcileTask {
     agoraMicStateCleanupService.cleanupUser(roomId, userId, reason);
   }
 
-  private Set<Long> scanSeatRoomIds() {
-    Set<Long> roomIds = new HashSet<>();
-    ScanOptions options = ScanOptions.scanOptions().match(SEAT_KEY_PREFIX + "*").count(200).build();
-    try (var cursor = redisTemplate.scan(options)) {
-      while (cursor.hasNext()) {
-        Object key = cursor.next();
-        Long roomId = parseRoomId(Objects.toString(key, null));
-        if (Objects.nonNull(roomId)) {
-          roomIds.add(roomId);
-        }
+  private void releaseLock(String lockToken) {
+    try {
+      Long removed = redisService.execute(
+          RELEASE_LOCK_SCRIPT,
+          Long.class,
+          List.of(LOCK_KEY),
+          lockToken
+      );
+      if (!Objects.equals(removed, 1L)) {
+        // TTL 到期后锁可能已由下一实例接管；compare-delete 返回 0 时绝不能删除新持有者的锁。
+        log.warn("声网麦位对账锁未由当前实例释放，等待持有者或 TTL 收敛");
       }
-    } catch (Exception e) {
-      log.error("扫描声网麦位对账 Redis key 失败", e);
+    } catch (RuntimeException e) {
+      // 对账是补偿任务；释放异常不应触发无 token DEL，保留锁直到 TTL 是更安全的失败方式。
+      log.error("释放声网麦位对账锁失败，等待 TTL 自动释放", e);
     }
-    return roomIds;
-  }
-
-  private Long parseRoomId(String key) {
-    if (Objects.isNull(key) || !key.startsWith(SEAT_KEY_PREFIX)) {
-      return null;
-    }
-    return DataTypeUtils.toLong(key.substring(SEAT_KEY_PREFIX.length()));
   }
 }

rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskIndexTest.java

@@ -0,0 +1,182 @@
+package com.red.circle.live.app.scheduler;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import com.red.circle.component.redis.service.RedisService;
+import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.junit.Test;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+import org.springframework.data.redis.core.script.RedisScript;
+
+public class AgoraMicStateReconcileTaskIndexTest {
+
+  private static final String LOCK_KEY = "agora:mic:reconcile:lock:v2";
+
+  @Test
+  public void schedulerShouldUseBoundedRoomIndexPruneEmptyHashAndReleaseOwnedLock() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(true, 1L);
+    RecordingMicCache micCache = new RecordingMicCache(Set.of(1001L));
+    AgoraMicStateReconcileTask task = task(redisTemplate, micCache.proxy());
+
+    task.reconcileAgoraMicState();
+
+    assertEquals(List.of(1001L), micCache.listedRoomIds);
+    assertEquals(List.of(1001L), micCache.prunedRoomIds);
+    assertEquals(LOCK_KEY, redisTemplate.lockKey);
+    assertNotNull(redisTemplate.lockToken);
+    assertEquals(5L, redisTemplate.lockTtl);
+    assertEquals(TimeUnit.MINUTES, redisTemplate.lockTtlUnit);
+    assertEquals(redisTemplate.lockToken, redisTemplate.releaseToken);
+    assertEquals(List.of(LOCK_KEY), redisTemplate.releaseKeys);
+    assertTrue(redisTemplate.releaseScript.contains("get"));
+    assertTrue(redisTemplate.releaseScript.contains("del"));
+    assertFalse(redisTemplate.directDeleteCalled);
+  }
+
+  @Test
+  public void schedulerShouldNotReadIndexOrReleaseLockWhenAnotherInstanceOwnsIt() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(false, 0L);
+    RecordingMicCache micCache = new RecordingMicCache(Set.of(1001L));
+    AgoraMicStateReconcileTask task = task(redisTemplate, micCache.proxy());
+
+    task.reconcileAgoraMicState();
+
+    assertTrue(micCache.listedRoomIds.isEmpty());
+    assertTrue(micCache.prunedRoomIds.isEmpty());
+    assertEquals("", redisTemplate.releaseScript);
+    assertFalse(redisTemplate.directDeleteCalled);
+  }
+
+  @Test
+  public void tokenMismatchShouldNeverFallBackToUnsafeDirectDelete() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(true, 0L);
+    RecordingMicCache micCache = new RecordingMicCache(Collections.emptySet());
+
+    task(redisTemplate, micCache.proxy()).reconcileAgoraMicState();
+
+    assertNotNull(redisTemplate.releaseToken);
+    assertFalse(redisTemplate.directDeleteCalled);
+  }
+
+  private static AgoraMicStateReconcileTask task(
+      RecordingRedisTemplate redisTemplate,
+      LiveMicCacheService micCache) {
+    return new AgoraMicStateReconcileTask(
+        new RedisService(redisTemplate),
+        micCache,
+        null,
+        null,
+        null
+    );
+  }
+
+  /**
+   * 仅实现调度器需要的两个索引动作。返回空麦位可验证残留 Set 成员会走原子 prune，
+   * 且测试类完全不提供 Redis SCAN 能力。
+   */
+  private static final class RecordingMicCache {
+
+    private final Set<Long> roomIds;
+    private final List<Long> listedRoomIds = new ArrayList<>();
+    private final List<Long> prunedRoomIds = new ArrayList<>();
+
+    private RecordingMicCache(Set<Long> roomIds) {
+      this.roomIds = new LinkedHashSet<>(roomIds);
+    }
+
+    private LiveMicCacheService proxy() {
+      return (LiveMicCacheService) Proxy.newProxyInstance(
+          LiveMicCacheService.class.getClassLoader(),
+          new Class<?>[] {LiveMicCacheService.class},
+          (proxy, method, args) -> switch (method.getName()) {
+            case "listReconcileRoomIds" -> new LinkedHashSet<>(roomIds);
+            case "listLiveMicrophone" -> {
+              listedRoomIds.add((Long) args[0]);
+              yield Collections.emptyList();
+            }
+            case "pruneReconcileRoomIndexIfEmpty" -> {
+              prunedRoomIds.add((Long) args[0]);
+              yield true;
+            }
+            case "toString" -> "RecordingMicCache";
+            case "hashCode" -> System.identityHashCode(proxy);
+            case "equals" -> proxy == args[0];
+            default -> throw new UnsupportedOperationException(method.getName());
+          }
+      );
+    }
+  }
+
+  private static final class RecordingRedisTemplate extends RedisTemplate<String, Object> {
+
+    private final boolean acquireLock;
+    private final Long releaseResult;
+    private String lockKey;
+    private Object lockToken;
+    private long lockTtl;
+    private TimeUnit lockTtlUnit;
+    private String releaseScript = "";
+    private List<String> releaseKeys = List.of();
+    private Object releaseToken;
+    private boolean directDeleteCalled;
+
+    private RecordingRedisTemplate(boolean acquireLock, Long releaseResult) {
+      this.acquireLock = acquireLock;
+      this.releaseResult = releaseResult;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public ValueOperations<String, Object> opsForValue() {
+      return (ValueOperations<String, Object>) Proxy.newProxyInstance(
+          ValueOperations.class.getClassLoader(),
+          new Class<?>[] {ValueOperations.class},
+          (proxy, method, args) -> {
+            if (Objects.equals("setIfAbsent", method.getName()) && args.length == 4) {
+              lockKey = Objects.toString(args[0]);
+              lockToken = args[1];
+              lockTtl = ((Number) args[2]).longValue();
+              lockTtlUnit = (TimeUnit) args[3];
+              return acquireLock;
+            }
+            if (method.getDeclaringClass().equals(Object.class)) {
+              return switch (method.getName()) {
+                case "toString" -> "RecordingValueOperations";
+                case "hashCode" -> System.identityHashCode(proxy);
+                case "equals" -> proxy == args[0];
+                default -> throw new UnsupportedOperationException(method.getName());
+              };
+            }
+            throw new UnsupportedOperationException(method.getName());
+          }
+      );
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <T> T execute(RedisScript<T> script, List<String> keys, Object... args) {
+      releaseScript = script.getScriptAsString();
+      releaseKeys = List.copyOf(keys);
+      releaseToken = args[0];
+      return (T) releaseResult;
+    }
+
+    @Override
+    public Boolean delete(String key) {
+      directDeleteCalled = true;
+      return Boolean.TRUE;
+    }
+  }
+}

rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java

@@ -32,7 +32,6 @@ public class AgoraMicStateReconcileTaskRtcProviderTest {
         null,
         null,
         null,
-        null,
         null);
   }
 

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/key/LiveMicKey.java

@@ -14,10 +14,18 @@ public enum LiveMicKey implements RedisKeys {
    */
   LOCK,
 
-  /**
-   * 位置.
-   */
-  SEAT;
+  /**
+   * 位置.
+   */
+  SEAT,
+
+  /**
+   * 当前存在麦位数据、需要参与 RTC 对账的房间集合.
+   *
+   * <p>它是 {@link #SEAT} Hash 的有界辅助索引，不是第二份麦位状态。成员只保存 roomId，
+   * 对账时仍以对应 SEAT Hash 为事实；索引残留会由原子 compare-prune 收敛。
+   */
+  SEAT_ROOM_INDEX;
 
 
   @Override

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/LiveMicCacheService.java

@@ -1,8 +1,9 @@
 package com.red.circle.live.infra.database.cache.service;
 
 import com.red.circle.live.domain.live.LiveMicrophone;
-import com.red.circle.tool.core.func.VoidConsumer;
-import java.util.List;
+import com.red.circle.tool.core.func.VoidConsumer;
+import java.util.List;
+import java.util.Set;
 
 /**
  * 麦克风用户缓存服务.
@@ -60,7 +61,24 @@ public interface LiveMicCacheService {
   /**
    * 下掉所有.
    */
-  void goDownAll(Long roomId);
+  void goDownAll(Long roomId);
+
+  /**
+   * 返回当前需要参与 RTC 对账的房间 id.
+   *
+   * <p>这里只读取 LiveMicCacheService 在写麦位时维护的有界 Set，禁止通过
+   * {@code SCAN LiveMick:SEAT:*} 从共享 Redis DB 反查房间。
+   */
+  Set<Long> listReconcileRoomIds();
+
+  /**
+   * 仅当房间 SEAT Hash 已为空时从对账索引移除 roomId.
+   *
+   * <p>判断和删除必须在一条 Lua 内完成，避免并发 goUp 已写入新麦位后被旧清理误删索引。
+   *
+   * @return true 表示本次确实移除了残留索引成员
+   */
+  boolean pruneReconcileRoomIndexIfEmpty(Long roomId);
 
   /**
    * 验证是否是自己在麦克风.

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImpl.java

@@ -5,11 +5,13 @@ import com.red.circle.component.redis.service.RedisService;
 import com.red.circle.live.domain.live.LiveMicrophone;
 import com.red.circle.live.infra.database.cache.key.LiveMicKey;
 import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
-import com.red.circle.tool.core.collection.CollectionUtils;
-import com.red.circle.tool.core.func.VoidConsumer;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
+import com.red.circle.tool.core.collection.CollectionUtils;
+import com.red.circle.tool.core.func.VoidConsumer;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -22,9 +24,22 @@ import org.springframework.stereotype.Service;
 @Slf4j
 @Service
 @RequiredArgsConstructor
-public class LiveMicCacheServiceImpl implements LiveMicCacheService {
-
-  private final RedisService redisService;
+public class LiveMicCacheServiceImpl implements LiveMicCacheService {
+
+  /**
+   * 删除索引前在 Redis 内重新检查 SEAT Hash，保证判断与 SREM 不留并发窗口。
+   *
+   * <p>goUp 使用 HSET -> SADD。若 goUp 先于本脚本执行，HLEN 会阻止误删；若 goUp 后执行，
+   * 它最终的 SADD 会恢复索引。因此无论命令如何串行化，都不会留下“有麦位但无索引”的稳态。
+   */
+  private static final String PRUNE_ROOM_INDEX_IF_EMPTY_SCRIPT = """
+      if redis.call('hlen', KEYS[1]) == 0 then
+        return redis.call('srem', KEYS[2], ARGV[1])
+      end
+      return 0
+      """;
+
+  private final RedisService redisService;
 
   @Override
   public boolean checkLockMic(Long roomId, Integer index) {
@@ -67,23 +82,70 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService {
     redisService.hashPut(LiveMicKey.SEAT.getKey(roomId),
         Objects.toString(index),
         microphone);
-  }
+    // HSET 必须先于 SADD：并发下麦的 compare-prune 与 HSET 在 Redis 内有确定顺序，
+    // 随后的幂等 SADD 会保证任何成功写入的麦位最终都可被定时对账发现。
+    addReconcileRoomIndex(roomId);
+  }
 
   @Override
-  public void goDown(Long roomId, Integer index) {
-    redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), Objects.toString(index));
-  }
+  public void goDown(Long roomId, Integer index) {
+    redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), Objects.toString(index));
+    pruneReconcileRoomIndexIfEmpty(roomId);
+  }
 
   @Override
-  public void goDown(Long roomId, List<Integer> index) {
-    redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId),
-        index.stream().map(Objects::toString).toArray(String[]::new));
-  }
+  public void goDown(Long roomId, List<Integer> index) {
+    redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId),
+        index.stream().map(Objects::toString).toArray(String[]::new));
+    pruneReconcileRoomIndexIfEmpty(roomId);
+  }
 
   @Override
-  public void goDownAll(Long roomId) {
-    redisService.delete(LiveMicKey.SEAT.getKey(roomId));
-  }
+  public void goDownAll(Long roomId) {
+    redisService.delete(LiveMicKey.SEAT.getKey(roomId));
+    // 即使与新 goUp 并发，也必须复用 HLEN compare-prune，不能无条件 SREM。
+    pruneReconcileRoomIndexIfEmpty(roomId);
+  }
+
+  @Override
+  public Set<Long> listReconcileRoomIds() {
+    Set<Object> members = redisService.setGet(reconcileRoomIndexKey());
+    if (CollectionUtils.isEmpty(members)) {
+      return new LinkedHashSet<>();
+    }
+
+    Set<Long> roomIds = new LinkedHashSet<>();
+    for (Object member : members) {
+      Long roomId = parseRoomId(member);
+      if (Objects.isNull(roomId)) {
+        // 只有合法 roomId 才能定位 SEAT Hash；损坏成员无法 compare-prune，直接从辅助索引剔除。
+        redisService.setRemove(reconcileRoomIndexKey(), member);
+        continue;
+      }
+      roomIds.add(roomId);
+    }
+    return roomIds;
+  }
+
+  @Override
+  public boolean pruneReconcileRoomIndexIfEmpty(Long roomId) {
+    if (Objects.isNull(roomId)) {
+      return false;
+    }
+    try {
+      Long removed = redisService.execute(
+          PRUNE_ROOM_INDEX_IF_EMPTY_SCRIPT,
+          Long.class,
+          List.of(LiveMicKey.SEAT.getKey(roomId), reconcileRoomIndexKey()),
+          roomId.toString()
+      );
+      return Objects.nonNull(removed) && removed > 0L;
+    } catch (RuntimeException e) {
+      // 麦位 Hash 已先完成主操作；辅助索引失败只能造成可重试的残留，不能让客户端误以为上下麦失败。
+      log.error("原子清理麦位房间索引失败, roomId={}", roomId, e);
+      return false;
+    }
+  }
 
   @Override
   public boolean checkEqSelf(Long roomId, Integer index, Long userId) {
@@ -130,6 +192,8 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService {
 
     List<LiveMicrophone> liveMicrophone = listLiveMicrophone(roomId);
     if (CollectionUtils.isEmpty(liveMicrophone)) {
+      // Hash 可能已经因 120 秒 TTL 自然过期；这里同步收敛 Set 中的残留成员。
+      pruneReconcileRoomIndexIfEmpty(roomId);
       return CollectionUtils.newArrayList();
     }
 
@@ -148,6 +212,9 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService {
     Long size = redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), index);
     log.info("清除不活跃的用户结果 = {} ", size);
 
+    // 删除最后一个不活跃麦位时立即移除索引；若并发上麦，Lua 的 HLEN 会保留新房间成员。
+    pruneReconcileRoomIndexIfEmpty(roomId);
+
     if (Objects.isNull(size) || size <= 0) {
       return CollectionUtils.newArrayList();
     }
@@ -155,8 +222,34 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService {
   }
 
 
-  private void setExpire(Long roomId) {
-    redisService.expire(LiveMicKey.SEAT.getKey(roomId), 120, TimeUnit.SECONDS);
-  }
-
-}
+  private void setExpire(Long roomId) {
+    redisService.expire(LiveMicKey.SEAT.getKey(roomId), 120, TimeUnit.SECONDS);
+  }
+
+  private String reconcileRoomIndexKey() {
+    return LiveMicKey.SEAT_ROOM_INDEX.getKey();
+  }
+
+  private void addReconcileRoomIndex(Long roomId) {
+    try {
+      // Set 成员和 Lua ARGV 统一使用十进制字符串，避免不同 Redis serializer 对 Long/String
+      // 产生不同字节，导致 compare-prune 看似成功却删不到实际成员。
+      redisService.setAdd(reconcileRoomIndexKey(), roomId.toString());
+    } catch (RuntimeException e) {
+      // HSET 已经成功时不回滚真实麦位；活跃用户下一次心跳 goUp 会继续幂等补写索引。
+      log.error("维护麦位房间索引失败，等待心跳重试, roomId={}", roomId, e);
+    }
+  }
+
+  private Long parseRoomId(Object member) {
+    if (Objects.isNull(member)) {
+      return null;
+    }
+    try {
+      return Long.valueOf(member.toString());
+    } catch (NumberFormatException ignored) {
+      return null;
+    }
+  }
+
+}

rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImplTest.java

@@ -0,0 +1,323 @@
+package com.red.circle.live.infra.database.cache.service.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.red.circle.component.redis.service.RedisService;
+import com.red.circle.live.domain.live.LiveMicrophone;
+import com.red.circle.live.infra.database.cache.key.LiveMicKey;
+import com.red.circle.tool.core.json.JacksonUtils;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.junit.Test;
+import org.springframework.data.redis.core.HashOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.SetOperations;
+import org.springframework.data.redis.core.script.RedisScript;
+
+public class LiveMicCacheServiceImplTest {
+
+  private static final Long ROOM_ID = 1001L;
+  private static final String SEAT_KEY = "LiveMick:SEAT:1001";
+  private static final String ROOM_INDEX_KEY = "LiveMick:SEAT_ROOM_INDEX";
+
+  @Test
+  public void goUpShouldWriteSeatBeforeIndexAndHeartbeatShouldBackfillLegacyRoom() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+
+    // 模拟滚动升级前已存在、但尚未进入新 Set 的麦位 Hash。下一次正常心跳仍会调用 goUp。
+    redisTemplate.seedSeat(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+    assertTrue(redisTemplate.indexMembers().isEmpty());
+
+    service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+
+    assertEquals(List.of("HSET " + SEAT_KEY + " 0", "SADD " + ROOM_INDEX_KEY + " 1001"),
+        redisTemplate.events.subList(0, 2));
+    assertEquals(Set.of(ROOM_ID.toString()), redisTemplate.indexMembers());
+    assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds());
+  }
+
+  @Test
+  public void goDownShouldKeepIndexUntilLastSeatIsRemoved() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+    service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+    service.goUp(ROOM_ID, 1, microphone(1, System.currentTimeMillis()));
+
+    service.goDown(ROOM_ID, 0);
+    assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds());
+
+    service.goDown(ROOM_ID, 1);
+    assertTrue(service.listReconcileRoomIds().isEmpty());
+    assertEquals(0L, redisTemplate.hashSize(SEAT_KEY));
+  }
+
+  @Test
+  public void comparePruneShouldNotRemoveIndexWhenConcurrentGoUpWritesNewSeat() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+    service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+
+    // 精确模拟 HDEL 已完成、Lua 尚未执行时另一个请求上麦。Lua 内的 HLEN 必须看到新 Hash。
+    redisTemplate.beforeNextScript = () ->
+        service.goUp(ROOM_ID, 1, microphone(1, System.currentTimeMillis()));
+    service.goDown(ROOM_ID, 0);
+
+    assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds());
+    assertEquals(1L, redisTemplate.hashSize(SEAT_KEY));
+    assertTrue(redisTemplate.scriptCalls.get(0).script().contains("hlen"));
+    assertTrue(redisTemplate.scriptCalls.get(0).script().contains("srem"));
+    assertEquals(List.of(SEAT_KEY, ROOM_INDEX_KEY), redisTemplate.scriptCalls.get(0).keys());
+    assertEquals(List.of(ROOM_ID.toString()), redisTemplate.scriptCalls.get(0).args());
+  }
+
+  @Test
+  public void batchAndAllGoDownShouldPruneIndexOnlyThroughCompareScript() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+    service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+    service.goUp(ROOM_ID, 1, microphone(1, System.currentTimeMillis()));
+
+    service.goDown(ROOM_ID, List.of(0, 1));
+    assertTrue(service.listReconcileRoomIds().isEmpty());
+
+    service.goUp(ROOM_ID, 2, microphone(2, System.currentTimeMillis()));
+    service.goDownAll(ROOM_ID);
+    assertTrue(service.listReconcileRoomIds().isEmpty());
+    assertEquals(2, redisTemplate.scriptCalls.size());
+  }
+
+  @Test
+  public void inactiveCleanupShouldKeepLegacyTtlAndRemoveLastRoomIndex() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+    long expiredRefreshTime = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(121L);
+    service.goUp(ROOM_ID, 0, microphone(0, expiredRefreshTime));
+
+    List<LiveMicrophone> removed = service.refreshNotActiveUserAndReturn(ROOM_ID);
+
+    assertEquals(1, removed.size());
+    assertEquals(List.of(new ExpireCall(SEAT_KEY, 120L, TimeUnit.SECONDS)),
+        redisTemplate.expireCalls);
+    assertTrue(service.listReconcileRoomIds().isEmpty());
+  }
+
+  @Test
+  public void listReconcileRoomIdsShouldDiscardMalformedMembersWithoutScanningKeyspace() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+    redisTemplate.setMembers.computeIfAbsent(ROOM_INDEX_KEY, ignored -> new LinkedHashSet<>())
+        .addAll(List.of("bad-room-id", ROOM_ID.toString()));
+
+    assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds());
+    assertEquals(Set.of(ROOM_ID.toString()), redisTemplate.indexMembers());
+    // RecordingRedisTemplate 只实现 Hash/Set/expire/script；若生产代码重新引入 SCAN，测试会直接失败。
+    assertFalse(redisTemplate.events.stream().anyMatch(event -> event.startsWith("SCAN")));
+  }
+
+  @Test
+  public void auxiliaryIndexFailureShouldNotReverseCompletedSeatMutation() {
+    RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
+    LiveMicCacheServiceImpl service = service(redisTemplate);
+    redisTemplate.setFailure = new IllegalStateException("set unavailable");
+
+    service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+    assertEquals(1L, redisTemplate.hashSize(SEAT_KEY));
+
+    redisTemplate.setFailure = null;
+    service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis()));
+    redisTemplate.scriptFailure = new IllegalStateException("script unavailable");
+    service.goDown(ROOM_ID, 0);
+    assertEquals(0L, redisTemplate.hashSize(SEAT_KEY));
+  }
+
+  private static LiveMicCacheServiceImpl service(RecordingRedisTemplate redisTemplate) {
+    return new LiveMicCacheServiceImpl(new RedisService(redisTemplate));
+  }
+
+  private static LiveMicrophone microphone(int index, long refreshTime) {
+    return new LiveMicrophone().setMicIndex(index).setRefreshTime(refreshTime);
+  }
+
+  private record ExpireCall(String key, long ttl, TimeUnit unit) {
+  }
+
+  private record ScriptCall(String script, List<String> keys, List<Object> args) {
+  }
+
+  /**
+   * 用最小内存模型执行 RedisService 会使用的命令，并真实模拟 compare-prune Lua 的原子结果。
+   * 未实现的命令一律抛错，防止测试悄悄放过新的全库遍历或非预期读写。
+   */
+  private static final class RecordingRedisTemplate extends RedisTemplate<String, Object> {
+
+    private final Map<String, LinkedHashMap<Object, Object>> hashes = new LinkedHashMap<>();
+    private final Map<String, LinkedHashSet<Object>> setMembers = new LinkedHashMap<>();
+    private final List<String> events = new ArrayList<>();
+    private final List<ExpireCall> expireCalls = new ArrayList<>();
+    private final List<ScriptCall> scriptCalls = new ArrayList<>();
+    private Runnable beforeNextScript;
+    private RuntimeException setFailure;
+    private RuntimeException scriptFailure;
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <HK, HV> HashOperations<String, HK, HV> opsForHash() {
+      return (HashOperations<String, HK, HV>) proxy(HashOperations.class, (method, args) -> {
+        String key = Objects.toString(args[0]);
+        LinkedHashMap<Object, Object> hash = hashes.computeIfAbsent(key,
+            ignored -> new LinkedHashMap<>());
+        return switch (method) {
+          case "put" -> {
+            hash.put(args[1], args[2]);
+            events.add("HSET " + key + " " + args[1]);
+            yield null;
+          }
+          case "delete" -> {
+            long removed = 0L;
+            Object[] fields = (Object[]) args[1];
+            for (Object field : fields) {
+              if (hash.remove(field) != null) {
+                removed++;
+              }
+            }
+            events.add("HDEL " + key);
+            yield removed;
+          }
+          case "size" -> (long) hash.size();
+          case "values" -> new ArrayList<>(hash.values());
+          case "get" -> hash.get(args[1]);
+          case "hasKey" -> hash.containsKey(args[1]);
+          default -> throw new UnsupportedOperationException(method);
+        };
+      });
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public SetOperations<String, Object> opsForSet() {
+      return proxy(SetOperations.class, (method, args) -> {
+        String key = Objects.toString(args[0]);
+        LinkedHashSet<Object> members = setMembers.computeIfAbsent(key,
+            ignored -> new LinkedHashSet<>());
+        return switch (method) {
+          case "add" -> {
+            if (setFailure != null) {
+              throw setFailure;
+            }
+            long added = 0L;
+            Object[] values = (Object[]) args[1];
+            for (Object value : values) {
+              if (members.add(value)) {
+                added++;
+              }
+              events.add("SADD " + key + " " + value);
+            }
+            yield added;
+          }
+          case "members" -> new LinkedHashSet<>(members);
+          case "remove" -> {
+            long removed = 0L;
+            Object[] values = (Object[]) args[1];
+            for (Object value : values) {
+              if (members.remove(value)) {
+                removed++;
+              }
+            }
+            events.add("SREM " + key);
+            yield removed;
+          }
+          default -> throw new UnsupportedOperationException(method);
+        };
+      });
+    }
+
+    @Override
+    public Boolean delete(String key) {
+      events.add("DEL " + key);
+      return hashes.remove(key) != null || setMembers.remove(key) != null;
+    }
+
+    @Override
+    public Boolean expire(String key, long timeout, TimeUnit unit) {
+      expireCalls.add(new ExpireCall(key, timeout, unit));
+      return Boolean.TRUE;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <T> T execute(RedisScript<T> script, List<String> keys, Object... args) {
+      if (scriptFailure != null) {
+        throw scriptFailure;
+      }
+      Runnable hook = beforeNextScript;
+      beforeNextScript = null;
+      if (hook != null) {
+        hook.run();
+      }
+      scriptCalls.add(new ScriptCall(
+          script.getScriptAsString(),
+          List.copyOf(keys),
+          List.of(args)
+      ));
+
+      String seatKey = keys.get(0);
+      String indexKey = keys.get(1);
+      long removed = 0L;
+      if (hashSize(seatKey) == 0L) {
+        LinkedHashSet<Object> members = setMembers.computeIfAbsent(indexKey,
+            ignored -> new LinkedHashSet<>());
+        if (members.remove(args[0])) {
+          removed = 1L;
+        }
+      }
+      events.add("LUA compare-prune " + seatKey);
+      return (T) Long.valueOf(removed);
+    }
+
+    private void seedSeat(Long roomId, int index, LiveMicrophone microphone) {
+      String key = LiveMicKey.SEAT.getKey(roomId);
+      hashes.computeIfAbsent(key, ignored -> new LinkedHashMap<>())
+          .put(Integer.toString(index), JacksonUtils.toJson(microphone));
+    }
+
+    private long hashSize(String key) {
+      return hashes.getOrDefault(key, new LinkedHashMap<>()).size();
+    }
+
+    private Set<Object> indexMembers() {
+      return new LinkedHashSet<>(setMembers.getOrDefault(ROOM_INDEX_KEY, new LinkedHashSet<>()));
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T proxy(Class<T> type, Operation operation) {
+      return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[] {type},
+          (proxy, method, args) -> {
+            if (method.getDeclaringClass().equals(Object.class)) {
+              return switch (method.getName()) {
+                case "toString" -> type.getSimpleName() + "RecordingProxy";
+                case "hashCode" -> System.identityHashCode(proxy);
+                case "equals" -> proxy == args[0];
+                default -> throw new UnsupportedOperationException(method.getName());
+              };
+            }
+            return operation.invoke(method.getName(), args == null ? new Object[0] : args);
+          });
+    }
+  }
+
+  @FunctionalInterface
+  private interface Operation {
+
+    Object invoke(String method, Object[] args);
+  }
+}