声网问题修复

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java

@@ -7,6 +7,8 @@ import com.red.circle.component.mq.service.Action;
 import com.red.circle.component.mq.service.ConsumerMessage;
 import com.red.circle.component.mq.service.MessageListener;
 import com.red.circle.component.redis.service.RedisService;
+import com.red.circle.common.business.dto.rtc.RtcProviderEnum;
+import com.red.circle.common.business.dto.rtc.RtcVersionSupport;
 import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import com.red.circle.framework.dto.ResultResponse;
 import com.red.circle.live.app.dto.cmd.HeartbeatStatusEnum;
@@ -218,18 +220,23 @@ public class UserHeartbeatListener implements MessageListener {
     }
 
     UserProfileDTO actualUserProfile = ensureUserProfile(userProfile);
-    AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId);
+    boolean validateAgoraState = shouldValidateAgoraState(roomId);
-    if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) {
+    AgoraChannelStateDTO channelState = null;
-      handleAgoraUserMissing(roomId, actualUserProfile.getId(), channelState,
+    if (validateAgoraState) {
-          "heartbeat agora user missing");
+      channelState = agoraRoomStateService.getChannelState(roomId);
-      return;
+      if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) {
+        handleAgoraUserMissing(roomId, actualUserProfile.getId(), channelState,
+            "heartbeat agora user missing");
+        return;
+      }
+      clearAgoraMissingAfterPresent(channelState, roomId, actualUserProfile.getId());
     }
-    clearAgoraMissingAfterPresent(channelState, roomId, actualUserProfile.getId());
 
     // 添加房间在线用户
     addRoomOnlineUser(actualUserProfile, roomProfile);
 
