修复麦位任务

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

@@ -213,6 +213,9 @@ public class UserHeartbeatListener implements MessageListener {
       log.warn("没有找到房间ID: {}", event.getSessionId());
       return;
     }
+    if (Objects.nonNull(event.getSysOrigin())) {
+      roomProfile.setSysOrigin(event.getSysOrigin().name());
+    }
 
     UserProfileDTO actualUserProfile = ensureUserProfile(userProfile);
     AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId);

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/client/TaskCenterGoClient.java

@@ -109,9 +109,21 @@ public class TaskCenterGoClient {
       }
 
       int status = connection.getResponseCode();
+      String responseBody = readResponse(connection);
       if (status < 200 || status >= 300) {
         log.warn("Task center go report failed. status={}, body={}, eventId={}",
-            status, readResponse(connection), request.getEventId());
+            status, responseBody, request.getEventId());
+        return;
+      }
+      TaskCenterEventResult result = parseEventResult(responseBody);
+      if (Objects.nonNull(result) && !Boolean.TRUE.equals(result.getProcessed())) {
+        log.warn("Task center go report ignored. eventId={}, eventType={}, userId={}, updated={}, reason={}, body={}",
+            request.getEventId(),
+            request.getEventType(),
+            request.getUserId(),
+            result.getUpdated(),
+            result.getReason(),
+            responseBody);
       }
     } catch (Exception e) {
       log.warn("Task center go report error. eventId={}", request.getEventId(), e);
@@ -135,6 +147,18 @@ public class TaskCenterGoClient {
     }
   }
 
+  private TaskCenterEventResult parseEventResult(String responseBody) {
+    try {
+      TaskCenterResponse response = JSON.parseObject(responseBody, TaskCenterResponse.class);
+      if (Objects.isNull(response)) {
+        return null;
+      }
+      return Objects.nonNull(response.getBody()) ? response.getBody() : response.getData();
+    } catch (Exception e) {
+      return null;
+    }
+  }
+
   private String trimRightSlash(String value) {
     while (value.endsWith("/")) {
       value = value.substring(0, value.length() - 1);
@@ -153,4 +177,17 @@ public class TaskCenterGoClient {
     private String occurredAt;
     private Map<String, Object> payload;
   }
+
+  @Data
+  private static class TaskCenterResponse {
+    private TaskCenterEventResult body;
+    private TaskCenterEventResult data;
+  }
+
+  @Data
+  private static class TaskCenterEventResult {
+    private Boolean processed;
+    private Integer updated;
+    private String reason;
+  }
 }

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

@@ -89,4 +89,9 @@ public interface LiveMicCacheService {
    */
   boolean refreshNotActiveUser(Long roomId);
 
+  /**
+   * 刷新移除不活跃用户, 返回被移除的麦位.
+   */
+  List<LiveMicrophone> refreshNotActiveUserAndReturn(Long roomId);
+
 }

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

@@ -60,10 +60,9 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService {
 
   @Override
   public void goUp(Long roomId, Integer index, LiveMicrophone microphone) {
-//    if (Objects.isNull(microphone.getRefreshTime())) {
+    if (Objects.isNull(microphone.getRefreshTime())) {
-//      microphone.updateLastActivityTime();
-//    }
       microphone.updateLastActivityTime();
+    }
     log.info("用户上麦 index = {} microphone = {} ", index, JSON.toJSONString(microphone));
     redisService.hashPut(LiveMicKey.SEAT.getKey(roomId),
         Objects.toString(index),
@@ -122,24 +121,37 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService {
 
   @Override
   public boolean refreshNotActiveUser(Long roomId) {
+    return CollectionUtils.isNotEmpty(refreshNotActiveUserAndReturn(roomId));
+  }
+
+  @Override
+  public List<LiveMicrophone> refreshNotActiveUserAndReturn(Long roomId) {
     setExpire(roomId);
 
     List<LiveMicrophone> liveMicrophone = listLiveMicrophone(roomId);
     if (CollectionUtils.isEmpty(liveMicrophone)) {
-      return false;
+      return CollectionUtils.newArrayList();
     }
 
+    List<LiveMicrophone> inactiveMicrophones =
+        liveMicrophone.stream().filter(LiveMicrophone::checkInactiveExpire).toList();
     String[] index =
-        liveMicrophone.stream().filter(LiveMicrophone::checkInactiveExpire)
+        inactiveMicrophones.stream()
             .map(LiveMicrophone::getMicIndex)
             .map(Objects::toString)
             .toArray(String[]::new);
+    if (index.length == 0) {
+      return CollectionUtils.newArrayList();
+    }
 
     log.info("清除不活跃的用户入参 = {} ", JSON.toJSONString(index));
     Long size = redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), index);
     log.info("清除不活跃的用户结果 = {} ", size);
 
-    return Objects.nonNull(size) && size > 0;
+    if (Objects.isNull(size) || size <= 0) {
+      return CollectionUtils.newArrayList();
+    }
+    return inactiveMicrophones;
   }
 
 

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java

@@ -181,7 +181,7 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
       ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, liveMicrophone.getUser());
 
       if (liveMicCacheService.checkEqSelf(roomId, micIndex, userId)) {
-        liveMicCacheService.refreshNotActiveUser(roomId);
+        refreshNotActiveUserAndReport(roomProfile);
         return micUserChangeNotice(roomProfile);
       }
 
@@ -213,7 +213,7 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
       micUpdated = true;
 
       // 清理活跃用户
-      liveMicCacheService.refreshNotActiveUser(roomId);
+      refreshNotActiveUserAndReport(roomProfile);
 
       // 发送通知
       return micUserChangeNotice(roomProfile);
@@ -307,7 +307,7 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
     // 下麦克风
     liveMicCacheService.goDown(roomId, micIndex);
     // 清理活跃用户
-    liveMicCacheService.refreshNotActiveUser(roomId);
+    refreshNotActiveUserAndReport(roomProfile);
     // 发送通知
     micUserChangeNotice(roomProfile);
   }
@@ -379,7 +379,7 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
       liveMicCacheService.goDown(roomId, mickIndex);
       reportMicDuration(roomProfile, roomId, liveMicrophone, "MIC_LOCK");
       // 清理活跃用户
-      liveMicCacheService.refreshNotActiveUser(roomId);
+      refreshNotActiveUserAndReport(roomProfile);
       // 发送通知
       micUserChangeNotice(roomProfile, settingSeat);
     } finally {
@@ -762,11 +762,24 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
 
   @Override
   public void refreshNotActiveUser(RoomProfileDTO roomProfile) {
-    if (liveMicCacheService.refreshNotActiveUser(roomProfile.getId())) {
+    if (refreshNotActiveUserAndReport(roomProfile)) {
       micUserChangeNotice(roomProfile);
     }
   }
 
+  private boolean refreshNotActiveUserAndReport(RoomProfileDTO roomProfile) {
+    if (Objects.isNull(roomProfile) || Objects.isNull(roomProfile.getId())) {
+      return false;
+    }
+    List<LiveMicrophone> removedMicrophones =
+        liveMicCacheService.refreshNotActiveUserAndReturn(roomProfile.getId());
+    if (CollectionUtils.isEmpty(removedMicrophones)) {
+      return false;
+    }
+    reportMicDuration(roomProfile, roomProfile.getId(), removedMicrophones, "INACTIVE_CLEANUP");
+    return true;
+  }
+
   /**
    * 获取缓存房间roomAccount
    *