rtc修复

rc-service/rc-inner-api/external-inner/external-inner-api/src/main/java/com/red/circle/external/inner/endpoint/message/api/AgoraClientApi.java

@@ -1,6 +1,7 @@
 package com.red.circle.external.inner.endpoint.message.api;
 
 import com.red.circle.framework.dto.ResultResponse;
+import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 
@@ -39,4 +40,13 @@ public interface AgoraClientApi {
      */
     @GetMapping("/channel/user")
     ResultResponse<List<Long>> channelUser(@RequestParam("cname") String cname);
+
+    /**
+     * 查询频道状态.
+     *
+     * @param cname 频道名
+     * @return 频道状态
+     */
+    @GetMapping("/channel/state")
+    ResultResponse<AgoraChannelStateDTO> channelState(@RequestParam("cname") String cname);
 }

rc-service/rc-inner-api/external-inner/external-inner-model/src/main/java/com/red/circle/external/inner/model/dto/AgoraChannelStateDTO.java

@@ -0,0 +1,52 @@
+package com.red.circle.external.inner.model.dto;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * 声网频道状态.
+ */
+@Data
+@Accessors(chain = true)
+public class AgoraChannelStateDTO implements Serializable {
+
+  @Serial
+  private static final long serialVersionUID = 1L;
+
+  /**
+   * 查询是否成功.
+   */
+  private Boolean querySuccess;
+
+  /**
+   * 频道是否存在.
+   */
+  private Boolean channelExist;
+
+  /**
+   * 主播 uid 列表.
+   */
+  private List<Long> broadcasters = new ArrayList<>();
+
+  /**
+   * 观众 uid 列表.
+   */
+  private List<Long> audience = new ArrayList<>();
+
+  /**
+   * 原始返回或错误信息.
+   */
+  private String rawResponse;
+
+  public boolean querySuccess() {
+    return Boolean.TRUE.equals(querySuccess);
+  }
+
+  public boolean channelExist() {
+    return Boolean.TRUE.equals(channelExist);
+  }
+}

rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/endpoint/message/AgoraClientEndpoint.java

@@ -1,5 +1,6 @@
 package com.red.circle.external.inner.endpoint.message;
 
+import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import com.red.circle.external.inner.endpoint.message.api.AgoraClientApi;
 import com.red.circle.external.inner.service.message.AgoraClientService;
 import com.red.circle.framework.dto.ResultResponse;
@@ -41,4 +42,9 @@ public class AgoraClientEndpoint implements AgoraClient{
     public ResultResponse<List<Long>> channelUser(String cname) {
         return ResultResponse.success(agoraClientService.channelUser(cname));
     }
+
+    @Override
+    public ResultResponse<AgoraChannelStateDTO> channelState(String cname) {
+        return ResultResponse.success(agoraClientService.channelState(cname));
+    }
 }

rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/AgoraClientService.java

@@ -1,6 +1,7 @@
 package com.red.circle.external.inner.service.message;
 
 
+import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import java.util.List;
 
 /**
@@ -30,4 +31,12 @@ public interface AgoraClientService {
      * @return
      */
     List<Long> channelUser(String cname);
+
+    /**
+     * 查询频道状态.
+     *
+     * @param cname 频道名
+     * @return 频道状态
+     */
+    AgoraChannelStateDTO channelState(String cname);
 }

rc-service/rc-service-external/external-inner-endpoint/src/main/java/com/red/circle/external/inner/service/message/impl/AgoraClientServiceImpl.java

@@ -3,6 +3,7 @@ package com.red.circle.external.inner.service.message.impl;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.red.circle.external.infra.props.RcMessageAgoraProperties;
+import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import com.red.circle.external.inner.service.message.AgoraClientService;
 import com.red.circle.tool.core.collection.CollectionUtils;
 import lombok.RequiredArgsConstructor;
@@ -124,4 +125,46 @@ public class AgoraClientServiceImpl implements AgoraClientService {
         }
     }
 
+    @Override
+    public AgoraChannelStateDTO channelState(String cname) {
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+        HttpGet httpGet = new HttpGet("https://api.sd-rtn.com/dev/v1/channel/user/" + rcMessageAgoraProperties.getAppId() + "/" + cname);
+        httpGet.addHeader("Authorization", rcMessageAgoraProperties.getAuthKey());
+        try {
+            CloseableHttpResponse response = httpClient.execute(httpGet);
+            String result = EntityUtils.toString(response.getEntity());
+            log.info("声网频道状态返回信息 cname={}, result={}", cname, result);
+            JSONObject resultObject = JSON.parseObject(result);
+            Boolean success = resultObject.getBoolean("success");
+            AgoraChannelStateDTO state = new AgoraChannelStateDTO()
+                .setQuerySuccess(Boolean.TRUE.equals(success))
+                .setChannelExist(Boolean.FALSE)
+                .setRawResponse(result);
+            if (!Boolean.TRUE.equals(success)) {
+                return state;
+            }
+            JSONObject dataObject = resultObject.getJSONObject("data");
+            if (Objects.isNull(dataObject)) {
+                return state;
+            }
+            state.setChannelExist(Boolean.TRUE.equals(dataObject.getBoolean("channel_exist")));
+            if (!state.channelExist()) {
+                return state;
+            }
+            if (Objects.nonNull(dataObject.getJSONArray("broadcasters"))) {
+                state.setBroadcasters(dataObject.getJSONArray("broadcasters").toJavaList(Long.class));
+            }
+            if (Objects.nonNull(dataObject.getJSONArray("audience"))) {
+                state.setAudience(dataObject.getJSONArray("audience").toJavaList(Long.class));
+            }
+            return state;
+        } catch (Exception e) {
+            log.error("声网获取频道状态失败 cname={}:{}", cname, e.getMessage());
+            return new AgoraChannelStateDTO()
+                .setQuerySuccess(Boolean.FALSE)
+                .setChannelExist(Boolean.FALSE)
+                .setRawResponse(e.getMessage());
+        }
+    }
+
 }

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java

