From 8b497fa65516218c6325255df066d9421cbef349 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Tue, 12 May 2026 13:17:04 +0800 Subject: [PATCH] ✨ feat(im): 完善 rtc 的后端代码逻辑，各种代码风格的优化 --- yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java | 16 ++++++++++++++++ yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/http/HttpUtils.java | 10 ++++++++++ yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java | 85 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcLiveKitController.java | 70 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcWebhookController.java | 107 ----------------------------------------------------------------------------------------------------------- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java | 3 --- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java | 5 +++++ yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java | 2 +- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java | 1 + yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java | 2 +- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java | 43 +++++++++++++++++++++++++++++++++++++++++++ yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java | 23 +++++++++++++++++++---- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallService.java | 86 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------- yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantConnectedNotification.java | 66 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantDisconnectedNotification.java | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 15 files changed, 422 insertions(+), 153 deletions(-) create mode 100644 yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcLiveKitController.java delete mode 100644 yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcWebhookController.java create mode 100644 yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantConnectedNotification.java create mode 100644 yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantDisconnectedNotification.java diff --git a/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java b/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java index 7194b5e..b2db78b 100644 --- a/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java +++ b/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java @@ -124,6 +124,22 @@ public class CollectionUtils { return from.stream().filter(filter).map(func).filter(Objects::nonNull).collect(Collectors.toSet()); } + public static Set convertLinkedSet(Collection from, Function func) { + if (CollUtil.isEmpty(from)) { + return new LinkedHashSet<>(); + } + return from.stream().map(func).filter(Objects::nonNull) + .collect(Collectors.toCollection(LinkedHashSet::new)); + } + + public static Set convertLinkedSet(Collection from, Function func, Predicate 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 Map convertMapByFilter(Collection from, Predicate filter, Function keyFunc) { if (CollUtil.isEmpty(from)) { return new HashMap<>(); diff --git a/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/http/HttpUtils.java b/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/http/HttpUtils.java index 9de0377..ae04801 100644 --- a/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/http/HttpUtils.java +++ b/yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/http/HttpUtils.java @@ -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; + } + } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java index b668d6e..9ec95f5 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java @@ -1,10 +1,15 @@ 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 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 joinCall(@RequestParam("roomName") String roomName) { + ImRtcCallDO call = rtcCallService.joinCall(getLoginUserId(), roomName); + return success(buildCallResp(call, getLoginUserId())); } @PostMapping("/invite-more") @Operation(summary = "通话中添加成员；仅群通话可用") public CommonResult 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 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 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 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 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 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 getGroupActiveCall(@RequestParam("groupId") Long groupId) { - return success(rtcCallService.getGroupActiveCall(getLoginUserId(), groupId)); + public CommonResult 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 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 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 filterUserIds(List participants, + ImRtcParticipantStatusEnum status) { + return CollectionUtils.convertLinkedSet(participants, ImRtcParticipantDO::getUserId, + p -> Objects.equals(p.getStatus(), status.getStatus())); } } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcLiveKitController.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcLiveKitController.java new file mode 100644 index 0000000..140583b --- /dev/null +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcLiveKitController.java @@ -0,0 +1,70 @@ +package cn.iocoder.yudao.module.im.controller.admin.rtc; + +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +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.framework.rtc.core.LiveKitWebhookEventDTO; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.annotation.Resource; +import jakarta.annotation.security.PermitAll; +import jakarta.servlet.http.HttpServletRequest; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.MediaType; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success; + +/** + * LiveKit 厂商对接入口（Webhook 回调） + *

+ * 安全由请求签名（JWT + body sha256）保证，不走登录鉴权；伪造请求会被签名校验直接拒绝 + * + * @author 芋道源码 + */ +@Tag(name = "LiveKit Webhook 回调") +@RestController +@RequestMapping("/im/livekit") +@Slf4j +public class ImRtcLiveKitController { + + @Resource + private LiveKitClient liveKitClient; + @Resource + private ImRtcCallService rtcCallService; + + @PostMapping(value = "/webhook", consumes = MediaType.APPLICATION_JSON_VALUE) + @Operation(summary = "LiveKit Webhook 回调；接收成员离开 / 房间结束等事件做业务态兜底清理") + @PermitAll + @TenantIgnore + public CommonResult webhook(HttpServletRequest request, @RequestBody String 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); + } + // 1.2 解析事件载荷；非法 / 空 event 直接忽略 + LiveKitWebhookEventDTO event = JsonUtils.parseObject(rawBody, LiveKitWebhookEventDTO.class); + if (event == null || StrUtil.isBlank(event.getEvent())) { + return success(false); + } + + // 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("[webhook][事件处理失败 event={} body={}]", event.getEvent(), rawBody, e); + } + return success(true); + } + +} diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcWebhookController.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcWebhookController.java deleted file mode 100644 index 6de096d..0000000 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcWebhookController.java +++ /dev/null @@ -1,107 +0,0 @@ -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.module.im.service.rtc.ImRtcCallService; -import cn.iocoder.yudao.module.im.service.rtc.dto.LiveKitWebhookEventDTO; -import io.swagger.v3.oas.annotations.Operation; -import io.swagger.v3.oas.annotations.tags.Tag; -import jakarta.annotation.Resource; -import jakarta.annotation.security.PermitAll; -import jakarta.servlet.http.HttpServletRequest; -import lombok.extern.slf4j.Slf4j; -import org.springframework.http.MediaType; -import org.springframework.web.bind.annotation.PostMapping; -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 入口 - *

- * 安全由请求签名（JWT + body sha256）保证，不走登录鉴权；伪造请求会被签名校验直接拒绝 - * - * @author 芋道源码 - */ -@Tag(name = "LiveKit Webhook 回调") -@RestController -@RequestMapping("/im/livekit") -@Slf4j -public class ImRtcWebhookController { - - @Resource - private ImProperties imProperties; - @Resource - private ImRtcCallService rtcCallService; - - @PostMapping(value = "/webhook", consumes = MediaType.APPLICATION_JSON_VALUE) - @Operation(summary = "LiveKit Webhook 回调；接收成员离开 / 房间结束等事件做业务态兜底清理") - @PermitAll - public CommonResult 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); - return success(false); - } - // 2. 解析事件并交给 service 处理；幂等由 service 自己保证 - LiveKitWebhookEventDTO event = JsonUtils.parseObject(rawBody, LiveKitWebhookEventDTO.class); - if (event == null || StrUtil.isBlank(event.getEvent())) { - return success(false); - } - - // TODO @AI：2. ；这里是不是要打个 logger info？ - try { - rtcCallService.handleLiveKitEvent(event); - } catch (Exception e) { - log.error("[livekit-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; - } - } - -} diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java index c286891..42ce0e4 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java @@ -42,7 +42,4 @@ public class ImRtcCallRespVO { @Schema(description = "已加入房间的成员编号集合") private Set joinedUserIds; - @Schema(description = "是否本次新建；true 主叫弹「等待对方接受」；false 直接进通话中（已有房间）") - private Boolean newCreated; - } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java index ae56894..4ae62b3 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java @@ -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 diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java index 4cee275..960280e 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java @@ -33,7 +33,7 @@ public interface ImRtcParticipantMapper extends BaseMapperX .eq(ImRtcParticipantDO::getStatus, status)); } - default ImRtcParticipantDO selectLastByUserIdAndStatusIn(Long userId, Collection statuses) { + default ImRtcParticipantDO selectLastByUserIdAndStatus(Long userId, Collection statuses) { return selectLast(new LambdaQueryWrapperX() .eq(ImRtcParticipantDO::getUserId, userId) .in(ImRtcParticipantDO::getStatus, statuses)); diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java index 519c63e..311ac1d 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java @@ -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, "该群已有进行中通话，请通过胶囊条加入"); } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java index 7bc5b6e..15e9b2b 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java @@ -82,7 +82,7 @@ public enum ImMessageTypeEnum implements ArrayValuable { /** * 对应 OpenIM：SignalingNotification 1601（通话信令统一入口） * 对应自己的类：ImRtcCallNotification - * 场景：通话信令；不入库，走 imWebSocketService 仅推参与方；signalingType 区分 INVITE / REJECT（预留 ACCEPT / CANCEL / HUNGUP） + * 场景：通话信令；不入库，走 imWebSocketService 仅推参与方；status 复用参与者状态枚举区分 INVITING / JOINED / REJECTED / NO_ANSWER / LEFT */ RTC_CALL(1601, "通话信令", false, false), /** diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java index 40d44aa..10b7562 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java @@ -1,8 +1,10 @@ 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 签名；流程参见 + * webhook 文档 + *

