✨ feat(im): 完善 rtc 的后端代码逻辑，各种代码风格的优化

YunaiV
1 parent 530d3b3a
Showing 14 changed files with 368 additions and 99 deletions
yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java
yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/http/HttpUtils.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallService.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantConnectedNotification.java
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantDisconnectedNotification.java
@@ -124,6 +124,22 @@ public class CollectionUtils {
         return from.stream().filter(filter).map(func).filter(Objects::nonNull).collect(Collectors.toSet());
     }
+    public static <T, U> Set<U> convertLinkedSet(Collection<T> from, Function<T, U> func) {
+        if (CollUtil.isEmpty(from)) {
+            return new LinkedHashSet<>();
+        }
+        return from.stream().map(func).filter(Objects::nonNull)
+                .collect(Collectors.toCollection(LinkedHashSet::new));
+    }
+
+    public static <T, U> Set<U> convertLinkedSet(Collection<T> from, Function<T, U> func, Predicate<T> filter) {
+        if (CollUtil.isEmpty(from)) {
+            return new LinkedHashSet<>();
+        }
+        return from.stream().filter(filter).map(func).filter(Objects::nonNull)
+                .collect(Collectors.toCollection(LinkedHashSet::new));
+    }
+
     public static <T, K> Map<K, T> convertMapByFilter(Collection<T> from, Predicate<T> filter, Function<T, K> keyFunc) {
         if (CollUtil.isEmpty(from)) {
             return new HashMap<>();
@@ -206,4 +206,14 @@ public class HttpUtils {
         }
     }
+    /**
+     * WebSocket URL 切换成 HTTP URL：ws:// → http://；wss:// → https://；其它格式原样保留
+     *
+     * @param url 原始 URL
+     * @return 切换协议后的 URL
+     */
+    public static String wsUrlToHttp(String url) {
+        return StrUtil.startWithIgnoreCase(url, "ws") ? "http" + url.substring(2) : url;
+    }
+
 }
 package cn.iocoder.yudao.module.im.controller.admin.rtc;
 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.framework.common.util.collection.CollectionUtils;
 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteMoreReqVO;
 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteReqVO;
 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallRespVO;
 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcGroupCallRespVO;
+import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcParticipantDO;
+import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantStatusEnum;
+import cn.iocoder.yudao.module.im.framework.config.ImProperties;
 import cn.iocoder.yudao.module.im.service.rtc.ImRtcCallService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -14,6 +19,10 @@ import jakarta.validation.Valid;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
 import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
 import static cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils.getLoginUserId;
