Commit 530d3b3a5f2ca5857db38298c37706af2dcd0b79

Authored by YunaiV
1 parent fe77884c

✨ feat(im): rtc 增加数据库存储(从单体 =》集群)

Showing 35 changed files with 1113 additions and 516 deletions
yudao-framework/yudao-spring-boot-starter-mybatis/src/main/java/cn/iocoder/yudao/framework/mybatis/core/mapper/BaseMapperX.java
@@ -145,6 +145,17 @@ public interface BaseMapperX<T> extends MPJBaseMapper<T> { @@ -145,6 +145,17 @@ public interface BaseMapperX<T> extends MPJBaseMapper<T> {
145 return CollUtil.getFirst(list); 145 return CollUtil.getFirst(list);
146 } 146 }
147 147
  148 + /**
  149 + * 获取满足条件的最新一条记录
  150 + * <p>
  151 + * 目的:解决并发场景下,插入多条记录后,使用 selectOne 会报错的问题
  152 + *
  153 + * @param queryWrapper 查询条件
  154 + * @return 最新一条;不存在返回 null
  155 + */
  156 + default T selectLast(LambdaQueryWrapper<T> queryWrapper) {
  157 + return CollUtil.getLast(selectList(queryWrapper));
  158 + }
148 159
149 default Long selectCount() { 160 default Long selectCount() {
150 return selectCount(new QueryWrapper<>()); 161 return selectCount(new QueryWrapper<>());
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/ImRtcCallController.java
@@ -14,8 +14,6 @@ import jakarta.validation.Valid; @@ -14,8 +14,6 @@ import jakarta.validation.Valid;
14 import org.springframework.validation.annotation.Validated; 14 import org.springframework.validation.annotation.Validated;
15 import org.springframework.web.bind.annotation.*; 15 import org.springframework.web.bind.annotation.*;
16 16
17 -import java.util.List;  
18 -  
19 import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success; 17 import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
20 import static cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils.getLoginUserId; 18 import static cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils.getLoginUserId;
21 19
@@ -79,17 +77,11 @@ public class ImRtcCallController { @@ -79,17 +77,11 @@ public class ImRtcCallController {
79 return success(rtcCallService.refreshToken(getLoginUserId(), roomName)); 77 return success(rtcCallService.refreshToken(getLoginUserId(), roomName));
80 } 78 }
81 79
82 - @GetMapping("/active-sessions")  
83 - @Operation(summary = "查询当前用户活跃通话;用于 App 冷启动 / 推送点开恢复")  
84 - public CommonResult<List<ImRtcCallRespVO>> getActiveSessions() {  
85 - return success(rtcCallService.getActiveSessions(getLoginUserId()));  
86 - }  
87 -  
88 @GetMapping("/group-active-call") 80 @GetMapping("/group-active-call")
89 @Operation(summary = "查询群当前进行中的通话;用于群聊顶部「N 人正在通话」胶囊条") 81 @Operation(summary = "查询群当前进行中的通话;用于群聊顶部「N 人正在通话」胶囊条")
90 @Parameter(name = "groupId", description = "群编号", required = true, example = "2048") 82 @Parameter(name = "groupId", description = "群编号", required = true, example = "2048")
91 public CommonResult<ImRtcGroupCallRespVO> getGroupActiveCall(@RequestParam("groupId") Long groupId) { 83 public CommonResult<ImRtcGroupCallRespVO> getGroupActiveCall(@RequestParam("groupId") Long groupId) {
92 - return success(rtcCallService.getGroupActiveCall(groupId)); 84 + return success(rtcCallService.getGroupActiveCall(getLoginUserId(), groupId));
93 } 85 }
94 86
95 } 87 }
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/controller/admin/rtc/vo/ImRtcCallRespVO.java
@@ -5,7 +5,7 @@ import lombok.Data; @@ -5,7 +5,7 @@ import lombok.Data;
5 5
6 import java.util.Set; 6 import java.util.Set;
7 7
8 -@Schema(description = "管理后台 - 通话会话 Response VO;invite / accept / refreshToken / getActiveSessions 共用") 8 +@Schema(description = "管理后台 - 通话会话 Response VO;invite / accept / refreshToken 共用")
9 @Data 9 @Data
10 public class ImRtcCallRespVO { 10 public class ImRtcCallRespVO {
11 11
@@ -18,7 +18,7 @@ public class ImRtcCallRespVO { @@ -18,7 +18,7 @@ public class ImRtcCallRespVO {
18 @Schema(description = "LiveKit Server WebSocket 地址;前端 connect 用") 18 @Schema(description = "LiveKit Server WebSocket 地址;前端 connect 用")
19 private String livekitUrl; 19 private String livekitUrl;
20 20
21 - @Schema(description = "LiveKit 接入 Token;getActiveSessions 列表里为 null,需要的话单独调 refreshToken") 21 + @Schema(description = "LiveKit 接入 Token;需要时调 refreshToken 重新签发")
22 private String token; 22 private String token;
23 23
24 @Schema(description = "会话场景") 24 @Schema(description = "会话场景")
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcCallDO.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.dal.dataobject.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
  4 +import cn.iocoder.yudao.module.im.dal.dataobject.group.ImGroupDO;
  5 +import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
  6 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallEndReasonEnum;
  7 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallMediaTypeEnum;
  8 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallStatusEnum;
  9 +import com.baomidou.mybatisplus.annotation.KeySequence;
  10 +import com.baomidou.mybatisplus.annotation.TableId;
  11 +import com.baomidou.mybatisplus.annotation.TableName;
  12 +import lombok.*;
  13 +
  14 +import java.time.LocalDateTime;
  15 +
  16 +/**
  17 + * IM 通话记录 DO(房间级 / 主表)
  18 + * <p>
  19 + * 一通通话一行;状态机 CREATED → RUNNING → ENDED;和明细表 {@link ImRtcParticipantDO} 通过 callId 关联
  20 + *
  21 + * @author 芋道源码
  22 + */
  23 +@TableName("im_rtc_call")
  24 +@KeySequence("im_rtc_call_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增;MySQL 等数据库可不写
  25 +@Data
  26 +@EqualsAndHashCode(callSuper = true)
  27 +@ToString(callSuper = true)
  28 +@Builder
  29 +@NoArgsConstructor
  30 +@AllArgsConstructor
  31 +public class ImRtcCallDO extends BaseDO {
  32 +
  33 + /**
  34 + * 编号
  35 + */
  36 + @TableId
  37 + private Long id;
  38 + /**
  39 + * 业务通话编号;唯一;UUID 生成
  40 + */
  41 + private String callId;
  42 + /**
  43 + * LiveKit 房间名
  44 + *
  45 + * 格式 {@code im_rtc_call_{callId}}
  46 + */
  47 + private String roomName;
  48 + /**
  49 + * 会话类型
  50 + *
  51 + * 枚举 {@link ImConversationTypeEnum}
  52 + */
  53 + private Integer conversationType;
  54 + /**
  55 + * 媒体类型
  56 + *
  57 + * 枚举 {@link ImRtcCallMediaTypeEnum}
  58 + */
  59 + private Integer mediaType;
  60 + /**
  61 + * 发起人用户编号
  62 + * <p>
  63 + * 关联 AdminUserDO 的 id 字段
  64 + */
  65 + private Long inviterUserId;
  66 + /**
  67 + * 群编号;私聊为 NULL
  68 + * <p>
  69 + * 关联 {@link ImGroupDO#getId()}
  70 + */
  71 + private Long groupId;
  72 + /**
  73 + * 通话状态
  74 + *
  75 + * 枚举 {@link ImRtcCallStatusEnum}
  76 + */
  77 + private Integer status;
  78 + /**
  79 + * 结束原因;通话未结束时为 NULL
  80 + *
  81 + * 枚举 {@link ImRtcCallEndReasonEnum}
  82 + */
  83 + private Integer endReason;
  84 + /**
  85 + * 发起时间
  86 + */
  87 + private LocalDateTime startTime;
  88 + /**
  89 + * 接通时间
  90 + *
  91 + * 首个非发起人加入时写入;未接通保持 NULL
  92 + */
  93 + private LocalDateTime acceptTime;
  94 + /**
  95 + * 通话结束时间
  96 + */
  97 + private LocalDateTime endTime;
  98 +
  99 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/dataobject/rtc/ImRtcParticipantDO.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.dal.dataobject.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
  4 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantRoleEnum;
  5 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcParticipantStatusEnum;
  6 +import com.baomidou.mybatisplus.annotation.KeySequence;
  7 +import com.baomidou.mybatisplus.annotation.TableId;
  8 +import com.baomidou.mybatisplus.annotation.TableName;
  9 +import lombok.*;
  10 +
  11 +import java.time.LocalDateTime;
  12 +
  13 +/**
  14 + * IM 通话参与者 DO(用户级 / 明细表)
  15 + * <p>
  16 + * 一通通话每个参与者一行;通过 callId 关联主表 {@link ImRtcCallDO}
  17 + * <p>
  18 + * 终态闭合:通话 ENDED 时所有明细 status 必属 {LEFT / REJECTED / NO_ANSWER} 之一
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@TableName("im_rtc_participant")
  23 +@KeySequence("im_rtc_participant_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增;MySQL 等数据库可不写
  24 +@Data
  25 +@EqualsAndHashCode(callSuper = true)
  26 +@ToString(callSuper = true)
  27 +@Builder
  28 +@NoArgsConstructor
  29 +@AllArgsConstructor
  30 +public class ImRtcParticipantDO extends BaseDO {
  31 +
  32 + /**
  33 + * 编号
  34 + */
  35 + @TableId
  36 + private Long id;
  37 + /**
  38 + * 业务通话编号
  39 + * <p>
  40 + * 关联 {@link ImRtcCallDO#getCallId()}
  41 + */
  42 + private String callId;
  43 + /**
  44 + * 参与者用户编号
  45 + * <p>
  46 + * 关联 AdminUserDO 的 id 字段
  47 + */
  48 + private Long userId;
  49 + /**
  50 + * 参与角色
  51 + *
  52 + * 枚举 {@link ImRtcParticipantRoleEnum}
  53 + */
  54 + private Integer role;
  55 + /**
  56 + * 参与状态
  57 + *
  58 + * 枚举 {@link ImRtcParticipantStatusEnum}
  59 + */
  60 + private Integer status;
  61 + /**
  62 + * 被邀请时间;发起人取通话 startTime
  63 + */
  64 + private LocalDateTime inviteTime;
  65 + /**
  66 + * 接听时间;未接听 NULL
  67 + */
  68 + private LocalDateTime acceptTime;
  69 + /**
  70 + * 离开时间;未加入 NULL
  71 + */
  72 + private LocalDateTime leaveTime;
  73 +
  74 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcCallMapper.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.dal.mysql.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
  4 +import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
  5 +import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
  6 +import com.baomidou.mybatisplus.core.toolkit.Wrappers;
  7 +import org.apache.ibatis.annotations.Mapper;
  8 +
  9 +import java.time.LocalDateTime;
  10 +import java.util.Collection;
  11 +import java.util.List;
  12 +
  13 +/**
  14 + * IM 通话记录 Mapper
  15 + *
  16 + * @author 芋道源码
  17 + */
  18 +@Mapper
  19 +public interface ImRtcCallMapper extends BaseMapperX<ImRtcCallDO> {
  20 +
  21 + default ImRtcCallDO selectByCallId(String callId) {
  22 + return selectOne(ImRtcCallDO::getCallId, callId);
  23 + }
  24 +
  25 + default ImRtcCallDO selectByRoomName(String roomName) {
  26 + return selectOne(ImRtcCallDO::getRoomName, roomName);
  27 + }
  28 +
  29 + default ImRtcCallDO selectLastByGroupIdAndStatusIn(Long groupId, Collection<Integer> statuses) {
  30 + return selectLast(new LambdaQueryWrapperX<ImRtcCallDO>()
  31 + .eq(ImRtcCallDO::getGroupId, groupId)
  32 + .in(ImRtcCallDO::getStatus, statuses));
  33 + }
  34 +
  35 + default int updateByIdAndStatus(Long id, Integer oldStatus, ImRtcCallDO updateObj) {
  36 + return update(updateObj, Wrappers.<ImRtcCallDO>lambdaUpdate()
  37 + .eq(ImRtcCallDO::getId, id)
  38 + .eq(ImRtcCallDO::getStatus, oldStatus));
  39 + }
  40 +
  41 + @SuppressWarnings("UnusedReturnValue")
  42 + default int updateByIdAndStatusIn(Long id, Collection<Integer> statuses, ImRtcCallDO updateObj) {
  43 + return update(updateObj, Wrappers.<ImRtcCallDO>lambdaUpdate()
  44 + .eq(ImRtcCallDO::getId, id)
  45 + .in(ImRtcCallDO::getStatus, statuses));
  46 + }
  47 +
  48 + default List<ImRtcCallDO> selectListByStatusInAndStartTimeBefore(Collection<Integer> statuses,
  49 + LocalDateTime startTimeBefore) {
  50 + return selectList(new LambdaQueryWrapperX<ImRtcCallDO>()
  51 + .in(ImRtcCallDO::getStatus, statuses)
  52 + .lt(ImRtcCallDO::getStartTime, startTimeBefore));
  53 + }
  54 +
  55 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/mysql/rtc/ImRtcParticipantMapper.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.dal.mysql.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
  4 +import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
  5 +import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcParticipantDO;
  6 +import com.baomidou.mybatisplus.core.toolkit.Wrappers;
  7 +import org.apache.ibatis.annotations.Mapper;
  8 +
  9 +import java.util.Collection;
  10 +import java.util.List;
  11 +
  12 +/**
  13 + * IM 通话参与者 Mapper
  14 + *
  15 + * @author 芋道源码
  16 + */
  17 +@Mapper
  18 +public interface ImRtcParticipantMapper extends BaseMapperX<ImRtcParticipantDO> {
  19 +
  20 + default ImRtcParticipantDO selectByCallIdAndUserId(String callId, Long userId) {
  21 + return selectOne(new LambdaQueryWrapperX<ImRtcParticipantDO>()
  22 + .eq(ImRtcParticipantDO::getCallId, callId)
  23 + .eq(ImRtcParticipantDO::getUserId, userId));
  24 + }
  25 +
  26 + default List<ImRtcParticipantDO> selectListByCallId(String callId) {
  27 + return selectList(ImRtcParticipantDO::getCallId, callId);
  28 + }
  29 +
  30 + default List<ImRtcParticipantDO> selectListByCallIdAndStatus(String callId, Integer status) {
  31 + return selectList(new LambdaQueryWrapperX<ImRtcParticipantDO>()
  32 + .eq(ImRtcParticipantDO::getCallId, callId)
  33 + .eq(ImRtcParticipantDO::getStatus, status));
  34 + }
  35 +
  36 + default ImRtcParticipantDO selectLastByUserIdAndStatusIn(Long userId, Collection<Integer> statuses) {
  37 + return selectLast(new LambdaQueryWrapperX<ImRtcParticipantDO>()
  38 + .eq(ImRtcParticipantDO::getUserId, userId)
  39 + .in(ImRtcParticipantDO::getStatus, statuses));
  40 + }
  41 +
  42 + @SuppressWarnings("UnusedReturnValue")
  43 + default int updateByIdAndStatus(Long id, Integer oldStatus, ImRtcParticipantDO updateObj) {
  44 + return update(updateObj, Wrappers.<ImRtcParticipantDO>lambdaUpdate()
  45 + .eq(ImRtcParticipantDO::getId, id)
  46 + .eq(ImRtcParticipantDO::getStatus, oldStatus));
  47 + }
  48 +
  49 + @SuppressWarnings("UnusedReturnValue")
  50 + default int updateByCallIdAndStatus(String callId, Integer oldStatus, ImRtcParticipantDO updateObj) {
  51 + return update(updateObj, Wrappers.<ImRtcParticipantDO>lambdaUpdate()
  52 + .eq(ImRtcParticipantDO::getCallId, callId)
  53 + .eq(ImRtcParticipantDO::getStatus, oldStatus));
  54 + }
  55 +
  56 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/RedisKeyConstants.java
@@ -41,4 +41,15 @@ public interface RedisKeyConstants { @@ -41,4 +41,15 @@ public interface RedisKeyConstants {
41 */ 41 */
42 String GROUP_MEMBER_IDS = "group_member_ids"; 42 String GROUP_MEMBER_IDS = "group_member_ids";
43 43
  44 + /**
  45 + * 通话同对 / 同群活跃唯一性的分布式锁
  46 + * <p>
  47 + * KEY 格式:im_rtc_call:{conversationType}:{suffix}
  48 + * 私聊(conversationType=1):suffix = {小 userId}_{大 userId}
  49 + * 群聊(conversationType=2):suffix = {groupId}
  50 + * VALUE 数据格式:HASH // RLock.class:Redisson 的 Lock 锁,使用 Hash 数据结构
  51 + * 过期时间:不固定(lock 时显式传 timeoutMillis)
  52 + */
  53 + String IM_RTC_CALL_LOCK = "im_rtc_call:%d:%s";
  54 +
44 } 55 }
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/dal/redis/rtc/ImRtcCallLockRedisDAO.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.dal.redis.rtc;
  2 +
  3 +import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
  4 +import jakarta.annotation.Resource;
  5 +import lombok.extern.slf4j.Slf4j;
  6 +import org.redisson.api.RLock;
  7 +import org.redisson.api.RedissonClient;
  8 +import org.springframework.stereotype.Repository;
  9 +
  10 +import java.util.concurrent.Callable;
  11 +import java.util.concurrent.TimeUnit;
  12 +
  13 +import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
  14 +import static cn.iocoder.yudao.module.im.dal.redis.RedisKeyConstants.IM_RTC_CALL_LOCK;
  15 +import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.RTC_INVITE_BUSY;
  16 +
  17 +/**
  18 + * IM 通话同对 / 同群活跃唯一性的锁 Redis DAO
  19 + * <p>
  20 + * invite 入口包一层;锁内做「SELECT 已有活跃通话 → 命中即加入分支;否则 INSERT 新通话」
  21 + *
  22 + * @author 芋道源码
  23 + */
  24 +@Repository
  25 +@Slf4j
  26 +public class ImRtcCallLockRedisDAO {
  27 +
  28 + /**
  29 + * 等待获取锁的最长时间;超时抛 RTC_INVITE_BUSY
  30 + */
  31 + private static final long LOCK_WAIT_MS = 5_000L;
  32 + /**
  33 + * 持有锁的最长时间;自动释放兜底;给 LiveKit / DB / 推送偶发慢留余地
  34 + */
  35 + private static final long LOCK_LEASE_MS = 30_000L;
  36 +
  37 + @Resource
  38 + private RedissonClient redissonClient;
  39 +
  40 + /**
  41 + * 私聊通话锁;按 userId 排序拼 key 保对称(同对呼叫和反向呼叫共用一把锁)
  42 + */
  43 + public <V> V lockPrivate(Long userIdA, Long userIdB, Callable<V> callable) throws Exception {
  44 + String key = String.format(IM_RTC_CALL_LOCK, ImConversationTypeEnum.PRIVATE.getType(),
  45 + Math.min(userIdA, userIdB)+ "_" + Math.max(userIdA, userIdB));
  46 + return doLock(key, callable);
  47 + }
  48 +
  49 + /**
  50 + * 群通话锁;同群所有 invite 串行
  51 + */
  52 + public <V> V lockGroup(Long groupId, Callable<V> callable) throws Exception {
  53 + String key = String.format(IM_RTC_CALL_LOCK, ImConversationTypeEnum.GROUP.getType(), groupId);
  54 + return doLock(key, callable);
  55 + }
  56 +
  57 + /**
  58 + * tryLock(waitTime, leaseTime, unit):waitTime 内拿不到锁直接抛繁忙;拿到后 leaseTime 自动释放
  59 + * <p>
  60 + * unlock 前用 isHeldByCurrentThread 兜底;业务超过 leaseTime 时锁已自动释放,不再抛 IllegalMonitorStateException
  61 + */
  62 + private <V> V doLock(String lockKey, Callable<V> callable) throws Exception {
  63 + RLock lock = redissonClient.getLock(lockKey);
  64 + boolean acquired = lock.tryLock(LOCK_WAIT_MS, LOCK_LEASE_MS, TimeUnit.MILLISECONDS);
  65 + if (!acquired) {
  66 + log.error("[doLock][lockKey={} 等待 {}ms 仍未获取到锁]", lockKey, LOCK_WAIT_MS);
  67 + throw exception(RTC_INVITE_BUSY);
  68 + }
  69 + try {
  70 + return callable.call();
  71 + } finally {
  72 + if (lock.isHeldByCurrentThread()) {
  73 + lock.unlock();
  74 + } else {
  75 + log.error("[doLock][lockKey={} 业务超过 leaseTime={}ms,锁已被 Redisson 自动释放]", lockKey, LOCK_LEASE_MS);
  76 + }
  77 + }
  78 + }
  79 +
  80 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/ErrorCodeConstants.java
@@ -89,5 +89,6 @@ public interface ErrorCodeConstants { @@ -89,5 +89,6 @@ public interface ErrorCodeConstants {
89 ErrorCode RTC_INVITE_SELF = new ErrorCode(1_040_900_005, "不能呼叫自己"); 89 ErrorCode RTC_INVITE_SELF = new ErrorCode(1_040_900_005, "不能呼叫自己");
90 ErrorCode RTC_PRIVATE_INVITEE_REQUIRED = new ErrorCode(1_040_900_006, "私聊通话必须指定对方"); 90 ErrorCode RTC_PRIVATE_INVITEE_REQUIRED = new ErrorCode(1_040_900_006, "私聊通话必须指定对方");
91 ErrorCode RTC_GROUP_REQUIRED = new ErrorCode(1_040_900_007, "群聊通话必须指定群编号"); 91 ErrorCode RTC_GROUP_REQUIRED = new ErrorCode(1_040_900_007, "群聊通话必须指定群编号");
  92 + ErrorCode RTC_INVITE_BUSY = new ErrorCode(1_040_900_008, "通话发起繁忙,请稍后再试");
92 93
93 } 94 }
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/message/ImMessageTypeEnum.java
@@ -78,35 +78,42 @@ public enum ImMessageTypeEnum implements ArrayValuable&lt;Integer&gt; { @@ -78,35 +78,42 @@ public enum ImMessageTypeEnum implements ArrayValuable&lt;Integer&gt; {
78 */ 78 */
79 READ(2201, "已读", false, false), 79 READ(2201, "已读", false, false),
80 80
81 - // ========== 实时通话信令(2300-2302) ==========  
82 - // TODO @AI:openim 的信号,是怎么设计的?感觉对齐下,会好一点点,不然用户不好拓展; 81 + // ========== 实时通话信令(1601-1605 段位与 OpenIM 对齐;1610+ 自有扩展) ==========
83 /** 82 /**
84 - * 通话邀请:主叫发起后推给被叫弹「来电」界面;payload 为 RtcInviteNotification(roomName / mediaType / scene / inviter / token) 83 + * 对应 OpenIM:SignalingNotification 1601(通话信令统一入口)
  84 + * 对应自己的类:ImRtcCallNotification
  85 + * 场景:通话信令;不入库,走 imWebSocketService 仅推参与方;signalingType 区分 INVITE / REJECT(预留 ACCEPT / CANCEL / HUNGUP)
85 */ 86 */
86 - RTC_INVITE(2300, "通话邀请", false, false), 87 + RTC_CALL(1601, "通话信令", false, false),
87 /** 88 /**
88 - * 通话已接通:被叫接听后推给主叫切换到「通话中」UI;payload 为 RtcAcceptNotification(roomName / acceptorId) 89 + * 对应 OpenIM:RoomParticipantsConnectedNotification 1602
  90 + * 对应自己的类:ImRtcParticipantConnectedNotification
  91 + * 场景:通话参与者加入;LiveKit webhook participant_joined 触发;私聊推 peer 多端 + inviter 多端,群聊全群广播;不入库
89 */ 92 */
90 - RTC_ACCEPT(2301, "通话接通", false, false), 93 + RTC_PARTICIPANT_CONNECTED(1602, "通话参与者加入", false, false),
91 /** 94 /**
92 - * 通话结束:拒绝 / 取消 / 挂断 / 超时 / 异常 统一走这一条;payload 为 RtcEndNotification(roomName / reason / operatorId) 95 + * 对应 OpenIM:RoomParticipantsDisconnectedNotification 1603
  96 + * 对应自己的类:ImRtcParticipantDisconnectedNotification
  97 + * 场景:通话参与者离开;LiveKit webhook participant_left 触发;推送范围同 1602;不入库
93 */ 98 */
94 - RTC_END(2302, "通话结束", false, false),  
95 -  
96 - // ========== 群通话广播信号(2310-2312):让所有群成员能感知 / 主动加入 ==========  
97 - /**  
98 - * 群通话开始:群通话发起时给所有群成员广播;payload 为 ImRtcGroupNotification(callId / roomName / mediaType / inviterId / joinedUserIds)  
99 - */  
100 - RTC_GROUP_STARTED(2310, "群通话开始", false, false), 99 + RTC_PARTICIPANT_DISCONNECTED(1603, "通话参与者离开", false, false),
  100 + // 1604-1609 OpenIM 已用 / 留作扩展,本系统暂不使用
101 /** 101 /**
102 - * 群通话结束:最后一人离开 / 异常关房时广播;payload 仅含 roomName / groupId 102 + * 对应 OpenIM:无(自有扩展,OpenIM 通话事件不入消息流)
  103 + * 对应自己的类:ImRtcCallStartNotification
  104 + * 场景:群通话开始;入 im_group_message + 全群广播;前端渲染聊天 tip「{inviterNickname} 发起了{voice/video}通话」
  105 + * <p>
  106 + * 与 RTC_CALL_END(1611) 两段式配对:START 一定先于 END 入库(START 在 invite 接口事务里、END 在 cancel/leave 接口事务里,自然按请求顺序串行)
103 */ 107 */
104 - RTC_GROUP_ENDED(2311, "群通话结束", false, false), 108 + RTC_CALL_START(1610, "通话开始", true, false),
105 /** 109 /**
106 - * 群通话成员变更:有人加入 / 离开时广播;payload 为 ImRtcGroupNotification(含最新 joinedUserIds);  
107 - * 用于胶囊条上「N 人正在通话」实时刷新 110 + * 对应 OpenIM:无(自有扩展,OpenIM 通话事件不入消息流)
  111 + * 对应自己的类:ImRtcCallEndNotification
  112 + * 场景:通话结束;入 im_private_message / im_group_message;私聊渲染准气泡,群聊渲染 tip「{voice/video}通话已结束 [时长 X]」
  113 + * <p>
  114 + * 群通话两段式配对前导是 RTC_CALL_START(1610);私聊场景没有 START,单 END 准气泡承载完整生命周期
108 */ 115 */
109 - RTC_GROUP_UPDATED(2312, "群通话成员变更", false, false), 116 + RTC_CALL_END(1611, "通话结束", true, false),
110 117
111 // ========== 好友通知(1201-1210 直接复用 OpenIM 段位编号) ========== 118 // ========== 好友通知(1201-1210 直接复用 OpenIM 段位编号) ==========
112 // TODO @芋艿:FRIEND_REQUEST_* 与 GROUP_REQUEST_* 都是 persistent=false 的 SysMsg,离线 pull 拉不到, 119 // TODO @芋艿:FRIEND_REQUEST_* 与 GROUP_REQUEST_* 都是 persistent=false 的 SysMsg,离线 pull 拉不到,
@@ -224,7 +231,7 @@ public enum ImMessageTypeEnum implements ArrayValuable&lt;Integer&gt; { @@ -224,7 +231,7 @@ public enum ImMessageTypeEnum implements ArrayValuable&lt;Integer&gt; {
224 /** 231 /**
225 * 对应 OpenIM:sdkws.MemberInvitedTips(MemberInvitedNotification 1509) 232 * 对应 OpenIM:sdkws.MemberInvitedTips(MemberInvitedNotification 1509)
226 * 对应自己的类:GroupMemberInviteNotification 233 * 对应自己的类:GroupMemberInviteNotification
227 - * 场景:成员邀请新人入群,全员广播(含被邀请者);被邀请人前端按 memberUserIds 含自己自判 fetchGroupInfo + fetchGroupMembers bootstrap 234 + * 场景:成员邀请新人入群,全员广播(含被邀请者);被邀请人前端按 memberUserIds 含自己自判,初次拉取 fetchGroupInfo + fetchGroupMembers
228 */ 235 */
229 GROUP_MEMBER_INVITE(1509, "成员加入", true, false), 236 GROUP_MEMBER_INVITE(1509, "成员加入", true, false),
230 /** 237 /**
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImCallStatusEnum.java deleted
1 -package cn.iocoder.yudao.module.im.enums.rtc;  
2 -  
3 -import cn.iocoder.yudao.framework.common.core.ArrayValuable;  
4 -import lombok.Getter;  
5 -import lombok.RequiredArgsConstructor;  
6 -  
7 -import java.util.Arrays;  
8 -import java.util.Objects;  
9 -  
10 -// TODO @AI:是否一定要有 status 状态的记录????【可以在调研下】  
11 -/**  
12 - * IM 通话状态枚举  
13 - *  
14 - * @author 芋道源码  
15 - */  
16 -@RequiredArgsConstructor  
17 -@Getter  
18 -public enum ImCallStatusEnum implements ArrayValuable<Integer> {  
19 -  
20 - INVITING(10, "邀请中"), // 邀请中:1v1 主叫等待被叫接听  
21 - ONGOING(20, "通话中"), // 通话中:至少一名被叫已接听;群通话从发起人加入即进入此状态  
22 - ENDED(30, "已结束"); // 已结束:会话即将销毁;写入消息流后从内存移除  
23 -  
24 - public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImCallStatusEnum::getType).toArray(Integer[]::new);  
25 -  
26 - private final Integer type;  
27 - private final String name;  
28 -  
29 - @Override  
30 - public Integer[] array() {  
31 - return ARRAYS;  
32 - }  
33 -  
34 - public static boolean isInviting(Integer type) {  
35 - return Objects.equals(INVITING.type, type);  
36 - }  
37 -  
38 - public static boolean isOngoing(Integer type) {  
39 - return Objects.equals(ONGOING.type, type);  
40 - }  
41 -  
42 - public static boolean isEnded(Integer type) {  
43 - return Objects.equals(ENDED.type, type);  
44 - }  
45 -  
46 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcCallEndReasonEnum.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.enums.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.common.core.ArrayValuable;
  4 +import lombok.Getter;
  5 +import lombok.RequiredArgsConstructor;
  6 +
  7 +import java.util.Arrays;
  8 +
  9 +/**
  10 + * IM 通话结束原因枚举;落历史消息时计算 content 文案
  11 + * <p>
  12 + * 不含 TIMEOUT;产品决策走微信式不超时,振铃由用户主动 cancel / reject
  13 + *
  14 + * @author 芋道源码
  15 + */
  16 +@RequiredArgsConstructor
  17 +@Getter
  18 +public enum ImRtcCallEndReasonEnum implements ArrayValuable<Integer> {
  19 +
  20 + HANGUP(1, "通话结束"), // 接通后任一方主动挂断
  21 + REJECT(2, "已拒绝"), // 被叫接通前点拒接
  22 + CANCEL(3, "已取消"), // 主叫接通前主动取消
  23 + BUSY(5, "对方正忙"), // 私聊呼叫时对方在另一通话
  24 + ERROR(9, "通话异常"); // 网络中断、设备失败等
  25 +
  26 + public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImRtcCallEndReasonEnum::getReason).toArray(Integer[]::new);
  27 +
  28 + /**
  29 + * 原因值
  30 + */
  31 + private final Integer reason;
  32 + /**
  33 + * 名字
  34 + */
  35 + private final String name;
  36 +
  37 + @Override
  38 + public Integer[] array() {
  39 + return ARRAYS;
  40 + }
  41 +
  42 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImCallMediaTypeEnum.java renamed to yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcCallMediaTypeEnum.java
@@ -5,6 +5,7 @@ import lombok.Getter; @@ -5,6 +5,7 @@ import lombok.Getter;
5 import lombok.RequiredArgsConstructor; 5 import lombok.RequiredArgsConstructor;
6 6
7 import java.util.Arrays; 7 import java.util.Arrays;
  8 +import java.util.Objects;
8 9
9 /** 10 /**
10 * IM 通话媒体类型枚举 11 * IM 通话媒体类型枚举
@@ -13,14 +14,20 @@ import java.util.Arrays; @@ -13,14 +14,20 @@ import java.util.Arrays;
13 */ 14 */
14 @RequiredArgsConstructor 15 @RequiredArgsConstructor
15 @Getter 16 @Getter
16 -public enum ImCallMediaTypeEnum implements ArrayValuable<Integer> { 17 +public enum ImRtcCallMediaTypeEnum implements ArrayValuable<Integer> {
17 18
18 - VOICE(1, "语音"),  
19 - VIDEO(2, "视频"); 19 + VOICE(1, "语音"), // 仅音频
  20 + VIDEO(2, "视频"); // 音频 + 视频
20 21
21 - public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImCallMediaTypeEnum::getType).toArray(Integer[]::new); 22 + public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImRtcCallMediaTypeEnum::getType).toArray(Integer[]::new);
22 23
  24 + /**
  25 + * 类型
  26 + */
23 private final Integer type; 27 private final Integer type;
  28 + /**
  29 + * 名字
  30 + */
24 private final String name; 31 private final String name;
25 32
26 @Override 33 @Override
@@ -28,4 +35,12 @@ public enum ImCallMediaTypeEnum implements ArrayValuable&lt;Integer&gt; { @@ -28,4 +35,12 @@ public enum ImCallMediaTypeEnum implements ArrayValuable&lt;Integer&gt; {
28 return ARRAYS; 35 return ARRAYS;
29 } 36 }
30 37
  38 + public static boolean isVoice(Integer type) {
  39 + return Objects.equals(VOICE.type, type);
  40 + }
  41 +
  42 + public static boolean isVideo(Integer type) {
  43 + return Objects.equals(VIDEO.type, type);
  44 + }
  45 +
31 } 46 }
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcCallStatusEnum.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.enums.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.common.core.ArrayValuable;
  4 +import lombok.Getter;
  5 +import lombok.RequiredArgsConstructor;
  6 +
  7 +import java.util.Arrays;
  8 +import java.util.List;
  9 +import java.util.Objects;
  10 +
  11 +/**
  12 + * IM 通话状态枚举(主表 status)
  13 + * <p>
  14 + * 状态机:CREATED → RUNNING → ENDED;CREATED 直接到 ENDED 表示无人接听 / 主叫取消
  15 + *
  16 + * @author 芋道源码
  17 + */
  18 +@RequiredArgsConstructor
  19 +@Getter
  20 +public enum ImRtcCallStatusEnum implements ArrayValuable<Integer> {
  21 +
  22 + CREATED(10, "创建"), // 通话已创建;私聊:等被叫接听;群聊:发起人已进房,等其他人加入
  23 + RUNNING(20, "进行中"), // 第一个非发起人接通后进入
  24 + ENDED(30, "已结束"); // 任一方挂断 / 主叫 cancel / Webhook 兜底
  25 +
  26 + public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImRtcCallStatusEnum::getStatus).toArray(Integer[]::new);
  27 +
  28 + /**
  29 + * 活跃状态集合;通话未结束(CREATED / RUNNING)
  30 + */
  31 + public static final List<Integer> ACTIVE_STATUSES = List.of(CREATED.getStatus(), RUNNING.getStatus());
  32 +
  33 + /**
  34 + * 状态值
  35 + */
  36 + private final Integer status;
  37 + /**
  38 + * 状态名
  39 + */
  40 + private final String name;
  41 +
  42 + @Override
  43 + public Integer[] array() {
  44 + return ARRAYS;
  45 + }
  46 +
  47 + public static boolean isCreated(Integer status) {
  48 + return Objects.equals(CREATED.status, status);
  49 + }
  50 +
  51 + public static boolean isRunning(Integer status) {
  52 + return Objects.equals(RUNNING.status, status);
  53 + }
  54 +
  55 + public static boolean isEnded(Integer status) {
  56 + return Objects.equals(ENDED.status, status);
  57 + }
  58 +
  59 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImCallEndReasonEnum.java renamed to yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcParticipantRoleEnum.java
@@ -7,32 +7,27 @@ import lombok.RequiredArgsConstructor; @@ -7,32 +7,27 @@ import lombok.RequiredArgsConstructor;
7 import java.util.Arrays; 7 import java.util.Arrays;
8 8
9 /** 9 /**
10 - * IM 通话结束原因枚举;用于落历史消息时计算 content 文案 10 + * IM 通话参与者角色枚举
11 * 11 *
12 * @author 芋道源码 12 * @author 芋道源码
13 */ 13 */
14 @RequiredArgsConstructor 14 @RequiredArgsConstructor
15 @Getter 15 @Getter
16 -public enum ImCallEndReasonEnum implements ArrayValuable<Integer> {  
17 -  
18 - // TODO @AI:这个对齐的哪个东西;  
19 - // TODO @AI:尾注释;  
20 - /** 正常挂断:双方在通话中后由任一方主动挂断 */  
21 - HANGUP(1, "通话结束"),  
22 - /** 拒绝:被叫接通前点拒接 */  
23 - REJECT(2, "已拒绝"),  
24 - /** 取消:主叫接通前主动取消 */  
25 - CANCEL(3, "已取消"),  
26 - /** 超时:振铃 30 秒未应答 */  
27 - TIMEOUT(4, "未接通"),  
28 - /** 对方忙线 */  
29 - BUSY(5, "对方正忙"),  
30 - /** 异常:网络中断、设备失败等 */  
31 - ERROR(9, "通话异常");  
32 -  
33 - public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImCallEndReasonEnum::getType).toArray(Integer[]::new);  
34 -  
35 - private final Integer type; 16 +public enum ImRtcParticipantRoleEnum implements ArrayValuable<Integer> {
  17 +
  18 + INVITER(1, "发起人"),
  19 + INVITEE(2, "被邀请者"),
  20 + ACTIVE_JOIN(3, "主动加入者"); // 仅群通话场景:旁观者点胶囊条加入已有通话
  21 +
  22 + public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImRtcParticipantRoleEnum::getRole).toArray(Integer[]::new);
  23 +
  24 + /**
  25 + * 角色
  26 + */
  27 + private final Integer role;
  28 + /**
  29 + * 名字
  30 + */
36 private final String name; 31 private final String name;
37 32
38 @Override 33 @Override
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/enums/rtc/ImRtcParticipantStatusEnum.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.enums.rtc;
  2 +
  3 +import cn.iocoder.yudao.framework.common.core.ArrayValuable;
  4 +import lombok.Getter;
  5 +import lombok.RequiredArgsConstructor;
  6 +
  7 +import java.util.Arrays;
  8 +import java.util.List;
  9 +import java.util.Objects;
  10 +
  11 +/**
  12 + * IM 通话参与者状态枚举(明细表 status)
  13 + * <p>
  14 + * 终态闭合:通话 ENDED 时所有明细 status 必属 {LEFT / REJECTED / NO_ANSWER}
  15 + * <p>
  16 + * 1、INVITING → JOINED → LEFT(接通后挂断 / 离开);
  17 + * 2、INVITING → REJECTED(接通前点拒接);
  18 + * 3、INVITING → NO_ANSWER(通话结束仍未应答)
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@RequiredArgsConstructor
  23 +@Getter
  24 +public enum ImRtcParticipantStatusEnum implements ArrayValuable<Integer> {
  25 +
  26 + // TODO @芋艿:hand up 要不要也搞下。
  27 + INVITING(10, "邀请中"), // 已发出 invite,等被叫响应
  28 + JOINED(20, "已加入"), // 已 connect 进 LiveKit 房间
  29 + REJECTED(30, "已拒绝"), // 接通前点拒接
  30 + NO_ANSWER(40, "未应答"), // 通话已结束仍未应答;endSession 批量改
  31 + LEFT(50, "已离开"); // 接通后挂断 / Webhook 兜底
  32 +
  33 + public static final Integer[] ARRAYS = Arrays.stream(values()).map(ImRtcParticipantStatusEnum::getStatus).toArray(Integer[]::new);
  34 +
  35 + /**
  36 + * 活跃状态集合;尚未离开通话(INVITING / JOINED)
  37 + */
  38 + public static final List<Integer> ACTIVE_STATUSES = List.of(INVITING.getStatus(), JOINED.getStatus());
  39 +
  40 + /**
  41 + * 状态值
  42 + */
  43 + private final Integer status;
  44 + /**
  45 + * 状态名
  46 + */
  47 + private final String name;
  48 +
  49 + @Override
  50 + public Integer[] array() {
  51 + return ARRAYS;
  52 + }
  53 +
  54 + public static boolean isInviting(Integer status) {
  55 + return Objects.equals(INVITING.status, status);
  56 + }
  57 +
  58 + public static boolean isJoined(Integer status) {
  59 + return Objects.equals(JOINED.status, status);
  60 + }
  61 +
  62 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/config/ImProperties.java
1 package cn.iocoder.yudao.module.im.framework.config; 1 package cn.iocoder.yudao.module.im.framework.config;
2 2
  3 +import jakarta.validation.Valid;
  4 +import jakarta.validation.constraints.NotBlank;
  5 +import jakarta.validation.constraints.Size;
3 import lombok.Data; 6 import lombok.Data;
4 import org.springframework.boot.context.properties.ConfigurationProperties; 7 import org.springframework.boot.context.properties.ConfigurationProperties;
5 import org.springframework.stereotype.Component; 8 import org.springframework.stereotype.Component;
  9 +import org.springframework.validation.annotation.Validated;
6 10
7 -// TODO @AI:参考别的模块;是不是也要分 im;config 和 core 这样;在瞅瞅;  
8 /** 11 /**
9 * IM 模块全局配置 12 * IM 模块全局配置
10 * <p> 13 * <p>
@@ -15,6 +18,7 @@ import org.springframework.stereotype.Component; @@ -15,6 +18,7 @@ import org.springframework.stereotype.Component;
15 */ 18 */
16 @Component 19 @Component
17 @ConfigurationProperties(prefix = "yudao.im") 20 @ConfigurationProperties(prefix = "yudao.im")
  21 +@Validated
18 @Data 22 @Data
19 public class ImProperties { 23 public class ImProperties {
20 24
@@ -24,6 +28,7 @@ public class ImProperties { @@ -24,6 +28,7 @@ public class ImProperties {
24 28
25 private Message message = new Message(); 29 private Message message = new Message();
26 30
  31 + @Valid
27 private Rtc rtc = new Rtc(); 32 private Rtc rtc = new Rtc();
28 33
29 /** 34 /**
@@ -134,16 +139,20 @@ public class ImProperties { @@ -134,16 +139,20 @@ public class ImProperties {
134 /** 139 /**
135 * LiveKit Server WebSocket 地址;客户端 connect 时使用,通常 ws://host:7880 或 wss://host 140 * LiveKit Server WebSocket 地址;客户端 connect 时使用,通常 ws://host:7880 或 wss://host
136 */ 141 */
  142 + @NotBlank(message = "LiveKit URL 不能为空")
137 private String livekitUrl = "ws://127.0.0.1:7880"; 143 private String livekitUrl = "ws://127.0.0.1:7880";
138 144
139 /** 145 /**
140 * LiveKit API Key 146 * LiveKit API Key
141 */ 147 */
  148 + @NotBlank(message = "LiveKit API Key 不能为空")
142 private String apiKey = "devkey"; 149 private String apiKey = "devkey";
143 150
144 /** 151 /**
145 * LiveKit API Secret;生产必须改为强随机值 152 * LiveKit API Secret;生产必须改为强随机值
146 */ 153 */
  154 + @NotBlank(message = "LiveKit API Secret 不能为空")
  155 + @Size(min = 32, message = "LiveKit API Secret 长度需 ≥ 32 位")
147 private String apiSecret = "secret-poc-key-min-32-chars-required-here"; 156 private String apiSecret = "secret-poc-key-min-32-chars-required-here";
148 157
149 /** 158 /**
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/config/ImRtcConfiguration.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.framework.rtc.config;
  2 +
  3 +import cn.iocoder.yudao.module.im.framework.config.ImProperties;
  4 +import cn.iocoder.yudao.module.im.framework.rtc.core.LiveKitClient;
  5 +import org.springframework.context.annotation.Bean;
  6 +import org.springframework.context.annotation.Configuration;
  7 +
  8 +/**
  9 + * im 模块的 RTC 组件的 Configuration
  10 + *
  11 + * @author 芋道源码
  12 + */
  13 +@Configuration(proxyBeanMethods = false)
  14 +public class ImRtcConfiguration {
  15 +
  16 + /**
  17 + * LiveKit 客户端
  18 + */
  19 + @Bean
  20 + public LiveKitClient liveKitClient(ImProperties imProperties) {
  21 + return new LiveKitClient(imProperties);
  22 + }
  23 +
  24 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitClient.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.framework.rtc.core;
  2 +
  3 +import cn.hutool.core.collection.CollUtil;
  4 +import cn.hutool.core.lang.Assert;
  5 +import cn.hutool.core.util.StrUtil;
  6 +import cn.hutool.http.HttpRequest;
  7 +import cn.hutool.http.HttpResponse;
  8 +import cn.hutool.json.JSONArray;
  9 +import cn.hutool.json.JSONUtil;
  10 +import cn.hutool.jwt.JWT;
  11 +import cn.hutool.jwt.signers.JWTSignerUtil;
  12 +import cn.iocoder.yudao.framework.common.util.http.HttpUtils;
  13 +import cn.iocoder.yudao.module.im.framework.config.ImProperties;
  14 +import lombok.extern.slf4j.Slf4j;
  15 +
  16 +import java.time.Duration;
  17 +import java.time.Instant;
  18 +import java.util.Date;
  19 +import java.util.HashMap;
  20 +import java.util.Map;
  21 +
  22 +/**
  23 + * LiveKit 客户端
  24 + * <p>
  25 + * 厂商绑定层封装:Token 签发(join / admin)、Server API(DeleteRoom / ListParticipants)、identity 拼接 / 解析
  26 + * <p>
  27 + * 由 {@link cn.iocoder.yudao.module.im.framework.rtc.config.ImRtcConfiguration} 注册为 Bean,本类不带 {@code @Component}
  28 + *
  29 + * @author 芋道源码
  30 + */
  31 +@Slf4j
  32 +public class LiveKitClient {
  33 +
  34 + /**
  35 + * Twirp 端点路径
  36 + */
  37 + private static final String TWIRP_DELETE_ROOM = "/twirp/livekit.RoomService/DeleteRoom";
  38 + private static final String TWIRP_LIST_PARTICIPANTS = "/twirp/livekit.RoomService/ListParticipants";
  39 +
  40 + /**
  41 + * 管理 Token 有效期;Server API 一次调用即弃,10 秒足够
  42 + */
  43 + private static final Duration ADMIN_TOKEN_TTL = Duration.ofSeconds(10);
  44 +
  45 + /**
  46 + * Server API HTTP 调用的超时上限;超时后直接报错,避免 LiveKit 异常 / 网络抖动时业务长时间阻塞
  47 + */
  48 + private static final int SERVER_API_TIMEOUT_MS = 10_000;
  49 +
  50 + private final ImProperties imProperties;
  51 +
  52 + public LiveKitClient(ImProperties imProperties) {
  53 + this.imProperties = imProperties;
  54 + }
  55 +
  56 + /**
  57 + * 签发客户端进房 Token;有效期从 {@link ImProperties.Rtc#getTokenTtlHours()} 读取
  58 + *
  59 + * @param identity 用户唯一标识;写入 sub claim;同 identity 重连会踢前一个连接
  60 + * @param displayName 客户端展示名;可空
  61 + * @param roomName 房间名
  62 + * @return JWT 字符串
  63 + */
  64 + public String signJoinToken(String identity, String displayName, String roomName) {
  65 + Assert.notBlank(identity, "identity 不可为空");
  66 + Assert.notBlank(roomName, "roomName 不可为空");
  67 + ImProperties.Rtc cfg = imProperties.getRtc();
  68 +
  69 + // video claim:限定客户端能在该房间内做什么
  70 + Map<String, Object> video = new HashMap<>();
  71 + video.put("roomJoin", true); // 允许加入房间
  72 + video.put("room", roomName); // 限定只能加入这个房间
  73 + video.put("canPublish", true); // 允许发布媒体(推流)
  74 + video.put("canSubscribe", true); // 允许订阅媒体(拉流)
  75 + video.put("canPublishData", true); // 允许发送 data channel 消息
  76 +
  77 + long nowSec = Instant.now().getEpochSecond();
  78 + long ttlSec = Duration.ofHours(cfg.getTokenTtlHours()).getSeconds();
  79 + JWT jwt = JWT.create()
  80 + .setIssuer(cfg.getApiKey())
  81 + .setSubject(identity)
  82 + .setNotBefore(new Date(nowSec * 1000))
  83 + .setExpiresAt(new Date((nowSec + ttlSec) * 1000))
  84 + .setPayload("video", video)
  85 + .setSigner(JWTSignerUtil.hs256(cfg.getApiSecret().getBytes()));
  86 + if (StrUtil.isNotEmpty(displayName)) {
  87 + jwt.setPayload("name", displayName);
  88 + }
  89 + return jwt.sign();
  90 + }
  91 +
  92 + /**
  93 + * 调用 LiveKit Server API 删除房间:用于通话结束时,强制断开异常残留客户端
  94 + *
  95 + * @param roomName 房间名
  96 + */
  97 + @SuppressWarnings("EmptyTryBlock")
  98 + public void deleteRoom(String roomName) {
  99 + try (HttpResponse ignored = postTwirp(TWIRP_DELETE_ROOM, roomName)) {
  100 + // 状态码不区分;调用方失败即兜底
  101 + }
  102 + }
  103 +
  104 + /**
  105 + * 调用 LiveKit Server API 查询某房间内的参与者数量:用于定时扫描僵尸通话
  106 + * <p>
  107 + * 房间不存在 LiveKit 返回 404,视同 0 人
  108 + *
  109 + * @param roomName 房间名
  110 + * @return 参与者数量;HTTP 失败返回 -1
  111 + */
  112 + public int listParticipants(String roomName) {
  113 + try (HttpResponse response = postTwirp(TWIRP_LIST_PARTICIPANTS, roomName)) {
  114 + if (response.getStatus() == 404) {
  115 + return 0;
  116 + }
  117 + if (!response.isOk()) {
  118 + log.warn("[listParticipants][LiveKit 返回非 2xx status={} roomName={} body={}]",
  119 + response.getStatus(), roomName, response.body());
  120 + return -1;
  121 + }
  122 + JSONArray participants = JSONUtil.parseObj(response.body()).getJSONArray("participants");
  123 + return CollUtil.size(participants);
  124 + }
  125 + }
  126 +
  127 + /**
  128 + * 拼接 LiveKit identity;当前单端 = userId 字符串
  129 + * <p>
  130 + * 多端扩展时改 {@code userId + "#" + terminal} 格式,调用方无需改
  131 + *
  132 + * @param userId 用户编号
  133 + * @return identity 字符串
  134 + */
  135 + public String buildIdentity(Long userId) {
  136 + Assert.notNull(userId, "userId 不可为空");
  137 + return String.valueOf(userId);
  138 + }
  139 +
  140 + /**
  141 + * 从 LiveKit identity 解析业务 userId
  142 + * <p>
  143 + * 当前 identity 直接是 userId 字符串;预留 {@code userId#terminal} 多端格式
  144 + *
  145 + * @param identity LiveKit identity
  146 + * @return 用户编号;解析失败返回 null
  147 + */
  148 + public Long parseUserId(String identity) {
  149 + if (StrUtil.isBlank(identity)) {
  150 + return null;
  151 + }
  152 + int sep = identity.indexOf('#');
  153 + String idPart = sep >= 0 ? identity.substring(0, sep) : identity;
  154 + try {
  155 + return Long.parseLong(idPart);
  156 + } catch (NumberFormatException e) {
  157 + return null;
  158 + }
  159 + }
  160 +
  161 + /**
  162 + * 签发管理 Token;用于调 Server API(DeleteRoom / ListParticipants / RemoveParticipant 等)
  163 + *
  164 + * @return JWT 字符串
  165 + */
  166 + private String signAdminToken() {
  167 + ImProperties.Rtc cfg = imProperties.getRtc();
  168 + // roomAdmin claim 给管理类 API 必备
  169 + long nowSec = Instant.now().getEpochSecond();
  170 + return JWT.create()
  171 + .setIssuer(cfg.getApiKey())
  172 + .setNotBefore(new Date(nowSec * 1000))
  173 + .setExpiresAt(new Date((nowSec + ADMIN_TOKEN_TTL.getSeconds()) * 1000))
  174 + .setPayload("video", Map.of("roomAdmin", true))
  175 + .setSigner(JWTSignerUtil.hs256(cfg.getApiSecret().getBytes()))
  176 + .sign();
  177 + }
  178 +
  179 + /**
  180 + * Twirp 协议 POST 调用;统一处理 ws→http 协议切换、签 admin token、Bearer 头、JSON body、超时
  181 + *
  182 + * @param path Twirp 端点路径,例如 {@code /twirp/livekit.RoomService/DeleteRoom}
  183 + * @param roomName 房间名;写入 JSON body 的 room 字段
  184 + * @return HTTP 响应;调用方负责 close 与状态码判断
  185 + */
  186 + private HttpResponse postTwirp(String path, String roomName) {
  187 + Assert.notBlank(roomName, "roomName 不可为空");
  188 + String token = signAdminToken();
  189 + return HttpRequest.post(HttpUtils.wsUrlToHttp(imProperties.getRtc().getLivekitUrl()) + path)
  190 + .header("Authorization", "Bearer " + token)
  191 + .header("Content-Type", "application/json")
  192 + .body(JSONUtil.toJsonStr(Map.of("room", roomName)))
  193 + .timeout(SERVER_API_TIMEOUT_MS)
  194 + .execute();
  195 + }
  196 +
  197 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitTokenUtils.java deleted
1 -package cn.iocoder.yudao.module.im.framework.rtc.core;  
2 -  
3 -import cn.hutool.core.lang.Assert;  
4 -import cn.hutool.jwt.JWT;  
5 -import cn.hutool.jwt.signers.JWTSignerUtil;  
6 -import lombok.AccessLevel;  
7 -import lombok.NoArgsConstructor;  
8 -  
9 -import java.time.Duration;  
10 -import java.time.Instant;  
11 -import java.util.Date;  
12 -import java.util.HashMap;  
13 -import java.util.Map;  
14 -  
15 -// TODO @AI:是不是搞成 LiveKitClient;然后 config 类去创建起来;通过 @Resource 注入进去?  
16 -/**  
17 - * LiveKit Access Token 签发工具  
18 - * <p>  
19 - * LiveKit 协议把权限放在 video claim 里;这里只暴露通话场景需要的最小权限:roomJoin / canPublish /  
20 - * canSubscribe / canPublishData,避免 PoC 阶段误开 roomAdmin。生产想细分会议主持人时再扩字段。  
21 - *  
22 - * @author 芋道源码  
23 - */  
24 -@NoArgsConstructor(access = AccessLevel.PRIVATE)  
25 -public class LiveKitTokenUtils {  
26 -  
27 - /**  
28 - * 签发客户端进房 Token  
29 - *  
30 - * @param apiKey LiveKit API Key;写入 iss claim  
31 - * @param apiSecret LiveKit API Secret;HS256 签名密钥;长度必须 ≥ 32  
32 - * @param identity 用户唯一标识;写入 sub claim;同 identity 重连会踢前一个连接  
33 - * @param displayName 客户端展示名;可空  
34 - * @param roomName 房间名  
35 - * @param ttl 有效期  
36 - * @return JWT 字符串  
37 - */  
38 - public static String signJoinToken(String apiKey, String apiSecret, String identity,  
39 - String displayName, String roomName, Duration ttl) {  
40 - Assert.notBlank(apiKey, "apiKey 不可为空");  
41 - Assert.notBlank(apiSecret, "apiSecret 不可为空");  
42 - Assert.isTrue(apiSecret.length() >= 32, "apiSecret 长度需 ≥ 32 位");  
43 - Assert.notBlank(identity, "identity 不可为空");  
44 - Assert.notBlank(roomName, "roomName 不可为空");  
45 - Assert.notNull(ttl, "ttl 不可为空");  
46 -  
47 - // TODO @AI:这里注释下;  
48 - Map<String, Object> video = new HashMap<>();  
49 - video.put("roomJoin", true); // TODO @AI:这里帮忙注释下  
50 - video.put("room", roomName); // TODO @AI:这里帮忙注释下  
51 - video.put("canPublish", true); // TODO @AI:这里帮忙注释下  
52 - video.put("canSubscribe", true); // TODO @AI:这里帮忙注释下  
53 - video.put("canPublishData", true); // TODO @AI:这里帮忙注释下  
54 -  
55 - // TODO @AI:这里注释下;  
56 - // TODO @AI:是不是 new Date;然后下面去处理,更合理?或者 System currentTime 这种?  
57 - long now = Instant.now().getEpochSecond();  
58 - JWT jwt = JWT.create()  
59 - .setIssuer(apiKey)  
60 - .setSubject(identity)  
61 - .setNotBefore(new Date(now * 1000))  
62 - .setExpiresAt(new Date((now + ttl.getSeconds()) * 1000))  
63 - .setPayload("video", video)  
64 - .setSigner(JWTSignerUtil.hs256(apiSecret.getBytes()));  
65 - // TODO @AI:empty 处理下;  
66 - if (displayName != null && !displayName.isEmpty()) {  
67 - jwt.setPayload("name", displayName);  
68 - }  
69 - return jwt.sign();  
70 - }  
71 -  
72 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/dto/LiveKitWebhookEventDTO.java renamed to yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/framework/rtc/core/LiveKitWebhookEventDTO.java
1 -package cn.iocoder.yudao.module.im.service.rtc.dto; 1 +package cn.iocoder.yudao.module.im.framework.rtc.core;
2 2
3 import com.fasterxml.jackson.annotation.JsonIgnoreProperties; 3 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
4 import lombok.Data; 4 import lombok.Data;
5 5
6 -// TODO @AI:注释风格;(这个是不是 VO 哈。不放在 VO )  
7 /** 6 /**
8 * LiveKit Webhook 事件载荷;只反序列化我们关心的字段,其余忽略 7 * LiveKit Webhook 事件载荷;只反序列化我们关心的字段,其余忽略
9 * <p> 8 * <p>
10 - * 文档参考:https://docs.livekit.io/home/server/webhook/ 9 + * 文档参考:<a href="https://docs.livekit.io/home/server/webhook/">webhook 文档</a>
11 * 10 *
12 * @author 芋道源码 11 * @author 芋道源码
13 */ 12 */
@@ -15,35 +14,70 @@ import lombok.Data; @@ -15,35 +14,70 @@ import lombok.Data;
15 @JsonIgnoreProperties(ignoreUnknown = true) 14 @JsonIgnoreProperties(ignoreUnknown = true)
16 public class LiveKitWebhookEventDTO { 15 public class LiveKitWebhookEventDTO {
17 16
18 - /** 事件类型;room_started / room_finished / participant_joined / participant_left / track_published 等 */  
19 - private String event; 17 + /**
  18 + * 房间创建:首个 participant 加入时触发
  19 + */
  20 + public static final String EVENT_ROOM_STARTED = "room_started";
  21 + /**
  22 + * 房间销毁:最后一个 participant 离开 / 显式 DeleteRoom 后触发;用于业务侧兜底关房
  23 + */
  24 + public static final String EVENT_ROOM_FINISHED = "room_finished";
  25 + /**
  26 + * 参与者加入房间
  27 + */
  28 + public static final String EVENT_PARTICIPANT_JOINED = "participant_joined";
  29 + /**
  30 + * 参与者离开房间:关 tab / 网络断 / 显式 disconnect 都会触发;用于业务侧兜底清理
  31 + */
  32 + public static final String EVENT_PARTICIPANT_LEFT = "participant_left";
  33 + /**
  34 + * 参与者发布媒体轨道(摄像头 / 麦克风 / 屏幕共享)
  35 + */
  36 + public static final String EVENT_TRACK_PUBLISHED = "track_published";
20 37
21 - /** 事件 id;用于幂等去重 */ 38 + /**
  39 + * 事件 id:用于幂等去重
  40 + */
22 private String id; 41 private String id;
23 42
24 - /** 房间元信息;room_started / room_finished 必填 */ 43 + /**
  44 + * 事件类型;取值参见本类 EVENT_* 常量
  45 + */
  46 + private String event;
  47 +
  48 + /**
  49 + * 房间元信息:room_started / room_finished 必填
  50 + */
25 private RoomInfo room; 51 private RoomInfo room;
26 52
27 - /** 参与者元信息;participant_* 事件必填 */ 53 + /**
  54 + * 参与者元信息:participant_* 事件必填
  55 + */
28 private ParticipantInfo participant; 56 private ParticipantInfo participant;
29 57
30 @Data 58 @Data
31 @JsonIgnoreProperties(ignoreUnknown = true) 59 @JsonIgnoreProperties(ignoreUnknown = true)
32 public static class RoomInfo { 60 public static class RoomInfo {
  61 +
33 private String sid; 62 private String sid;
34 private String name; 63 private String name;
35 private Long creationTime; 64 private Long creationTime;
36 private Integer numParticipants; 65 private Integer numParticipants;
  66 +
37 } 67 }
38 68
39 @Data 69 @Data
40 @JsonIgnoreProperties(ignoreUnknown = true) 70 @JsonIgnoreProperties(ignoreUnknown = true)
41 public static class ParticipantInfo { 71 public static class ParticipantInfo {
  72 +
42 private String sid; 73 private String sid;
43 - /** 用户身份;我们签 token 时写入 userId 字符串,后续支持多端会拼成 userId#terminal */ 74 + /**
  75 + * 用户身份:签 token 时写入 userId 字符串,后续支持多端会拼成 userId#terminal
  76 + */
44 private String identity; 77 private String identity;
45 private String name; 78 private String name;
46 private Long joinedAt; 79 private Long joinedAt;
  80 +
47 } 81 }
48 82
49 } 83 }
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/job/rtc/ImRtcCallCleanupJob.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.job.rtc;
  2 +
  3 +import cn.hutool.core.util.NumberUtil;
  4 +import cn.hutool.core.util.StrUtil;
  5 +import cn.iocoder.yudao.framework.quartz.core.handler.JobHandler;
  6 +import cn.iocoder.yudao.framework.tenant.core.job.TenantJob;
  7 +import cn.iocoder.yudao.module.im.service.rtc.ImRtcCallService;
  8 +import jakarta.annotation.Resource;
  9 +import lombok.extern.slf4j.Slf4j;
  10 +import org.springframework.stereotype.Component;
  11 +
  12 +/**
  13 + * 僵尸通话清理 Job:兜底 LiveKit Webhook 丢失 / 客户端异常关闭等未调 leave 等场景
  14 + *
  15 + * @author 芋道源码
  16 + */
  17 +@Component
  18 +@Slf4j
  19 +public class ImRtcCallCleanupJob implements JobHandler {
  20 +
  21 + @Resource
  22 + private ImRtcCallService rtcCallService;
  23 +
  24 + /**
  25 + * 执行清理
  26 + *
  27 + * @param param 阈值(分钟);为空走 Service 默认(5 分钟)
  28 + * @return 清理数量描述
  29 + */
  30 + @Override
  31 + @TenantJob
  32 + public String execute(String param) {
  33 + Integer thresholdMinutes = StrUtil.isNotBlank(param) && NumberUtil.isInteger(param)
  34 + ? Integer.parseInt(param) : null;
  35 + int cleaned = rtcCallService.cleanupZombieCalls(thresholdMinutes);
  36 + log.info("[execute][清理僵尸通话数量 ({}) 个]", cleaned);
  37 + return String.format("清理僵尸通话 %s 个", cleaned);
  38 + }
  39 +
  40 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/bo/ImCallSessionBO.java deleted
1 -package cn.iocoder.yudao.module.im.service.rtc.bo;  
2 -  
3 -import lombok.Builder;  
4 -import lombok.Data;  
5 -  
6 -import java.util.LinkedHashSet;  
7 -import java.util.Set;  
8 -  
9 -// TODO @AI:后续讨论;抽成 DO;写到 mysql 里;  
10 -/**  
11 - * IM 通话会话 BO;存放在 {@link ImCallSessionStore} 内存里,通话结束即移除  
12 - *  
13 - * @author 芋道源码  
14 - */  
15 -@Data  
16 -@Builder  
17 -public class ImCallSessionBO {  
18 -  
19 - /**  
20 - * 业务通话编号;雪花 ID 生成;用于消息流落库与 Webhook 关联  
21 - */  
22 - private String callId;  
23 - /**  
24 - * LiveKit 房间名;私聊由好友对派生 call_friend_{minId}_{maxId};群聊为 call_group_{groupId}  
25 - */  
26 - private String roomName;  
27 - /**  
28 - * 会话场景;取自 {@link cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum} 的 type  
29 - */  
30 - private Integer scene;  
31 - /**  
32 - * 媒体类型;取自 {@link cn.iocoder.yudao.module.im.enums.rtc.ImCallMediaTypeEnum} 的 type  
33 - */  
34 - private Integer mediaType;  
35 - /**  
36 - * 发起人编号  
37 - */  
38 - private Long inviterId;  
39 - /**  
40 - * 被邀请人集合;私聊为单元素;群聊为发起时所有候选成员(含发起人本人,方便统一遍历)  
41 - */  
42 - private Set<Long> inviteeIds;  
43 - /**  
44 - * 群编号;群聊场景必填  
45 - */  
46 - private Long groupId;  
47 - /**  
48 - * 状态;取自 {@link cn.iocoder.yudao.module.im.enums.rtc.ImCallStatusEnum} 的 type  
49 - */  
50 - private Integer status;  
51 - /**  
52 - * 通话开始时间戳;millis  
53 - */  
54 - private Long startTime;  
55 - /**  
56 - * 已加入房间的成员集合  
57 - */  
58 - @Builder.Default  
59 - private Set<Long> joinedUserIds = new LinkedHashSet<>();  
60 - /**  
61 - * 已拒绝的成员集合(接通前点拒接)  
62 - */  
63 - @Builder.Default  
64 - private Set<Long> rejectedUserIds = new LinkedHashSet<>();  
65 - /**  
66 - * 已离开的成员集合(接通后挂断)  
67 - */  
68 - @Builder.Default  
69 - private Set<Long> leftUserIds = new LinkedHashSet<>();  
70 -  
71 - // TODO @AI:bo 里,不要有这种方法;  
72 - /**  
73 - * 当前会话中所有相关用户;用于 RTC_END 信号扇出与忙线清理  
74 - */  
75 - public Set<Long> getAllUserIds() {  
76 - Set<Long> result = new LinkedHashSet<>();  
77 - result.add(inviterId);  
78 - if (inviteeIds != null) {  
79 - result.addAll(inviteeIds);  
80 - }  
81 - result.addAll(joinedUserIds);  
82 - return result;  
83 - }  
84 -  
85 - // TODO @AI:这个方法,是不是没在使用???  
86 - /**  
87 - * 判断用户是否参与该通话;不在被邀请也不在已加入即非参与方  
88 - */  
89 - public boolean containsUser(Long userId) {  
90 - if (userId == null) {  
91 - return false;  
92 - }  
93 - if (userId.equals(inviterId)) {  
94 - return true;  
95 - }  
96 - if (inviteeIds != null && inviteeIds.contains(userId)) {  
97 - return true;  
98 - }  
99 - return joinedUserIds.contains(userId);  
100 - }  
101 -  
102 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/bo/ImCallSessionStore.java deleted
1 -package cn.iocoder.yudao.module.im.service.rtc.bo;  
2 -  
3 -import org.springframework.stereotype.Component;  
4 -  
5 -import java.util.ArrayList;  
6 -import java.util.List;  
7 -import java.util.Map;  
8 -import java.util.concurrent.ConcurrentHashMap;  
9 -import java.util.function.Function;  
10 -  
11 -/**  
12 - * IM 通话会话内存仓储;MVP 阶段不落库,全量存内存  
13 - * <p>  
14 - * 双层索引:  
15 - * - sessions: roomName -> 通话会话;invite 时 computeIfAbsent 保证同一好友对 / 群只有一个进行中通话;  
16 - * - userActiveRoom: userId -> 当前所在 roomName;用于"对方正忙"提示与冷启动恢复。  
17 - * 通话结束时 {@link #remove} 一次性清两个索引;同一用户的并发挂断走 ConcurrentHashMap 原子操作收敛。  
18 - *  
19 - * @author 芋道源码  
20 - */  
21 -@Component  
22 -public class ImCallSessionStore {  
23 -  
24 - private final Map<String, ImCallSessionBO> sessions = new ConcurrentHashMap<>();  
25 -  
26 - private final Map<Long, String> userActiveRoom = new ConcurrentHashMap<>();  
27 -  
28 - public ImCallSessionBO get(String roomName) {  
29 - if (roomName == null) {  
30 - return null;  
31 - }  
32 - return sessions.get(roomName);  
33 - }  
34 -  
35 - /**  
36 - * 原子获取或创建会话;在并发 invite 下保证同一 roomName 只 build 一次  
37 - *  
38 - * @param roomName 房间名(已由 service 派生)  
39 - * @param builder 仅当不存在时调用;返回已组装好的 ImCallSessionBO  
40 - * @return 会话实例(新建或已有)  
41 - */  
42 - public ImCallSessionBO getOrCreate(String roomName, Function<String, ImCallSessionBO> builder) {  
43 - return sessions.computeIfAbsent(roomName, builder);  
44 - }  
45 -  
46 - /**  
47 - * 移除会话并清理参与者的 active 索引;调用方应在会话状态切到 ENDED 后调用  
48 - */  
49 - public void remove(String roomName) {  
50 - if (roomName == null) {  
51 - return;  
52 - }  
53 - ImCallSessionBO session = sessions.remove(roomName);  
54 - if (session == null) {  
55 - return;  
56 - }  
57 - for (Long userId : session.getAllUserIds()) {  
58 - // 仅当该用户的 active room 仍是当前 room 时才解绑;避免误清用户后续新建的通话  
59 - userActiveRoom.remove(userId, roomName);  
60 - }  
61 - }  
62 -  
63 - /**  
64 - * 绑定用户到房间;invite / accept 时调用  
65 - */  
66 - public void bindUser(Long userId, String roomName) {  
67 - if (userId == null || roomName == null) {  
68 - return;  
69 - }  
70 - userActiveRoom.put(userId, roomName);  
71 - }  
72 -  
73 - /**  
74 - * 仅在用户当前 active room 等于给定 roomName 时解绑;用于 reject / leave 单方退出  
75 - */  
76 - public void unbindUserFromRoom(Long userId, String roomName) {  
77 - if (userId == null || roomName == null) {  
78 - return;  
79 - }  
80 - userActiveRoom.remove(userId, roomName);  
81 - }  
82 -  
83 - /**  
84 - * 用户是否处于通话中;用于 invite 时判定"对方正忙"  
85 - */  
86 - public boolean isBusy(Long userId) {  
87 - return userId != null && userActiveRoom.containsKey(userId);  
88 - }  
89 -  
90 - /**  
91 - * 用户当前所在的 roomName;冷启动恢复用  
92 - */  
93 - public String getActiveRoom(Long userId) {  
94 - return userId == null ? null : userActiveRoom.get(userId);  
95 - }  
96 -  
97 - /**  
98 - * 列出当前所有进行中会话;用于运维 / 单元测试观察;线上调用频次需谨慎  
99 - */  
100 - public List<ImCallSessionBO> listAll() {  
101 - return new ArrayList<>(sessions.values());  
102 - }  
103 -  
104 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/dto/ImRtcAcceptNotification.java deleted
1 -package cn.iocoder.yudao.module.im.service.rtc.dto;  
2 -  
3 -import lombok.Data;  
4 -  
5 -// TODO @AI:一些 notification 应该挪到 message 那边,你在  
6 -// TODO @AI:注释风格;  
7 -/**  
8 - * RTC 接通通知;推给主叫与已在房成员,提示某用户已加入;前端把"等待对方接受"切换成通话中 UI  
9 - *  
10 - * @author 芋道源码  
11 - */  
12 -@Data  
13 -public class ImRtcAcceptNotification {  
14 -  
15 - /** 业务通话编号 */  
16 - // TODO @AI:是不是后续使用 id Long?(后续在改)  
17 - private String callId;  
18 - /** LiveKit 房间名 */  
19 - private String roomName;  
20 - /** 接听者编号 */  
21 - private Long acceptorId;  
22 -  
23 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/dto/ImRtcEndNotification.java deleted
1 -package cn.iocoder.yudao.module.im.service.rtc.dto;  
2 -  
3 -import lombok.Data;  
4 -  
5 -// TODO @AI:注释风格;  
6 -/**  
7 - * RTC 结束通知;拒绝 / 取消 / 挂断 / 超时 / 异常 统一走这一条;前端关闭通话界面  
8 - *  
9 - * @author 芋道源码  
10 - */  
11 -@Data  
12 -public class ImRtcEndNotification {  
13 -  
14 - /** 业务通话编号 */  
15 - private String callId;  
16 - /** LiveKit 房间名 */  
17 - private String roomName;  
18 - /** 操作者编号;TIMEOUT 等系统触发场景为 null */  
19 - private Long operatorId;  
20 - /** 结束原因;取自 {@link cn.iocoder.yudao.module.im.enums.rtc.ImCallEndReasonEnum} 的 type */  
21 - private Integer reason;  
22 - /** 通话时长(秒);接通过为 endTime - acceptTime;未接通为 null */  
23 - private Long durationSeconds;  
24 -  
25 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/dto/ImRtcGroupNotification.java deleted
1 -package cn.iocoder.yudao.module.im.service.rtc.dto;  
2 -  
3 -import lombok.Data;  
4 -  
5 -import java.util.Set;  
6 -  
7 -// TODO @AI:注释风格;  
8 -/**  
9 - * RTC 群通话广播载荷;让群里所有成员(不仅是被邀请人)能感知通话存在,并支持主动加入  
10 - *  
11 - * @author 芋道源码  
12 - */  
13 -@Data  
14 -public class ImRtcGroupNotification {  
15 -  
16 - /** 业务通话编号 */  
17 - private String callId;  
18 - /** LiveKit 房间名 */  
19 - private String roomName;  
20 - /** 群编号 */  
21 - private Long groupId;  
22 - /** 媒体类型;1 语音;2 视频 */  
23 - private Integer mediaType;  
24 - /** 发起人编号 */  
25 - private Long inviterId;  
26 - /** 当前已加入房间的用户编号集合 */  
27 - private Set<Long> joinedUserIds;  
28 - /** 当前被邀请池;展开胶囊条时用来展示「待加入」灰色头像 */  
29 - private Set<Long> inviteeIds;  
30 -  
31 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/rtc/dto/ImRtcInviteNotification.java deleted
1 -package cn.iocoder.yudao.module.im.service.rtc.dto;  
2 -  
3 -import lombok.Data;  
4 -  
5 -// TODO @AI:注释风格;  
6 -/**  
7 - * RTC 来电邀请通知;推给被叫弹「来电」界面  
8 - *  
9 - * @author 芋道源码  
10 - */  
11 -@Data  
12 -public class ImRtcInviteNotification {  
13 -  
14 - /** 业务通话编号 */  
15 - private String callId;  
16 - /** LiveKit 房间名 */  
17 - private String roomName;  
18 - /** LiveKit Server WebSocket 地址 */  
19 - private String livekitUrl;  
20 - /** 该被叫专属的 LiveKit 接听 Token;接通后直接 connect 用 */  
21 - private String token;  
22 - /** 会话场景;1 私聊;2 群聊 */  
23 - private Integer scene;  
24 - /** 媒体类型;1 语音;2 视频 */  
25 - private Integer mediaType;  
26 - /** 发起人编号 */  
27 - private Long inviterId;  
28 - /** 发起人昵称;前端来电界面展示 */  
29 - private String inviterNickname;  
30 - /** 发起人头像 */  
31 - private String inviterAvatar;  
32 - /** 群编号;群通话场景必填 */  
33 - private Long groupId;  
34 -  
35 -}  
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/ImWebSocketService.java
@@ -21,7 +21,19 @@ public interface ImWebSocketService { @@ -21,7 +21,19 @@ public interface ImWebSocketService {
21 * @param userId 目标用户编号 21 * @param userId 目标用户编号
22 * @param dto 私聊消息 DTO 22 * @param dto 私聊消息 DTO
23 */ 23 */
24 - void sendPrivateMessageAsync(Long userId, ImPrivateMessageDTO dto); 24 + default void sendPrivateMessageAsync(Long userId, ImPrivateMessageDTO dto) {
  25 + sendPrivateMessageAsync(Collections.singleton(userId), dto);
  26 + }
  27 +
  28 + /**
  29 + * 异步批量推送私聊消息给多个用户;用于一条 payload 扇出到多个收件人
  30 + * <p>
  31 + * 相比逐个 sendPrivateMessageAsync,仅注册一个 afterCommit 回调 + 一个 @Async 任务,大群参与者事件(1602 / 1603)下避免 N 次线程池调度
  32 + *
  33 + * @param userIds 目标用户编号列表
  34 + * @param dto 私聊消息 DTO
  35 + */
  36 + void sendPrivateMessageAsync(Collection<Long> userIds, ImPrivateMessageDTO dto);
25 37
26 /** 38 /**
27 * 异步推送群聊消息给指定用户 39 * 异步推送群聊消息给指定用户
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/ImWebSocketServiceImpl.java
@@ -32,10 +32,10 @@ public class ImWebSocketServiceImpl implements ImWebSocketService { @@ -32,10 +32,10 @@ public class ImWebSocketServiceImpl implements ImWebSocketService {
32 private WebSocketMessageSender webSocketMessageSender; 32 private WebSocketMessageSender webSocketMessageSender;
33 33
34 @Override 34 @Override
35 - public void sendPrivateMessageAsync(Long userId, ImPrivateMessageDTO dto) { 35 + public void sendPrivateMessageAsync(Collection<Long> userIds, ImPrivateMessageDTO dto) {
36 // 说明:通过 executeAfterCommitOrNow 保证事务提交后再推送,避免客户端收到消息后查询数据库时事务尚未提交 36 // 说明:通过 executeAfterCommitOrNow 保证事务提交后再推送,避免客户端收到消息后查询数据库时事务尚未提交
37 // 通过 getSelf() 获取 Spring 代理对象调用 @Async 方法,确保异步 AOP 生效(直接 this 调用会绕过代理) 37 // 通过 getSelf() 获取 Spring 代理对象调用 @Async 方法,确保异步 AOP 生效(直接 this 调用会绕过代理)
38 - executeAfterTransaction(() -> getSelf().doSendPrivateMessage(userId, dto)); 38 + executeAfterTransaction(() -> getSelf().doSendPrivateMessage(userIds, dto));
39 } 39 }
40 40
41 @Override 41 @Override
@@ -44,15 +44,17 @@ public class ImWebSocketServiceImpl implements ImWebSocketService { @@ -44,15 +44,17 @@ public class ImWebSocketServiceImpl implements ImWebSocketService {
44 } 44 }
45 45
46 /** 46 /**
47 - * 异步发送私聊 WebSocket 消息 47 + * 异步发送私聊 WebSocket 消息;多收件人共享同一 dto,避免按收件人重复注册 afterCommit 回调
48 */ 48 */
49 @Async 49 @Async
50 - public void doSendPrivateMessage(Long userId, ImPrivateMessageDTO dto) {  
51 - try {  
52 - webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,  
53 - ImPrivateMessageDTO.TYPE, dto);  
54 - } catch (Exception e) {  
55 - log.error("[doSendPrivateMessage][userId({}) dto({}) 发送失败]", userId, dto, e); 50 + public void doSendPrivateMessage(Collection<Long> userIds, ImPrivateMessageDTO dto) {
  51 + for (Long userId : userIds) {
  52 + try {
  53 + webSocketMessageSender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
  54 + ImPrivateMessageDTO.TYPE, dto);
  55 + } catch (Exception e) {
  56 + log.error("[doSendPrivateMessage][userId({}) dto({}) 发送失败]", userId, dto, e);
  57 + }
56 } 58 }
57 } 59 }
58 60
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/ImPrivateMessageDTO.java
@@ -123,7 +123,7 @@ public class ImPrivateMessageDTO { @@ -123,7 +123,7 @@ public class ImPrivateMessageDTO {
123 123
124 // ==================== 群定向私聊通知 ==================== 124 // ==================== 群定向私聊通知 ====================
125 125
126 - // TODO @AI:能不能走群聊的通道????只是接收人不同。 126 + // TODO @AI:能不能走群聊的通道????只是接收人不同。【统一优化的】
127 /** 127 /**
128 * 构建群通知推送 DTO(走私聊通道定向推送,不入群消息流) 128 * 构建群通知推送 DTO(走私聊通道定向推送,不入群消息流)
129 * <p> 129 * <p>
@@ -143,4 +143,25 @@ public class ImPrivateMessageDTO { @@ -143,4 +143,25 @@ public class ImPrivateMessageDTO {
143 .setContent(JsonUtils.toJsonString(payload)).setSendTime(LocalDateTime.now()); 143 .setContent(JsonUtils.toJsonString(payload)).setSendTime(LocalDateTime.now());
144 } 144 }
145 145
  146 + // ==================== 实时通话信令 ====================
  147 +
  148 + // TODO @AI:能不能走群聊的通道????只是接收人不同。【统一优化的】
  149 + /**
  150 + * 构建通话信令推送 DTO(走私聊通道仅推参与方,不入消息流)
  151 + * <p>
  152 + * 用于 RTC_CALL(INVITE / REJECT 等)/ RTC_PARTICIPANT_CONNECTED / RTC_PARTICIPANT_DISCONNECTED
  153 + *
  154 + * @param type 消息类型;取自 {@link ImMessageTypeEnum} 中的 RTC_* 段
  155 + * @param senderId 发送人编号;INVITE 时为发起人,REJECT 时为拒绝者,参与者事件时为加入 / 离开者
  156 + * @param receiverUserId 推送目标用户编号
  157 + * @param payload 通话事件 payload(任意 RTC 通知 DTO)
  158 + * @return 私聊 DTO
  159 + */
  160 + public static ImPrivateMessageDTO ofRtcNotification(Integer type, Long senderId,
  161 + Long receiverUserId, Object payload) {
  162 + return new ImPrivateMessageDTO().setType(type)
  163 + .setSenderId(senderId).setReceiverId(receiverUserId)
  164 + .setContent(JsonUtils.toJsonString(payload)).setSendTime(LocalDateTime.now());
  165 + }
  166 +
146 } 167 }
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/group/GroupMemberEnterNotification.java
@@ -7,7 +7,7 @@ import lombok.EqualsAndHashCode; @@ -7,7 +7,7 @@ import lombok.EqualsAndHashCode;
7 * 自由进群事件通知 7 * 自由进群事件通知
8 * <p> 8 * <p>
9 * 用户经搜索 / 二维码 / 分享链接自由进群(FREE 模式或审批通过后),全员广播; 9 * 用户经搜索 / 二维码 / 分享链接自由进群(FREE 模式或审批通过后),全员广播;
10 - * 进群者前端按 entrantUserId === self 自判 fetchGroupInfo + fetchGroupMembers bootstrap,其余成员局部插入新成员 10 + * 进群者前端按 entrantUserId === self 自判,初次拉取 fetchGroupInfo + fetchGroupMembers;其余成员局部插入新成员
11 */ 11 */
12 @EqualsAndHashCode(callSuper = true) 12 @EqualsAndHashCode(callSuper = true)
13 @Data 13 @Data
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcCallEndNotification.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.service.websocket.dto.notification.rtc;
  2 +
  3 +import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
  4 +import cn.iocoder.yudao.module.im.enums.rtc.ImRtcCallEndReasonEnum;
  5 +import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
  6 +import lombok.Data;
  7 +
  8 +/**
  9 + * RTC_CALL_END 通话结束通知
  10 + * <p>
  11 + * 入 im_private_message / im_group_message;接收方关闭通话窗 + 渲染聊天 tip
  12 + * <p>
  13 + * 文案分场景:
  14 + * 群聊:「{voice/video}通话已结束 [时长 X]」;发起人信息走配对的 {@link ImRtcCallStartNotification}
  15 + * 私聊:仿微信准气泡,按 endReason × selfSend 视角转换文案(HANGUP / CANCEL / REJECT / BUSY / ERROR)
  16 + * <p>
  17 + * 与 {@link ImRtcCallStartNotification} 两段式配对(仅群通话):START 在 invite 事务里 INSERT,END 在 cancel / leave 事务里 INSERT;
  18 + * 两段位于不同请求 / 事务,自增 id 保证聊天流顺序
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@Data
  23 +public class ImRtcCallEndNotification {
  24 +
  25 + /**
  26 + * 业务通话编号
  27 + */
  28 + private String callId;
  29 + /**
  30 + * 会话类型
  31 + */
  32 + private Integer conversationType;
  33 + /**
  34 + * 媒体类型
  35 + */
  36 + private Integer mediaType;
  37 + /**
  38 + * 结束原因
  39 + */
  40 + private Integer endReason;
  41 + /**
  42 + * 通话时长(秒):接通过为 endTime - acceptTime;未接通为 null
  43 + */
  44 + private Long durationSeconds;
  45 + /**
  46 + * 操作者用户编号:HANGUP / CANCEL / REJECT 是触发结束的人;HANGUP webhook 兜底为 null
  47 + * <p>
  48 + * 用于前端「被某某挂断」类文案;普通文案不依赖此字段
  49 + */
  50 + private Long operatorUserId;
  51 + /**
  52 + * 操作者昵称:前端按需展示(被某某挂断 / 头像 tip);操作者为空时随之为空
  53 + */
  54 + private String operatorNickname;
  55 + /**
  56 + * 操作者头像:前端按需展示;操作者为空时随之为空
  57 + */
  58 + private String operatorAvatar;
  59 +
  60 + public static ImRtcCallEndNotification of(ImRtcCallDO call, ImRtcCallEndReasonEnum reason, Long durationSeconds,
  61 + Long operatorId, AdminUserRespDTO operator) {
  62 + ImRtcCallEndNotification notification = new ImRtcCallEndNotification();
  63 + notification.callId = call.getCallId();
  64 + notification.conversationType = call.getConversationType();
  65 + notification.mediaType = call.getMediaType();
  66 + notification.endReason = reason.getReason();
  67 + notification.durationSeconds = durationSeconds;
  68 + notification.operatorUserId = operatorId;
  69 + if (operator != null) {
  70 + notification.operatorNickname = operator.getNickname();
  71 + notification.operatorAvatar = operator.getAvatar();
  72 + }
  73 + return notification;
  74 + }
  75 +
  76 +}
yudao-module-im/src/main/java/cn/iocoder/yudao/module/im/service/websocket/dto/notification/rtc/ImRtcCallStartNotification.java 0 → 100644
  1 +package cn.iocoder.yudao.module.im.service.websocket.dto.notification.rtc;
  2 +
  3 +import cn.iocoder.yudao.module.im.dal.dataobject.rtc.ImRtcCallDO;
  4 +import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
  5 +import lombok.Data;
  6 +
  7 +/**
  8 + * RTC_CALL_START 通话开始通知
  9 + * <p>
  10 + * 仅群通话场景;入 im_group_message + 全群广播;前端渲染聊天 tip「{inviterNickname} 发起了{voice/video}通话」
  11 + * <p>
  12 + * 与 {@link ImRtcCallEndNotification} 两段式配对:
  13 + * START 在 invite 接口事务里 INSERT,END 在 cancel / leave 接口事务里 INSERT,
  14 + * 两段位于不同请求 / 事务,自增 id 保证聊天流顺序;后续如果合并到同一事务里 push,需要额外保证 START 先于 END
  15 + * <p>
  16 + * 私聊场景没有 START(单 END 准气泡足够),本通知不会用于 conversationType=PRIVATE
  17 + *
  18 + * @author 芋道源码
  19 + */
  20 +@Data
  21 +public class ImRtcCallStartNotification {
  22 +
  23 + /**
  24 + * 业务通话编号
  25 + */
  26 + private String callId;
  27 + /**
  28 + * 会话类型:当前固定 GROUP(私聊无 START)
  29 + */
  30 + private Integer conversationType;
  31 + /**
  32 + * 媒体类型
  33 + */
  34 + private Integer mediaType;
  35 + /**
  36 + * 发起人用户编号
  37 + */
  38 + private Long inviterUserId;
  39 + /**
  40 + * 发起人昵称:用于聊天 tip 文案,可空走前端 fallback
  41 + */
  42 + private String inviterNickname;
  43 + /**
  44 + * 发起人头像:可空,预留给点击 tip 展示发起人卡片
  45 + */
  46 + private String inviterAvatar;
  47 +
  48 + public static ImRtcCallStartNotification of(ImRtcCallDO call, AdminUserRespDTO inviter) {
  49 + ImRtcCallStartNotification notification = new ImRtcCallStartNotification();
  50 + notification.callId = call.getCallId();
  51 + notification.conversationType = call.getConversationType();
  52 + notification.mediaType = call.getMediaType();
  53 + notification.inviterUserId = call.getInviterUserId();
  54 + if (inviter != null) {
  55 + notification.inviterNickname = inviter.getNickname();
  56 + notification.inviterAvatar = inviter.getAvatar();
  57 + }
  58 + return notification;
  59 + }
  60 +
  61 +}