BJ_ZTEITS-Developers / parking-cloud-yudao

Commit b68303152808a56636fbf59e3e69a0ab44252c4f

Authored by YunaiV
1 parent 8b497fa6

✨ feat(im): rtc 交流通信的部分

Inline Side-by-side
Showing 1 changed file with 610 additions and 498 deletions
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/ImRtcCallServiceImpl.java
  View file @b683031
... ... @@ -2,31 +2,36 @@ package cn.iocoder.yudao.module.im.service.rtc;
2 2  
3 3 import cn.hutool.core.collection.CollUtil;
4 4 import cn.hutool.core.util.IdUtil;
5   -import cn.hutool.core.util.StrUtil;
6   -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
7   -import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
  5 +import cn.iocoder.yudao.framework.common.util.collection.CollectionUtils;
8 6 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteMoreReqVO;
9 7 import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteReqVO;
10   -import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallRespVO;
11   -import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcGroupCallRespVO;
12   -import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupMemberDO;
  8 +import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
  9 +import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcParticipantDO;
  10 +import cn.iocoder.yudao.module.im.dal.mysql.rtc.ImRtcCallMapper;
  11 +import cn.iocoder.yudao.module.im.dal.mysql.rtc.ImRtcParticipantMapper;
  12 +import cn.iocoder.yudao.module.im.dal.redis.rtc.ImRtcCallLockRedisDAO;
13 13 import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
14   -import cn.iocoder.yudao.module.im.enums.friend.ImFriendStateEnum;
15 14 import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
16   -import cn.iocoder.yudao.module.im.enums.rtc.ImCallEndReasonEnum;
17   -import cn.iocoder.yudao.module.im.enums.rtc.ImCallStatusEnum;
  15 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallEndReasonEnum;
  16 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantRoleEnum;
  17 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantStatusEnum;
  18 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallStatusEnum;
18 19 import cn.iocoder.yudao.module.im.framework.config.ImProperties;
19   -import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitTokenUtils;
  20 +import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitClient;
20 21 import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
21 22 import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService;
22   -import cn.iocoder.yudao.module.im.service.rtc.bo.ImCallSessionBO;
23   -import cn.iocoder.yudao.module.im.service.rtc.bo.ImCallSessionStore;
24   -import cn.iocoder.yudao.module.im.service.rtc.dto.*;
  23 +import cn.iocoder.yudao.module.im.service.message.ImGroupMessageService;
  24 +import cn.iocoder.yudao.module.im.service.message.ImPrivateMessageService;
  25 +import cn.iocoder.yudao.module.im.service.message.dto.ImGroupMessageSendDTO;
  26 +import cn.iocoder.yudao.module.im.service.message.dto.ImPrivateMessageSendDTO;
  27 +import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitWebhookEventDTO;
  28 +import cn.iocoder.yudao.module.im.service.websocket.dto.notification.rtc.*;
25 29 import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
26 30 import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
27 31 import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
28 32 import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
29 33 import jakarta.annotation.Resource;
  34 +import lombok.SneakyThrows;
