package cn.iocoder.yudao.module.im.service.rtc;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.collection.CollectionUtils;
import cn.iocoder.yudao.module.im.controller.admin.manager.rtc.vo.ImRtcCallManagerPageReqVO;
import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallCreateReqVO;
import cn.iocoder.yudao.module.im.controller.admin.rtc.vo.ImRtcCallInviteReqVO;
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.dal.mysql.rtc.ImRtcCallMapper;
import cn.iocoder.yudao.module.im.dal.mysql.rtc.ImRtcParticipantMapper;
import cn.iocoder.yudao.module.im.dal.redis.rtc.ImRtcCallLockRedisDAO;
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallEndReasonEnum;
import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallStatusEnum;
import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantRoleEnum;
import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantStatusEnum;
import cn.iocoder.yudao.module.im.framework.config.ImProperties;
import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitClient;
import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitWebhookEventDTO;
import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
import cn.iocoder.yudao.module.im.service.group.ImGroupMemberService;
import cn.iocoder.yudao.module.im.service.message.ImGroupMessageService;
import cn.iocoder.yudao.module.im.service.message.ImPrivateMessageService;
import cn.iocoder.yudao.module.im.service.message.dto.ImGroupMessageSendDTO;
import cn.iocoder.yudao.module.im.service.message.dto.ImPrivateMessageSendDTO;
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
import cn.iocoder.yudao.module.im.service.websocket.notification.rtc.*;
import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.validation.annotation.Validated;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
/**
* IM 实时通话 Service 实现
*
* 存储模型：DB 单一存储（im_rtc_call 主表 + im_rtc_participant 明细表）
*
* 并发幂等：同好友对 / 同群活跃唯一性走 {@link ImRtcCallLockRedisDAO} 分布式锁 + 锁内 SELECT 兜底；webhook 兜底走条件 UPDATE；
*
* 推送通道分流：
* 1601 RTC_CALL（INVITING / JOINED / REJECTED / NO_ANSWER / LEFT 子类型）→ {@link ImWebSocketService#sendNotificationAsync} 仅推参与方；
* 1602 / 1603 PARTICIPANT_CONNECTED / DISCONNECTED → {@link ImWebSocketService} 推参与方 + 群通话场景广播全群；
* 1610 RTC_CALL_START + 1611 RTC_CALL_END → {@link ImPrivateMessageService} / {@link ImGroupMessageService} 入消息流当聊天 tip
* （START 仅群通话；两者分别在 invite / cancel(leave) 事务里 INSERT，自增 id 自然保证顺序）
*
* 职责边界：媒体协商完全交给 LiveKit；后端只做会话状态机、Token 签发、来电信令推送、通话历史落消息流；房内媒体流变化交给 LiveKit 客户端事件（TrackSubscribed 等），后端不重复推
*
* @author 芋道源码
*/
@Service
@Validated
@Slf4j
public class ImRtcCallServiceImpl implements ImRtcCallService {
@Resource
private ImRtcCallMapper rtcCallMapper;
@Resource
private ImRtcParticipantMapper rtcParticipantMapper;
@Resource
private ImRtcCallLockRedisDAO rtcCallLockRedisDAO;
@Resource
private ImGroupMemberService groupMemberService;
@Resource
private ImFriendService friendService;
@Resource
private ImWebSocketService webSocketService;
@Resource
private ImPrivateMessageService privateMessageService;
@Resource
private ImGroupMessageService groupMessageService;
@Resource
private LiveKitClient liveKitClient;
@Resource
private AdminUserApi adminUserApi;
@Resource
private ImProperties imProperties;
// ========== 业务接口 ==========
@Override
@SneakyThrows
@Transactional(rollbackFor = Exception.class)
public ImRtcCallDO createCall(Long userId, ImRtcCallCreateReqVO reqVO) {
validateEnabled();
// 1. 校验入参与场景
validateCreateCall(userId, reqVO);
// 2. 加锁后跑业务主体；同好友对 / 同群串行，避免并发各开一通的竞态
if (ImConversationTypeEnum.isGroup(reqVO.getConversationType())) {
return rtcCallLockRedisDAO.lockGroup(reqVO.getGroupId(), () -> createGroupCall(userId, reqVO));
}
Long peerUserId = CollUtil.getFirst(reqVO.getInviteeIds());
return rtcCallLockRedisDAO.lockPrivate(userId, peerUserId, () -> createPrivateCall(userId, reqVO, peerUserId));
}
/**
* 群通话创建锁内主体：群有活跃通话直接抛（引导走 inviteCall / joinCall）；
* 否则走完整生命周期，若发起人自身忙线立即 end(BUSY) 留下通话记录
*
* @param userId 发起人编号
* @param reqVO 创建请求
* @return 通话主表（可能 status=ENDED 表示自身忙线）
*/
private ImRtcCallDO createGroupCall(Long userId, ImRtcCallCreateReqVO reqVO) {
// 1.1 同群有活跃通话 → 直接抛异常（UI 应已拦截），引导用户走 inviteCall / joinCall；避免重复开通
ImRtcCallDO active = rtcCallMapper.selectLastOneByGroupIdAndStatusIn(
reqVO.getGroupId(), ImRtcCallStatusEnum.ACTIVE_STATUSES);
if (active != null) {
throw exception(RTC_GROUP_CALL_ACTIVE);
}
// 1.2 先检测发起人忙线状态；不抛，留给下方 end 决定
boolean selfBusy = rtcParticipantMapper.selectLastOneByUserIdAndStatus(
userId, ImRtcParticipantStatusEnum.ACTIVE_STATUSES) != null;
// 2. 完整生命周期：INSERT + INVITE × N + START 全推
ImRtcCallDO call = createCall0(userId, reqVO);
// 3. 自身忙线立即 end(BUSY)；END 推送给群，群成员看到完整 START + END
if (selfBusy) {
endSession(call, userId, ImRtcCallEndReasonEnum.BUSY);
}
return call;
}
/**
* 私聊创建锁内主体：双方忙线时仍走完整生命周期（create + 立即 end(BUSY)），
* 同一对正在通话视作数据异常直接抛
*
* @param userId 发起人编号
* @param reqVO 创建请求
* @param peerUserId 对端编号；来自 reqVO.inviteeIds 的唯一元素
* @return 通话主表（可能 status=ENDED 表示忙线）
*/
private ImRtcCallDO createPrivateCall(Long userId, ImRtcCallCreateReqVO reqVO, Long peerUserId) {
// 1.1 双方已在同一通话 → 数据异常（UI 应已拦截），直接抛
if (getActivePrivateCallByPair(userId, peerUserId) != null) {
throw exception(RTC_SELF_BUSY);
}
// 1.2 忙线检测：self 优先（更可执行的提示）；不抛，留给下方 end 决定
boolean selfBusy = rtcParticipantMapper.selectLastOneByUserIdAndStatus(
userId, ImRtcParticipantStatusEnum.ACTIVE_STATUSES) != null;
boolean peerBusy = !selfBusy && rtcParticipantMapper.selectLastOneByUserIdAndStatus(
peerUserId, ImRtcParticipantStatusEnum.ACTIVE_STATUSES) != null;
// 2. 完整生命周期：INSERT + INVITE 全推
ImRtcCallDO call = createCall0(userId, reqVO);
// 3. 忙线立即 end(BUSY)；operator 决定两端看到的文案（self busy → operator=自己；peer busy → operator=对端）
if (selfBusy) {
endSession(call, userId, ImRtcCallEndReasonEnum.BUSY);
} else if (peerBusy) {
endSession(call, peerUserId, ImRtcCallEndReasonEnum.BUSY);
}
return call;
}
@Override
@SneakyThrows
@Transactional(rollbackFor = Exception.class)
public void inviteCall(Long userId, ImRtcCallInviteReqVO reqVO) {
validateEnabled();
// 1.1 校验通话存在且活跃
ImRtcCallDO call = validateCallActive(reqVO.getRoom());
// 1.2 仅群通话支持追加邀请
if (!ImConversationTypeEnum.isGroup(call.getConversationType())) {
throw exception(RTC_GROUP_REQUIRED);
}
// 1.3 操作者必须是房内 JOINED 参与者
ImRtcParticipantDO operator = rtcParticipantMapper.selectByRoomAndUserId(call.getRoom(), userId);
if (operator == null || !ImRtcParticipantStatusEnum.isJoined(operator.getStatus())) {
throw exception(RTC_NOT_PARTICIPANT);
}
// 2. 加群锁后执行追加邀请；避免与新建 / 重复追加的竞态
rtcCallLockRedisDAO.lockGroup(call.getGroupId(), () -> {
addInvitees(call, userId, reqVO.getInviteeIds());
return null;
});
}
/**
* 新建通话实体；INSERT 主表 + 参与表 + 推送 INVITE / START
*
* @param inviterId 发起人编号
* @param reqVO 创建请求
* @return 通话主表
*/
private ImRtcCallDO createCall0(Long inviterId, ImRtcCallCreateReqVO reqVO) {
// 1. 构造参数：room 用 UUID；解析被邀请池
String room = IdUtil.fastSimpleUUID();
LocalDateTime now = LocalDateTime.now();
Set invitees = resolveInvitees(reqVO, inviterId);
// 2.1 INSERT 主表；群聊发起人即时 JOINED 但通话仍处 CREATED，等首个非发起人接通才切 RUNNING
ImRtcCallDO call = new ImRtcCallDO().setRoom(room)
.setConversationType(reqVO.getConversationType()).setMediaType(reqVO.getMediaType())
.setInviterUserId(inviterId).setGroupId(reqVO.getGroupId())
.setStatus(ImRtcCallStatusEnum.CREATED.getStatus()).setStartTime(now);
rtcCallMapper.insert(call);
// 2.2 批量 INSERT 参与表：发起人即时 JOINED，被邀请人 INVITING 等接通
List participants = new ArrayList<>(invitees.size() + 1);
participants.add(new ImRtcParticipantDO().setCallId(call.getId()).setRoom(room).setUserId(inviterId)
.setRole(ImRtcParticipantRoleEnum.INVITER.getRole())
.setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus())
.setInviteTime(now).setAcceptTime(now));
for (Long inviteeId : invitees) {
participants.add(new ImRtcParticipantDO()
.setCallId(call.getId()).setRoom(room).setUserId(inviteeId)
.setRole(ImRtcParticipantRoleEnum.INVITEE.getRole())
.setStatus(ImRtcParticipantStatusEnum.INVITING.getStatus())
.setInviteTime(now));
}
rtcParticipantMapper.insertBatch(participants);
// 3.1 推送通知：RTC_CALL(INVITE) 给每个被邀请人
AdminUserRespDTO inviterUser = adminUserApi.getUser(inviterId);
Map inviteeMap = adminUserApi.getUserMap(invitees);
for (Long inviteeId : invitees) {
pushCallInviteNotification(call, inviterUser, inviteeId, inviteeMap.get(inviteeId), invitees);
}
// 3.2 向消息流写入 RTC_CALL_START；群聊全群广播，私聊定向给被叫，作为会话列表预览的依据
pushCallStartNotification(call, inviterUser, invitees);
return call;
}
@Override
@Transactional(rollbackFor = Exception.class)
public ImRtcCallDO joinCall(Long userId, String room) {
validateEnabled();
// 1.1 校验通话存在且活跃
ImRtcCallDO call = validateCallActive(room);
// 1.2 仅群通话支持「旁观者加入」
if (!ImConversationTypeEnum.isGroup(call.getConversationType())) {
throw exception(RTC_GROUP_REQUIRED);
}
// 1.3 校验当前用户是该群有效成员；防止仅凭 room 就拿到 LiveKit token 越权入会
groupMemberService.validateMemberInGroup(call.getGroupId(), userId);
// 1.4 校验当前用户没有其它活跃通话
validateUserNotInOtherCall(userId, call.getRoom());
// 2. 入参与表：已有记录切回 JOINED；不在记录则以 JOINER 角色 INSERT
LocalDateTime now = LocalDateTime.now();
joinParticipant(call, userId, now);
// 3. 主表 CREATED → RUNNING（首次有非发起人加入）
maybeMarkOngoing(call, userId, now);
return call;
}
/**
* 给已存在的活跃群通话追加邀请；批量校验群成员 + 去重已在通话池 + 推送 INVITE
*
* @param call 活跃通话主表
* @param inviterId 本次追加邀请的发起人；已是房内 JOINED 参与者
* @param inviteeIds 本次追加的被邀请人编号
*/
private void addInvitees(ImRtcCallDO call, Long inviterId, Collection inviteeIds) {
// 1.1 校验被邀请人都是群活跃成员
groupMemberService.validateMembersInGroup(call.getGroupId(), inviteeIds);
// 1.2 排除已在通话池的；剩余即本次新邀请
List existingParticipants = rtcParticipantMapper.selectListByRoom(call.getRoom());
Set existingUserIds = CollectionUtils.convertSet(existingParticipants, ImRtcParticipantDO::getUserId);
Set incomingUserIds = new LinkedHashSet<>(inviteeIds);
incomingUserIds.removeAll(existingUserIds);
if (CollUtil.isEmpty(incomingUserIds)) {
return;
}
long activeCount = existingParticipants.stream()
.filter(participant -> ImRtcParticipantStatusEnum.ACTIVE_STATUSES.contains(participant.getStatus()))
.count();
if (activeCount + incomingUserIds.size() > imProperties.getRtc().getGroupMaxParticipants()) {
throw exception(RTC_GROUP_INVITEE_OVER_LIMIT);
}
// 2. 批量 INSERT 新邀请人
LocalDateTime now = LocalDateTime.now();
List participants = CollectionUtils.convertList(incomingUserIds, inviteeId ->
new ImRtcParticipantDO().setCallId(call.getId()).setRoom(call.getRoom()).setUserId(inviteeId)
.setRole(ImRtcParticipantRoleEnum.INVITEE.getRole())
.setStatus(ImRtcParticipantStatusEnum.INVITING.getStatus()).setInviteTime(now));
rtcParticipantMapper.insertBatch(participants);
// 3. 推送通知：RTC_CALL(INVITE) 给每个新邀请人
AdminUserRespDTO inviter = adminUserApi.getUser(inviterId);
Map inviteeMap = adminUserApi.getUserMap(incomingUserIds);
for (Long inviteeId : incomingUserIds) {
pushCallInviteNotification(call, inviter, inviteeId, inviteeMap.get(inviteeId), incomingUserIds);
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public ImRtcCallDO acceptCall(Long userId, String room) {
validateEnabled();
// 1.1 校验通话存在且活跃
ImRtcCallDO call = validateCallActive(room);
// 1.2 校验本人是该通话的参与者
ImRtcParticipantDO participant = validateParticipant(call, userId);
// 2.1 已 JOINED 直接幂等返回
if (ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
return call;
}
validateUserNotInOtherCall(userId, call.getRoom());
// 2.2 仅 INVITING → JOINED；其它状态拒
if (!ImRtcParticipantStatusEnum.isInviting(participant.getStatus())) {
throw exception(RTC_SESSION_NOT_EXISTS);
}
LocalDateTime now = LocalDateTime.now();
int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(), ImRtcParticipantStatusEnum.INVITING.getStatus(),
new ImRtcParticipantDO().setId(participant.getId()).setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus()).setAcceptTime(now));
if (updated == 0) {
throw exception(RTC_SESSION_NOT_EXISTS);
}
// 3. 主表 CREATED → RUNNING（首次有非发起人接通）
maybeMarkOngoing(call, userId, now);
return call;
}
@Override
@Transactional(rollbackFor = Exception.class)
public void rejectCall(Long userId, String room) {
validateEnabled();
// 1.1 校验通话存在且活跃
ImRtcCallDO call = validateCallActive(room);
// 1.2 校验本人是该通话的参与者
ImRtcParticipantDO participant = validateParticipant(call, userId);
// 1.3 仅 INVITING 状态可拒
if (!ImRtcParticipantStatusEnum.isInviting(participant.getStatus())) {
throw exception(RTC_SESSION_NOT_EXISTS);
}
// 2. INVITING → REJECTED；并发已变更则忽略
int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(), ImRtcParticipantStatusEnum.INVITING.getStatus(),
new ImRtcParticipantDO().setId(participant.getId()).setStatus(ImRtcParticipantStatusEnum.REJECTED.getStatus()).setLeaveTime(LocalDateTime.now()));
if (updated == 0) {
return;
}
// 3.1 群通话拒绝：推 RTC_CALL(REJECT) 给主叫，再判定是否要关房（全员拒接 / 只剩主叫一人时收敛）
if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
pushCallRejectNotification(call, userId);
endSessionIfTerminal(call, userId);
return;
}
// 3.2 私聊拒绝：走 endSession（推 RTC_CALL_END(reason=REJECT)）
endSession(call, userId, ImRtcCallEndReasonEnum.REJECT);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void cancelCall(Long userId, String room) {
validateEnabled();
// 1.1 校验通话存在且活跃
ImRtcCallDO call = validateCallActive(room);
// 1.2 仅主叫可取消
if (ObjUtil.notEqual(call.getInviterUserId(), userId)) {
throw exception(RTC_NOT_PARTICIPANT);
}
// 1.3 仅 CREATED 状态可取消（RUNNING 应走 leave）
if (!ImRtcCallStatusEnum.isCreated(call.getStatus())) {
throw exception(RTC_SESSION_NOT_EXISTS);
}
// 2. 关会话并推 RTC_CALL_END(reason=CANCEL)
endSession(call, userId, ImRtcCallEndReasonEnum.CANCEL);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void leaveCall(Long userId, String room) {
validateEnabled();
// 1.1 校验通话存在且活跃
ImRtcCallDO call = validateCallActive(room);
// 1.2 校验本人是该通话的参与者
ImRtcParticipantDO participant = validateParticipant(call, userId);
// 2. 当前状态 → LEFT；条件 UPDATE 防并发反复改
LocalDateTime now = LocalDateTime.now();
int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(), participant.getStatus(),
new ImRtcParticipantDO().setId(participant.getId()).setStatus(ImRtcParticipantStatusEnum.LEFT.getStatus()).setLeaveTime(now));
if (updated == 0) {
return;
}
// 3. 群通话已入会参与者离开时推送离线通知
if (ImConversationTypeEnum.isGroup(call.getConversationType())
&& ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
pushParticipantDisconnectedNotification(call, userId);
}
// 4. 触发关房判定：私聊任一方离开必关；群通话仅在「无人在房 + 无人响铃」时关
endSessionIfTerminal(call, userId);
}
/**
* 群通话是否应该关闭
*
* 两种终态：
* 1. JOINED 数 = 0（房内已没人）
* 2. JOINED 数 = 1 且 INVITING 数 = 0（只剩 1 人独守，无后续可加入者）
*
* 关房后 endSession 会把残留 INVITING 批量改 NO_ANSWER 并推 RTC_CALL_END，响铃端 UI 自动收敛
*
* @param room 业务通话编号
* @return true 表示无法继续通话，应关房
*/
private boolean shouldCloseGroupRoom(String room) {
int joined = 0;
int inviting = 0;
for (ImRtcParticipantDO p : rtcParticipantMapper.selectListByRoom(room)) {
if (ImRtcParticipantStatusEnum.isJoined(p.getStatus())) {
joined++;
} else if (ImRtcParticipantStatusEnum.isInviting(p.getStatus())) {
inviting++;
}
}
return joined == 0 || (joined == 1 && inviting == 0);
}
/**
* 查询两人共同所在的活跃私聊通话
*
* 拆成 3 段简单查询：拿 A 的活跃 participant → 拿主表判私聊未结束 → 看 B 是否在同 call 活跃
*
* @param userIdA 用户 A 编号
* @param userIdB 用户 B 编号
* @return 活跃私聊通话；不存在返回 null
*/
private ImRtcCallDO getActivePrivateCallByPair(Long userIdA, Long userIdB) {
ImRtcParticipantDO participantA = rtcParticipantMapper.selectLastOneByUserIdAndStatus(userIdA, ImRtcParticipantStatusEnum.ACTIVE_STATUSES);
if (participantA == null) {
return null;
}
ImRtcCallDO call = rtcCallMapper.selectByRoom(participantA.getRoom());
if (call == null
|| !ImConversationTypeEnum.isPrivate(call.getConversationType())
|| ImRtcCallStatusEnum.isEnded(call.getStatus())) {
return null;
}
ImRtcParticipantDO participantB = rtcParticipantMapper.selectByRoomAndUserId(call.getRoom(), userIdB);
if (participantB == null
|| !ImRtcParticipantStatusEnum.ACTIVE_STATUSES.contains(participantB.getStatus())) {
return null;
}
return call;
}
@Override
public ImRtcCallDO getActiveCall(Long userId, Long groupId) {
validateEnabled();
// 1. 鉴权：仅群活跃成员能查（走单行 SQL，不依赖成员列表缓存）
groupMemberService.validateMemberInGroup(groupId, userId);
// 2. 查询活跃通话
return rtcCallMapper.selectLastOneByGroupIdAndStatusIn(groupId, ImRtcCallStatusEnum.ACTIVE_STATUSES);
}
@Override
public List getCallParticipantList(String room) {
return rtcParticipantMapper.selectListByRoom(room);
}
@Override
public String signCallToken(Long userId, String room) {
validateEnabled();
ImRtcCallDO call = validateCallActive(room);
ImRtcParticipantDO participant = validateParticipant(call, userId);
if (!ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
throw exception(RTC_NOT_PARTICIPANT);
}
return signToken(userId, resolveDisplayName(adminUserApi.getUser(userId), userId), room);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void handleLiveKitEvent(LiveKitWebhookEventDTO event) {
if (event == null || event.getEvent() == null) {
return;
}
switch (event.getEvent()) {
case LiveKitWebhookEventDTO.EVENT_PARTICIPANT_JOINED:
handleParticipantJoined(event);
break;
case LiveKitWebhookEventDTO.EVENT_PARTICIPANT_LEFT:
handleParticipantLeft(event);
break;
case LiveKitWebhookEventDTO.EVENT_ROOM_FINISHED:
handleRoomFinished(event);
break;
default:
// 其它事件忽略；track_published 等业务态由 LiveKit 自身分发驱动
}
}
/**
* 处理 LiveKit 「成员加入」事件：DB 接通态由 {@link #acceptCall(Long, String)} 接口写入，此处仅做 1602 转推
*
* @param event 事件
*/
private void handleParticipantJoined(LiveKitWebhookEventDTO event) {
if (ObjUtil.hasNull(event.getRoom(), event.getParticipant())) {
return;
}
// 1. 前置检查：通话存在且活跃
String room = event.getRoom().getName();
ImRtcCallDO call = rtcCallMapper.selectByRoom(room);
if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
return;
}
Long userId = liveKitClient.parseUserId(event.getParticipant().getIdentity());
if (userId == null) {
return;
}
// 2. 推 1602 通知参与方 / 全群参与方
pushParticipantConnectedNotification(call, userId);
}
/**
* 处理 LiveKit 「成员离开」事件：正常 {@link #leaveCall(Long, String)} 接口已经清理过的话，条件 UPDATE 自动幂等
*
* @param event 事件
*/
private void handleParticipantLeft(LiveKitWebhookEventDTO event) {
if (ObjUtil.hasNull(event.getRoom(), event.getParticipant())) {
return;
}
// 1. 前置检查：通话存在且活跃 + 参与者仍在 JOINED
String room = event.getRoom().getName();
ImRtcCallDO call = rtcCallMapper.selectByRoom(room);
if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
return;
}
Long userId = liveKitClient.parseUserId(event.getParticipant().getIdentity());
if (userId == null) {
return;
}
ImRtcParticipantDO participant = rtcParticipantMapper.selectByRoomAndUserId(call.getRoom(), userId);
if (participant == null || !ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
return;
}
// 2. JOINED → LEFT；正常 leave 接口已改过则 update 影响 0 行直接退出
int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(), ImRtcParticipantStatusEnum.JOINED.getStatus(),
new ImRtcParticipantDO().setId(participant.getId()).setStatus(ImRtcParticipantStatusEnum.LEFT.getStatus()).setLeaveTime(LocalDateTime.now()));
if (updated == 0) {
return;
}
log.info("[handleParticipantLeft][room={} userId={} 由 LiveKit Webhook 兜底]", room, userId);
// 3. 推 1603 通知参与方 / 全群参与方
pushParticipantDisconnectedNotification(call, userId);
// 4. 触发关房判定：私聊任一方离开必关；群通话仅在「无人在房 + 无人响铃」时关
endSessionIfTerminal(call, userId);
}
/**
* 处理 LiveKit 「房间结束」事件：兜底把 call 推到 ENDED
*
* @param event 事件
*/
private void handleRoomFinished(LiveKitWebhookEventDTO event) {
if (event.getRoom() == null) {
return;
}
// 1. 前置检查：通话存在且活跃
String room = event.getRoom().getName();
ImRtcCallDO call = rtcCallMapper.selectByRoom(room);
if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
return;
}
// 2. 关会话
log.info("[handleRoomFinished][room={} 由 LiveKit Webhook 兜底]", room);
endSession(call, null, ImRtcCallEndReasonEnum.HANGUP);
}
@Override
@Transactional(rollbackFor = Exception.class)
public int cleanupZombieCalls(int thresholdMinutes) {
// 阈值由调用方（Job）保证 > 0；低于 1 分钟会误杀刚发起的合理零人态
LocalDateTime threshold = LocalDateTime.now().minusMinutes(thresholdMinutes);
List candidates = rtcCallMapper.selectListByStatusInAndStartTimeBefore(
ImRtcCallStatusEnum.ACTIVE_STATUSES, threshold);
if (CollUtil.isEmpty(candidates)) {
return 0;
}
// 2. 逐个查 LiveKit 房间真实 participant 数
int cleaned = 0;
for (ImRtcCallDO call : candidates) {
int count;
try {
count = liveKitClient.listParticipants(call.getRoom());
} catch (Exception e) {
log.warn("[cleanupZombieCalls][查询 LiveKit 失败 room={}]", call.getRoom(), e);
continue;
}
if (count != 0) {
continue;
}
log.info("[cleanupZombieCalls][清理僵尸通话 room={}]", call.getRoom());
endSession(call, null, ImRtcCallEndReasonEnum.HANGUP);
cleaned++;
}
return cleaned;
}
@Override
@Transactional(rollbackFor = Exception.class)
public int timeoutInvitingParticipants(int thresholdMinutes) {
// 阈值由调用方（Job）保证 > 0；低于 1 分钟可能误杀刚发起还在响铃的合理 INVITING 态
LocalDateTime threshold = LocalDateTime.now().minusMinutes(thresholdMinutes);
return noAnswerCallCheck0(rtcParticipantMapper.selectListByStatusAndInviteTimeBefore(
ImRtcParticipantStatusEnum.INVITING.getStatus(), threshold));
}
@Override
@Transactional(rollbackFor = Exception.class)
public void noAnswerCallCheck(Long userId, String room) {
// 鉴权：仅该 room 参与者可触发；失败静默，不暴露错误
ImRtcParticipantDO operator = rtcParticipantMapper.selectByRoomAndUserId(room, userId);
if (operator == null) {
return;
}
// 阈值取后端配置，避免前后端配置不一致；前端 timer 仅是触发时机
LocalDateTime threshold = LocalDateTime.now()
.minusMinutes(imProperties.getRtc().getInviteTimeoutMinutes());
List candidates = rtcParticipantMapper.selectListByRoomAndStatusAndInviteTimeBefore(
room, ImRtcParticipantStatusEnum.INVITING.getStatus(), threshold);
noAnswerCallCheck0(candidates);
}
/**
* 批量超时处理：循环单参与者；同 room 复用 call、批量预查 user 避免 N+1；返回成功处理数
*
* @param candidates 已过滤的超时 INVITING 候选
* @return 成功处理（CAS 抢占）的数量
*/
private int noAnswerCallCheck0(List candidates) {
if (CollUtil.isEmpty(candidates)) {
return 0;
}
Map callCache = new HashMap<>();
Map userMap = adminUserApi.getUserMap(
CollectionUtils.convertSet(candidates, ImRtcParticipantDO::getUserId));
int timedOut = 0;
for (ImRtcParticipantDO participant : candidates) {
if (timeoutInvitingParticipant(participant, callCache, userMap)) {
timedOut++;
}
}
return timedOut;
}
/**
* 单参与者振铃超时：CAS 把 INVITING 切到 NO_ANSWER，群通话推 RTC_CALL(NO_ANSWER) + 级联关房判定；
* 私聊场景被叫超时即整通话结束，走 endSession(NO_ANSWER) 推 RTC_CALL_END
*
* @param participant INVITING 状态的超时候选
* @param callCache 按 room 缓存 call 对象，避免同批次多人重复查询
* @param userMap 候选用户预查 map；避免逐个 adminUserApi.getUser 走 N+1
* @return 是否成功处理（CAS 失败 / 通话主表缺失等场景返回 false）
*/
private boolean timeoutInvitingParticipant(ImRtcParticipantDO participant,
Map callCache,
Map userMap) {
// 1. CAS：INVITING → NO_ANSWER；并发已变（用户刚接 / 拒，或 endSession 整体改）跳过
int updated = rtcParticipantMapper.updateByIdAndStatus(participant.getId(),
ImRtcParticipantStatusEnum.INVITING.getStatus(),
new ImRtcParticipantDO().setId(participant.getId())
.setStatus(ImRtcParticipantStatusEnum.NO_ANSWER.getStatus())
.setLeaveTime(LocalDateTime.now()));
if (updated == 0) {
return false;
}
Long userId = participant.getUserId();
log.info("[timeoutInvitingParticipant][参与者振铃超时 room={} userId={}]", participant.getRoom(), userId);
// 2. 查询通话主表；同 room 复用 callCache，避免同批次多人重复查询
ImRtcCallDO call = callCache.computeIfAbsent(participant.getRoom(), rtcCallMapper::selectByRoom);
if (call == null) {
log.warn("[timeoutInvitingParticipant][通话主表缺失 room={} userId={}]", participant.getRoom(), userId);
return false;
}
// 3.1 群通话：推 RTC_CALL(NO_ANSWER) 让前端 banner 移除该人 + 级联关房判定
if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
pushCallNoAnswerNotification(call, userId, userMap.get(userId));
endSessionIfTerminal(call, userId);
return true;
}
// 3.2 私聊：被叫超时 = 整通话无人接听，走 endSession 推 RTC_CALL_END(NO_ANSWER)
endSession(call, userId, ImRtcCallEndReasonEnum.NO_ANSWER);
return true;
}
// ========== 内部辅助 ==========
private void validateEnabled() {
if (!imProperties.getRtc().isEnabled()) {
throw exception(RTC_NOT_ENABLED);
}
}
/**
* 校验通话存在且活跃：不存在 / 已 ENDED 直接抛
*
* @param room 业务通话编号
* @return 通话主表
*/
private ImRtcCallDO validateCallActive(String room) {
ImRtcCallDO call = rtcCallMapper.selectByRoom(room);
if (call == null || ImRtcCallStatusEnum.isEnded(call.getStatus())) {
throw exception(RTC_SESSION_NOT_EXISTS);
}
return call;
}
/**
* 校验本人是该通话的参与者：accept / reject / leave / refreshToken 的共用前置
*
* @param call 通话主表
* @param userId 用户编号
* @return 参与者记录
*/
private ImRtcParticipantDO validateParticipant(ImRtcCallDO call, Long userId) {
ImRtcParticipantDO participant = rtcParticipantMapper.selectByRoomAndUserId(call.getRoom(), userId);
if (participant == null) {
throw exception(RTC_NOT_PARTICIPANT);
}
return participant;
}
/**
* 校验用户不在其它活跃通话中
*
* @param userId 用户编号
* @param room 当前房间标识
*/
private void validateUserNotInOtherCall(Long userId, String room) {
// 查询当前房间外的活跃参与记录
ImRtcParticipantDO participant = rtcParticipantMapper.selectLastOneByUserIdAndStatusInAndRoomNot(
userId, ImRtcParticipantStatusEnum.ACTIVE_STATUSES, room);
// 存在活跃参与记录，则当前用户忙线
if (participant != null) {
throw exception(RTC_SELF_BUSY);
}
}
/**
* 加入群通话参与者列表
*
* @param call 通话主表
* @param userId 用户编号
* @param now 当前时间
*/
private void joinParticipant(ImRtcCallDO call, Long userId, LocalDateTime now) {
// 1. 已有参与记录：切回 JOINED
ImRtcParticipantDO existing = rtcParticipantMapper.selectByRoomAndUserId(call.getRoom(), userId);
if (existing != null) {
updateParticipantJoined(existing, now);
return;
}
// 2. 无参与记录：以主动加入者身份新增
try {
rtcParticipantMapper.insert(new ImRtcParticipantDO()
.setCallId(call.getId()).setRoom(call.getRoom())
.setUserId(userId).setRole(ImRtcParticipantRoleEnum.JOINER.getRole())
.setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus()).setInviteTime(now).setAcceptTime(now));
} catch (DuplicateKeyException ex) {
// 3. 唯一键冲突：回查并复用并发写入的记录
existing = rtcParticipantMapper.selectByRoomAndUserId(call.getRoom(), userId);
if (existing == null) {
throw ex;
}
updateParticipantJoined(existing, now);
}
}
/**
* 将参与者更新为已加入
*
* @param participant 参与者记录
* @param now 当前时间
*/
private void updateParticipantJoined(ImRtcParticipantDO participant, LocalDateTime now) {
// 已是 JOINED 直接返回
if (ImRtcParticipantStatusEnum.isJoined(participant.getStatus())) {
return;
}
// 更新状态和接听时间
rtcParticipantMapper.updateById(new ImRtcParticipantDO().setId(participant.getId())
.setStatus(ImRtcParticipantStatusEnum.JOINED.getStatus()).setAcceptTime(now));
}
/**
* 关房判定收口：私聊任一方离开必关；群通话仅在「无人在房 + 无人响铃」时关
*
* end reason 按 call.status 自动推：CREATED（没人接通过）= CANCEL；RUNNING（有人接通过）= HANGUP
*
* @param call 通话主表
* @param operatorId 操作者用户编号；webhook 兜底场景可空
*/
private void endSessionIfTerminal(ImRtcCallDO call, Long operatorId) {
if (!ImConversationTypeEnum.isPrivate(call.getConversationType())
&& !shouldCloseGroupRoom(call.getRoom())) {
return;
}
ImRtcCallEndReasonEnum reason = ImRtcCallStatusEnum.isCreated(call.getStatus())
? ImRtcCallEndReasonEnum.CANCEL : ImRtcCallEndReasonEnum.HANGUP;
endSession(call, operatorId, reason);
}
/**
* 校验创建通话入参；按场景区分必填字段，私聊补好友校验，群聊补群成员校验
*
* @param userId 发起人编号
* @param reqVO 创建请求
*/
private void validateCreateCall(Long userId, ImRtcCallCreateReqVO reqVO) {
Integer conversationType = reqVO.getConversationType();
if (ImConversationTypeEnum.isPrivate(conversationType)) {
// 私聊必须 1 个对端
if (CollUtil.size(reqVO.getInviteeIds()) != 1) {
throw exception(RTC_PRIVATE_INVITEE_REQUIRED);
}
Long peerUserId = CollUtil.getFirst(reqVO.getInviteeIds());
if (ObjUtil.equal(userId, peerUserId)) {
throw exception(RTC_INVITE_SELF);
}
friendService.validateFriend(userId, peerUserId);
return;
}
if (ImConversationTypeEnum.isGroup(conversationType)) {
if (reqVO.getGroupId() == null) {
throw exception(RTC_GROUP_REQUIRED);
}
// 群通话必须前端选中被邀请人（对齐微信）；空集合直接拒
if (CollUtil.isEmpty(reqVO.getInviteeIds())) {
throw exception(RTC_GROUP_INVITEE_REQUIRED);
}
groupMemberService.validateMemberInGroup(reqVO.getGroupId(), userId);
return;
}
throw new IllegalArgumentException("非法的 conversationType: " + conversationType);
}
/**
* 解析被邀请池：私聊为 peerUserId 单元素；群聊为前端选中子集（超量抛错）
*
* @param reqVO 创建请求
* @param inviterId 发起人编号；自己不进被邀请池
* @return 被邀请人 userId 集合
*/
private Set resolveInvitees(ImRtcCallCreateReqVO reqVO, Long inviterId) {
// 1. 私聊：inviteeIds 已在 validateCreateCall 校验仅 1 个对端，直接复用
if (ImConversationTypeEnum.isPrivate(reqVO.getConversationType())) {
return new LinkedHashSet<>(reqVO.getInviteeIds());
}
// 2. 群聊校验：被邀请人必须是该群活跃成员，防止恶意客户端塞任意 userId
groupMemberService.validateMembersInGroup(reqVO.getGroupId(), reqVO.getInviteeIds());
Set initial = new LinkedHashSet<>(reqVO.getInviteeIds());
// 发起人本人不进被邀请池
initial.remove(inviterId);
if (CollUtil.isEmpty(initial)) {
throw exception(RTC_GROUP_INVITEE_REQUIRED);
}
int max = imProperties.getRtc().getGroupMaxParticipants();
if (initial.size() + 1 > max) {
throw exception(RTC_GROUP_INVITEE_OVER_LIMIT);
}
return initial;
}
/**
* 主表 CREATED → RUNNING；仅当首个非发起人加入时推进；条件 UPDATE 保幂等
*
* @param call 通话主表；推进成功后会同步内存字段
* @param acceptorId 加入者用户编号；发起人加入不算「首次接通」
* @param now 当前时间
*/
private void maybeMarkOngoing(ImRtcCallDO call, Long acceptorId, LocalDateTime now) {
// 1. 已 RUNNING / ENDED 直接退出；发起人加入不算「首次有人接通」
if (!ImRtcCallStatusEnum.isCreated(call.getStatus())) {
return;
}
if (ObjUtil.equal(call.getInviterUserId(), acceptorId)) {
return;
}
// 2. CREATED → RUNNING 条件 UPDATE；多人并发只有一人成功
int updated = rtcCallMapper.updateByIdAndStatus(call.getId(), ImRtcCallStatusEnum.CREATED.getStatus(),
new ImRtcCallDO().setStatus(ImRtcCallStatusEnum.RUNNING.getStatus()).setAcceptTime(now));
if (updated == 0) {
// 3. 竞争失败：reload 看真实终态；已 ENDED 抛错，否则同步内存
ImRtcCallDO latest = rtcCallMapper.selectById(call.getId());
if (latest == null || ImRtcCallStatusEnum.isEnded(latest.getStatus())) {
throw exception(RTC_SESSION_NOT_EXISTS);
}
call.setStatus(latest.getStatus()).setAcceptTime(latest.getAcceptTime());
return;
}
// 4. 推进成功：同步内存给后续判断
call.setStatus(ImRtcCallStatusEnum.RUNNING.getStatus()).setAcceptTime(now);
}
/**
* 关闭会话：主表条件 UPDATE 推到 ENDED → 残留 INVITING 批量改 NO_ANSWER → 推 RTC_CALL_END
*
* @param call 通话主表
* @param operatorId 操作者用户编号；webhook 兜底 / Job 清理场景可空
* @param reason 结束原因
*/
private void endSession(ImRtcCallDO call, Long operatorId, ImRtcCallEndReasonEnum reason) {
// 1.1 更新通话主表为已结束；条件 UPDATE 仅在 status 还活跃时生效
LocalDateTime now = LocalDateTime.now();
int updated = rtcCallMapper.updateByIdAndStatusIn(call.getId(), ImRtcCallStatusEnum.ACTIVE_STATUSES,
new ImRtcCallDO().setStatus(ImRtcCallStatusEnum.ENDED.getStatus())
.setEndReason(reason.getReason()).setEndTime(now));
if (updated == 0) {
log.info("[endSession][已被另一路径终结，跳过 room={} operator={} reason={}]",
call.getRoom(), operatorId, reason);
return;
}
// 【特殊】同步内存 call：让 createCall 这类调用方拿到的 DO 立即反映 ENDED 终态，Controller 拼 RespVO 能直接用
call.setStatus(ImRtcCallStatusEnum.ENDED.getStatus()).setEndReason(reason.getReason()).setEndTime(now);
// 1.2 更新参与表为已结束：残留 INVITING 改 NO_ANSWER；残留 JOINED 改 LEFT 并写 leaveTime
rtcParticipantMapper.updateByRoomAndStatus(call.getRoom(), ImRtcParticipantStatusEnum.INVITING.getStatus(),
new ImRtcParticipantDO().setStatus(ImRtcParticipantStatusEnum.NO_ANSWER.getStatus()));
rtcParticipantMapper.updateByRoomAndStatus(call.getRoom(), ImRtcParticipantStatusEnum.JOINED.getStatus(),
new ImRtcParticipantDO().setStatus(ImRtcParticipantStatusEnum.LEFT.getStatus()).setLeaveTime(now));
// 2. 推 RTC_CALL_END；先于 deleteRoom 异步发出，让前端按业务语义 reset（NO_ANSWER / CANCEL 等），
// 避免随后 LiveKit Disconnected 事件抢先触发前端 "通话已断开" 兜底 toast
Long durationSeconds = call.getAcceptTime() != null ?
Duration.between(call.getAcceptTime(), now).getSeconds() : null;
pushCallEndNotification(call, operatorId, reason, durationSeconds);
// 3. 兜底删除 LiveKit 房间，强制断开异常残留客户端；失败仅记日志，不阻断业务
try {
liveKitClient.deleteRoom(call.getRoom());
} catch (Exception e) {
log.warn("[endSession][删除 LiveKit 房间失败 room={} operator={} reason={}]",
call.getRoom(), operatorId, reason, e);
}
log.info("[endSession][room={} operator={} reason={}]", call.getRoom(), operatorId, reason);
}
/**
* 通话事件的收件人池：1）私聊为双方参与者；2）群聊为群活跃成员
*
* @param call 通话主表
* @return 收件人 userId 集合
*/
private Collection getCallAudienceUserIdList(ImRtcCallDO call) {
if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
return groupMemberService.getActiveGroupMemberUserIdsByGroupId(call.getGroupId());
}
return CollectionUtils.convertSet(
rtcParticipantMapper.selectListByRoom(call.getRoom()), ImRtcParticipantDO::getUserId);
}
// ========== 通知推送 ==========
/**
* RTC_CALL(INVITE)：走 webSocketService 直推到被邀请人；persistent=false 不入消息流
*
* @param call 通话主表
* @param inviter 发起人；可空，缺失时 payload 的 inviterNickname / Avatar 留空
* @param inviteeId 被邀请人用户编号
* @param invitee 被邀请人；可空，缺失时 token 内 displayName 降级为 userId
* @param inviteeIds 本次被邀请人列表；前端来电界面展示「邀请的其他人」用，包含 inviteeId 自身
*/
private void pushCallInviteNotification(ImRtcCallDO call, AdminUserRespDTO inviter,
Long inviteeId, AdminUserRespDTO invitee,
Collection inviteeIds) {
String token = signToken(inviteeId, resolveDisplayName(invitee, inviteeId), call.getRoom());
ImRtcCallNotification payload = ImRtcCallNotification.ofInvite(
call, inviter, imProperties.getRtc().getLivekitUrl(), token, inviteeIds);
webSocketService.sendNotificationAsync(inviteeId, ImConversationTypeEnum.NONE.getType(),
ImContentTypeEnum.RTC_CALL.getType(), payload);
}
/**
* RTC_CALL(REJECT)：仅群通话场景；走 webSocketService 推给群通话受众
*
* 私聊拒绝走 endSession → RTC_CALL_END(reason=REJECT) 入消息流，不在此推
*
* @param call 通话主表
* @param operatorUserId 拒接者用户编号
*/
private void pushCallRejectNotification(ImRtcCallDO call, Long operatorUserId) {
AdminUserRespDTO operator = operatorUserId != null ? adminUserApi.getUser(operatorUserId) : null;
ImRtcCallNotification payload = ImRtcCallNotification.ofReject(call, operatorUserId, operator);
for (Long receiverUserId : getCallAudienceUserIdList(call)) {
webSocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
ImContentTypeEnum.RTC_CALL.getType(), payload);
}
}
/**
* RTC_CALL(NO_ANSWER)：仅群通话场景；振铃超时由 Job 触发；走 webSocketService 推给群通话受众
*
* 私聊未接听走 endSession → RTC_CALL_END(reason=NO_ANSWER) 入消息流，不在此推
*
* @param call 通话主表
* @param operatorUserId 未接听者用户编号
* @param operator 未接听者预查结果；调用方批量查避免 N+1，可空
*/
private void pushCallNoAnswerNotification(ImRtcCallDO call, Long operatorUserId, AdminUserRespDTO operator) {
ImRtcCallNotification payload = ImRtcCallNotification.ofNoAnswer(call, operatorUserId, operator);
for (Long receiverUserId : getCallAudienceUserIdList(call)) {
webSocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
ImContentTypeEnum.RTC_CALL.getType(), payload);
}
}
/**
* 通话参与者加入：LiveKit webhook participant_joined 触发；私聊推双方多端、群聊推全群成员（胶囊条 + 1）
*
* @param call 通话主表
* @param userId 加入的参与者用户编号
*/
private void pushParticipantConnectedNotification(ImRtcCallDO call, Long userId) {
pushParticipantNotification(call, ImContentTypeEnum.RTC_PARTICIPANT_CONNECTED.getType(), userId,
ImRtcParticipantConnectedNotification.of(call, userId));
}
/**
* 通话参与者离开：LiveKit webhook participant_left 触发；推送范围同 {@link #pushParticipantConnectedNotification}
*
* @param call 通话主表
* @param userId 离开的参与者用户编号
*/
private void pushParticipantDisconnectedNotification(ImRtcCallDO call, Long userId) {
pushParticipantNotification(call, ImContentTypeEnum.RTC_PARTICIPANT_DISCONNECTED.getType(), userId,
ImRtcParticipantDisconnectedNotification.of(call, userId));
}
/**
* 推送参与者事件的公共骨架；按会话类型决定收件人，单次 batch 推送扇出
*
* @param call 通话主表
* @param type 消息类型；1602 / 1603
* @param actorUserId 触发本次事件的用户编号
* @param payload 业务 payload
*/
private void pushParticipantNotification(ImRtcCallDO call, Integer type, Long actorUserId, Object payload) {
Collection receivers = getCallAudienceUserIdList(call);
if (CollUtil.isEmpty(receivers)) {
return;
}
webSocketService.sendNotificationAsync(receivers, ImConversationTypeEnum.NONE.getType(), type, payload);
}
/**
* RTC_CALL_START：群聊走 imGroupMessageService.send 全群广播；私聊走 imPrivateMessageService.send 定向给被叫
*
* 私聊 peer 直接复用 createCall 解析好的 invitees，避免再查参与表；用于会话列表预览展示「[语音通话]」
*
* @param call 通话主表
* @param inviter 发起人；可空
* @param invitees 本次邀请池；私聊场景取首个作为 peer
*/
private void pushCallStartNotification(ImRtcCallDO call, AdminUserRespDTO inviter, Set invitees) {
ImRtcCallStartNotification payload = ImRtcCallStartNotification.of(call, inviter);
Long peerUserId = ImConversationTypeEnum.isGroup(call.getConversationType()) ? null : CollUtil.getFirst(invitees);
pushCallChatMessage(call, ImContentTypeEnum.RTC_CALL_START, payload, peerUserId);
}
/**
* RTC_CALL_END：私聊走 imPrivateMessageService.send；群通话走 imGroupMessageService.send
*
* senderId 始终用通话发起人，让前端按「谁发起通话」决定气泡左右；操作者从 payload.operatorUserId 拿
*
* @param call 通话主表
* @param operatorId 操作者用户编号；webhook 兜底 / Job 清理场景可空
* @param reason 结束原因
* @param durationSeconds 通话时长（秒）；未接通时为 null
*/
private void pushCallEndNotification(ImRtcCallDO call, Long operatorId, ImRtcCallEndReasonEnum reason,
Long durationSeconds) {
AdminUserRespDTO operator = operatorId != null ? adminUserApi.getUser(operatorId) : null;
ImRtcCallEndNotification payload = ImRtcCallEndNotification.of(call, reason, durationSeconds, operatorId, operator);
Long peerUserId = null;
if (!ImConversationTypeEnum.isGroup(call.getConversationType())) {
ImRtcParticipantDO peer = CollUtil.findOne(
rtcParticipantMapper.selectListByRoom(call.getRoom()),
p -> ObjUtil.notEqual(p.getUserId(), call.getInviterUserId()));
peerUserId = peer != null ? peer.getUserId() : null;
}
pushCallChatMessage(call, ImContentTypeEnum.RTC_CALL_END, payload, peerUserId);
}
/**
* RTC 通话事件入消息流：群聊走 groupMessageService 全群广播，私聊走 privateMessageService 定向给 peer
*
* senderId 固定取 inviterUserId，让前端按「谁发起」决定气泡左右；私聊 peer 缺失时降级为发给自己作兜底
*
* @param call 通话主表
* @param type 消息类型；RTC_CALL_START / RTC_CALL_END
* @param payload 推送 payload
* @param peerUserId 私聊对端用户编号；群聊忽略，私聊缺失时回退为 senderId
*/
private void pushCallChatMessage(ImRtcCallDO call, ImContentTypeEnum type, Object payload, Long peerUserId) {
Long senderId = call.getInviterUserId();
if (ImConversationTypeEnum.isGroup(call.getConversationType())) {
ImGroupMessageSendDTO dto = new ImGroupMessageSendDTO().setGroupId(call.getGroupId())
.setType(type.getType()).setContent(payload);
groupMessageService.sendGroupMessage(senderId, dto);
return;
}
Long receiverId = peerUserId != null ? peerUserId : senderId;
ImPrivateMessageSendDTO dto = new ImPrivateMessageSendDTO().setReceiverId(receiverId)
.setType(type.getType()).setContent(payload);
privateMessageService.sendPrivateMessage(senderId, dto);
}
// ========== Token / VO ==========
/**
* 签 LiveKit Token
*
* @param userId 用户编号；token 内 identity 由此派生
* @param displayName 房内显示名
* @param room LiveKit 房间名
* @return JWT 字符串
*/
private String signToken(Long userId, String displayName, String room) {
return liveKitClient.signJoinToken(liveKitClient.buildIdentity(userId), displayName, room);
}
/**
* 解析房内显示名：优先取 user.nickname，缺失时降级为 userId 字符串（LiveKit displayName 不可空）
*
* @param user 用户信息；可空
* @param userId 用户编号；displayName 兜底
* @return 房内显示名
*/
private static String resolveDisplayName(AdminUserRespDTO user, Long userId) {
return StrUtil.blankToDefault(
user == null ? null : user.getNickname(), String.valueOf(userId));
}
// ==================== 管理后台 ====================
@Override
public PageResult getCallPage(ImRtcCallManagerPageReqVO reqVO) {
return rtcCallMapper.selectPage(reqVO);
}
@Override
public ImRtcCallDO getCall(Long id) {
return rtcCallMapper.selectById(id);
}
@Override
public List getCallParticipantListByCallId(Long id) {
return rtcParticipantMapper.selectListByCallId(id);
}
}