@@ -25,17 +34,28 @@ public class ImRtcCallController {
     @Resource
     private ImRtcCallService rtcCallService;
+    @Resource
+    private ImProperties imProperties;
     @PostMapping("/invite")
     @Operation(summary = "发起通话；私聊或群聊根据 scene 区分")
     public CommonResult<ImRtcCallRespVO> invite(@Valid @RequestBody ImRtcCallInviteReqVO reqVO) {
-        return success(rtcCallService.invite(getLoginUserId(), reqVO));
+        ImRtcCallDO call = rtcCallService.inviteCall(getLoginUserId(), reqVO);
+        return success(buildCallResp(call, getLoginUserId()));
+    }
+
+    @PostMapping("/join")
+    @Operation(summary = "加入已有群通话；用于胶囊条「加入」按钮")
+    @Parameter(name = "roomName", description = "LiveKit 房间名", required = true, example = "im_rtc_call_xxxx")
+    public CommonResult<ImRtcCallRespVO> joinCall(@RequestParam("roomName") String roomName) {
+        ImRtcCallDO call = rtcCallService.joinCall(getLoginUserId(), roomName);
+        return success(buildCallResp(call, getLoginUserId()));
     }
     @PostMapping("/invite-more")
     @Operation(summary = "通话中添加成员；仅群通话可用")
     public CommonResult<Boolean> inviteMore(@Valid @RequestBody ImRtcCallInviteMoreReqVO reqVO) {
-        rtcCallService.inviteMore(getLoginUserId(), reqVO);
+        rtcCallService.inviteMoreCall(getLoginUserId(), reqVO);
         return success(true);
     }
@@ -43,14 +63,15 @@ public class ImRtcCallController {
     @Operation(summary = "接听通话")
     @Parameter(name = "roomName", description = "LiveKit 房间名", required = true, example = "call_friend_1_2")
     public CommonResult<ImRtcCallRespVO> accept(@RequestParam("roomName") String roomName) {
-        return success(rtcCallService.accept(getLoginUserId(), roomName));
+        ImRtcCallDO call = rtcCallService.acceptCall(getLoginUserId(), roomName);
+        return success(buildCallResp(call, getLoginUserId()));
     }
     @PostMapping("/reject")
     @Operation(summary = "拒绝通话")
     @Parameter(name = "roomName", description = "LiveKit 房间名", required = true, example = "call_friend_1_2")
     public CommonResult<Boolean> reject(@RequestParam("roomName") String roomName) {
-        rtcCallService.reject(getLoginUserId(), roomName);
+        rtcCallService.rejectCall(getLoginUserId(), roomName);
         return success(true);
     }
@@ -58,7 +79,7 @@ public class ImRtcCallController {
     @Operation(summary = "取消邀请；主叫接通前调用")
     @Parameter(name = "roomName", description = "LiveKit 房间名", required = true, example = "call_friend_1_2")
     public CommonResult<Boolean> cancel(@RequestParam("roomName") String roomName) {
-        rtcCallService.cancel(getLoginUserId(), roomName);
+        rtcCallService.cancelCall(getLoginUserId(), roomName);
         return success(true);
     }
@@ -66,7 +87,7 @@ public class ImRtcCallController {
     @Operation(summary = "离开通话；接通后调用")
     @Parameter(name = "roomName", description = "LiveKit 房间名", required = true, example = "call_friend_1_2")
     public CommonResult<Boolean> leave(@RequestParam("roomName") String roomName) {
-        rtcCallService.leave(getLoginUserId(), roomName);
+        rtcCallService.leaveCall(getLoginUserId(), roomName);
         return success(true);
     }
@@ -74,14 +95,56 @@ public class ImRtcCallController {
     @Operation(summary = "重新签发 Token；客户端重连或 Token 过期续期")
     @Parameter(name = "roomName", description = "LiveKit 房间名", required = true, example = "call_friend_1_2")
     public CommonResult<ImRtcCallRespVO> refreshToken(@RequestParam("roomName") String roomName) {
-        return success(rtcCallService.refreshToken(getLoginUserId(), roomName));
+        ImRtcCallDO call = rtcCallService.refreshCallToken(getLoginUserId(), roomName);
+        return success(buildCallResp(call, getLoginUserId()));
     }
-    @GetMapping("/group-active-call")
-    @Operation(summary = "查询群当前进行中的通话；用于群聊顶部「N 人正在通话」胶囊条")
+    @GetMapping("/get-active-call")
+    @Operation(summary = "查询当前进行中的通话；用于群聊顶部「N 人正在通话」胶囊条")
     @Parameter(name = "groupId", description = "群编号", required = true, example = "2048")
-    public CommonResult<ImRtcGroupCallRespVO> getGroupActiveCall(@RequestParam("groupId") Long groupId) {
-        return success(rtcCallService.getGroupActiveCall(getLoginUserId(), groupId));
+    public CommonResult<ImRtcGroupCallRespVO> getActiveCall(@RequestParam("groupId") Long groupId) {
+        ImRtcCallDO call = rtcCallService.getActiveCall(getLoginUserId(), groupId);
+        return success(buildGroupActiveResp(call));
+    }
+
+    // ========== VO 拼装 ==========
+
+    private ImRtcCallRespVO buildCallResp(ImRtcCallDO call, Long userId) {
+        if (call == null) {
+            return null;
+        }
+        List<ImRtcParticipantDO> participants = rtcCallService.getCallParticipants(call.getCallId());
+        return new ImRtcCallRespVO()
+                .setCallId(call.getCallId()).setRoomName(call.getRoomName())
+                .setLivekitUrl(imProperties.getRtc().getLivekitUrl())
+                .setToken(rtcCallService.signCallToken(userId, call.getRoomName()))
+                .setScene(call.getConversationType()).setMediaType(call.getMediaType())
+                .setStatus(call.getStatus()).setInviterId(call.getInviterUserId())
+                .setGroupId(call.getGroupId())
+                .setInviteeIds(filterUserIds(participants, ImRtcParticipantStatusEnum.INVITING))
+                .setJoinedUserIds(filterUserIds(participants, ImRtcParticipantStatusEnum.JOINED));
+    }
+
+    private ImRtcGroupCallRespVO buildGroupActiveResp(ImRtcCallDO call) {
+        if (call == null) {
+            return null;
+        }
+        List<ImRtcParticipantDO> participants = rtcCallService.getCallParticipants(call.getCallId());
+        return new ImRtcGroupCallRespVO()
+                .setCallId(call.getCallId()).setRoomName(call.getRoomName())
+                .setGroupId(call.getGroupId()).setMediaType(call.getMediaType())
+                .setInviterId(call.getInviterUserId())
+                .setJoinedUserIds(filterUserIds(participants, ImRtcParticipantStatusEnum.JOINED))
+                .setInviteeIds(filterUserIds(participants, ImRtcParticipantStatusEnum.INVITING));
+    }
+
+    /**
+     * 按状态过滤参与者 userId；用 LinkedHashSet 保留前端展示顺序
+     */
+    private static Set<Long> filterUserIds(List<ImRtcParticipantDO> participants,
+                                           ImRtcParticipantStatusEnum status) {
+        return CollectionUtils.convertLinkedSet(participants, ImRtcParticipantDO::getUserId,
+                p -> Objects.equals(p.getStatus(), status.getStatus()));
     }
 }
 package cn.iocoder.yudao.module.im.controller.admin.rtc;
-import cn.hutool.core.codec.Base64;
 import cn.hutool.core.util.StrUtil;
-import cn.hutool.jwt.JWT;
 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
-import cn.iocoder.yudao.module.im.framework.config.ImProperties;
+import cn.iocoder.yudao.framework.tenant.core.aop.TenantIgnore;
+import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitClient;
 import cn.iocoder.yudao.module.im.service.rtc.ImRtcCallService;
-import cn.iocoder.yudao.module.im.service.rtc.dto.LiveKitWebhookEventDTO;
+import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitWebhookEventDTO;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.annotation.Resource;
@@ -20,15 +19,10 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
-import java.nio.charset.StandardCharsets;
-import java.security.MessageDigest;
-import java.util.Objects;
-
 import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
-// TODO @AI：是不是改成 ImRtcLiveKitController 更合适？？？
 /**
- * LiveKit Webhook 入口
+ * LiveKit 厂商对接入口（Webhook 回调）
  * <p>
  * 安全由请求签名（JWT + body sha256）保证，不走登录鉴权；伪造请求会被签名校验直接拒绝
  *
@@ -38,70 +32,39 @@ import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
 @RestController
 @RequestMapping("/im/livekit")
 @Slf4j
-public class ImRtcWebhookController {
+public class ImRtcLiveKitController {
     @Resource
-    private ImProperties imProperties;
+    private LiveKitClient liveKitClient;
     @Resource
     private ImRtcCallService rtcCallService;
     @PostMapping(value = "/webhook", consumes = MediaType.APPLICATION_JSON_VALUE)
     @Operation(summary = "LiveKit Webhook 回调；接收成员离开 / 房间结束等事件做业务态兜底清理")
     @PermitAll
+    @TenantIgnore
     public CommonResult<Boolean> webhook(HttpServletRequest request, @RequestBody String rawBody) {
-        // TODO @AI：1.1 1.2 这种
-        // 1. 校验签名；伪造请求直接 200 但忽略，避免给攻击者反馈
-        if (!verifySignature(request, rawBody)) {
-            log.warn("[livekit-webhook][签名校验失败 ip={} body={}]", request.getRemoteAddr(), rawBody);
+        // 1.1 校验签名；伪造请求直接 200 但忽略，避免给攻击者反馈
+        if (!liveKitClient.verifyWebhookSignature(request.getHeader("Authorization"), rawBody)) {
+            log.warn("[webhook][签名校验失败 ip={} bodyLength={}]",
+                    request.getRemoteAddr(), rawBody == null ? 0 : rawBody.length());
             return success(false);
         }
-        // 2. 解析事件并交给 service 处理；幂等由 service 自己保证
+        // 1.2 解析事件载荷；非法 / 空 event 直接忽略
         LiveKitWebhookEventDTO event = JsonUtils.parseObject(rawBody, LiveKitWebhookEventDTO.class);
         if (event == null || StrUtil.isBlank(event.getEvent())) {
             return success(false);
         }
-        // TODO @AI：2. ；这里是不是要打个 logger info？
+        // 2. 交给 service 处理；幂等由 service 自己保证
+        log.info("[webhook][事件处理 event={} roomName={}]", event.getEvent(),
+                event.getRoom() == null ? null : event.getRoom().getName());
         try {
             rtcCallService.handleLiveKitEvent(event);
         } catch (Exception e) {
-            log.error("[livekit-webhook][事件处理失败 event={} body={}]", event.getEvent(), rawBody, e);
+            log.error("[webhook][事件处理失败 event={} body={}]", event.getEvent(), rawBody, e);
         }
         return success(true);
     }
-    // TODO @AI：是不是给 LiveKitClient 校验？
-    /**
-     * 校验 LiveKit Webhook 签名；流程参见
-     * https://docs.livekit.io/home/server/webhook/#receiving-webhooks
-     */
-    private boolean verifySignature(HttpServletRequest request, String rawBody) {
-        String authHeader = request.getHeader("Authorization");
-        if (StrUtil.isBlank(authHeader)) {
-            return false;
-        }
-        String token = StrUtil.removePrefix(authHeader, "Bearer ").trim();
-        ImProperties.Rtc cfg = imProperties.getRtc();
-        try {
-            JWT jwt = JWT.of(token);
-            // 1. JWT HS256 签名；密钥使用 LiveKit API Secret
-            if (!jwt.setKey(cfg.getApiSecret().getBytes(StandardCharsets.UTF_8)).verify()) {
-                return false;
-            }
-            // 2. body sha256 一致性校验；防止抓到 token 后篡改 body
-            Object expectedSha = jwt.getPayload("sha256");
-            if (expectedSha == null) {
-                return false;
-            }
-            byte[] digest = MessageDigest.getInstance("SHA-256")
-                    .digest(rawBody.getBytes(StandardCharsets.UTF_8));
-            String actualSha = Base64.encode(digest);
-            return Objects.equals(expectedSha.toString(), actualSha);
-        } catch (Exception e) {
-            // TODO @AI：是不是要加个什么 request 的参数，方便溯源？
-            log.warn("[livekit-webhook][签名解析异常]", e);
-            return false;
-        }
-    }
-
 }
@@ -42,7 +42,4 @@ public class ImRtcCallRespVO {
     @Schema(description = "已加入房间的成员编号集合")
     private Set<Long> joinedUserIds;
-    @Schema(description = "是否本次新建；true 主叫弹「等待对方接受」；false 直接进通话中（已有房间）")
-    private Boolean newCreated;
-
 }
@@ -31,6 +31,11 @@ import java.time.LocalDateTime;
 public class ImRtcCallDO extends BaseDO {
     /**
+     * LiveKit 房间名前缀；roomName 格式固定为 {@code ROOM_NAME_PREFIX + callId}
+     */
+    public static final String ROOM_NAME_PREFIX = "im_rtc_call_";
+
+    /**
      * 编号
      */
     @TableId
@@ -33,7 +33,7 @@ public interface ImRtcParticipantMapper extends BaseMapperX&lt;ImRtcParticipantDO&gt; 
                 .eq(ImRtcParticipantDO::getStatus, status));
     }
-    default ImRtcParticipantDO selectLastByUserIdAndStatusIn(Long userId, Collection<Integer> statuses) {
+    default ImRtcParticipantDO selectLastByUserIdAndStatus(Long userId, Collection<Integer> statuses) {
         return selectLast(new LambdaQueryWrapperX<ImRtcParticipantDO>()
                 .eq(ImRtcParticipantDO::getUserId, userId)
                 .in(ImRtcParticipantDO::getStatus, statuses));
@@ -90,5 +90,6 @@ public interface ErrorCodeConstants {
     ErrorCode RTC_PRIVATE_INVITEE_REQUIRED = new ErrorCode(1_040_900_006, "私聊通话必须指定对方");
     ErrorCode RTC_GROUP_REQUIRED = new ErrorCode(1_040_900_007, "群聊通话必须指定群编号");
     ErrorCode RTC_INVITE_BUSY = new ErrorCode(1_040_900_008, "通话发起繁忙，请稍后再试");
+    ErrorCode RTC_GROUP_CALL_ACTIVE = new ErrorCode(1_040_900_009, "该群已有进行中通话，请通过胶囊条加入");
 }
@@ -82,7 +82,7 @@ public enum ImMessageTypeEnum implements ArrayValuable&lt;Integer&gt; {
     /**
      * 对应 OpenIM：SignalingNotification 1601（通话信令统一入口）
      * 对应自己的类：ImRtcCallNotification
-     * 场景：通话信令；不入库，走 imWebSocketService 仅推参与方；signalingType 区分 INVITE / REJECT（预留 ACCEPT / CANCEL / HUNGUP）
+     * 场景：通话信令；不入库，走 imWebSocketService 仅推参与方；status 复用参与者状态枚举区分 INVITING / JOINED / REJECTED / NO_ANSWER / LEFT
      */
     RTC_CALL(1601, "通话信令", false, false),
     /**
 package cn.iocoder.yudao.module.im.framework.rtc.core;
+import cn.hutool.core.codec.Base64;
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.lang.Assert;
 import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.digest.DigestUtil;
 import cn.hutool.http.HttpRequest;
 import cn.hutool.http.HttpResponse;
 import cn.hutool.json.JSONArray;
@@ -13,11 +15,13 @@ import cn.iocoder.yudao.framework.common.util.http.HttpUtils;
 import cn.iocoder.yudao.module.im.framework.config.ImProperties;
 import lombok.extern.slf4j.Slf4j;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 /**
  * LiveKit 客户端
@@ -159,6 +163,45 @@ public class LiveKitClient {
     }
     /**
+     * 校验 LiveKit Webhook 签名；流程参见
+     * <a href="https://docs.livekit.io/home/server/webhook/#receiving-webhooks">webhook 文档</a>
+     * <p>
+     * 校验两步：
+     * 1.1 JWT HS256 签名验证；密钥使用 LiveKit API Secret
+     * 1.2 body 的 sha256 与 JWT 内 claim 一致；防止抓到 token 后篡改 body
+     *
+     * @param authHeader 请求头 Authorization 原值（含 "Bearer " 前缀）
+     * @param rawBody    请求原始 body
+     * @return 是否通过；签名异常一律视为不通过
+     */
+    public boolean verifyWebhookSignature(String authHeader, String rawBody) {
+        if (StrUtil.isBlank(authHeader)) {
+            return false;
+        }
+        String token = StrUtil.removePrefix(authHeader, "Bearer ").trim();
+        ImProperties.Rtc cfg = imProperties.getRtc();
+        try {
+            JWT jwt = JWT.of(token);
+            // JWT HS256 签名验证
+            if (!jwt.setKey(cfg.getApiSecret().getBytes(StandardCharsets.UTF_8)).verify()) {
+                return false;
+            }
+            // body sha256 一致性校验
+            Object expectedSha = jwt.getPayload("sha256");
+            if (expectedSha == null) {
+                return false;
+            }
+            // 计算 body 的 sha256，并与 JWT 内 claim 对比
+            String actualSha = Base64.encode(DigestUtil.sha256(rawBody));
+            return Objects.equals(expectedSha.toString(), actualSha);
+        } catch (Exception e) {
+            log.warn("[verifyWebhookSignature][签名解析异常 bodyLength={}]",
+                    rawBody == null ? 0 : rawBody.length(), e);
+            return false;
+        }
+    }
+
+    /**
      * 签发管理 Token；用于调 Server API（DeleteRoom / ListParticipants / RemoveParticipant 等）
      *
      * @return JWT 字符串
@@ -10,7 +10,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 /**
- * 僵尸通话清理 Job：兜底 LiveKit Webhook 丢失 / 客户端异常关闭等未调 leave 等场景
+ * 僵尸通话清理 Job：兜底 LiveKit Webhook 丢失 / 客户端异常关闭等未调 leave 的场景
  *
  * @author 芋道源码
  */
@@ -18,23 +18,38 @@ import org.springframework.stereotype.Component;
 @Slf4j
 public class ImRtcCallCleanupJob implements JobHandler {
+    /**
+     * 默认扫描阈值（分钟）；通话创建超过此值仍未结束才纳入扫描，避开「刚发起还在响铃」的合理零人态
+     */
+    private static final int DEFAULT_THRESHOLD_MINUTES = 5;
+
     @Resource
     private ImRtcCallService rtcCallService;
     /**
      * 执行清理
      *
-     * @param param 阈值（分钟）；为空走 Service 默认（5 分钟）
+     * @param param 阈值（分钟）；为空 / 非法走默认 {@link #DEFAULT_THRESHOLD_MINUTES}
      * @return 清理数量描述
      */
     @Override
     @TenantJob
     public String execute(String param) {
-        Integer thresholdMinutes = StrUtil.isNotBlank(param) && NumberUtil.isInteger(param)
-                ? Integer.parseInt(param) : null;
+        int thresholdMinutes = resolveThresholdMinutes(param);
         int cleaned = rtcCallService.cleanupZombieCalls(thresholdMinutes);
         log.info("[execute][清理僵尸通话数量 ({}) 个]", cleaned);
         return String.format("清理僵尸通话 %s 个", cleaned);
     }
+    /**
+     * 解析 quartz param 为分钟阈值；非法 / 非正数走默认值
+     */
+    private static int resolveThresholdMinutes(String param) {
+        if (StrUtil.isBlank(param) || !NumberUtil.isInteger(param)) {
+            return DEFAULT_THRESHOLD_MINUTES;
+        }
+        int minutes = Integer.parseInt(param);
+        return minutes > 0 ? minutes : DEFAULT_THRESHOLD_MINUTES;
+    }
+
 }
@@ -2,77 +2,103 @@ package cn.iocoder.yudao.module.im.service.rtc;
 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteMoreReqVO;
 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteReqVO;
-import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallRespVO;
-import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcGroupCallRespVO;
+import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
+import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcParticipantDO;
 import java.util.List;
-// TODO @AI：一个好友，只能一个通话；一个群，只能一个通话；（同一时间）；
 /**
  * IM 实时通话 Service
  * <p>
- * 业务模型：一个好友对 / 一个群同时只能有一个进行中的通话；invite 兼具「创建」与「加入已有」两种语义。
- * 媒体协商完全交给 LiveKit；后端只做：会话状态机、Token 签发、来电信令推送、通话历史落消息流。
+ * 业务模型：一个好友对 / 一个群同时只能有一个进行中的通话；inviteCall 仅负责「创建新通话」，加入已有群通话走 {@link #joinCall}
  *
  * @author 芋道源码
  */
 public interface ImRtcCallService {
-    // TODO @AI：是不是风格上，还是有 create 和 invite；这样两种逻辑上更好理解；
     /**
-     * 发起通话；同一 roomName 已有进行中通话时返回该会话并标记 newCreated=false
+     * 发起新通话；同好友对 / 同群已有进行中通话直接抛错（群通话场景应改走 {@link #joinCall}）
      *
      * @param userId 操作人编号；通常是当前登录用户
      * @param reqVO  请求参数（scene / mediaType / peerUserId 或 groupId）
-     * @return 包含 token / roomName / livekitUrl 的会话信息
+     * @return 通话主表
      */
-    ImRtcCallRespVO invite(Long userId, ImRtcCallInviteReqVO reqVO);
+    ImRtcCallDO inviteCall(Long userId, ImRtcCallInviteReqVO reqVO);
     /**
-     * 通话中追加成员；仅群通话场景可用；给新邀请人推 RTC_INVITE
+     * 加入已有群通话：用于群胶囊条「加入」按钮；旁观者作为 ACTIVE_JOIN 加入，邀请池内成员转 JOINED
+     *
+     * @param userId   加入者用户编号
+     * @param roomName LiveKit 房间名；从胶囊条 activeCall 拿
+     * @return 通话主表
+     */
+    ImRtcCallDO joinCall(Long userId, String roomName);
+
+    /**
+     * 通话中追加成员：仅群通话场景可用；给新邀请人推 RTC_CALL(INVITE)
      *
      * @param userId 操作人编号；必须是当前会话参与者
      * @param reqVO  房间名 + 新邀请的用户编号集合
      */
-    void inviteMore(Long userId, ImRtcCallInviteMoreReqVO reqVO);
+    void inviteMoreCall(Long userId, ImRtcCallInviteMoreReqVO reqVO);
     /**
-     * 接听通话；私聊状态切换 INVITING → ONGOING；群通话直接加入已有 ONGOING 房间
+     * 接听通话：私聊状态切换 INVITING → RUNNING；群通话直接加入已有 RUNNING 房间
+     *
+     * @return 通话主表
      */
-    ImRtcCallRespVO accept(Long userId, String roomName);
+    ImRtcCallDO acceptCall(Long userId, String roomName);
     /**
      * 拒绝通话；仅 INVITING 状态可拒；群通话拒绝等同于不参与，房间仍存在
      */
-    void reject(Long userId, String roomName);
+    void rejectCall(Long userId, String roomName);
     /**
      * 取消邀请；主叫在 INVITING 状态主动取消
      */
-    void cancel(Long userId, String roomName);
+    void cancelCall(Long userId, String roomName);
     /**
-     * 离开通话；ONGOING 状态下离开；私聊任一方离开 = 通话结束；群通话最后一人离开才结束
+     * 离开通话；RUNNING 状态下离开；私聊任一方离开 = 通话结束；群通话最后一人离开才结束
+     */
+    void leaveCall(Long userId, String roomName);
+
+    /**
+     * 校验用户仍是该通话的参与者，用于客户端重连或 Token 过期续期前的合法性检查；
+     * 签发新 Token 由 Controller 调 {@link #signCallToken} 完成
+     *
+     * @return 通话主表
      */
-    void leave(Long userId, String roomName);
+    ImRtcCallDO refreshCallToken(Long userId, String roomName);
     /**
-     * 重新签发 Token；用于客户端重连或 Token 过期续期
+     * 查询当前正在进行的通话；目前仅群聊场景（胶囊条），私聊未来扩展再补 peerUserId 参数
+     * <p>
+     * 鉴权：仅群活跃成员可查；防止任意用户探测群通话状态 / 拿到 inviter / inviteeIds 等敏感信息
+     *
+     * @param userId  操作人编号；通常是当前登录用户
+     * @param groupId 群编号
+     * @return 通话主表；不存在返回 null
      */
-    ImRtcCallRespVO refreshToken(Long userId, String roomName);
+    ImRtcCallDO getActiveCall(Long userId, Long groupId);
     /**
-     * 列出用户当前的活跃通话；用于 App 冷启动 / 推送点开恢复
+     * 查询某通话的全部参与者明细；交给 Controller 拼装 inviteeIds / joinedUserIds
+     *
+     * @param callId 业务通话编号
+     * @return 参与者明细列表
      */
-    List<ImRtcCallRespVO> getActiveSessions(Long userId);
+    List<ImRtcParticipantDO> getCallParticipants(String callId);
     /**
-     * 查询群当前正在进行的通话；用于群聊顶部「N 人正在通话」胶囊条
+     * 签发指定用户进入该通话的 LiveKit Token；供 Controller 拼接到响应 VO
      *
-     * @param groupId 群编号
-     * @return 群活跃通话；不存在返回 null；不含 token，仅展示用
+     * @param userId   进房用户编号；token 内 displayName 取该用户昵称
+     * @param roomName LiveKit 房间名
+     * @return JWT 字符串
      */
-    ImRtcGroupCallRespVO getGroupActiveCall(Long groupId);
+    String signCallToken(Long userId, String roomName);
     /**
      * 处理 LiveKit Webhook 事件；用于关 tab / 强杀 / 网络断开等异常退出场景的兜底清理
@@ -80,6 +106,14 @@ public interface ImRtcCallService {
      * 关键事件：participant_left（成员离开） / room_finished（房间结束）。前端正常 leave 时
      * 也会触发同样的 LiveKit 事件；此处需做幂等处理，session 已被业务接口移除时直接忽略。
      */
-    void handleLiveKitEvent(cn.iocoder.yudao.module.im.service.rtc.dto.LiveKitWebhookEventDTO event);
+    void handleLiveKitEvent(cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitWebhookEventDTO event);
+
+    /**
+     * 【定时任务调用】清理僵尸通话
+     *
+     * @param thresholdMinutes 通话创建超过此分钟数才纳入扫描；调用方保证 > 0
+     * @return 清理数量
+     */
+    int cleanupZombieCalls(int thresholdMinutes);
 }
+package cn.iocoder.yudao.module.im.service.websocket.dto.notification.rtc;
+
+import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
+import lombok.Data;
+
+/**
+ * RTC_PARTICIPANT_CONNECTED 通话参与者加入通知
+ * <p>
+ * 不入库；LiveKit webhook participant_joined 触发；私聊推 peer 多端 + inviter 多端，群聊全群广播
+ * <p>
+ * 前端 callStore 把 userId 追加进 joinedUserIds；胶囊条人数 +1；群聊场景首条 1602 携带通话元信息以便 首次填充胶囊条
+ *
+ * @author 芋道源码
+ */
+@Data
+public class ImRtcParticipantConnectedNotification {
+
+    /**
+     * 业务通话编号
+     */
+    private String callId;
+    /**
+     * LiveKit 房间名
+     */
+    private String roomName;
+    /**
+     * 加入的参与者用户编号
+     */
+    private Long userId;
+    /**
+     * 会话类型
+     */
+    private Integer conversationType;
+    /**
+     * 群编号；群通话场景必填
+     */
+    private Long groupId;
+    /**
+     * 媒体类型；群聊场景的非邀请成员靠这个字段 首次填充胶囊条
+     */
+    private Integer mediaType;
+    /**
+     * 发起人用户编号；群聊场景的非邀请成员靠这个字段 首次填充胶囊条
+     */
+    private Long inviterUserId;
+
+    /**
+     * 构造参与者加入通知；按 {@link ImRtcCallDO} 抽通话上下文，仅 userId 是变量
+     *
+     * @param call    通话主表
+     * @param userId  加入的参与者用户编号
+     * @return 通知载荷
+     */
+    public static ImRtcParticipantConnectedNotification of(ImRtcCallDO call, Long userId) {
+        ImRtcParticipantConnectedNotification notification = new ImRtcParticipantConnectedNotification();
+        notification.callId = call.getCallId();
+        notification.roomName = call.getRoomName();
+        notification.userId = userId;
+        notification.conversationType = call.getConversationType();
+        notification.groupId = call.getGroupId();
+        notification.mediaType = call.getMediaType();
+        notification.inviterUserId = call.getInviterUserId();
+        return notification;
+    }
+
+}
+package cn.iocoder.yudao.module.im.service.websocket.dto.notification.rtc;
+
+import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
+import lombok.Data;
+
+/**
+ * RTC_PARTICIPANT_DISCONNECTED 通话参与者离开通知
+ * <p>
+ * 不入库；LiveKit webhook participant_left 触发；推送范围同 {@link ImRtcParticipantConnectedNotification}
+ * <p>
+ * 前端 callStore 把 userId 从 joinedUserIds 移除；胶囊条人数 -1
+ *
+ * @author 芋道源码
+ */
+@Data
+public class ImRtcParticipantDisconnectedNotification {
+
+    /**
+     * 业务通话编号
+     */
+    private String callId;
+    /**
+     * LiveKit 房间名
+     */
+    private String roomName;
+    /**
+     * 离开的参与者用户编号
+     */
+    private Long userId;
+    /**
+     * 会话类型
+     */
+    private Integer conversationType;
+    /**
+     * 群编号；群通话场景必填
+     */
+    private Long groupId;
+
+    /**
+     * 构造参与者离开通知；按 {@link ImRtcCallDO} 抽通话上下文，仅 userId 是变量
+     *
+     * @param call    通话主表
+     * @param userId  离开的参与者用户编号
+     * @return 通知载荷
+     */
+    public static ImRtcParticipantDisconnectedNotification of(ImRtcCallDO call, Long userId) {
+        ImRtcParticipantDisconnectedNotification notification = new ImRtcParticipantDisconnectedNotification();
+        notification.callId = call.getCallId();
+        notification.roomName = call.getRoomName();
+        notification.userId = userId;
+        notification.conversationType = call.getConversationType();
+        notification.groupId = call.getGroupId();
+        return notification;
+    }
+
+}