+ * 校验两步： + * 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 字符串 diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java index 670601e..9cce516 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java @@ -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; + } + } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallService.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallService.java index 61f2c7e..0b78fa8 100644 --- a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallService.java +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallService.java @@ -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 *

- * 业务模型：一个好友对 / 一个群同时只能有一个进行中的通话；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 参数 + *

+ * 鉴权：仅群活跃成员可查；防止任意用户探测群通话状态 / 拿到 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 getActiveSessions(Long userId); + List 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); } diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantConnectedNotification.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantConnectedNotification.java new file mode 100644 index 0000000..283bc35 --- /dev/null +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantConnectedNotification.java @@ -0,0 +1,66 @@ +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 通话参与者加入通知 + *

+ * 不入库；LiveKit webhook participant_joined 触发；私聊推 peer 多端 + inviter 多端，群聊全群广播 + *

+ * 前端 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; + } + +} diff --git a/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantDisconnectedNotification.java b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantDisconnectedNotification.java new file mode 100644 index 0000000..ffe2671 --- /dev/null +++ b/yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcParticipantDisconnectedNotification.java @@ -0,0 +1,56 @@ +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 通话参与者离开通知 + *

+ * 不入库；LiveKit webhook participant_left 触发；推送范围同 {@link ImRtcParticipantConnectedNotification} + *

+ * 前端 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; + } + +} -- libgit2 0.21.4