-    refreshMicActiveTime(event, roomProfile, actualUserProfile, channelState);
+    refreshMicActiveTime(event, roomProfile, actualUserProfile, channelState,
+        validateAgoraState);
   }
 
   private RoomProfileDTO getRoomProfileDto(Long roomId) {
@@ -281,13 +288,15 @@ public class UserHeartbeatListener implements MessageListener {
   private void refreshMicActiveTime(UserHeartbeatEvent event,
       RoomProfileDTO roomProfile,
       UserProfileDTO userProfile) {
-    refreshMicActiveTime(event, roomProfile, userProfile, null);
+    refreshMicActiveTime(event, roomProfile, userProfile, null,
+        shouldValidateAgoraState(roomProfile.getId()));
   }
 
   private void refreshMicActiveTime(UserHeartbeatEvent event,
       RoomProfileDTO roomProfile,
       UserProfileDTO userProfile,
-      AgoraChannelStateDTO checkedChannelState) {
+      AgoraChannelStateDTO checkedChannelState,
+      boolean validateAgoraState) {
 
     // 如果不在麦克风, true&null兼容历史版本
     if (Objects.equals(event.getUpMick(), Boolean.FALSE)) {
@@ -310,6 +319,12 @@ public class UserHeartbeatListener implements MessageListener {
       return;
     }
 
+    if (!validateAgoraState) {
+      agoraMicMissingConfirmService.clearMissing(roomProfile.getId(), userProfile.getId());
+      refreshMicActiveTimeWithoutAgoraCheck(event, roomProfile, liveMicrophone);
+      return;
+    }
+
     AgoraChannelStateDTO channelState = Objects.nonNull(checkedChannelState)
         ? checkedChannelState
         : agoraRoomStateService.getChannelState(roomProfile.getId());
@@ -405,6 +420,20 @@ public class UserHeartbeatListener implements MessageListener {
     }
   }
 
+  boolean shouldValidateAgoraState(Long roomId) {
+    if (Objects.isNull(roomId)) {
+      return true;
+    }
+    try {
+      RtcProviderEnum provider = RtcProviderEnum.of(
+          redisService.getString(RtcVersionSupport.roomProviderKey(roomId)));
+      return !Objects.equals(RtcProviderEnum.TRTC, provider);
+    } catch (Exception e) {
+      log.warn("读取房间 RTC provider 失败，按 Agora 校验处理 roomId={}", roomId, e);
+      return true;
+    }
+  }
+
   private String appendChannelStateReason(AgoraChannelStateDTO channelState, String reason) {
     if (Objects.isNull(channelState)) {
       return reason;

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

@@ -1,6 +1,8 @@
 package com.red.circle.live.app.scheduler;
 
 import com.red.circle.component.redis.service.RedisService;
+import com.red.circle.common.business.dto.rtc.RtcProviderEnum;
+import com.red.circle.common.business.dto.rtc.RtcVersionSupport;
 import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import com.red.circle.live.app.service.AgoraMicMissingConfirmService;
 import com.red.circle.live.app.service.AgoraMicStateCleanupService;
@@ -57,6 +59,10 @@ public class AgoraMicStateReconcileTask {
     if (Objects.isNull(roomId)) {
       return;
     }
+    if (!shouldReconcileAgoraRoom(roomId)) {
+      log.debug("声网麦位对账跳过非 Agora 房间 roomId={}", roomId);
+      return;
+    }
 
     AgoraChannelStateDTO state = agoraRoomStateService.getChannelState(roomId);
     if (!state.querySuccess()) {
@@ -96,6 +102,17 @@ public class AgoraMicStateReconcileTask {
     }
   }
 
+  boolean shouldReconcileAgoraRoom(Long roomId) {
+    try {
+      RtcProviderEnum provider = RtcProviderEnum.of(
+          redisService.getString(RtcVersionSupport.roomProviderKey(roomId)));
+      return !Objects.equals(RtcProviderEnum.TRTC, provider);
+    } catch (Exception e) {
+      log.warn("读取房间 RTC provider 失败，按 Agora 对账处理 roomId={}", roomId, e);
+      return true;
+    }
+  }
+
   private void cleanupMissingMicrophoneUser(Long roomId,
       LiveMicrophone microphone,
       AgoraChannelStateDTO state,

rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/listener/UserHeartbeatListenerRtcProviderTest.java

@@ -0,0 +1,79 @@
+package com.red.circle.live.app.listener;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.red.circle.component.redis.service.RedisService;
+import java.lang.reflect.Proxy;
+import org.junit.Test;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+public class UserHeartbeatListenerRtcProviderTest {
+
+  @Test
+  public void trtcRoomShouldSkipAgoraStateValidation() {
+    assertFalse(listener("TRTC").shouldValidateAgoraState(1001026L));
+  }
+
+  @Test
+  public void agoraRoomShouldKeepAgoraStateValidation() {
+    assertTrue(listener("AGORA").shouldValidateAgoraState(1001026L));
+  }
+
+  @Test
+  public void unknownRoomProviderShouldKeepAgoraStateValidation() {
+    assertTrue(listener(null).shouldValidateAgoraState(1001026L));
+  }
+
+  private static UserHeartbeatListener listener(String provider) {
+    return new UserHeartbeatListener(
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        redisService(provider),
+        null,
+        null,
+        null,
+        null,
+        null);
+  }
+
+  private static RedisService redisService(String provider) {
+    return new RedisService(new RedisTemplate<>() {
+      @Override
+      public ValueOperations<String, Object> opsForValue() {
+        return valueOperations(provider);
+      }
+    });
+  }
+
+  @SuppressWarnings("unchecked")
+  private static ValueOperations<String, Object> valueOperations(String provider) {
+    return (ValueOperations<String, Object>) Proxy.newProxyInstance(
+        ValueOperations.class.getClassLoader(),
+        new Class<?>[] {ValueOperations.class},
+        (proxy, method, args) -> {
+          if ("get".equals(method.getName())) {
+            return provider;
+          }
+          if ("toString".equals(method.getName())) {
+            return "ValueOperations(" + provider + ")";
+          }
+          if ("hashCode".equals(method.getName())) {
+            return System.identityHashCode(proxy);
+          }
+          if ("equals".equals(method.getName())) {
+            return proxy == args[0];
+          }
+          throw new UnsupportedOperationException(method.getName());
+        });
+  }
+}

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

@@ -0,0 +1,69 @@
+package com.red.circle.live.app.scheduler;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.red.circle.component.redis.service.RedisService;
+import java.lang.reflect.Proxy;
+import org.junit.Test;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+public class AgoraMicStateReconcileTaskRtcProviderTest {
+
+  @Test
+  public void trtcRoomShouldSkipAgoraReconcile() {
+    assertFalse(task("TRTC").shouldReconcileAgoraRoom(1001026L));
+  }
+
+  @Test
+  public void agoraRoomShouldKeepAgoraReconcile() {
+    assertTrue(task("AGORA").shouldReconcileAgoraRoom(1001026L));
+  }
+
+  @Test
+  public void unknownRoomProviderShouldKeepAgoraReconcile() {
+    assertTrue(task(null).shouldReconcileAgoraRoom(1001026L));
+  }
+
+  private static AgoraMicStateReconcileTask task(String provider) {
+    return new AgoraMicStateReconcileTask(
+        redisService(provider),
+        null,
+        null,
+        null,
+        null,
+        null);
+  }
+
+  private static RedisService redisService(String provider) {
+    return new RedisService(new RedisTemplate<>() {
+      @Override
+      public ValueOperations<String, Object> opsForValue() {
+        return valueOperations(provider);
+      }
+    });
+  }
+
+  @SuppressWarnings("unchecked")
+  private static ValueOperations<String, Object> valueOperations(String provider) {
+    return (ValueOperations<String, Object>) Proxy.newProxyInstance(
+        ValueOperations.class.getClassLoader(),
+        new Class<?>[] {ValueOperations.class},
+        (proxy, method, args) -> {
+          if ("get".equals(method.getName())) {
+            return provider;
+          }
+          if ("toString".equals(method.getName())) {
+            return "ValueOperations(" + provider + ")";
+          }
+          if ("hashCode".equals(method.getName())) {
+            return System.identityHashCode(proxy);
+          }
+          if ("equals".equals(method.getName())) {
+            return proxy == args[0];
+          }
+          throw new UnsupportedOperationException(method.getName());
+        });
+  }
+}