30 35 import lombok.extern.slf4j.Slf4j;
31 36 import org.springframework.stereotype.Service;
32 37 import org.springframework.validation.annotation.Validated;
... ... @@ -41,6 +46,18 @@ import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
41 46  
42 47 /**
43 48 * IM 实时通话 Service 实现
  49 + * <p>
  50 + * 存储模型：DB 单一存储（im_rtc_call 主表 + im_rtc_participant 明细表）
  51 + * <p>
  52 + * 并发幂等：同好友对 / 同群活跃唯一性走 {@link ImRtcCallLockRedisDAO} 分布式锁 + 锁内 SELECT 兜底；webhook 兜底走条件 UPDATE；
  53 + * <p>
  54 + * 推送通道分流：
  55 + * 1601 RTC_CALL（INVITING / JOINED / REJECTED / NO_ANSWER / LEFT 子类型）→ {@link ImWebSocketService#sendPrivateMessageAsync} 仅推参与方；
  56 + * 1602 / 1603 PARTICIPANT_CONNECTED / DISCONNECTED → {@link ImWebSocketService} 推参与方 + 群通话场景广播全群；
  57 + * 1610 RTC_CALL_START + 1611 RTC_CALL_END → {@link ImPrivateMessageService} / {@link ImGroupMessageService} 入消息流当聊天 tip
  58 + * （START 仅群通话；两者分别在 invite / cancel(leave) 事务里 INSERT，自增 id 自然保证顺序）
  59 + * <p>
  60 + * 职责边界：媒体协商完全交给 LiveKit；后端只做会话状态机、Token 签发、来电信令推送、通话历史落消息流；房内媒体流变化交给 LiveKit 客户端事件（TrackSubscribed 等），后端不重复推
44 61 *
45 62 * @author 芋道源码
46 63 */
... ... @@ -49,9 +66,13 @@ import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
49 66 @Slf4j
50 67 public class ImRtcCallServiceImpl implements ImRtcCallService {
51 68  
52   - // TODO @AI：后续存储到 DB 里；（晚点改）
53 69 @Resource
54   - private ImCallSessionStore sessionStore;
  70 + private ImRtcCallMapper rtcCallMapper;
  71 + // TODO @AI：可以搞个 ImRtcParticipantService，从而简化这个类的大小么？
  72 + @Resource
  73 + private ImRtcParticipantMapper rtcParticipantMapper;
  74 + @Resource
  75 + private ImRtcCallLockRedisDAO rtcCallLockRedisDAO;
55 76  
56 77 @Resource
57 78 private ImGroupMemberService groupMemberService;
... ... @@ -59,6 +80,13 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
59 80 private ImFriendService friendService;
60 81 @Resource
61 82 private ImWebSocketService webSocketService;
  83 + @Resource
  84 + private ImPrivateMessageService privateMessageService;
  85 + @Resource
  86 + private ImGroupMessageService groupMessageService;
  87 +
  88 + @Resource
  89 + private LiveKitClient liveKitClient;
62 90  
63 91 @Resource
64 92 private AdminUserApi adminUserApi;
... ... @@ -66,411 +94,507 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
66 94 @Resource
67 95 private ImProperties imProperties;
68 96  
  97 + // ========== 业务接口 ==========
  98 +
69 99 @Override
70   - public ImRtcCallRespVO invite(Long userId, ImRtcCallInviteReqVO reqVO) {
71   - // TODO @AI：按照 1.1 1.2 1.3 这样搞；
  100 + @SneakyThrows
  101 + public ImRtcCallDO inviteCall(Long userId, ImRtcCallInviteReqVO reqVO) {
72 102 ensureEnabled();
73   - Integer scene = reqVO.getScene();
74 103 // 1. 校验入参与场景
75 104 validateInviteScene(userId, reqVO);
76   - // 2. 派生 roomName；同好友对 / 同群只对应一个 roomName
77   - String roomName = deriveRoomName(scene, userId, reqVO.getPeerUserId(), reqVO.getGroupId());
78   - // 3. 用户级忙线检查；当前用户已在另一通话且不是同房间则拒绝
79   - String selfActive = sessionStore.getActiveRoom(userId);
80   - // TODO @AI：notEquals 方案；
81   - if (selfActive != null && !selfActive.equals(roomName)) {
  105 + // TODO @AI：2 这个注释，有点复杂。略微重复了。应该就是加锁 + 所内逻辑。为什么加锁：xxxx
  106 + // 2. 包锁：同好友 / 同群串行；锁内 SELECT 命中已有活跃通话改走加入分支，否则 INSERT 新通话
  107 + if (ImConversationTypeEnum.isGroup(reqVO.getScene())) {
  108 + return rtcCallLockRedisDAO.lockGroup(reqVO.getGroupId(), () -> inviteCall0(userId, reqVO));
  109 + }
  110 + return rtcCallLockRedisDAO.lockPrivate(userId, reqVO.getPeerUserId(), () -> inviteCall0(userId, reqVO));
  111 + }
  112 +
  113 + /**
  114 + * invite 锁内主体；SELECT 已有活跃通话 → 直接抛错（群走 joinCall，私聊用户已是参与者）；否则 INSERT 新通话
  115 + */
  116 + private ImRtcCallDO inviteCall0(Long userId, ImRtcCallInviteReqVO reqVO) {
  117 + // TODO @AI：是不是可以改成：私聊的检测、群聊的检测；通用的检测；现在 if else 有点混乱。
  118 + // 1.1 已有活跃通话：群通话提示走加入按钮；私聊一定是当前用户已在通话中（自身忙线）
  119 + ImRtcCallDO active = ImConversationTypeEnum.isGroup(reqVO.getScene())
  120 + // TODO @AI：selectLastOneByXXX；然后 basemapperx 也改成这样的 selectLastOne；这样整体风格对齐；
  121 + ? rtcCallMapper.selectLastByGroupIdAndStatusIn(reqVO.getGroupId(), ImRtcCallStatusEnum.ACTIVE_STATUSES)
  122 + : getActivePrivateCallByPair(userId, reqVO.getPeerUserId());
  123 + if (active != null) {
  124 + throw exception(ImConversationTypeEnum.isGroup(reqVO.getScene())
  125 + ? RTC_GROUP_CALL_ACTIVE : RTC_SELF_BUSY);
  126 + }
  127 + // 1.2 自身忙线：当前用户已在另一通话；私聊场景再额外检查对端 TODO @AI：是不是要拆成 2 段注释。就是“私聊场景再额外检查对端”下面的“ImConversationTypeEnum.isPrivate(reqVO.getScene())”
  128 + ImRtcParticipantDO selfActive = rtcParticipantMapper.selectLastByUserIdAndStatus(userId, ImRtcParticipantStatusEnum.ACTIVE_STATUSES);
  129 + if (selfActive != null) {
82 130 throw exception(RTC_SELF_BUSY);
83 131 }
84   - // 4. 私聊场景下检查被叫忙线；群通话候选成员各自决定接听，不在此拦截
85   - // TODO @AI：需要在看看，群聊会不会出现这个情况？感觉应该也会，可以调研下微信，以及其他 im
86   - if (ImConversationTypeEnum.isPrivate(scene)
87   - && sessionStore.isBusy(reqVO.getPeerUserId())) {
88   - String peerActive = sessionStore.getActiveRoom(reqVO.getPeerUserId());
89   - // TODO @AI：notEquals 方案；
90   - if (peerActive != null && !peerActive.equals(roomName)) {
  132 + if (ImConversationTypeEnum.isPrivate(reqVO.getScene())) {
  133 + ImRtcParticipantDO peerActive = rtcParticipantMapper.selectLastByUserIdAndStatus(reqVO.getPeerUserId(), ImRtcParticipantStatusEnum.ACTIVE_STATUSES);
  134 + if (peerActive != null) {
91 135 throw exception(RTC_PEER_BUSY);
92 136 }
93 137 }
94 138  
95   - // 5. 原子获取或创建会话；并发同时点呼叫只会建一份
96   - boolean[] newCreated = {false};
97   - ImCallSessionBO session = sessionStore.getOrCreate(roomName, key -> {
98   - newCreated[0] = true;
99   - return buildNewSession(key, userId, reqVO);
100   - });
101   - // 6. 老 session：本次属于"加入已有通话"；新 session：发邀请信令并把发起人加入房间
102   - sessionStore.bindUser(userId, roomName);
103   - session.getJoinedUserIds().add(userId);
104   - if (newCreated[0]) {
105   - pushInviteNotifications(session, userId);
106   - // 群通话首次发起：向全群广播 STARTED，胶囊条立刻在所有成员客户端出现
107   - if (ImConversationTypeEnum.isGroup(session.getScene())) {
108   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_STARTED.getType(), userId);
109   - }
110   - } else if (ImConversationTypeEnum.isGroup(session.getScene())) {
111   - // 群通话主动加入已有通话：广播 UPDATED 让胶囊条人数刷新
112   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_UPDATED.getType(), userId);
  139 + // 2. 新建通话
  140 + return doInsertNewCall(userId, reqVO);
  141 + }
  142 +
  143 + // TODO @AI：改成 createCall 方法；
  144 + // TODO @AI：希望逻辑是：1 构造参数，2.1 插入主表、2.2 插入参与表；3. 推送通知；
  145 + private ImRtcCallDO doInsertNewCall(Long inviterId, ImRtcCallInviteReqVO reqVO) {
  146 + // 1. 派生 callId + roomName
  147 + String callId = IdUtil.fastSimpleUUID();
  148 + String roomName = ImRtcCallDO.ROOM_NAME_PREFIX + callId;
  149 + LocalDateTime now = LocalDateTime.now();
  150 + // 2. 构建被邀请池（私聊为 peerUserId 单元素；群聊取 invitee 子集 / 全员）
  151 + Set<Long> invitees = resolveInvitees(reqVO, inviterId);
  152 +
  153 + // 3. INSERT 主表；群聊发起人即时 JOINED 但通话仍处 CREATED，等首个非发起人接通才切 RUNNING
  154 + ImRtcCallDO call = new ImRtcCallDO()
  155 + .setCallId(callId).setRoomName(roomName)
  156 + .setConversationType(reqVO.getScene()).setMediaType(reqVO.getMediaType())
  157 + .setInviterUserId(inviterId).setGroupId(reqVO.getGroupId())
  158 + .setStatus(ImRtcCallStatusEnum.CREATED.getStatus())
  159 + .setStartTime(now);
  160 + rtcCallMapper.insert(call);
  161 + // TODO @AI：是不是 4 和 5，改成 批量插入；
  162 + // 4. INSERT 发起人 participant
  163 + rtcParticipantMapper.insert(new ImRtcParticipantDO()
  164 + .setCallId(callId).setUserId(inviterId)
  165 + .setRole(ImRtcParticipantRoleEnum.INVITER.getRole())
  166 + .setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus())
  167 + .setInviteTime(now).setAcceptTime(now));
  168 + // 5. INSERT 被邀请人 participants
  169 + for (Long inviteeId : invitees) {
  170 + rtcParticipantMapper.insert(new ImRtcParticipantDO()
  171 + .setCallId(callId).setUserId(inviteeId)
  172 + .setRole(ImRtcParticipantRoleEnum.INVITEE.getRole())
  173 + .setStatus(ImRtcParticipantStatusEnum.INVITING.getStatus())
  174 + .setInviteTime(now));
  175 + }
  176 + // 6. 推送 RTC_CALL(INVITE) 给被邀请人；群通话再向全群广播 RTC_CALL_START 入聊天流
  177 + AdminUserRespDTO inviterUser = adminUserApi.getUser(inviterId);
  178 + for (Long inviteeId : invitees) {
  179 + pushCallInviteNotification(call, inviterUser, inviteeId);
  180 + }
  181 + if (ImConversationTypeEnum.isGroup(reqVO.getScene())) {
  182 + pushCallStartNotification(call, inviterUser);
  183 + }
  184 + return call;
  185 + }
  186 +
  187 + @Override
  188 + public ImRtcCallDO joinCall(Long userId, String roomName) {
  189 + ensureEnabled();
  190 + // 1.1 校验通话存在且活跃
  191 + ImRtcCallDO call = requireActiveCall(roomName);
  192 + // 1.2 仅群通话支持「旁观者加入」
  193 + if (!ImConversationTypeEnum.isGroup(call.getConversationType())) {
  194 + throw exception(RTC_GROUP_REQUIRED);
113 195 }
114   - // 7. 给当前调用方签 token
115   - return buildResp(session, userId, newCreated[0]);
  196 +
  197 + // 2. TODO 把 joinExistingCall 拿进来？避免方法太多太碎。
  198 + return joinExistingCall(call, userId);
  199 + }
  200 +
  201 + /**
  202 + * 命中已有活跃通话；当前用户加入或重新进入；不再触发 INVITE / START 广播
  203 + */
  204 + private ImRtcCallDO joinExistingCall(ImRtcCallDO active, Long userId) {
  205 + ImRtcParticipantDO existing = rtcParticipantMapper.selectByCallIdAndUserId(active.getCallId(), userId);
  206 + LocalDateTime now = LocalDateTime.now();
  207 + if (existing != null) {
  208 + // 非 JOINED 状态切回 JOINED + accept_time；INVITER / INVITEE 重连 / 旁观点胶囊条都走这条
  209 + if (!ImRtcParticipantStatusEnum.isJoined(existing.getStatus())) {
  210 + rtcParticipantMapper.updateById(new ImRtcParticipantDO()
  211 + .setId(existing.getId())
  212 + .setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus())
  213 + .setAcceptTime(now));
  214 + }
  215 + } else {
  216 + // 旁观者主动加入：INSERT role=ACTIVE_JOIN
  217 + rtcParticipantMapper.insert(new ImRtcParticipantDO()
  218 + .setCallId(active.getCallId()).setUserId(userId)
  219 + .setRole(ImRtcParticipantRoleEnum.ACTIVE_JOIN.getRole())
  220 + .setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus())
  221 + .setInviteTime(now).setAcceptTime(now));
  222 + }
  223 + // 主表 CREATED → RUNNING（首次有非发起人加入）
  224 + maybeMarkOngoing(active, userId, now);
  225 + return active;
116 226 }
117 227  
118 228 @Override
119   - public ImRtcCallRespVO accept(Long userId, String roomName) {
120   - // TODO @AI：按照 1.1 1.2 1.3 这样搞；（注释分层一点）
  229 + public void inviteMoreCall(Long userId, ImRtcCallInviteMoreReqVO reqVO) {
121 230 ensureEnabled();
122   - ImCallSessionBO session = requireSession(roomName);
123   - // TODO @AI：减少 ONGOING、INVITING 的混用；最好是中文（英文），大家更好理解。
124   - // 群通话处于 ONGOING；私聊必须 INVITING 才能接听
125   - if (ImConversationTypeEnum.isPrivate(session.getScene())
126   - && !ImCallStatusEnum.isInviting(session.getStatus())) {
127   - throw exception(RTC_SESSION_NOT_EXISTS);
  231 + ImRtcCallDO call = requireActiveCall(reqVO.getRoomName());
  232 + // 仅群通话支持追加邀请
  233 + if (!ImConversationTypeEnum.isGroup(call.getConversationType())) {
  234 + throw exception(RTC_GROUP_REQUIRED);
128 235 }
129   - if (!session.containsUser(userId)) {
  236 + // 操作者必须是已在房间的参与者
  237 + ImRtcParticipantDO operator = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userId);
  238 + if (operator == null || !ImRtcParticipantStatusEnum.isJoined(operator.getStatus())) {
130 239 throw exception(RTC_NOT_PARTICIPANT);
131 240 }
132   - // 私聊接听 → 切到 ONGOING；群通话已经是 ONGOING；幂等：重复 accept 不报错
133   - if (ImConversationTypeEnum.isPrivate(session.getScene())) {
134   - session.setStatus(ImCallStatusEnum.ONGOING.getType());
135   - // 振铃定时器在前端，后端不做 cancel 动作
  241 + // 校验被邀请人必须是该群活跃成员；且未在通话池
  242 + Set<Long> validMemberIds = new LinkedHashSet<>(
  243 + groupMemberService.getActiveGroupMemberUserIdsByGroupId(call.getGroupId()));
  244 + Set<Long> existingUserIds = rtcParticipantMapper.selectListByCallId(call.getCallId()).stream()
  245 + .map(ImRtcParticipantDO::getUserId)
  246 + .collect(Collectors.toSet());
  247 + List<Long> incoming = reqVO.getInviteeIds().stream()
  248 + .filter(id -> validMemberIds.contains(id) && !existingUserIds.contains(id))
  249 + .toList();
  250 + if (CollUtil.isEmpty(incoming)) {
  251 + return;
  252 + }
  253 + // INSERT 新邀请人 + 推 INVITE
  254 + LocalDateTime now = LocalDateTime.now();
  255 + for (Long inviteeId : incoming) {
  256 + rtcParticipantMapper.insert(new ImRtcParticipantDO()
  257 + .setCallId(call.getCallId()).setUserId(inviteeId)
  258 + .setRole(ImRtcParticipantRoleEnum.INVITEE.getRole())
  259 + .setStatus(ImRtcParticipantStatusEnum.INVITING.getStatus())
  260 + .setInviteTime(now));
136 261 }
137   - session.getJoinedUserIds().add(userId);
138   - sessionStore.bindUser(userId, roomName);
139   - // 推 RTC_ACCEPT 给除自己外的全部相关方
140   - pushAcceptNotification(session, userId);
141   - // 群通话有人接听 → 全群广播 UPDATED 刷新胶囊条
142   - if (ImConversationTypeEnum.isGroup(session.getScene())) {
143   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_UPDATED.getType(), userId);
  262 + AdminUserRespDTO inviter = adminUserApi.getUser(userId);
  263 + for (Long inviteeId : incoming) {
  264 + pushCallInviteNotification(call, inviter, inviteeId);
144 265 }
145   - return buildResp(session, userId, false);
146 266 }
147 267  
148 268 @Override
149   - public void reject(Long userId, String roomName) {
150   - // TODO @AI：按照 1.1 1.2 1.3 这样搞；（注释分层一点）
151   - ensureEnabled(); // TODO @AI：是不是只有发起校验，其它地方先不校验了；
152   - ImCallSessionBO session = requireSession(roomName);
153   - if (!session.containsUser(userId)) {
  269 + public ImRtcCallDO acceptCall(Long userId, String roomName) {
  270 + ensureEnabled();
  271 + ImRtcCallDO call = requireActiveCall(roomName);
  272 + ImRtcParticipantDO participant = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userId);
  273 + if (participant == null) {
154 274 throw exception(RTC_NOT_PARTICIPANT);
155 275 }
156   - // 群通话拒绝：不影响其他成员，房间继续存在
157   - if (ImConversationTypeEnum.isGroup(session.getScene())) {
158   - session.getRejectedUserIds().add(userId);
159   - sessionStore.unbindUserFromRoom(userId, roomName);
160   - // 仅广播 UPDATED；之前向主叫推 RTC_END 会被前端错误识别为「整个通话结束」并 reset 主叫的通话窗
161   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_UPDATED.getType(), userId);
162   - return;
  276 + // 已 JOINED 视为重复 accept，幂等返回
  277 + LocalDateTime now = LocalDateTime.now();
  278 + if (!ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
  279 + // 仅 INVITING → JOINED；其它状态拒
  280 + if (!ImRtcParticipantStatusEnum.isInviting(participant.getStatus())) {
  281 + throw exception(RTC_SESSION_NOT_EXISTS);
  282 + }
  283 + int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(),
  284 + ImRtcParticipantStatusEnum.INVITING.getStatus(),
  285 + new ImRtcParticipantDO()
  286 + .setId(participant.getId())
  287 + .setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus())
  288 + .setAcceptTime(now));
  289 + if (updated == 0) {
  290 + throw exception(RTC_SESSION_NOT_EXISTS);
  291 + }
163 292 }
164   - // 私聊拒绝：仅 INVITING 状态可拒；推 RTC_END 给所有相关方并关房
165   - if (!ImCallStatusEnum.isInviting(session.getStatus())) {
  293 + // 主表 CREATED → RUNNING（首次有非发起人接通）
  294 + maybeMarkOngoing(call, userId, now);
  295 + return call;
  296 + }
  297 +
  298 + @Override
  299 + public void rejectCall(Long userId, String roomName) {
  300 + ensureEnabled();
  301 + ImRtcCallDO call = requireActiveCall(roomName);
  302 + ImRtcParticipantDO participant = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userId);
  303 + if (participant == null) {
  304 + throw exception(RTC_NOT_PARTICIPANT);
  305 + }
  306 + // 仅 INVITING 状态可拒
  307 + if (!ImRtcParticipantStatusEnum.isInviting(participant.getStatus())) {
166 308 throw exception(RTC_SESSION_NOT_EXISTS);
167 309 }
168   - endSession(session, userId, ImCallEndReasonEnum.REJECT);
  310 + int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(),
  311 + ImRtcParticipantStatusEnum.INVITING.getStatus(),
  312 + new ImRtcParticipantDO()
  313 + .setId(participant.getId())
  314 + .setStatus(ImRtcParticipantStatusEnum.REJECTED.getStatus())
  315 + .setLeaveTime(LocalDateTime.now()));
  316 + if (updated == 0) {
  317 + return;
  318 + }
  319 + // 群通话拒绝：仅推 RTC_CALL(REJECT) 给主叫；不影响通话整体
  320 + if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
  321 + pushCallRejectNotification(call, userId);
  322 + return;
  323 + }
  324 + // 私聊拒绝：走 endSession（推 RTC_CALL_END(reason=REJECT)）
  325 + endSession(call, userId, ImRtcCallEndReasonEnum.REJECT);
169 326 }
170 327  
171 328 @Override
172   - public void cancel(Long userId, String roomName) {
173   - // TODO @AI：按照 1.1 1.2 1.3 这样搞；（注释分层一点）
  329 + public void cancelCall(Long userId, String roomName) {
174 330 ensureEnabled();
175   - ImCallSessionBO session = requireSession(roomName);
176   - if (!Objects.equals(session.getInviterId(), userId)) {
  331 + ImRtcCallDO call = requireActiveCall(roomName);
  332 + if (!Objects.equals(call.getInviterUserId(), userId)) {
177 333 throw exception(RTC_NOT_PARTICIPANT);
178 334 }
179   - // 仅 INVITING 状态可取消；ONGOING 应走 leave
180   - if (!ImCallStatusEnum.isInviting(session.getStatus())) {
  335 + // 仅 CREATED 状态可取消（RUNNING 应走 leave）
  336 + if (!ImRtcCallStatusEnum.isCreated(call.getStatus())) {
181 337 throw exception(RTC_SESSION_NOT_EXISTS);
182 338 }
183   - endSession(session, userId, ImCallEndReasonEnum.CANCEL);
  339 + endSession(call, userId, ImRtcCallEndReasonEnum.CANCEL);
184 340 }
185 341  
186 342 @Override
187   - public void leave(Long userId, String roomName) {
188   - // TODO @AI：按照 1.1 1.2 1.3 这样搞；（注释分层一点）
  343 + public void leaveCall(Long userId, String roomName) {
189 344 ensureEnabled();
190   - ImCallSessionBO session = requireSession(roomName);
191   - if (!session.containsUser(userId)) {
  345 + ImRtcCallDO call = requireActiveCall(roomName);
  346 + ImRtcParticipantDO participant = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userId);
  347 + if (participant == null) {
192 348 throw exception(RTC_NOT_PARTICIPANT);
193 349 }
194   - session.getLeftUserIds().add(userId);
195   - session.getJoinedUserIds().remove(userId);
196   - sessionStore.unbindUserFromRoom(userId, roomName);
197   - // 关房条件：私聊任一方离开 = 关；群通话仅当所有被邀请人都已应答 + 在房无人时才关
198   - // 发起人离开后被邀请人仍可继续接听，房间不应当立即销毁
199   - // TODO @AI：单聊、群聊，是否可以复用一个逻辑；private 本质，只是一个特殊的方式而已；如果 private 后续支持邀请人，那 private 这个判断，就是 bug 了呀。其它地方，看看是不是也调整下；
200   - if (ImConversationTypeEnum.isPrivate(session.getScene())
201   - || shouldCloseGroupRoom(session)) {
202   - endSession(session, userId, ImCallEndReasonEnum.HANGUP);
203   - return;
204   - }
205   - // 群通话单人离开：仅广播 UPDATED；不推 RTC_END 避免主叫误关通话窗
206   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_UPDATED.getType(), userId);
  350 + LocalDateTime now = LocalDateTime.now();
  351 + rtcParticipantMapper.updateByIdAndStatus(participant.getId(), participant.getStatus(),
  352 + new ImRtcParticipantDO()
  353 + .setId(participant.getId())
  354 + .setStatus(ImRtcParticipantStatusEnum.LEFT.getStatus())
  355 + .setLeaveTime(now));
  356 + // 关房条件：私聊任一方离开 = 关；群通话仅在「无人在房 + 无人响铃」时关
  357 + if (ImConversationTypeEnum.isPrivate(call.getConversationType())
  358 + || shouldCloseGroupRoom(call.getCallId())) {
  359 + endSession(call, userId, ImRtcCallEndReasonEnum.HANGUP);
  360 + }
  361 + // 群通话单人离开：LiveKit `ParticipantDisconnected` 自动通知房内成员；业务后端不另外推
  362 + }
  363 +
  364 + /**
  365 + * 群通话是否应该关闭：房内没有 JOINED 就关
  366 + * <p>
  367 + * 群通话至少需要 1 个 JOINED 在房才有意义；如果只剩 INVITING（响铃中）的，接通后也只是 1 人独房，无价值
  368 + * <p>
  369 + * 关房后 endSession 会把残留 INVITING 批量改 NO_ANSWER 并推 RTC_CALL_END，响铃端 UI 自动收敛
  370 + */
  371 + private boolean shouldCloseGroupRoom(String callId) {
  372 + List<ImRtcParticipantDO> joined = rtcParticipantMapper.selectListByCallIdAndStatus(callId,
  373 + ImRtcParticipantStatusEnum.JOINED.getStatus());
  374 + return CollUtil.isEmpty(joined);
207 375 }
208 376  
209 377 /**
210   - * 群通话是否应该关闭：在房成员空 + 没有还在响铃的被邀请人
  378 + * 查询两人共同所在的活跃私聊通话；由应用层 Redisson 锁串行化，最多一个
211 379 * <p>
212   - * 仅 INVITED 状态下未接受 / 拒绝 / 离开的成员视为「还在响铃」；这样发起人提前离开后，被邀请人仍能接听并独立组成通话，符合微信群通话语义
  380 + * 拆成 3 段简单查询：拿 A 的活跃 participant → 拿主表判私聊未结束 → 看 B 是否在同 call 活跃
213 381 */
214   - private boolean shouldCloseGroupRoom(ImCallSessionBO session) {
215   - // TODO @AI：CollUtil
216   - if (!session.getJoinedUserIds().isEmpty()) {
217   - return false;
218   - }
219   - // TODO @AI：CollUtil？
220   - if (session.getInviteeIds() == null) {
221   - return true;
222   - }
223   - // TODO @AI：exists 或者哪种？这样更简洁一点。
224   - for (Long id : session.getInviteeIds()) {
225   - boolean responded = session.getRejectedUserIds().contains(id)
226   - || session.getLeftUserIds().contains(id);
227   - if (!responded) {
228   - return false;
229   - }
  382 + private ImRtcCallDO getActivePrivateCallByPair(Long userIdA, Long userIdB) {
  383 + ImRtcParticipantDO aPart = rtcParticipantMapper.selectLastByUserIdAndStatus(userIdA, ImRtcParticipantStatusEnum.ACTIVE_STATUSES);
  384 + if (aPart == null) {
  385 + return null;
230 386 }
231   - return true;
  387 + ImRtcCallDO call = rtcCallMapper.selectByCallId(aPart.getCallId());
  388 + if (call == null
  389 + || !ImConversationTypeEnum.isPrivate(call.getConversationType())
  390 + || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
  391 + return null;
  392 + }
  393 + ImRtcParticipantDO bPart = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userIdB);
  394 + if (bPart == null
  395 + || !ImRtcParticipantStatusEnum.ACTIVE_STATUSES.contains(bPart.getStatus())) {
  396 + return null;
  397 + }
  398 + return call;
232 399 }
233 400  
234 401 @Override
235   - public ImRtcCallRespVO refreshToken(Long userId, String roomName) {
236   - ensureEnabled(); // TODO @AI：简化下；
237   - ImCallSessionBO session = requireSession(roomName);
238   - if (!session.containsUser(userId)) {
  402 + public ImRtcCallDO refreshCallToken(Long userId, String roomName) {
  403 + ensureEnabled();
  404 + ImRtcCallDO call = requireActiveCall(roomName);
  405 + ImRtcParticipantDO participant = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userId);
  406 + if (participant == null) {
239 407 throw exception(RTC_NOT_PARTICIPANT);
240 408 }
241   - return buildResp(session, userId, false);
  409 + return call;
242 410 }
243 411  
244 412 @Override
245   - public void inviteMore(Long userId, ImRtcCallInviteMoreReqVO reqVO) {
  413 + public ImRtcCallDO getActiveCall(Long userId, Long groupId) {
246 414 ensureEnabled();
247   - ImCallSessionBO session = requireSession(reqVO.getRoomName());
248   - // 仅群通话支持追加邀请
249   - // TODO @AI：如果放开私聊的邀请，代码改动量大么？（这里只讨论，不用改）
250   - if (!ImConversationTypeEnum.isGroup(session.getScene())) {
251   - throw exception(RTC_GROUP_REQUIRED);
252   - }
253   - // 操作者必须是已在房间的参与者
254   - if (!session.getJoinedUserIds().contains(userId)) {
255   - throw exception(RTC_NOT_PARTICIPANT);
256   - }
257   - // 过滤已在被邀请池或已加入的人；仅给新成员发邀请
258   - Set<Long> existing = new LinkedHashSet<>();
259   - if (session.getInviteeIds() != null) {
260   - existing.addAll(session.getInviteeIds());
261   - }
262   - existing.addAll(session.getJoinedUserIds());
263   - // 校验被邀请人必须是该群活跃成员；防止操作者夹带任意 userId 进群房间
264   - // TODO @AI：如果放开私聊的邀请，代码改动量大么？（这里只讨论，不用改）
265   - Set<Long> validMemberIds = new LinkedHashSet<>(
266   - groupMemberService.getActiveGroupMemberUserIdsByGroupId(session.getGroupId()));
267   - Set<Long> incoming = new LinkedHashSet<>();
268   - for (Long id : reqVO.getInviteeIds()) {
269   - if (!existing.contains(id) && validMemberIds.contains(id)) {
270   - incoming.add(id);
271   - }
272   - }
273   - // TODO @AI:CollUtil isempty 判断；
274   - if (incoming.isEmpty()) {
275   - return;
276   - }
277   - if (session.getInviteeIds() == null) {
278   - session.setInviteeIds(new LinkedHashSet<>());
279   - }
280   - session.getInviteeIds().addAll(incoming);
281   - // 给新人推 RTC_INVITE；签 token 复用 invite 的逻辑
282   - AdminUserRespDTO inviter = adminUserApi.getUser(userId);
283   - for (Long inviteeId : incoming) {
284   - ImRtcInviteNotification payload = buildInvitePayload(session, userId, inviter, inviteeId);
285   - // TODO @AI：链式调用，相同的在一行；
286   - ImPrivateMessageDTO dto = new ImPrivateMessageDTO()
287   - .setType(ImMessageTypeEnum.RTC_INVITE.getType())
288   - .setSenderId(userId)
289   - .setReceiverId(inviteeId)
290   - .setContent(JsonUtils.toJsonString(payload))
291   - .setSendTime(LocalDateTime.now());
292   - webSocketService.sendPrivateMessageAsync(inviteeId, dto);
293   - }
294   - // 全群广播 UPDATED；让胶囊条上「等待加入」头像同步
295   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_UPDATED.getType(), userId);
  415 + // 1. 鉴权：仅群活跃成员能查（走单行 SQL，不依赖成员列表缓存）
  416 + groupMemberService.validateMemberInGroup(groupId, userId);
  417 + // 2. 查询活跃通话
  418 + return rtcCallMapper.selectLastByGroupIdAndStatusIn(groupId, ImRtcCallStatusEnum.ACTIVE_STATUSES);
296 419 }
297 420  
298 421 @Override
299   - public List<ImRtcCallRespVO> getActiveSessions(Long userId) {
300   - // TODO @AI：报错，前端就不应该发起；
301   - if (!imProperties.getRtc().isEnabled()) {
302   - return Collections.emptyList();
303   - }
304   - String roomName = sessionStore.getActiveRoom(userId);
305   - if (StrUtil.isBlank(roomName)) {
306   - return Collections.emptyList();
307   - }
308   - ImCallSessionBO session = sessionStore.get(roomName);
309   - if (session == null) {
310   - return Collections.emptyList();
311   - }
  422 + public List<ImRtcParticipantDO> getCallParticipants(String callId) {
  423 + return rtcParticipantMapper.selectListByCallId(callId);
  424 + }
312 425  
313   - // TODO @AI：返回 ImCallSessionBO？controller 拼接？（后面在改，所有涉及到各种翻译的地方）
314   - // 列表场景不签 token；前端要 token 时再调 refreshToken
315   - ImRtcCallRespVO vo = toBaseVO(session);
316   - return Collections.singletonList(vo);
  426 + @Override
  427 + public String signCallToken(Long userId, String roomName) {
  428 + AdminUserRespDTO user = adminUserApi.getUser(userId);
  429 + String displayName = user == null ? null : user.getNickname();
  430 + return signToken(userId, displayName, roomName);
317 431 }
318 432  
319 433 @Override
  434 + // TODO @AI：这个方法的编写，需要兼容 JDK8
320 435 public void handleLiveKitEvent(LiveKitWebhookEventDTO event) {
321   - // TODO @AI：需要 try catch 记录 hook + 成功 or 失败的日志么？
322 436 if (event == null || event.getEvent() == null) {
323 437 return;
324 438 }
325   - // TODO @AI：需要检验 jdk8 语法；
326   - // TODO @AI：livekit webhookevent 是不是搞个枚举类？
327 439 switch (event.getEvent()) {
328   - case "participant_left" -> handleParticipantLeft(event);
329   - case "room_finished" -> handleRoomFinished(event);
  440 + case LiveKitWebhookEventDTO.EVENT_PARTICIPANT_JOINED -> handleParticipantJoined(event);
  441 + case LiveKitWebhookEventDTO.EVENT_PARTICIPANT_LEFT -> handleParticipantLeft(event);
  442 + case LiveKitWebhookEventDTO.EVENT_ROOM_FINISHED -> handleRoomFinished(event);
330 443 default -> {
331   - // TODO @AI：仅仅 logger，后续按需搞；
332   - /* 其它事件忽略；participant_joined / track_published 等业务态由 LiveKit 自身分发驱动 */
  444 + /* 其它事件忽略；track_published 等业务态由 LiveKit 自身分发驱动 */
333 445 }
334 446 }
335 447 }
336 448  
337   - // TODO @AI：方法注释，里面的注释，按照 1. 2. 3. 阅读性更好
338   - // TODO @AI：处理必要的代码空行；阅读性更好（但是不要太多空行，层次感缺少）
339 449 /**
340   - * 处理 LiveKit「成员离开」事件；正常 leave 接口已经清理过的话此处幂等忽略
341   - * 私聊任一方离开 → 结束通话；群通话仅一人离开 → 广播；最后一人离开 → 结束通话
  450 + * 处理 LiveKit 「成员加入」事件；DB 接通态由 accept 接口写入，此处仅做 1602 转推
342 451 */
343   - private void handleParticipantLeft(LiveKitWebhookEventDTO event) {
  452 + private void handleParticipantJoined(LiveKitWebhookEventDTO event) {
344 453 if (event.getRoom() == null || event.getParticipant() == null) {
345 454 return;
346 455 }
347 456 String roomName = event.getRoom().getName();
348   - ImCallSessionBO session = sessionStore.get(roomName);
349   - if (session == null) {
  457 + ImRtcCallDO call = rtcCallMapper.selectByRoomName(roomName);
  458 + if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
350 459 return;
351 460 }
352   - Long userId = parseUserIdFromIdentity(event.getParticipant().getIdentity());
  461 + Long userId = liveKitClient.parseUserId(event.getParticipant().getIdentity());
353 462 if (userId == null) {
354 463 return;
355 464 }
356   - // 业务态可能已被 leave 接口清掉；contains 判断保证幂等
357   - boolean wasInRoom = session.getJoinedUserIds().remove(userId);
358   - sessionStore.unbindUserFromRoom(userId, roomName);
359   - if (!wasInRoom) {
  465 + pushParticipantConnectedNotification(call, userId);
  466 + }
  467 +
  468 + /**
  469 + * 处理 LiveKit 「成员离开」事件；正常 leave 接口已经清理过的话条件 UPDATE 自动幂等
  470 + */
  471 + private void handleParticipantLeft(LiveKitWebhookEventDTO event) {
  472 + if (event.getRoom() == null || event.getParticipant() == null) {
360 473 return;
361 474 }
362   - session.getLeftUserIds().add(userId);
363   - log.info("[handleParticipantLeft][roomName={} userId={} 由 LiveKit Webhook 兜底清理]", roomName, userId);
364   - // 关房条件与 leave 一致；群通话仅在「无人在房 + 无人响铃」时关
365   - if (ImConversationTypeEnum.isPrivate(session.getScene())
366   - || shouldCloseGroupRoom(session)) {
367   - endSession(session, userId, ImCallEndReasonEnum.HANGUP);
  475 + String roomName = event.getRoom().getName();
  476 + ImRtcCallDO call = rtcCallMapper.selectByRoomName(roomName);
  477 + if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
368 478 return;
369 479 }
370   - // 群通话单人异常离开：仅广播 UPDATED；同 leave 路径，避免主叫误关
371   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_UPDATED.getType(), userId);
372   - }
373   -
374   - // TODO @AI：方法注释，里面的注释，按照 1. 2. 3. 阅读性更好
375   - // TODO @AI：处理必要的代码空行；阅读性更好（但是不要太多空行，层次感缺少）
376   - /** 处理「房间结束」事件；兜底从 sessionStore 移除残留会话 */
377   - private void handleRoomFinished(LiveKitWebhookEventDTO event) {
378   - if (event.getRoom() == null) {
  480 + Long userId = liveKitClient.parseUserId(event.getParticipant().getIdentity());
  481 + if (userId == null) {
379 482 return;
380 483 }
381   - String roomName = event.getRoom().getName();
382   - ImCallSessionBO session = sessionStore.get(roomName);
383   - if (session == null) {
  484 + ImRtcParticipantDO participant = rtcParticipantMapper.selectByCallIdAndUserId(call.getCallId(), userId);
  485 + if (participant == null || !ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
  486 + return;
  487 + }
  488 + int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(),
  489 + ImRtcParticipantStatusEnum.JOINED.getStatus(),
  490 + new ImRtcParticipantDO()
  491 + .setId(participant.getId())
  492 + .setStatus(ImRtcParticipantStatusEnum.LEFT.getStatus())
  493 + .setLeaveTime(LocalDateTime.now()));
  494 + if (updated == 0) {
384 495 return;
385 496 }
386   - log.info("[handleRoomFinished][roomName={} 由 LiveKit Webhook 兜底清理]", roomName);
387   - // 复用 endSession：状态切 ENDED + 推 RTC_END + 群通话广播 ENDED + 移除内存索引
388   - endSession(session, null, ImCallEndReasonEnum.HANGUP);
  497 + log.info("[handleParticipantLeft][roomName={} userId={} 由 LiveKit Webhook 兜底]", roomName, userId);
  498 + // 推 1603 通知参与方 / 全群参与方
  499 + pushParticipantDisconnectedNotification(call, userId);
  500 + // 关房条件与 leave 一致；群通话仅在「无人在房 + 无人响铃」时关
  501 + if (ImConversationTypeEnum.isPrivate(call.getConversationType())
  502 + || shouldCloseGroupRoom(call.getCallId())) {
  503 + endSession(call, userId, ImRtcCallEndReasonEnum.HANGUP);
  504 + }
389 505 }
390 506  
391   - // TODO @AI：这里多端是必须的么？如果非必须，就先不用了。。。
392 507 /**
393   - * 从 LiveKit identity 解析业务 userId
394   - * <p>
395   - * 当前 identity 直接是 userId 字符串；预留 userId#terminal 多端格式的解析支持
  508 + * 处理 LiveKit 「房间结束」事件；兜底把 call 推到 ENDED
396 509 */
397   - private Long parseUserIdFromIdentity(String identity) {
398   - if (identity == null || identity.isEmpty()) {
399   - return null;
  510 + private void handleRoomFinished(LiveKitWebhookEventDTO event) {
  511 + if (event.getRoom() == null) {
  512 + return;
400 513 }
401   - int sep = identity.indexOf('#');
402   - String idPart = sep >= 0 ? identity.substring(0, sep) : identity;
403   - try {
404   - return Long.parseLong(idPart);
405   - } catch (NumberFormatException e) {
406   - log.warn("[parseUserIdFromIdentity][非法 identity={}]", identity);
407   - return null;
  514 + String roomName = event.getRoom().getName();
  515 + ImRtcCallDO call = rtcCallMapper.selectByRoomName(roomName);
  516 + if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
  517 + return;
408 518 }
  519 + log.info("[handleRoomFinished][roomName={} 由 LiveKit Webhook 兜底]", roomName);
  520 + endSession(call, null, ImRtcCallEndReasonEnum.HANGUP);
409 521 }
410 522  
411 523 @Override
412   - public ImRtcGroupCallRespVO getGroupActiveCall(Long groupId) {
413   - if (!imProperties.getRtc().isEnabled() || groupId == null) {
414   - return null;
415   - }
416   - ImCallSessionBO session = sessionStore.get("call_group_" + groupId);
417   - if (session == null) {
418   - return null;
419   - }
420   - // TODO @AI：链式调用；
421   - // TODO @AI：后续看看，能不能改成 controller 处理；
422   - ImRtcGroupCallRespVO vo = new ImRtcGroupCallRespVO();
423   - vo.setCallId(session.getCallId());
424   - vo.setRoomName(session.getRoomName());
425   - vo.setGroupId(session.getGroupId());
426   - vo.setMediaType(session.getMediaType());
427   - vo.setInviterId(session.getInviterId());
428   - vo.setJoinedUserIds(new LinkedHashSet<>(session.getJoinedUserIds()));
429   - if (session.getInviteeIds() != null) {
430   - vo.setInviteeIds(new LinkedHashSet<>(session.getInviteeIds()));
  524 + public int cleanupZombieCalls(int thresholdMinutes) {
  525 + // 阈值由调用方（Job）保证 > 0；低于 1 分钟会误杀刚发起的合理零人态
  526 + LocalDateTime threshold = LocalDateTime.now().minusMinutes(thresholdMinutes);
  527 + List<ImRtcCallDO> candidates = rtcCallMapper.selectListByStatusInAndStartTimeBefore(
  528 + ImRtcCallStatusEnum.ACTIVE_STATUSES, threshold);
  529 + if (CollUtil.isEmpty(candidates)) {
  530 + return 0;
  531 + }
  532 +
  533 + // 2. 逐个查 LiveKit 房间真实 participant 数
  534 + int cleaned = 0;
  535 + for (ImRtcCallDO call : candidates) {
  536 + int count;
  537 + try {
  538 + count = liveKitClient.listParticipants(call.getRoomName());
  539 + } catch (Exception e) {
  540 + log.warn("[cleanupZombieCalls][查询 LiveKit 失败 callId={} roomName={}]",
  541 + call.getCallId(), call.getRoomName(), e);
  542 + continue;
  543 + }
  544 + if (count != 0) {
  545 + continue;
  546 + }
  547 + log.info("[cleanupZombieCalls][清理僵尸通话 callId={} roomName={}]",
  548 + call.getCallId(), call.getRoomName());
  549 + endSession(call, null, ImRtcCallEndReasonEnum.HANGUP);
  550 + cleaned++;
431 551 }
432   - return vo;
  552 + return cleaned;
433 553 }
434 554  
435 555 // ========== 内部辅助 ==========
436 556  
437   - /** 校验通话功能开关 */
  557 + /**
  558 + * 校验通话功能开关
  559 + */
438 560 private void ensureEnabled() {
439 561 if (!imProperties.getRtc().isEnabled()) {
440 562 throw exception(RTC_NOT_ENABLED);
441 563 }
442 564 }
443 565  
444   - /** 校验 invite 入参；按场景区分必填字段 + 群通话校验当前用户是群活跃成员 */
  566 + // TODO @AI：validateXXX
  567 + /**
  568 + * 取活跃通话；不存在 / 已 ENDED 直接抛
  569 + */
  570 + private ImRtcCallDO requireActiveCall(String roomName) {
  571 + ImRtcCallDO call = rtcCallMapper.selectByRoomName(roomName);
  572 + if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
  573 + throw exception(RTC_SESSION_NOT_EXISTS);
  574 + }
  575 + return call;
  576 + }
  577 +
  578 + /**
  579 + * 校验 invite 入参；按场景区分必填字段；私聊补好友 / 黑名单校验，群聊补群成员校验
  580 + */
445 581 private void validateInviteScene(Long userId, ImRtcCallInviteReqVO reqVO) {
446 582 Integer scene = reqVO.getScene();
447 583 if (ImConversationTypeEnum.isPrivate(scene)) {
448   - // TODO @AI：这个在 validator，参数那校验；vo 里写 asserttrue；
449 584 if (reqVO.getPeerUserId() == null) {
450 585 throw exception(RTC_PRIVATE_INVITEE_REQUIRED);
451 586 }
452 587 if (Objects.equals(userId, reqVO.getPeerUserId())) {
453 588 throw exception(RTC_INVITE_SELF);
454 589 }
455   - // 好友 / 黑名单校验：和私聊消息发送同一套语义；NONE 已删 / 未加，BLOCKED 被对方拉黑
456   - // TODO @AI：是不是抽一个方法？validateFriendXXX（因为有多处调用）；
457   - ImFriendStateEnum state = friendService.getFriendState(userId, reqVO.getPeerUserId());
458   - if (state == ImFriendStateEnum.NONE) {
459   - throw exception(FRIEND_NOT_FRIEND);
460   - }
461   - if (state == ImFriendStateEnum.BLOCKED) {
462   - throw exception(FRIEND_BLOCKED_BY_PEER);
463   - }
  590 + // TODO DONE @AI：抽到 ImFriendService#validateFriend；
  591 + friendService.validateFriend(userId, reqVO.getPeerUserId());
464 592 return;
465 593 }
466 594 if (ImConversationTypeEnum.isGroup(scene)) {
467   - // TODO @AI：这个在 validator，参数那校验；vo 里写 asserttrue；
468 595 if (reqVO.getGroupId() == null) {
469 596 throw exception(RTC_GROUP_REQUIRED);
470 597 }
471   - // 当前用户必须是群活跃成员（包括主动加入已有通话的场景）；非成员不允许进入群房间
472   - // TODO @AI：是不是校验下，被邀请人，是不是这个群里的？
473   - // TODO @AI：member那搞个 ids 校验？避免太大的查询。
474 598 List<Long> activeMemberIds = groupMemberService
475 599 .getActiveGroupMemberUserIdsByGroupId(reqVO.getGroupId());
476 600 if (!activeMemberIds.contains(userId)) {
... ... @@ -478,69 +602,35 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
478 602 }
479 603 return;
480 604 }
481   - // TODO @AI：抛出异常的 scene 类型；
482 605 throw exception(RTC_PRIVATE_INVITEE_REQUIRED);
483 606 }
484 607  
485   - // TODO @AI在：注释风格；
486   - // TODO @AI：方法名，我们一般叫 buildXXX；
487   - /** 派生 roomName；私聊用好友对小大 ID，双向同 key；群聊用 groupId */
488   - private String deriveRoomName(Integer scene, Long selfId, Long peerId, Long groupId) {
489   - if (ImConversationTypeEnum.isPrivate(scene)) {
490   - long min = Math.min(selfId, peerId);
491   - long max = Math.max(selfId, peerId);
492   - return "call_friend_" + min + "_" + max;
493   - }
494   - return "call_group_" + groupId;
495   - }
496   -
497   - // TODO @AI：注释风格？
498   - /** 构造新会话；私聊起始 INVITING；群聊起始 ONGOING */
499   - private ImCallSessionBO buildNewSession(String roomName, Long inviterId, ImRtcCallInviteReqVO reqVO) {
500   - Set<Long> invitees = resolveInvitees(reqVO, inviterId);
501   - long now = System.currentTimeMillis();
502   - return ImCallSessionBO.builder()
503   - // TODO @AI：链式调用，相同的放在一行；
504   - .callId(IdUtil.fastSimpleUUID())
505   - .roomName(roomName)
506   - .scene(reqVO.getScene())
507   - .mediaType(reqVO.getMediaType())
508   - .inviterId(inviterId)
509   - .inviteeIds(invitees)
510   - .groupId(reqVO.getGroupId())
511   - // TODO @AI：是不是
512   - .status(ImConversationTypeEnum.isPrivate(reqVO.getScene())
513   - ? ImCallStatusEnum.INVITING.getType()
514   - : ImCallStatusEnum.ONGOING.getType())
515   - .startTime(now)
516   - .joinedUserIds(new LinkedHashSet<>())
517   - .rejectedUserIds(new LinkedHashSet<>())
518   - .leftUserIds(new LinkedHashSet<>())
519   - .build();
520   - }
521   -
522   - /** 解析被邀请池：私聊为 peerUserId 单元素；群聊优先用前端选中的子集，否则取群活跃成员 */
  608 + /**
  609 + * 解析被邀请池：私聊为 peerUserId 单元素；群聊优先用前端选中的子集，否则取群活跃成员；超量截断
  610 + * <p>
  611 + * 群聊场景：被邀请人必须是该群活跃成员；防止恶意客户端塞任意 userId 进群房间
  612 + */
523 613 private Set<Long> resolveInvitees(ImRtcCallInviteReqVO reqVO, Long inviterId) {
524   - // TODO @AI：私聊。
525 614 if (ImConversationTypeEnum.isPrivate(reqVO.getScene())) {
526   - Set<Long> set = new LinkedHashSet<>();
527   - set.add(reqVO.getPeerUserId());
528   - return set;
  615 + return CollUtil.newLinkedHashSet(reqVO.getPeerUserId());
529 616 }
530   -
531   - // 群通话：前端有传 inviteeIds 就用，否则回退拉群当前活跃成员
  617 + List<Long> activeMemberIds = groupMemberService
  618 + .getActiveGroupMemberUserIdsByGroupId(reqVO.getGroupId());
532 619 Set<Long> initial = new LinkedHashSet<>();
533   - // TODO @AI：必须传递，否则报错；
534 620 if (CollUtil.isNotEmpty(reqVO.getInviteeIds())) {
535   - initial.addAll(reqVO.getInviteeIds());
  621 + Set<Long> activeMemberSet = new HashSet<>(activeMemberIds);
  622 + for (Long inviteeId : reqVO.getInviteeIds()) {
  623 + if (!activeMemberSet.contains(inviteeId)) {
  624 + throw exception(GROUP_MEMBER_NOT_IN_GROUP);
  625 + }
  626 + initial.add(inviteeId);
  627 + }
536 628 } else {
537   - groupMemberService.getActiveGroupMemberListByGroupId(reqVO.getGroupId())
538   - .forEach(member -> initial.add(member.getUserId()));
  629 + initial.addAll(activeMemberIds);
539 630 }
540 631 // 不论来源，发起人本人不进被邀请池
541 632 initial.remove(inviterId);
542 633 // 超出最大同时在房成员数；截断为 max-1 个，避免邀请整个 500 人大群
543   - // TODO @AI：超出，直接报错，不要裁剪；
544 634 int max = imProperties.getRtc().getGroupMaxParticipants();
545 635 if (initial.size() + 1 > max) {
546 636 return initial.stream().limit(max - 1L)
... ... @@ -549,176 +639,198 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
549 639 return initial;
550 640 }
551 641  
552   - /** 推送来电信令；给除发起人外的所有被邀请人独立签 token */
553   - private void pushInviteNotifications(ImCallSessionBO session, Long inviterId) {
554   - AdminUserRespDTO inviter = adminUserApi.getUser(inviterId);
555   - for (Long inviteeId : session.getInviteeIds()) {
556   - ImRtcInviteNotification payload = buildInvitePayload(session, inviterId, inviter, inviteeId);
557   - // TODO @AI：链式调用；在一行，相同的；
558   - ImPrivateMessageDTO dto = new ImPrivateMessageDTO()
559   - .setType(ImMessageTypeEnum.RTC_INVITE.getType())
560   - .setSenderId(inviterId)
561   - .setReceiverId(inviteeId)
562   - .setContent(JsonUtils.toJsonString(payload))
563   - .setSendTime(LocalDateTime.now());
564   - webSocketService.sendPrivateMessageAsync(inviteeId, dto);
565   - }
566   - }
567   -
568   - /** 组装单条来电载荷；invite / inviteMore 共用 */
569   - private ImRtcInviteNotification buildInvitePayload(ImCallSessionBO session, Long inviterId,
570   - AdminUserRespDTO inviter, Long inviteeId) {
571   - // TODO @AI：链式调用，相同的在一行；
572   - ImRtcInviteNotification payload = new ImRtcInviteNotification();
573   - payload.setCallId(session.getCallId());
574   - payload.setRoomName(session.getRoomName());
575   - payload.setLivekitUrl(imProperties.getRtc().getLivekitUrl());
576   - payload.setToken(signToken(inviteeId,
577   - inviter == null ? null : inviter.getNickname(), session.getRoomName()));
578   - payload.setScene(session.getScene());
579   - payload.setMediaType(session.getMediaType());
580   - payload.setInviterId(inviterId);
581   - payload.setInviterNickname(inviter == null ? null : inviter.getNickname());
582   - payload.setInviterAvatar(inviter == null ? null : inviter.getAvatar());
583   - payload.setGroupId(session.getGroupId());
584   - return payload;
585   - }
586   -
587   - /** 推送接通信令；给除接听人外的所有相关方 */
588   - private void pushAcceptNotification(ImCallSessionBO session, Long acceptorId) {
589   - ImRtcAcceptNotification payload = new ImRtcAcceptNotification();
590   - // TODO @AI：链式调用，相同的在一行；
591   - payload.setCallId(session.getCallId());
592   - payload.setRoomName(session.getRoomName());
593   - payload.setAcceptorId(acceptorId);
594   - for (Long target : session.getAllUserIds()) {
595   - if (Objects.equals(target, acceptorId)) {
596   - continue;
597   - }
598   - // TODO @AI：链式调用，相同的在一行；
599   - ImPrivateMessageDTO dto = new ImPrivateMessageDTO()
600   - .setType(ImMessageTypeEnum.RTC_ACCEPT.getType())
601   - .setSenderId(acceptorId)
602   - .setReceiverId(target)
603   - .setContent(JsonUtils.toJsonString(payload))
604   - .setSendTime(LocalDateTime.now());
605   - webSocketService.sendPrivateMessageAsync(target, dto);
606   - }
607   - }
608   -
609   - /** 推送结束信令 */
610   - private void pushEndNotification(ImCallSessionBO session, Long operatorId,
611   - ImCallEndReasonEnum reason, boolean roomTerminated) {
612   - // TODO @AI：链式调用，相同的在一行；
613   - ImRtcEndNotification payload = new ImRtcEndNotification();
614   - payload.setCallId(session.getCallId());
615   - payload.setRoomName(session.getRoomName());
616   - payload.setOperatorId(operatorId);
617   - payload.setReason(reason.getType());
618   - if (roomTerminated && session.getStartTime() != null) {
619   - payload.setDurationSeconds((System.currentTimeMillis() - session.getStartTime()) / 1000L);
620   - }
621   - Set<Long> targets = roomTerminated
622   - ? session.getAllUserIds()
623   - : CollUtil.newLinkedHashSet(session.getInviterId());
624   - for (Long target : targets) {
625   - // TODO @AI：链式调用，相同的在一行；
626   - ImPrivateMessageDTO dto = new ImPrivateMessageDTO()
627   - .setType(ImMessageTypeEnum.RTC_END.getType())
628   - .setSenderId(operatorId == null ? session.getInviterId() : operatorId)
629   - .setReceiverId(target)
630   - .setContent(JsonUtils.toJsonString(payload))
631   - .setSendTime(LocalDateTime.now());
632   - webSocketService.sendPrivateMessageAsync(target, dto);
633   - }
634   - }
635   -
636   - // TODO @AI：注释风格；
637   - /** 关闭会话：状态切 ENDED → 推 RTC_END → 群通话再向全员广播 RTC_GROUP_ENDED → 从内存移除 */
638   - private void endSession(ImCallSessionBO session, Long operatorId, ImCallEndReasonEnum reason) {
639   - session.setStatus(ImCallStatusEnum.ENDED.getType());
640   - pushEndNotification(session, operatorId, reason, true);
641   - if (ImConversationTypeEnum.isGroup(session.getScene())) {
642   - pushGroupBroadcast(session, ImMessageTypeEnum.RTC_GROUP_ENDED.getType(), operatorId);
643   - }
644   - sessionStore.remove(session.getRoomName());
645   - log.info("[endSession][roomName={} operator={} reason={}]",
646   - session.getRoomName(), operatorId, reason);
  642 + /**
  643 + * 主表 CREATED → RUNNING；仅当首个非发起人加入时推进；条件 UPDATE 保幂等
  644 + * <p>
  645 + * 与 cancel / endSession 并发时 update 可能影响 0 行；reload 看真实状态：
  646 + * ENDED → 通话已被另一路径终结，抛 RTC_SESSION_NOT_EXISTS 让 accept 失败而不是返回假成功 token
  647 + * RUNNING → 已被另一接听者抢先推进，幂等同步内存
  648 + */
  649 + private void maybeMarkOngoing(ImRtcCallDO call, Long acceptorId, LocalDateTime now) {
  650 + if (!ImRtcCallStatusEnum.isCreated(call.getStatus())) {
  651 + return;
  652 + }
  653 + if (Objects.equals(call.getInviterUserId(), acceptorId)) {
  654 + return;
  655 + }
  656 + int updated = rtcCallMapper.updateByIdAndStatus(call.getId(),
  657 + ImRtcCallStatusEnum.CREATED.getStatus(),
  658 + new ImRtcCallDO()
  659 + .setStatus(ImRtcCallStatusEnum.RUNNING.getStatus())
  660 + .setAcceptTime(now));
  661 + if (updated > 0) {
  662 + // 内存值同步给后续判断
  663 + call.setStatus(ImRtcCallStatusEnum.RUNNING.getStatus());
  664 + call.setAcceptTime(now);
  665 + return;
  666 + }
  667 + // updated == 0：与并发路径竞争失败；reload 看真实终态
  668 + ImRtcCallDO latest = rtcCallMapper.selectById(call.getId());
  669 + if (latest == null || ImRtcCallStatusEnum.isEnded(latest.getStatus())) {
  670 + throw exception(RTC_SESSION_NOT_EXISTS);
  671 + }
  672 + call.setStatus(latest.getStatus());
  673 + call.setAcceptTime(latest.getAcceptTime());
647 674 }
648 675  
649 676 /**
650   - * 群通话广播：给该群当前所有活跃成员推一条信号；用于胶囊条「N 人正在通话 / 开始 / 结束」实时同步
651   - *
652   - * @param session 会话；必须是群通话
653   - * @param messageType 信号类型；取自 RTC_GROUP_STARTED / RTC_GROUP_UPDATED / RTC_GROUP_ENDED
654   - * @param operatorId 触发本次广播的操作人；STARTED 为发起人，ENDED 为关房操作者，UPDATED 为加入 / 离开者
  677 + * 关闭会话：主表条件 UPDATE 推到 ENDED → 残留 INVITING 批量改 NO_ANSWER → 推 RTC_CALL_END
655 678 */
656   - private void pushGroupBroadcast(ImCallSessionBO session, Integer messageType, Long operatorId) {
657   - if (!ImConversationTypeEnum.isGroup(session.getScene()) || session.getGroupId() == null) {
  679 + private void endSession(ImRtcCallDO call, Long operatorId, ImRtcCallEndReasonEnum reason) {
  680 + // 1.1 更新通话主表为已结束；条件 UPDATE 仅在 status 还活跃时生效
  681 + LocalDateTime now = LocalDateTime.now();
  682 + int updated = rtcCallMapper.updateByIdAndStatusIn(call.getId(), ImRtcCallStatusEnum.ACTIVE_STATUSES,
  683 + new ImRtcCallDO().setStatus(ImRtcCallStatusEnum.ENDED.getStatus())
  684 + .setEndReason(reason.getReason()).setEndTime(now));
  685 + if (updated == 0) {
  686 + log.info("[endSession][已被另一路径终结，跳过 callId={} roomName={} operator={} reason={}]",
  687 + call.getCallId(), call.getRoomName(), operatorId, reason);
658 688 return;
659 689 }
660   - // TODO @AI：链式调用，相同的在一行；
661   - ImRtcGroupNotification payload = new ImRtcGroupNotification();
662   - payload.setCallId(session.getCallId());
663   - payload.setRoomName(session.getRoomName());
664   - payload.setGroupId(session.getGroupId());
665   - payload.setMediaType(session.getMediaType());
666   - payload.setInviterId(session.getInviterId());
667   - payload.setJoinedUserIds(new LinkedHashSet<>(session.getJoinedUserIds()));
668   - if (session.getInviteeIds() != null) {
669   - payload.setInviteeIds(new LinkedHashSet<>(session.getInviteeIds()));
670   - }
671   - Long senderId = operatorId != null ? operatorId : session.getInviterId();
672   - String content = JsonUtils.toJsonString(payload);
673   - LocalDateTime now = LocalDateTime.now();
674   - List<ImGroupMemberDO> members = groupMemberService
675   - .getActiveGroupMemberListByGroupId(session.getGroupId());
676   - for (ImGroupMemberDO member : members) {
677   - ImPrivateMessageDTO dto = new ImPrivateMessageDTO()
678   - .setType(messageType)
679   - .setSenderId(senderId)
680   - .setReceiverId(member.getUserId())
681   - .setContent(content)
682   - .setSendTime(now);
683   - webSocketService.sendPrivateMessageAsync(member.getUserId(), dto);
  690 + // 1.2 更新参与表为已结束：残留 INVITING 改 NO_ANSWER；残留 JOINED 改 LEFT 并写 leaveTime
  691 + rtcParticipantMapper.updateByCallIdAndStatus(call.getCallId(), ImRtcParticipantStatusEnum.INVITING.getStatus(),
  692 + new ImRtcParticipantDO().setStatus(ImRtcParticipantStatusEnum.NO_ANSWER.getStatus()));
  693 + rtcParticipantMapper.updateByCallIdAndStatus(call.getCallId(), ImRtcParticipantStatusEnum.JOINED.getStatus(),
  694 + new ImRtcParticipantDO().setStatus(ImRtcParticipantStatusEnum.LEFT.getStatus()).setLeaveTime(now));
  695 +
  696 + // 2. 兜底删除 LiveKit 房间，强制断开异常残留客户端；失败仅记日志，不阻断业务
  697 + try {
  698 + liveKitClient.deleteRoom(call.getRoomName());
  699 + } catch (Exception e) {
  700 + log.warn("[endSession][删除 LiveKit 房间失败 callId={} roomName={} operator={} reason={}]",
  701 + call.getCallId(), call.getRoomName(), operatorId, reason, e);
684 702 }
  703 +
  704 + // 3. 推 RTC_CALL_END
  705 + Long durationSeconds = call.getAcceptTime() != null ?
  706 + Duration.between(call.getAcceptTime(), now).getSeconds() : null;
  707 + pushCallEndNotification(call, operatorId, reason, durationSeconds);
  708 + log.info("[endSession][roomName={} operator={} reason={}]", call.getRoomName(), operatorId, reason);
685 709 }
686 710  
687   - // TODO @AI：注释风格；
688   - /** 根据会话取出（不含 token 的）基础 VO */
689   - private ImRtcCallRespVO toBaseVO(ImCallSessionBO session) {
690   - return BeanUtils.toBean(session, ImRtcCallRespVO.class)
691   - .setLivekitUrl(imProperties.getRtc().getLivekitUrl());
  711 +
  712 + /**
  713 + * 通话事件的收件人池：私聊为双方参与者，群聊为群活跃成员
  714 + */
  715 + private Collection<Long> getCallAudience(ImRtcCallDO call) {
  716 + if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
  717 + return groupMemberService.getActiveGroupMemberUserIdsByGroupId(call.getGroupId());
  718 + }
  719 + return CollectionUtils.convertSet(
  720 + rtcParticipantMapper.selectListByCallId(call.getCallId()), ImRtcParticipantDO::getUserId);
692 721 }
693 722  
694   - /** 构造完整响应：含 token + livekitUrl + newCreated */
695   - // TODO @AI：controller 组装；
696   - private ImRtcCallRespVO buildResp(ImCallSessionBO session, Long userId, boolean newCreated) {
697   - AdminUserRespDTO user = adminUserApi.getUser(userId);
698   - String displayName = user == null ? null : user.getNickname();
699   - ImRtcCallRespVO vo = toBaseVO(session);
700   - vo.setToken(signToken(userId, displayName, session.getRoomName()));
701   - vo.setNewCreated(newCreated);
702   - return vo;
  723 + // ========== 推送 ==========
  724 +
  725 + /**
  726 + * RTC_CALL(INVITE)：走 webSocketService 直推到被邀请人
  727 + * <p>
  728 + * 不能走 imPrivateMessageService.send：persistent=false 时 send 仅推 sender 多端，receiver 不感知
  729 + */
  730 + private void pushCallInviteNotification(ImRtcCallDO call, AdminUserRespDTO inviter, Long inviteeId) {
  731 + String token = signToken(inviteeId, inviter == null ? null : inviter.getNickname(), call.getRoomName());
  732 + ImRtcCallNotification payload = ImRtcCallNotification.ofInvite(
  733 + call, inviter, imProperties.getRtc().getLivekitUrl(), token);
  734 + webSocketService.sendPrivateMessageAsync(inviteeId, ImPrivateMessageDTO.ofRtcNotification(
  735 + ImMessageTypeEnum.RTC_CALL.getType(), call.getInviterUserId(), inviteeId, payload));
703 736 }
704 737  
705   - /** 签 LiveKit Token；identity 用 userId 字符串保证多端唯一 */
706   - private String signToken(Long userId, String displayName, String roomName) {
707   - ImProperties.Rtc cfg = imProperties.getRtc();
708   - return LiveKitTokenUtils.signJoinToken(
709   - cfg.getApiKey(), cfg.getApiSecret(),
710   - String.valueOf(userId), displayName,
711   - roomName, Duration.ofHours(cfg.getTokenTtlHours()));
712   - }
713   -
714   - // TODO @AI：方法名，改成项目里，符合预期的；
715   - /** 拉取必存在的会话；不存在直接抛 */
716   - private ImCallSessionBO requireSession(String roomName) {
717   - ImCallSessionBO session = sessionStore.get(roomName);
718   - if (session == null) {
719   - throw exception(RTC_SESSION_NOT_EXISTS);
  738 + /**
  739 + * RTC_CALL(REJECT)：仅群通话场景；走 webSocketService 直推主叫
  740 + */
  741 + private void pushCallRejectNotification(ImRtcCallDO call, Long operatorUserId) {
  742 + AdminUserRespDTO operator = operatorUserId != null ? adminUserApi.getUser(operatorUserId) : null;
  743 + ImRtcCallNotification payload = ImRtcCallNotification.ofReject(call, operatorUserId, operator);
  744 + webSocketService.sendPrivateMessageAsync(call.getInviterUserId(), ImPrivateMessageDTO.ofRtcNotification(
  745 + ImMessageTypeEnum.RTC_CALL.getType(), operatorUserId, call.getInviterUserId(), payload));
  746 + }
  747 +
  748 + /**
  749 + * 通话参与者加入：LiveKit webhook participant_joined 触发；私聊推双方多端、群聊推全群成员（胶囊条 + 1）
  750 + *
  751 + * @param call 通话主表
  752 + * @param userId 加入的参与者用户编号
  753 + */
  754 + private void pushParticipantConnectedNotification(ImRtcCallDO call, Long userId) {
  755 + pushParticipantNotification(call, ImMessageTypeEnum.RTC_PARTICIPANT_CONNECTED.getType(), userId,
  756 + ImRtcParticipantConnectedNotification.of(call, userId));
  757 + }
  758 +
  759 + /**
  760 + * 通话参与者离开：LiveKit webhook participant_left 触发；推送范围同 {@link #pushParticipantConnectedNotification}
  761 + *
  762 + * @param call 通话主表
  763 + * @param userId 离开的参与者用户编号
  764 + */
  765 + private void pushParticipantDisconnectedNotification(ImRtcCallDO call, Long userId) {
  766 + pushParticipantNotification(call, ImMessageTypeEnum.RTC_PARTICIPANT_DISCONNECTED.getType(), userId,
  767 + ImRtcParticipantDisconnectedNotification.of(call, userId));
  768 + }
  769 +
  770 + /**
  771 + * 推送参与者事件的公共骨架；按会话类型决定收件人，单次 batch 推送扇出
  772 + */
  773 + private void pushParticipantNotification(ImRtcCallDO call, Integer type, Long actorUserId, Object payload) {
  774 + Collection<Long> receivers = getCallAudience(call);
  775 + if (CollUtil.isEmpty(receivers)) {
  776 + return;
720 777 }
721   - return session;
  778 + ImPrivateMessageDTO dto = ImPrivateMessageDTO.ofRtcNotification(type, actorUserId, null, payload);
  779 + webSocketService.sendPrivateMessageAsync(receivers, dto);
  780 + }
  781 +
  782 + /**
  783 + * RTC_CALL_START：仅群通话场景；走 imGroupMessageService.send，入 im_group_message 全群广播
  784 + * <p>
  785 + * 与 RTC_CALL_END 两段式配对：START 在本接口（invite）事务里 INSERT，END 在 cancel / leave 接口事务里 INSERT，
  786 + * 自增 id 自然保证聊天流顺序
  787 + */
  788 + private void pushCallStartNotification(ImRtcCallDO call, AdminUserRespDTO inviter) {
  789 + ImRtcCallStartNotification payload = ImRtcCallStartNotification.of(call, inviter);
  790 + ImGroupMessageSendDTO dto = new ImGroupMessageSendDTO()
  791 + .setGroupId(call.getGroupId())
  792 + .setType(ImMessageTypeEnum.RTC_CALL_START.getType())
  793 + .setContent(payload);
  794 + groupMessageService.sendGroupMessage(call.getInviterUserId(), dto);
  795 + }
  796 +
  797 + /**
  798 + * RTC_CALL_END：私聊走 imPrivateMessageService.send；群通话走 imGroupMessageService.send
  799 + * <p>
  800 + * senderId 始终用通话发起人，让前端按「谁发起通话」决定气泡左右；操作者从 payload.operatorUserId 拿
  801 + */
  802 + private void pushCallEndNotification(ImRtcCallDO call, Long operatorId, ImRtcCallEndReasonEnum reason,
  803 + Long durationSeconds) {
  804 + AdminUserRespDTO operator = operatorId != null ? adminUserApi.getUser(operatorId) : null;
  805 + ImRtcCallEndNotification payload = ImRtcCallEndNotification.of(call, reason, durationSeconds, operatorId, operator);
  806 + Long senderId = call.getInviterUserId();
  807 + if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
  808 + ImGroupMessageSendDTO dto = new ImGroupMessageSendDTO()
  809 + .setGroupId(call.getGroupId())
  810 + .setType(ImMessageTypeEnum.RTC_CALL_END.getType())
  811 + .setContent(payload);
  812 + groupMessageService.sendGroupMessage(senderId, dto);
  813 + return;
  814 + }
  815 + // 私聊：发起人总是 sender；receiver = 通话另一方
  816 + ImRtcParticipantDO peer = CollUtil.findOne(
  817 + rtcParticipantMapper.selectListByCallId(call.getCallId()),
  818 + p -> !Objects.equals(p.getUserId(), senderId));
  819 + Long receiverId = peer != null ? peer.getUserId() : senderId;
  820 + ImPrivateMessageSendDTO dto = new ImPrivateMessageSendDTO()
  821 + .setReceiverId(receiverId)
  822 + .setType(ImMessageTypeEnum.RTC_CALL_END.getType())
  823 + .setContent(payload);
  824 + privateMessageService.sendPrivateMessage(senderId, dto);
  825 + }
  826 +
  827 + // ========== Token / VO ==========
  828 +
  829 + /**
  830 + * 签 LiveKit Token
  831 + */
  832 + private String signToken(Long userId, String displayName, String roomName) {
  833 + return liveKitClient.signJoinToken(liveKitClient.buildIdentity(userId), displayName, roomName);
722 834 }
723 835  
724 836 }
... ...