@@ -9,6 +9,7 @@ import com.red.circle.live.domain.live.LiveMicrophone;
 import com.red.circle.live.domain.live.LiveMusicStatus;
 import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
 import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService;
+import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService;
 import com.red.circle.other.inner.endpoint.live.ActiveVoiceRoomClient;
 import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO;
 import com.red.circle.tool.core.collection.CollectionUtils;
@@ -35,6 +36,7 @@ public class LiveQuitRoomExe {
   private final LiveMusicHeartbeatService liveMusicHeartbeatService;
   private final ActiveVoiceRoomClient activeVoiceRoomClient;
   private final RedisService redisService;
+  private final UserAgoraTokenCacheService userAgoraTokenCacheService;
 
   private static final String EMPTY_ROOM_CACHE_KEY = "empty_room_cache:";
   private static final int CACHE_EXPIRE_MINUTES = 10;
@@ -70,6 +72,9 @@ public class LiveQuitRoomExe {
       liveMusicHeartbeatService.removeMickUser(roomId);
     }
 
+    redisService.delete("IN_ROOM:" + userId);
+    userAgoraTokenCacheService.remove(userId, roomId);
+
     boolean finalMicChanged = micChanged;
     ThreadPoolManager.getInstance().execute(() -> {
       if (finalMicChanged) {

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

@@ -7,8 +7,11 @@ 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.external.inner.model.dto.AgoraChannelStateDTO;
 import com.red.circle.framework.dto.ResultResponse;
 import com.red.circle.live.app.dto.cmd.HeartbeatStatusEnum;
+import com.red.circle.live.app.service.AgoraMicStateCleanupService;
+import com.red.circle.live.app.service.AgoraRoomStateService;
 import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
 import com.red.circle.live.domain.live.LiveMicrophone;
 import com.red.circle.live.domain.online.LiveRoomUser;
@@ -74,6 +77,8 @@ public class UserHeartbeatListener implements MessageListener {
   private final RoomDailyTaskClient roomDailyTaskClient;
   private final RedisService redisService;
   private final RoomMemberClient roomMemberClient;
+  private final AgoraRoomStateService agoraRoomStateService;
+  private final AgoraMicStateCleanupService agoraMicStateCleanupService;
 
   private static final int HEARTBEAT_INTERVAL_SECONDS = 60;
 
@@ -206,10 +211,18 @@ public class UserHeartbeatListener implements MessageListener {
       return;
     }
 
-    // 添加房间在线用户
+    UserProfileDTO actualUserProfile = ensureUserProfile(userProfile);
-    addRoomOnlineUser(userProfile, roomProfile);
+    AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId);
+    if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) {
+      agoraMicStateCleanupService.cleanupUser(roomId, actualUserProfile.getId(),
+          "heartbeat agora user missing");
+      return;
+    }
 
-    refreshMicActiveTime(event, roomProfile, userProfile);
+    // 添加房间在线用户
+    addRoomOnlineUser(actualUserProfile, roomProfile);
+
+    refreshMicActiveTime(event, roomProfile, actualUserProfile, channelState);
   }
 
   private RoomProfileDTO getRoomProfileDto(Long roomId) {
@@ -261,6 +274,13 @@ public class UserHeartbeatListener implements MessageListener {
   private void refreshMicActiveTime(UserHeartbeatEvent event,
       RoomProfileDTO roomProfile,
       UserProfileDTO userProfile) {
+    refreshMicActiveTime(event, roomProfile, userProfile, null);
+  }
+
+  private void refreshMicActiveTime(UserHeartbeatEvent event,
+      RoomProfileDTO roomProfile,
+      UserProfileDTO userProfile,
+      AgoraChannelStateDTO checkedChannelState) {
 
     // 如果不在麦克风, true&null兼容历史版本
     if (Objects.equals(event.getUpMick(), Boolean.FALSE)) {
@@ -277,6 +297,30 @@ public class UserHeartbeatListener implements MessageListener {
     }
 
     LiveMicrophone liveMicrophone = liveMicrophones.get(0);
+    if (checkAccountNotAvailable(userProfile)) {
+      agoraMicStateCleanupService.cleanupUser(roomProfile.getId(), userProfile.getId(),
+          "heartbeat user account not available");
+      return;
+    }
+
+    AgoraChannelStateDTO channelState = Objects.nonNull(checkedChannelState)
+        ? checkedChannelState
+        : agoraRoomStateService.getChannelState(roomProfile.getId());
+    if (!channelState.querySuccess()) {
+      log.warn("心跳刷新麦位跳过，声网查询失败 roomId={}, userId={}, response={}",
+          roomProfile.getId(), userProfile.getId(), channelState.getRawResponse());
+      liveMicrophoneGateway.refreshNotActiveUser(roomProfile);
+      return;
+    }
+
+    Long agoraUid = DataTypeUtils.toLong(userProfile.getAccount());
+    if (!channelState.channelExist() || Objects.isNull(agoraUid)
+        || CollectionUtils.isEmpty(channelState.getBroadcasters())
+        || !channelState.getBroadcasters().contains(agoraUid)) {
+      agoraMicStateCleanupService.cleanupUser(roomProfile.getId(), userProfile.getId(),
+          "heartbeat agora broadcaster missing");
+      return;
+    }
 
     // 刷新活跃时间
     if (liveMicCacheService.checkLockMic(roomProfile.getId(), liveMicrophone.getMicIndex())) {
@@ -291,4 +335,36 @@ public class UserHeartbeatListener implements MessageListener {
     liveMicrophoneGateway.refreshNotActiveUser(roomProfile);
   }
 
+  private UserProfileDTO ensureUserProfile(UserProfileDTO userProfile) {
+    if (Objects.nonNull(userProfile) && StringUtils.isNotBlank(userProfile.getAccount())) {
+      return userProfile;
+    }
+    if (Objects.isNull(userProfile) || Objects.isNull(userProfile.getId())) {
+      return userProfile;
+    }
+    try {
+      return userProfileClient.getByUserId(userProfile.getId()).getBody();
+    } catch (Exception e) {
+      log.error("获取用户信息失败 userId={}", userProfile.getId(), e);
+      return userProfile;
+    }
+  }
+
+  private boolean isUserInAgoraChannel(AgoraChannelStateDTO channelState, UserProfileDTO userProfile) {
+    if (Objects.isNull(userProfile) || StringUtils.isBlank(userProfile.getAccount())) {
+      return false;
+    }
+    if (!channelState.channelExist()) {
+      return false;
+    }
+    Long agoraUid = DataTypeUtils.toLong(userProfile.getAccount());
+    if (Objects.isNull(agoraUid)) {
+      return false;
+    }
+    return (CollectionUtils.isNotEmpty(channelState.getBroadcasters())
+        && channelState.getBroadcasters().contains(agoraUid))
+        || (CollectionUtils.isNotEmpty(channelState.getAudience())
+        && channelState.getAudience().contains(agoraUid));
+  }
+
 }

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

@@ -1,13 +1,22 @@
 package com.red.circle.live.app.listener.agora;
 
+import com.red.circle.component.redis.service.RedisService;
 import com.red.circle.live.app.listener.agora.callback.AgoraChannelBase;
 import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserJoin;
 import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserLeave;
 import com.red.circle.live.app.listener.agora.callback.AgoraChannelUserRoleChange;
 import com.red.circle.live.app.listener.agora.callback.AgoraChannelWebhook;
 import com.red.circle.live.app.listener.agora.callback.AgoraChannelWebhookListener;
+import com.red.circle.live.app.service.AgoraMicStateCleanupService;
+import com.red.circle.other.inner.endpoint.live.RoomManagerClient;
+import com.red.circle.other.inner.endpoint.user.user.UserProfileClient;
 import com.red.circle.other.inner.model.dto.live.RoomProfileDTO;
 import com.red.circle.other.inner.model.dto.user.UserProfileDTO;
+import com.red.circle.tool.core.parse.DataTypeUtils;
+import com.red.circle.tool.core.regex.RegexConstant;
+import com.red.circle.tool.core.text.StringUtils;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 import lombok.Data;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.Accessors;
@@ -24,12 +33,10 @@ import org.springframework.stereotype.Component;
 @RequiredArgsConstructor
 public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListener {
 
-//  private final RedisService redisService;
+  private final RedisService redisService;
-//  private final LiveMicCommon liveMicCommon;
+  private final UserProfileClient userProfileClient;
-//  private final UserProfileClient userProfileClient;
+  private final RoomManagerClient roomManagerClient;
-//  private final RoomManagerClient roomManagerClient;
+  private final AgoraMicStateCleanupService agoraMicStateCleanupService;
-//  private final LiveMicCacheService liveMicCacheService;
-//  private final LiveMicUserCacheService liveMicUserCacheService;
 
   @Override
   public void channelCreate(AgoraChannelWebhook webhook, AgoraChannelBase body) {
@@ -38,7 +45,11 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe
 
   @Override
   public void channelDestroy(AgoraChannelWebhook webhook, AgoraChannelBase body) {
-
+    Long roomId = getRoomId(body);
+    if (Objects.isNull(roomId) || !checkProcess(webhook, body)) {
+      return;
+    }
+    agoraMicStateCleanupService.cleanupRoom(roomId, "agora channel destroy");
   }
 
   @Override
@@ -56,7 +67,7 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe
 
   @Override
   public void broadcasterLeaveChannel(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) {
-    // leaveChannelProcess(webhook, body);
+    leaveChannelProcess(webhook, body, body.getUid(), "agora broadcaster leave");
   }
 
 
@@ -67,7 +78,7 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe
 
   @Override
   public void audienceLeaveChannel(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) {
-    // leaveChannelProcess(webhook, body);
+    leaveChannelProcess(webhook, body, body.getUid(), "agora audience leave");
   }
 
   @Override
@@ -79,103 +90,72 @@ public class AgoraChannelWebhookImplListener implements AgoraChannelWebhookListe
   @Override
   public void clientRoleChangeToAudience(AgoraChannelWebhook webhook,
       AgoraChannelUserRoleChange body) {
+    leaveChannelProcess(webhook, body, body.getUid(), "agora role change to audience");
+  }
 
+  private void leaveChannelProcess(AgoraChannelWebhook webhook, AgoraChannelBase body, Long uid,
+      String reason) {
+    UserRoomProfile userRoomProfile = getUserRoomProfile(webhook, body, uid);
+    if (Objects.isNull(userRoomProfile)) {
+      return;
+    }
+
+    agoraMicStateCleanupService.cleanupUser(
+        userRoomProfile.getRoomProfile().getId(),
+        userRoomProfile.getUserProfile().getId(),
+        reason
+    );
+  }
+
+  private Long getRoomId(AgoraChannelBase body) {
+    if (Objects.isNull(body) || StringUtils.isBlank(body.getChannelName())
+        || !body.getChannelName().matches(RegexConstant.NUMBER)) {
+      log.warn("错误的声网频道名称: {}", Objects.nonNull(body) ? body.getChannelName() : null);
+      return null;
+    }
+    return DataTypeUtils.toLong(body.getChannelName());
+  }
+
+  private UserRoomProfile getUserRoomProfile(AgoraChannelWebhook webhook, AgoraChannelBase body,
+      Long uid) {
+    Long roomId = getRoomId(body);
+    if (Objects.isNull(roomId) || roomId <= 0 || Objects.isNull(uid)) {
+      return null;
+    }
+
+    if (!checkProcess(webhook, body)) {
+      return null;
+    }
+
+    RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody();
+    if (Objects.isNull(roomProfile)) {
+      return null;
+    }
+
+    UserProfileDTO userProfile = userProfileClient.getByAccount(roomProfile.getSysOrigin(),
+        Objects.toString(uid)).getBody();
+    if (Objects.isNull(userProfile)) {
+      log.warn("声网回调未找到用户 roomId={}, uid={}", roomId, uid);
+      return null;
+    }
+
+    return new UserRoomProfile()
+        .setRoomProfile(roomProfile)
+        .setUserProfile(userProfile);
   }
 
   /**
-   * 退出频道处理.
+   * true 未处理, false 已处理.
    */
-//  private void leaveChannelProcess(AgoraChannelWebhook webhook, AgoraChannelUserLeave body) {
+  private boolean checkProcess(AgoraChannelWebhook webhook, AgoraChannelBase body) {
-//
+    if (Objects.isNull(webhook)) {
-//    UserRoomProfile userRoomProfile = getUserRoomProfile(webhook, body, body.getUid());
+      return false;
-//
+    }
-//    if (Objects.isNull(userRoomProfile)) {
+    String uniqueKey = StringUtils.isNotBlank(webhook.getNoticeId())
-//      return;
+        ? webhook.getNoticeId()
-//    }
+        : webhook.getEventType() + ":" + (Objects.nonNull(body) ? body.getTs() : webhook.getNotifyMs());
-//
+    return redisService.setIfAbsent("AGORA:" + uniqueKey, 1, 1, TimeUnit.MINUTES);
-//    RoomProfileDTO roomProfile = userRoomProfile.getRoomProfile();
+  }
-//    UserProfileDTO userProfile = userRoomProfile.getUserProfile();
-//    Long roomId = roomProfile.getId();
-//
-//    liveMicCacheService.refreshNotActiveUser(roomId);
-//    liveMicUserCacheService.refreshNotActiveUser(roomId);
-//    liveMicUserCacheService.removeUser(roomProfile.getId(), userProfile.getId());
-//
-//    List<LiveMicrophone> liveMicrophones = liveMicCacheService.listLiveMicrophone(
-//        roomId, userProfile.getId());
-//    if (CollectionUtils.isEmpty(liveMicrophones)) {
-//      return;
-//    }
-//
-//    LiveMicrophone liveMicrophone = liveMicrophones.get(0);
-//
-//    liveMicCommon.tryMicOpsLock(roomProfile.getId(), liveMicrophone.getMicIndex(),
-//        () -> {
-//          log.warn("踢下麦克风: roomId={}, micIndex={}", roomProfile.getId(),
-//              liveMicrophone.getMicIndex());
-//          liveMicCacheService.goDown(roomProfile.getId(), liveMicrophone.getMicIndex());
-//          liveMicCommon.micUserChangeNotice(roomProfile);
-//        });
-//  }
-
-//  private Long getRoomId(String channelName) {
-//
-//    if (StringUtils.isBlank(channelName) || !channelName.matches(RegexConstant.NUMBER)) {
-//      log.warn("错误的频道名称: {}", channelName);
-//      return null;
-//    }
-//    return DataTypeUtils.toLong(channelName);
-//  }
-//
-//  private UserRoomProfile getUserRoomProfile(AgoraChannelWebhook webhook, AgoraChannelBase body,
-//      Long uid) {
-//    if (Objects.isNull(webhook) || Objects.isNull(body)) {
-//      log.warn("数据错误: {}, {}",
-//          JacksonUtils.toJson(webhook),
-//          JacksonUtils.toJson(body)
-//      );
-//      return null;
-//    }
-//
-//    if (Objects.isNull(uid)) {
-//      log.warn("数据错误, uid: {}", uid);
-//      return null;
-//    }
-//
-//    Long roomId = getRoomId(body.getChannelName());
-//    if (Objects.isNull(roomId) || roomId <= 0) {
-//      return null;
-//    }
-//
-//    if (!checkProcess(webhook, body.getTs())) {
-//      return null;
-//    }
-//
-//    // log.warn("leaveChannelProcess: {}", JacksonUtils.toJson(webhook));
-//
-//    RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody();
-//    if (Objects.isNull(roomProfile)) {
-//      return null;
-//    }
-//
-//    UserProfileDTO userProfile = userProfileClient.getByAccount(roomProfile.getSysOrigin(),
-//        Objects.toString(uid)).getBody();
-//    if (Objects.isNull(userProfile)) {
-//      return null;
-//    }
-//
-//    return new UserRoomProfile()
-//        .setRoomProfile(roomProfile)
-//        .setUserProfile(userProfile);
-//  }
-
-//  /**
-//   * true 未处理, false 已处理.
-//   */
-//  private boolean checkProcess(AgoraChannelWebhook webhook, Long ts) {
-//    return redisService.setIfAbsent("AGORA:" + webhook.getEventType() + ":" + ts, 1,
-//        TimeUnit.MINUTES);
-//  }
 
 
   @Data

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

@@ -0,0 +1,111 @@
+package com.red.circle.live.app.scheduler;
+
+import com.red.circle.component.redis.service.RedisService;
+import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
+import com.red.circle.live.app.service.AgoraMicStateCleanupService;
+import com.red.circle.live.app.service.AgoraRoomStateService;
+import com.red.circle.live.domain.live.LiveMicrophone;
+import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
+import com.red.circle.tool.core.collection.CollectionUtils;
+import com.red.circle.tool.core.parse.DataTypeUtils;
+import java.util.HashSet;
+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.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 声网真实状态和业务麦位状态补偿对账.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class AgoraMicStateReconcileTask {
+
+  private static final String LOCK_KEY = "agora:mic:reconcile:lock";
+  private static final String SEAT_KEY_PREFIX = "LiveMick:SEAT:";
+
+  private final RedisService redisService;
+  private final RedisTemplate<String, Object> redisTemplate;
+  private final LiveMicCacheService liveMicCacheService;
+  private final AgoraRoomStateService agoraRoomStateService;
+  private final AgoraMicStateCleanupService agoraMicStateCleanupService;
+
+  @Scheduled(fixedDelayString = "${red-circle.live.agora-mic-reconcile-delay-ms:30000}")
+  public void reconcileAgoraMicState() {
+    if (!redisService.setIfAbsent(LOCK_KEY, 1, 25, TimeUnit.SECONDS)) {
+      return;
+    }
+    try {
+      for (Long roomId : scanSeatRoomIds()) {
+        reconcileRoom(roomId);
+      }
+    } finally {
+      redisService.delete(LOCK_KEY);
+    }
+  }
+
+  private void reconcileRoom(Long roomId) {
+    if (Objects.isNull(roomId)) {
+      return;
+    }
+
+    AgoraChannelStateDTO state = agoraRoomStateService.getChannelState(roomId);
+    if (!state.querySuccess()) {
+      log.warn("声网麦位对账跳过，查询失败 roomId={}, response={}", roomId, state.getRawResponse());
+      return;
+    }
+
+    if (!state.channelExist()) {
+      agoraMicStateCleanupService.cleanupRoom(roomId, "reconcile agora channel not exist");
+      return;
+    }
+
+    List<LiveMicrophone> microphones = liveMicCacheService.listLiveMicrophone(roomId);
+    if (CollectionUtils.isEmpty(microphones)) {
+      return;
+    }
+
+    Set<Long> broadcasterUids = new HashSet<>(state.getBroadcasters());
+    for (LiveMicrophone microphone : microphones) {
+      if (Objects.isNull(microphone) || Objects.isNull(microphone.getUser())) {
+        continue;
+      }
+      Long agoraUid = DataTypeUtils.toLong(microphone.getUser().getAccount());
+      if (Objects.isNull(agoraUid) || !broadcasterUids.contains(agoraUid)) {
+        agoraMicStateCleanupService.cleanupUser(roomId, microphone.getUser().getId(),
+            "reconcile agora broadcaster missing");
+      }
+    }
+  }
+
+  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);
+        }
+      }
+    } catch (Exception e) {
+      log.error("扫描声网麦位对账 Redis key 失败", 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/main/java/com/red/circle/live/app/service/AgoraMicStateCleanupService.java

@@ -0,0 +1,84 @@
+package com.red.circle.live.app.service;
+
+import com.red.circle.component.redis.service.RedisService;
+import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
+import com.red.circle.live.domain.live.LiveMicrophone;
+import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
+import com.red.circle.live.infra.database.cache.service.LiveMicUserCacheService;
+import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService;
+import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService;
+import com.red.circle.tool.core.collection.CollectionUtils;
+import java.util.List;
+import java.util.Objects;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * 声网状态反向清理业务麦位和房间在线缓存.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AgoraMicStateCleanupService {
+
+  private final LiveMicCacheService liveMicCacheService;
+  private final LiveMicUserCacheService liveMicUserCacheService;
+  private final LiveMusicHeartbeatService liveMusicHeartbeatService;
+  private final LiveMicrophoneGateway liveMicrophoneGateway;
+  private final UserAgoraTokenCacheService userAgoraTokenCacheService;
+  private final RedisService redisService;
+
+  public boolean cleanupUser(Long roomId, Long userId, String reason) {
+    if (Objects.isNull(roomId) || Objects.isNull(userId)) {
+      return false;
+    }
+
+    boolean micChanged = false;
+    List<LiveMicrophone> liveMicrophones = liveMicCacheService.listLiveMicrophone(roomId, userId);
+    if (CollectionUtils.isNotEmpty(liveMicrophones)) {
+      for (LiveMicrophone mic : liveMicrophones) {
+        if (Objects.isNull(mic) || Objects.isNull(mic.getMicIndex())) {
+          continue;
+        }
+        final boolean[] removed = {false};
+        liveMicrophoneGateway.tryMicOpsLock(roomId, mic.getMicIndex(), () -> {
+          liveMicCacheService.goDown(roomId, mic.getMicIndex());
+          removed[0] = true;
+        });
+        micChanged = micChanged || removed[0];
+      }
+    }
+
+    liveMicUserCacheService.removeUser(roomId, userId);
+    redisService.delete("IN_ROOM:" + userId);
+    userAgoraTokenCacheService.remove(userId, roomId);
+
+    if (micChanged) {
+      notifyMicChanged(roomId, reason);
+    }
+
+    log.warn("声网状态清理用户 roomId={}, userId={}, micChanged={}, reason={}",
+        roomId, userId, micChanged, reason);
+    return micChanged;
+  }
+
+  public void cleanupRoom(Long roomId, String reason) {
+    if (Objects.isNull(roomId)) {
+      return;
+    }
+    liveMicCacheService.goDownAll(roomId);
+    liveMicUserCacheService.remove(roomId);
+    liveMusicHeartbeatService.removeMickUser(roomId);
+    notifyMicChanged(roomId, reason);
+    log.warn("声网状态清理房间 roomId={}, reason={}", roomId, reason);
+  }
+
+  private void notifyMicChanged(Long roomId, String reason) {
+    try {
+      liveMicrophoneGateway.micUserChangeNotice(roomId);
+    } catch (Exception e) {
+      log.error("声网状态清理后推送麦位变更失败 roomId={}, reason={}", roomId, reason, e);
+    }
+  }
+}

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

@@ -0,0 +1,65 @@
+package com.red.circle.live.app.service;
+
+import com.red.circle.external.inner.endpoint.message.AgoraClient;
+import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
+import com.red.circle.framework.dto.ResultResponse;
+import com.red.circle.tool.core.collection.CollectionUtils;
+import java.util.Objects;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * 声网房间状态查询.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AgoraRoomStateService {
+
+  private final AgoraClient agoraClient;
+
+  public AgoraChannelStateDTO getChannelState(Long roomId) {
+    if (Objects.isNull(roomId)) {
+      return queryFailed("roomId is null");
+    }
+    try {
+      ResultResponse<AgoraChannelStateDTO> response = agoraClient.channelState(roomId.toString());
+      if (Objects.isNull(response) || !response.checkSuccess()) {
+        return queryFailed("response not success");
+      }
+      AgoraChannelStateDTO body = response.getBody();
+      if (Objects.isNull(body)) {
+        return queryFailed("body is null");
+      }
+      return body;
+    } catch (Exception e) {
+      log.error("查询声网频道状态异常 roomId={}", roomId, e);
+      return queryFailed(e.getMessage());
+    }
+  }
+
+  public boolean isBroadcasterInChannel(Long roomId, String account) {
+    AgoraChannelStateDTO state = getChannelState(roomId);
+    if (!state.querySuccess() || !state.channelExist() || CollectionUtils.isEmpty(state.getBroadcasters())) {
+      return false;
+    }
+    Long agoraUid = parseAccount(account);
+    return Objects.nonNull(agoraUid) && state.getBroadcasters().contains(agoraUid);
+  }
+
+  private Long parseAccount(String account) {
+    try {
+      return Long.valueOf(account);
+    } catch (Exception e) {
+      return null;
+    }
+  }
+
+  private AgoraChannelStateDTO queryFailed(String reason) {
+    return new AgoraChannelStateDTO()
+        .setQuerySuccess(Boolean.FALSE)
+        .setChannelExist(Boolean.FALSE)
+        .setRawResponse(reason);
+  }
+}

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/gift/GiveGiftsListener.java

@@ -11,6 +11,7 @@ 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.external.inner.endpoint.message.ImGroupClient;
+import com.red.circle.external.inner.model.cmd.message.BroadcastGroupMsgBodyCmd;
 import com.red.circle.external.inner.model.cmd.message.CustomGroupMsgBodyCmd;
 import com.red.circle.external.inner.model.enums.message.GroupMessageTypeEnum;
 import com.red.circle.framework.core.dto.ReqSysOrigin;
@@ -64,6 +65,7 @@ import com.red.circle.tool.core.collection.CollectionUtils;
 import com.red.circle.tool.core.date.TimestampUtils;
 import com.red.circle.tool.core.json.JacksonUtils;
 import com.red.circle.tool.core.sequence.IdWorkerUtils;
+import com.red.circle.tool.core.text.StringUtils;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
@@ -377,6 +379,8 @@ public class GiveGiftsListener implements MessageListener {
             return;
         }
 
+        sendNoticeGiftBroadcast(event, runningWater);
+
         if (!SysOriginPlatformEnum.isVoiceSystem(runningWater.getSysOrigin())) {
             return;
         }
@@ -404,6 +408,51 @@ public class GiveGiftsListener implements MessageListener {
         offlineProcessGiftEvent(OfflineStatisticsGiftEventType.ROCKET_COUNT, runningWater);
     }
 
+    void sendNoticeGiftBroadcast(GiveAwayGiftBatchEvent event,
+                                 GiftGiveRunningWater runningWater) {
+        if (Objects.isNull(event) || Objects.isNull(event.getGiftConfig())
+                || Objects.isNull(runningWater)) {
+            return;
+        }
+        if (event.checkLuckyGift()) {
+            return;
+        }
+        String special = event.getGiftConfig().getSpecial();
+        if (StringUtils.isBlank(special) || !special.contains(GiftSpecialEnum.NOTICE.name())) {
+            return;
+        }
+
+        RoomProfile roomProfile = Objects.nonNull(event.getRoomId())
+                ? roomProfileManagerService.getProfileById(event.getRoomId())
+                : null;
+        SendGiftNotifyCO notify = new SendGiftNotifyCO()
+                .setGlobalNews(Boolean.TRUE)
+                .setSysOrigin(event.getSysOrigin().name())
+                .setRoomId(event.getRoomId())
+                .setRoomAccount(Objects.nonNull(roomProfile) ? roomProfile.getRoomAccount() : null)
+                .setSendUserId(event.getSendUserId())
+                .setAcceptUserIds(event.getAccepts().stream()
+                        .map(GiveAwayGiftRoomAccepts::getAcceptUserId)
+                        .filter(Objects::nonNull)
+                        .distinct()
+                        .collect(Collectors.toList()))
+                .setGiftId(event.getGiftConfig().getId())
+                .setGiftQuantity(event.getQuantity())
+                .setGiftCandy(event.getGiftConfig().getGiftCandy())
+                .setActualAmount(runningWater.getGiftValue().getActualAmount())
+                .setGiftType(runningWater.getGiftValue().getGiftType());
+
+        imGroupClient.sendMessageBroadcast(
+                BroadcastGroupMsgBodyCmd.builder()
+                        .toPlatform(event.getSysOrigin())
+                        .type(GroupMessageTypeEnum.SEND_GIFT)
+                        .data(notify)
+                        .build()
+        );
+        log.warn("NOTICE礼物全服通报成功, trackId={}, giftId={}, sendUserId={}",
+                event.getTrackId(), event.getGiftConfig().getId(), event.getSendUserId());
+    }
+
     private Pair<GiftGiveRunningWater,Boolean> saveRunningWater(GiveAwayGiftBatchEvent event) {
         List<Long> anchorUserIds = teamMemberService.listMemberIds(event.allUserId());
 

rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/listener/gift/GiveGiftsListenerTest.java

@@ -0,0 +1,141 @@
+package com.red.circle.other.app.listener.gift;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.red.circle.common.business.core.enums.SysOriginPlatformEnum;
+import com.red.circle.component.mq.MessageEventProcess;
+import com.red.circle.external.inner.endpoint.message.ImGroupClient;
+import com.red.circle.external.inner.model.cmd.message.BroadcastGroupMsgBodyCmd;
+import com.red.circle.external.inner.model.enums.message.GroupMessageTypeEnum;
+import com.red.circle.mq.business.model.event.gift.GiveAwayGiftBatchEvent;
+import com.red.circle.mq.business.model.event.gift.GiveAwayGiftRoomAccepts;
+import com.red.circle.mq.business.model.event.gift.GiveGiftConfig;
+import com.red.circle.mq.rocket.business.producer.GameMqMessage;
+import com.red.circle.mq.rocket.business.producer.GiftMqMessage;
+import com.red.circle.other.app.command.game.barrage.BarrageGameBaishunExe;
+import com.red.circle.other.app.command.game.burstcrystal.GameBurstCrystalScheduleTwoFunQueryExe;
+import com.red.circle.other.app.command.game.roompk.GameRoomPkUserCmdExe;
+import com.red.circle.other.app.command.game.teampk.GameTeamPkBaseInfoCmdExe;
+import com.red.circle.other.app.common.gift.GameLuckyGiftCommon;
+import com.red.circle.other.app.dto.clientobject.gift.SendGiftNotifyCO;
+import com.red.circle.other.domain.gateway.user.ability.UserRegionGateway;
+import com.red.circle.other.infra.database.cache.service.other.EnumConfigCacheService;
+import com.red.circle.other.infra.database.cache.service.other.LiveHeartbeatCacheService;
+import com.red.circle.other.infra.database.mongo.entity.gift.GiftGiveRunningWater;
+import com.red.circle.other.infra.database.mongo.entity.gift.GiftValue;
+import com.red.circle.other.infra.database.mongo.entity.live.RoomProfile;
+import com.red.circle.other.infra.database.mongo.service.gift.GiftCountAnchorDailyRunningWaterService;
+import com.red.circle.other.infra.database.mongo.service.gift.GiftGiveRunningWaterService;
+import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManagerService;
+import com.red.circle.other.infra.database.mongo.service.team.team.TeamBillCycleService;
+import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberService;
+import com.red.circle.other.infra.database.mongo.service.team.team.TeamPolicyManagerService;
+import com.red.circle.other.inner.enums.material.GiftSpecialEnum;
+import java.math.BigDecimal;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+class GiveGiftsListenerTest {
+
+  private final ImGroupClient imGroupClient = mock(ImGroupClient.class);
+  private final RoomProfileManagerService roomProfileManagerService = mock(
+      RoomProfileManagerService.class);
+  private final GiveGiftsListener listener = new GiveGiftsListener(
+      mock(GameMqMessage.class),
+      mock(GiftMqMessage.class),
+      mock(TeamMemberService.class),
+      mock(UserRegionGateway.class),
+      mock(MessageEventProcess.class),
+      mock(GameLuckyGiftCommon.class),
+      mock(TeamBillCycleService.class),
+      mock(TeamPolicyManagerService.class),
+      mock(EnumConfigCacheService.class),
+      mock(LiveHeartbeatCacheService.class),
+      mock(GiftGiveRunningWaterService.class),
+      imGroupClient,
+      roomProfileManagerService,
+      mock(GameTeamPkBaseInfoCmdExe.class),
+      mock(GameRoomPkUserCmdExe.class),
+      mock(BarrageGameBaishunExe.class),
+      mock(GameBurstCrystalScheduleTwoFunQueryExe.class),
+      mock(GiftCountAnchorDailyRunningWaterService.class)
+  );
+
+  @Test
+  void sendNoticeGiftBroadcastShouldBroadcastNoticeGift() {
+    RoomProfile roomProfile = new RoomProfile();
+    roomProfile.setRoomAccount("1002001");
+    when(roomProfileManagerService.getProfileById(2001L))
+        .thenReturn(roomProfile);
+
+    listener.sendNoticeGiftBroadcast(
+        event(GiftSpecialEnum.NOTICE.name(), "GOLD"),
+        runningWater()
+    );
+
+    ArgumentCaptor<BroadcastGroupMsgBodyCmd> captor = ArgumentCaptor.forClass(
+        BroadcastGroupMsgBodyCmd.class);
+    verify(imGroupClient).sendMessageBroadcast(captor.capture());
+
+    BroadcastGroupMsgBodyCmd message = captor.getValue();
+    assertEquals(GroupMessageTypeEnum.SEND_GIFT.name(), message.getType());
+    assertEquals(SysOriginPlatformEnum.LIKEI, message.getSysOrigin());
+
+    SendGiftNotifyCO data = (SendGiftNotifyCO) message.getData();
+    assertTrue(data.getGlobalNews());
+    assertEquals(2001L, data.getRoomId());
+    assertEquals("1002001", data.getRoomAccount());
+    assertEquals(3001L, data.getSendUserId());
+    assertEquals(List.of(4001L), data.getAcceptUserIds());
+    assertEquals(5001L, data.getGiftId());
+    assertEquals(2, data.getGiftQuantity());
+    assertEquals(new BigDecimal("20"), data.getGiftCandy());
+    assertEquals(new BigDecimal("40"), data.getActualAmount());
+    assertEquals("NORMAL", data.getGiftType());
+  }
+
+  @Test
+  void sendNoticeGiftBroadcastShouldIgnoreNonNoticeGift() {
+    listener.sendNoticeGiftBroadcast(event("ANIMATION", "GOLD"), runningWater());
+
+    verify(imGroupClient, never()).sendMessageBroadcast(any());
+  }
+
+  @Test
+  void sendNoticeGiftBroadcastShouldIgnoreLuckyGift() {
+    listener.sendNoticeGiftBroadcast(event(GiftSpecialEnum.NOTICE.name(), "LUCKY_GIFT"),
+        runningWater());
+
+    verify(imGroupClient, never()).sendMessageBroadcast(any());
+  }
+
+  private GiveAwayGiftBatchEvent event(String special, String giftTab) {
+    return new GiveAwayGiftBatchEvent()
+        .setSysOrigin(SysOriginPlatformEnum.LIKEI)
+        .setRoomId(2001L)
+        .setSendUserId(3001L)
+        .setAccepts(List.of(new GiveAwayGiftRoomAccepts().setAcceptUserId(4001L)))
+        .setQuantity(2)
+        .setTrackId(6001L)
+        .setGiftConfig(new GiveGiftConfig()
+            .setId(5001L)
+            .setSpecial(special)
+            .setType("GOLD")
+            .setGiftTab(giftTab)
+            .setGiftCandy(new BigDecimal("20")));
+  }
+
+  private GiftGiveRunningWater runningWater() {
+    return new GiftGiveRunningWater()
+        .setGiftValue(new GiftValue()
+            .setGiftType("NORMAL")
+            .setActualAmount(new BigDecimal("40")));
+  }
+}

rc-service/rc-service-other/other-client/src/main/java/com/red/circle/other/app/dto/clientobject/gift/SendGiftNotifyCO.java

@@ -1,8 +1,11 @@
 package com.red.circle.other.app.dto.clientobject.gift;
 
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
 import com.red.circle.other.app.dto.clientobject.game.GameBurstCrystalScheduleCO;
 import com.red.circle.other.app.dto.clientobject.game.GameIndoorTeamPkCO;
 import com.red.circle.other.app.dto.clientobject.game.GameRoomPkUserCO;
+import java.math.BigDecimal;
 import lombok.Data;
 import lombok.experimental.Accessors;
 
@@ -17,6 +20,64 @@ import java.util.List;
 @Accessors(chain = true)
 public class SendGiftNotifyCO {
 
+    /**
+     * 是否全局通报.
+     */
+    private Boolean globalNews;
+
+    /**
+     * 系统.
+     */
+    private String sysOrigin;
+
+    /**
+     * 房间ID.
+     */
+    @JsonSerialize(using = ToStringSerializer.class)
+    private Long roomId;
+
+    /**
+     * 房间账号.
+     */
+    private String roomAccount;
+
+    /**
+     * 送礼物人用户id.
+     */
+    @JsonSerialize(using = ToStringSerializer.class)
+    private Long sendUserId;
+
+    /**
+     * 接收用户ID集合.
+     */
+    private List<Long> acceptUserIds;
+
+    /**
+     * 礼物ID.
+     */
+    @JsonSerialize(using = ToStringSerializer.class)
+    private Long giftId;
+
+    /**
+     * 礼物数量.
+     */
+    private Integer giftQuantity;
+
+    /**
+     * 礼物价值/糖果.
+     */
+    private BigDecimal giftCandy;
+
+    /**
+     * 本次实际消费.
+     */
+    private BigDecimal actualAmount;
+
+    /**
+     * 礼物类型.
+     */
+    private String giftType;
+
     /**
      * PK详细信息
      */