main
use std::collections::HashMap;
use livekit_api::access_token::{AccessToken, VideoGrants};
use livekit_api::services::LiveKitApi;
use livekit_api::services::room::{CreateRoomOptions, UpdateParticipantOptions};
use livekit_protocol as proto;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::lookup::{lookup_space_by_name, require_room_readable};
use crate::share::realtime::RoomEvent;
/// 举手状态。审批通过后 approved，被收回后 revoked（可再次举手）。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum HandState {
Pending,
Approved,
Denied,
Revoked,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct HandEntry {
pub state: HandState,
pub requested_at: i64,
pub decided_by: Option<i64>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct HandDto {
pub user_id: i64,
pub display_name: String,
pub avatar_url: Option<String>,
pub state: HandState,
pub requested_at: i64,
}
/// 一个参与者窗口在画布里的摆位，坐标按画布宽高归一化（0~1），
/// 这样不同分辨率、不同窗口大小的成员看到的是同一套布局。
#[derive(Debug, Clone, Copy, Serialize, Deserialize, ToSchema)]
pub struct VoiceTileDto {
pub x: f64,
pub y: f64,
/// 归一化宽度。
pub w: f64,
/// 归一化高度，和 `w` 一样按画布的可用宽高分别归一，因此窗口可以自由
/// 拉伸而不是只能等比缩放。旧记录没有这个字段，读出来是 `None`，客户端
/// 按宽度的 3/4 补一个经典比例。
#[serde(default)]
pub h: Option<f64>,
/// 服务端写入时刻（秒）。请求里被忽略，响应里供客户端判断“谁更新”，
/// 这样并发拖拽时一个迟到的 GET 不会把刚摆好的窗口拽回旧位置。
#[serde(default)]
pub updated_at: i64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema)]
pub struct VoiceLayoutDto {
/// key 为参与者用户 id。
#[serde(default)]
pub tiles: HashMap<String, VoiceTileDto>,
}
/// 布局写入是**按 key 合并**的补丁而不是整表替换：每个客户端只提交自己刚拖的
/// 那一格，收回的格子里列在 `removed`，两个人同时摆不同窗口互不吞没。
#[derive(Debug, Clone, Default, Deserialize, ToSchema)]
pub struct VoiceLayoutPatchDto {
/// 需要新增或覆盖的窗口（key = 参与者用户 id）。
#[serde(default)]
pub tiles: HashMap<String, VoiceTileDto>,
/// 需要收回自动网格的窗口 key。
#[serde(default)]
pub removed: Vec<String>,
}
/// 本次写入的服务端时刻，客户端用它给本地副本盖戳。
#[derive(Debug, Clone, Copy, Serialize, ToSchema)]
pub struct VoiceLayoutSavedDto {
pub updated_at: i64,
}
/// 房间成员共享的窗口数量上限，避免布局被当成存储滥用。
const MAX_LAYOUT_TILES: usize = 64;
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceConfigDto {
pub enabled: bool,
pub url: String,
pub recording_enabled: bool,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceTokenDto {
pub token: String,
pub url: String,
pub room_name: String,
pub expires_in_secs: u64,
pub can_publish: bool,
pub is_stage: bool,
pub hand_state: Option<HandState>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct VoiceTrackDto {
pub sid: String,
pub kind: String,
pub source: String,
pub muted: bool,
}
/// 房间参与者快照。
///
/// 权威来源是 LiveKit 的 list_participants，Redis 只补主持人静音等本地状态；
/// 这是「谁在这个频道里」的服务端视图，用于侧栏人数与未加入时的预览。
/// 会中界面的发言高亮、静音图标与画面轨道一律直接读浏览器里的 Room，
/// 因为本接口不是事件流，返回值天然是快照。
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceParticipantDto {
pub user_id: i64,
pub display_name: String,
pub avatar_url: Option<String>,
pub joined_at: i64,
pub can_publish: bool,
pub moderator_muted: bool,
/// 快照值：list_participants 不返回发言状态，因此这里恒为 false；
/// 实时发言指示由前端从 Room 的 activeSpeakers 读取。
pub speaking: bool,
pub tracks: Vec<VoiceTrackDto>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceRoomStateDto {
pub room_id: Uuid,
pub count: u32,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceSpaceStatesDto {
pub rooms: Vec<VoiceRoomStateDto>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceHandsDto {
pub hands: Vec<HandDto>,
pub can_moderate: bool,
pub my_state: Option<HandState>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceModerationDto {
pub ok: bool,
pub action: String,
pub user_id: i64,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct VoiceRecordingDto {
pub egress_id: String,
pub active: bool,
}
/// Redis 中一位参与者的在场记录。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoiceMember {
pub user_id: i64,
pub display_name: String,
pub avatar_url: Option<String>,
pub joined_at: i64,
pub can_publish: bool,
#[serde(default)]
pub moderator_muted: bool,
#[serde(default)]
pub tracks: Vec<VoiceTrackDto>,
pub livekit_sid: Option<String>,
}
pub fn voice_key(room_id: Uuid) -> String {
format!("chat:voice:{room_id}")
}
pub fn hands_key(room_id: Uuid) -> String {
format!("chat:voice:{room_id}:hands")
}
/// 布局合并脚本：读写必须在 Redis 里原子完成，否则两个人同时摆不同窗口时，
/// 后写的那次会带着自己读到的旧快照覆盖掉前一次的结果。
const MERGE_LAYOUT_SCRIPT: &str = r"
local key = KEYS[1]
local stamp = tonumber(ARGV[1])
local max_tiles = tonumber(ARGV[3])
local ttl = tonumber(ARGV[4])
local patch = cjson.decode(ARGV[2])
local current = redis.call('GET', key)
local layout = { tiles = {} }
if current then
local ok, decoded = pcall(cjson.decode, current)
if ok and type(decoded) == 'table' and type(decoded.tiles) == 'table' then
layout = decoded
end
end
local removed = patch.removed
if type(removed) == 'table' then
for _, id in ipairs(removed) do layout.tiles[id] = nil end
end
local added = patch.tiles
if type(added) == 'table' then
for id, tile in pairs(added) do
layout.tiles[id] = {
x = tonumber(tile.x),
y = tonumber(tile.y),
w = tonumber(tile.w),
-- 旧客户端（或旧记录回写）不带 h：tonumber(nil) 是 nil，这个键就不会写进
-- JSON，读出来仍是“未指定”，而不是塌成 0。
h = tonumber(tile.h),
updated_at = stamp,
}
end
end
local count = 0
for _ in pairs(layout.tiles) do count = count + 1 end
if count > max_tiles then return { err = 'too many tiles' } end
redis.call('SET', key, cjson.encode(layout), 'EX', ttl)
return stamp
";
pub fn layout_key(room_id: Uuid) -> String {
format!("chat:voice:{room_id}:layout")
}
fn now_secs() -> i64 {
OffsetDateTime::now_utc().unix_timestamp()
}
fn livekit(service: &ChatService) -> Result<LiveKitApi, ChatError> {
let cfg = &service.settings.livekit;
if !cfg.enabled {
return Err(ChatError::BadRequest("livekit is not enabled".to_string()));
}
Ok(LiveKitApi::with_api_key(
cfg.api_url.trim(),
cfg.api_key.trim(),
cfg.api_secret.trim(),
))
}
/// 语音能力的全局限定：房间必须存在、必须是 Voice/Stage、调用者可读。
pub async fn require_voice_room(
service: &ChatService,
room_id: Uuid,
uid: i64,
) -> Result<models::room::rooms::Model, ChatError> {
let ctx = require_room_readable(service, room_id, uid).await?;
if !matches!(
ctx.room.kind,
models::room::rooms::RoomKind::Voice | models::room::rooms::RoomKind::Stage
) {
return Err(ChatError::BadRequest(
"room is not a voice channel".to_string(),
));
}
Ok(ctx.room)
}
/// 调用者是否能主持（房间角色 >= Moderator，或空间 manage_channels）。
pub async fn can_moderate_voice(
service: &ChatService,
room_id: Uuid,
uid: i64,
) -> Result<bool, ChatError> {
let ctx = crate::share::lookup::lookup_room_context(service, room_id).await?;
match &ctx.space {
Some(space) => crate::share::authz::can_manage_space(service, space, uid).await,
None => Ok(false),
}
}
/// Stage 成员是否拥有发言权：房间角色 >= Moderator，或已有 approved 举手。
pub async fn has_publish_grant(
service: &ChatService,
room_id: Uuid,
uid: i64,
) -> Result<bool, ChatError> {
if can_moderate_voice(service, room_id, uid).await? {
return Ok(true);
}
Ok(matches!(
read_hand(service, room_id, uid).await?.map(|h| h.state),
Some(HandState::Approved)
))
}
/// 参与者 id 的解析：identity 是 uid 字符串。
pub fn identity_to_uid(identity: &str) -> Option<i64> {
identity.parse::<i64>().ok()
}
pub async fn display_names(
service: &ChatService,
uids: &[i64],
) -> HashMap<i64, (String, Option<String>)> {
let mut map = HashMap::new();
let users = match service.audience.load_user_briefs(uids).await {
Ok(users) => users,
Err(_) => return map,
};
for (uid, user) in users {
let avatar_url = api::resolve_user_avatar_url(uid, &user.avatar_url);
let avatar = if avatar_url.trim().is_empty() {
None
} else {
Some(avatar_url)
};
let name = if user.display_name.trim().is_empty() {
user.username
} else {
user.display_name
};
map.insert(uid, (name, avatar));
}
map
}
/// 把 Redis 里的一位成员写回（webhook 与对账共用）。
pub async fn write_member(
service: &ChatService,
room_id: Uuid,
member: &VoiceMember,
) -> Result<(), ChatError> {
let ttl = service.settings.livekit.presence_ttl_secs.max(60);
let value = serde_json::to_string(member)?;
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
let key = voice_key(room_id);
let field = member.user_id.to_string();
let result: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => {
let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
.arg(&key)
.arg(&field)
.arg(&value)
.query_async(c)
.await;
if written.is_ok() {
let _: Result<i64, redis::RedisError> =
redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
}
written
}
RedisConnection::Cluster(c) => {
let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
.arg(&key)
.arg(&field)
.arg(&value)
.query_async(c)
.await;
if written.is_ok() {
let _: Result<i64, redis::RedisError> =
redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
}
written
}
};
result.map(|_| ()).map_err(ChatError::from)
}
pub async fn remove_member(
service: &ChatService,
room_id: Uuid,
uid: i64,
) -> Result<(), ChatError> {
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
let key = voice_key(room_id);
let field = uid.to_string();
let _: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => {
redis::cmd("HDEL")
.arg(&key)
.arg(&field)
.query_async(c)
.await
}
RedisConnection::Cluster(c) => {
redis::cmd("HDEL")
.arg(&key)
.arg(&field)
.query_async(c)
.await
}
};
let hkey = hands_key(room_id);
let _: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => {
redis::cmd("HDEL")
.arg(&hkey)
.arg(&field)
.query_async(c)
.await
}
RedisConnection::Cluster(c) => {
redis::cmd("HDEL")
.arg(&hkey)
.arg(&field)
.query_async(c)
.await
}
};
Ok(())
}
pub async fn clear_room(service: &ChatService, room_id: Uuid) -> Result<(), ChatError> {
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
for key in [voice_key(room_id), hands_key(room_id)] {
let _: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => redis::cmd("DEL").arg(&key).query_async(c).await,
RedisConnection::Cluster(c) => redis::cmd("DEL").arg(&key).query_async(c).await,
};
}
Ok(())
}
/// 读取房间当前在 Redis 中的成员快照。
pub async fn read_members(
service: &ChatService,
room_id: Uuid,
) -> Result<Vec<VoiceMember>, ChatError> {
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
let key = voice_key(room_id);
let raw: Result<HashMap<String, String>, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
RedisConnection::Cluster(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
};
let raw = raw.unwrap_or_default();
let mut members: Vec<VoiceMember> = raw
.into_values()
.filter_map(|value| serde_json::from_str::<VoiceMember>(&value).ok())
.collect();
members.sort_by_key(|m| m.joined_at);
Ok(members)
}
pub async fn read_hand(
service: &ChatService,
room_id: Uuid,
uid: i64,
) -> Result<Option<HandEntry>, ChatError> {
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
let key = hands_key(room_id);
let field = uid.to_string();
let raw: Result<Option<String>, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => {
redis::cmd("HGET")
.arg(&key)
.arg(&field)
.query_async(c)
.await
}
RedisConnection::Cluster(c) => {
redis::cmd("HGET")
.arg(&key)
.arg(&field)
.query_async(c)
.await
}
};
Ok(raw
.ok()
.flatten()
.and_then(|v| serde_json::from_str(&v).ok()))
}
pub async fn read_hands(
service: &ChatService,
room_id: Uuid,
) -> Result<HashMap<i64, HandEntry>, ChatError> {
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
let key = hands_key(room_id);
let raw: Result<HashMap<String, String>, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
RedisConnection::Cluster(c) => redis::cmd("HGETALL").arg(&key).query_async(c).await,
};
let mut map = HashMap::new();
for (field, value) in raw.unwrap_or_default() {
if let (Ok(uid), Ok(entry)) = (
field.parse::<i64>(),
serde_json::from_str::<HandEntry>(&value),
) {
map.insert(uid, entry);
}
}
Ok(map)
}
pub async fn write_hand(
service: &ChatService,
room_id: Uuid,
uid: i64,
entry: &HandEntry,
) -> Result<(), ChatError> {
let ttl = service.settings.livekit.hand_ttl_secs.max(60);
let value = serde_json::to_string(entry)?;
let mut conn = service.participants.redis.connection_clone();
use cache::RedisConnection;
let key = hands_key(room_id);
let field = uid.to_string();
let result: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => {
let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
.arg(&key)
.arg(&field)
.arg(&value)
.query_async(c)
.await;
if written.is_ok() {
let _: Result<i64, redis::RedisError> =
redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
}
written
}
RedisConnection::Cluster(c) => {
let written: Result<i64, redis::RedisError> = redis::cmd("HSET")
.arg(&key)
.arg(&field)
.arg(&value)
.query_async(c)
.await;
if written.is_ok() {
let _: Result<i64, redis::RedisError> =
redis::cmd("EXPIRE").arg(&key).arg(ttl).query_async(c).await;
}
written
}
};
result.map(|_| ()).map_err(ChatError::from)
}
impl ChatService {
/// 前端搭接参数。
pub fn voice_config(&self) -> VoiceConfigDto {
let cfg = &self.settings.livekit;
VoiceConfigDto {
enabled: cfg.enabled,
url: cfg.browser_url().to_string(),
recording_enabled: cfg.recording_enabled,
}
}
/// 为调用者签发 LiveKit 访问令牌。
pub async fn voice_token(
&self,
caller: i64,
room_id: Uuid,
) -> Result<VoiceTokenDto, ChatError> {
let cfg = self.settings.livekit.clone();
if !cfg.enabled {
return Err(ChatError::BadRequest("livekit is not enabled".to_string()));
}
let room = require_voice_room(self, room_id, caller).await?;
let is_stage = room.kind == models::room::rooms::RoomKind::Stage;
let can_publish = if is_stage {
has_publish_grant(self, room_id, caller).await?
} else {
true
};
let api = livekit(self)?;
let room_name = room_id.to_string();
let limit = room
.user_limit
.map(|v| v.max(1) as u32)
.unwrap_or(cfg.default_max_participants);
if let Ok(participants) = api.room().list_participants(&room_name).await {
let already_inside = participants
.iter()
.any(|p| identity_to_uid(&p.identity) == Some(caller));
if !already_inside && participants.len() as u32 >= limit {
return Err(ChatError::Conflict("voice room is full".to_string()));
}
}
let _ = api
.room()
.create_room(
&room_name,
CreateRoomOptions {
empty_timeout: cfg.empty_timeout_secs,
departure_timeout: cfg.departure_timeout_secs,
max_participants: limit,
..Default::default()
},
)
.await;
let names = display_names(self, &[caller]).await;
let (display_name, _) = names
.get(&caller)
.cloned()
.unwrap_or_else(|| (caller.to_string(), None));
let grants = VideoGrants {
room_join: true,
room: room_name.clone(),
can_subscribe: true,
can_publish,
can_publish_data: true,
can_publish_sources: if can_publish {
{
vec![
"camera".to_string(),
"microphone".to_string(),
"screen_share".to_string(),
"screen_share_audio".to_string(),
]
}
} else {
Vec::new()
},
..Default::default()
};
let token = AccessToken::with_api_key(cfg.api_key.trim(), cfg.api_secret.trim())
.with_identity(&caller.to_string())
.with_name(&display_name)
.with_ttl(std::time::Duration::from_secs(cfg.token_ttl_secs))
.with_grants(grants)
.to_jwt()
.map_err(|error| ChatError::Internal(format!("livekit token: {error}")))?;
let hand_state = if is_stage {
read_hand(self, room_id, caller).await?.map(|h| h.state)
} else {
None
};
Ok(VoiceTokenDto {
token,
url: cfg.browser_url().to_string(),
room_name,
expires_in_secs: cfg.token_ttl_secs,
can_publish,
is_stage,
hand_state,
})
}
/// 以 LiveKit 为权威源对账，并回写 Redis。
pub async fn voice_participants(
&self,
caller: i64,
room_id: Uuid,
) -> Result<Vec<VoiceParticipantDto>, ChatError> {
require_voice_room(self, room_id, caller).await?;
let api = livekit(self)?;
let room_name = room_id.to_string();
let participants = api
.room()
.list_participants(&room_name)
.await
.map_err(|error| ChatError::Internal(format!("livekit list participants: {error}")))?;
let uids: Vec<i64> = participants
.iter()
.filter_map(|p| identity_to_uid(&p.identity))
.collect();
let names = display_names(self, &uids).await;
let stored: HashMap<i64, VoiceMember> = read_members(self, room_id)
.await?
.into_iter()
.map(|m| (m.user_id, m))
.collect();
let mut out = Vec::with_capacity(participants.len());
for participant in participants {
let Some(uid) = identity_to_uid(&participant.identity) else {
continue;
};
let can_publish = participant
.permission
.as_ref()
.map(|p| p.can_publish)
.unwrap_or(true);
let moderator_muted = stored.get(&uid).map(|m| m.moderator_muted).unwrap_or(false);
let tracks: Vec<VoiceTrackDto> = participant
.tracks
.iter()
.map(|t| VoiceTrackDto {
sid: t.sid.clone(),
kind: proto::TrackType::try_from(t.r#type)
.map(|k| k.as_str_name().to_string())
.unwrap_or_else(|_| "UNKNOWN".to_string()),
source: proto::TrackSource::try_from(t.source)
.map(|s| s.as_str_name().to_string())
.unwrap_or_else(|_| "UNKNOWN".to_string()),
muted: t.muted,
})
.collect();
let (display_name, avatar_url) = names
.get(&uid)
.cloned()
.unwrap_or_else(|| (participant.name.clone(), None));
let member = VoiceMember {
user_id: uid,
display_name: display_name.clone(),
avatar_url: avatar_url.clone(),
joined_at: participant.joined_at,
can_publish,
moderator_muted,
tracks: tracks.clone(),
livekit_sid: Some(participant.sid.clone()),
};
write_member(self, room_id, &member).await?;
out.push(VoiceParticipantDto {
user_id: uid,
display_name,
avatar_url,
joined_at: participant.joined_at,
can_publish,
moderator_muted,
speaking: false,
tracks,
});
}
out.sort_by_key(|p| p.joined_at);
Ok(out)
}
/// 侧栏用的整空间语音人数（纯 Redis 读，缺房间不报错）。
pub async fn voice_space_states(
&self,
caller: i64,
space_name: &str,
) -> Result<VoiceSpaceStatesDto, ChatError> {
let space = lookup_space_by_name(self, space_name).await?;
crate::share::lookup::require_space_member(self, &space, caller).await?;
let rooms = models::room::rooms::Entity::find()
.filter(models::room::rooms::Column::SpaceId.eq(space.id()))
.filter(models::room::rooms::Column::IsActive.eq(true))
.filter(models::room::rooms::Column::Kind.is_in([
models::room::rooms::RoomKind::Voice,
models::room::rooms::RoomKind::Stage,
]))
.all(self.database.read())
.await?;
let mut out = Vec::with_capacity(rooms.len());
for room in rooms {
let count = read_members(self, room.id).await?.len() as u32;
out.push(VoiceRoomStateDto {
room_id: room.id,
count,
});
}
Ok(VoiceSpaceStatesDto { rooms: out })
}
/// 读取房间共享的窗口布局；没有存过就是空布局（前端回落到自动网格）。
pub async fn voice_layout(
&self,
caller: i64,
room_id: Uuid,
) -> Result<VoiceLayoutDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
self.voice_layout_raw(room_id).await
}
/// 纯读取，不做权限校验：调用方要么刚校验过，要么只用于合并写入。
async fn voice_layout_raw(&self, room_id: Uuid) -> Result<VoiceLayoutDto, ChatError> {
let mut conn = self.participants.redis.connection_clone();
use cache::RedisConnection;
let key = layout_key(room_id);
let raw: Result<Option<String>, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => redis::cmd("GET").arg(&key).query_async(c).await,
RedisConnection::Cluster(c) => redis::cmd("GET").arg(&key).query_async(c).await,
};
let raw = raw.map_err(ChatError::from)?;
let Some(raw) = raw else {
return Ok(VoiceLayoutDto::default());
};
Ok(serde_json::from_str(&raw).unwrap_or_default())
}
/// 按 key 合并布局补丁并广播给房间里的所有人。坐标一律夹到 0~1，越界写入
/// 不可能发生；返回写入时刻，让客户端能分辨“我发的”和“比我新的”。
pub async fn save_voice_layout(
&self,
caller: i64,
room_id: Uuid,
patch: VoiceLayoutPatchDto,
) -> Result<VoiceLayoutSavedDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
if patch.tiles.is_empty() && patch.removed.is_empty() {
return Err(ChatError::BadRequest("empty layout patch".to_string()));
}
let stamp = now_secs();
let payload = serde_json::json!({
"tiles": patch.tiles,
"removed": patch.removed,
})
.to_string();
let ttl = self.settings.livekit.presence_ttl_secs.max(60);
let mut conn = self.participants.redis.connection_clone();
use cache::RedisConnection;
let key = layout_key(room_id);
let mut command = redis::cmd("EVAL");
command
.arg(MERGE_LAYOUT_SCRIPT)
.arg(1_u8)
.arg(&key)
.arg(stamp)
.arg(&payload)
.arg(MAX_LAYOUT_TILES as i64)
.arg(ttl);
let written: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => command.clone().query_async(c).await,
RedisConnection::Cluster(c) => command.clone().query_async(c).await,
};
let written = match written {
Ok(updated_at) => Ok(updated_at),
// The script signals a full map with an error reply; turn it back into
// the same 400 the previous whole-map write produced.
Err(error) if error.to_string().contains("too many tiles") => Err(
ChatError::BadRequest(format!("at most {MAX_LAYOUT_TILES} windows can be placed")),
),
Err(error) => Err(ChatError::from(error)),
}?;
self.audience
.realtime
.publish_room(&RoomEvent::VoiceLayout { room_id })
.await;
Ok(VoiceLayoutSavedDto {
updated_at: written,
})
}
/// 举手。Stage 频道之外、已有发言权都不允许。
pub async fn raise_hand(&self, caller: i64, room_id: Uuid) -> Result<HandDto, ChatError> {
let room = require_voice_room(self, room_id, caller).await?;
if room.kind != models::room::rooms::RoomKind::Stage {
return Err(ChatError::BadRequest(
"hands are only available in stage channels".to_string(),
));
}
if has_publish_grant(self, room_id, caller).await? {
return Err(ChatError::Conflict("already able to speak".to_string()));
}
let entry = HandEntry {
state: HandState::Pending,
requested_at: now_secs(),
decided_by: None,
};
write_hand(self, room_id, caller, &entry).await?;
self.audience
.realtime
.publish_room(&RoomEvent::VoiceHands { room_id })
.await;
let names = display_names(self, &[caller]).await;
let (display_name, avatar_url) = names
.get(&caller)
.cloned()
.unwrap_or_else(|| (caller.to_string(), None));
Ok(HandDto {
user_id: caller,
display_name,
avatar_url,
state: entry.state,
requested_at: entry.requested_at,
})
}
pub async fn cancel_hand(&self, caller: i64, room_id: Uuid) -> Result<(), ChatError> {
require_voice_room(self, room_id, caller).await?;
let existing = read_hand(self, room_id, caller).await?;
if matches!(existing.map(|h| h.state), Some(HandState::Approved)) {
return Err(ChatError::Conflict(
"an approved speaker must be revoked by a moderator".to_string(),
));
}
let mut conn = self.participants.redis.connection_clone();
use cache::RedisConnection;
let key = hands_key(room_id);
let field = caller.to_string();
let _: Result<i64, redis::RedisError> = match &mut conn {
RedisConnection::Standalone(c) => {
redis::cmd("HDEL")
.arg(&key)
.arg(&field)
.query_async(c)
.await
}
RedisConnection::Cluster(c) => {
redis::cmd("HDEL")
.arg(&key)
.arg(&field)
.query_async(c)
.await
}
};
self.audience
.realtime
.publish_room(&RoomEvent::VoiceHands { room_id })
.await;
Ok(())
}
pub async fn list_hands(&self, caller: i64, room_id: Uuid) -> Result<VoiceHandsDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
let can_moderate = can_moderate_voice(self, room_id, caller).await?;
let stored = read_hands(self, room_id).await?;
let my_state = stored.get(&caller).map(|h| h.state);
let uids: Vec<i64> = stored.keys().copied().collect();
let names = display_names(self, &uids).await;
let mut hands: Vec<HandDto> = stored
.into_iter()
.filter(|(_, entry)| can_moderate || matches!(entry.state, HandState::Approved))
.map(|(uid, entry)| {
let (display_name, avatar_url) = names
.get(&uid)
.cloned()
.unwrap_or_else(|| (uid.to_string(), None));
HandDto {
user_id: uid,
display_name,
avatar_url,
state: entry.state,
requested_at: entry.requested_at,
}
})
.collect();
hands.sort_by_key(|h| h.requested_at);
Ok(VoiceHandsDto {
hands,
can_moderate,
my_state,
})
}
async fn decide_hand(
&self,
caller: i64,
room_id: Uuid,
target: i64,
approve: bool,
) -> Result<VoiceModerationDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
if !can_moderate_voice(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
let Some(existing) = read_hand(self, room_id, target).await? else {
return Err(ChatError::NotFound);
};
if !matches!(existing.state, HandState::Pending | HandState::Revoked) {
return Err(ChatError::Conflict("hand is not pending".to_string()));
}
let entry = HandEntry {
state: if approve {
HandState::Approved
} else {
HandState::Denied
},
requested_at: existing.requested_at,
decided_by: Some(caller),
};
write_hand(self, room_id, target, &entry).await?;
if approve {
let _ = self.apply_publish_permission(room_id, target, true).await;
}
self.audience
.realtime
.publish_room(&RoomEvent::VoiceHands { room_id })
.await;
let action = if approve {
"speak_granted"
} else {
"speak_denied"
};
self.audience
.realtime
.publish_room(&RoomEvent::VoiceModeration {
room_id,
user_id: target,
action: action.to_string(),
})
.await;
Ok(VoiceModerationDto {
ok: true,
action: action.to_string(),
user_id: target,
})
}
pub async fn approve_hand(
&self,
caller: i64,
room_id: Uuid,
target: i64,
) -> Result<VoiceModerationDto, ChatError> {
self.decide_hand(caller, room_id, target, true).await
}
pub async fn deny_hand(
&self,
caller: i64,
room_id: Uuid,
target: i64,
) -> Result<VoiceModerationDto, ChatError> {
self.decide_hand(caller, room_id, target, false).await
}
/// 收回发言权：翻转 can_publish 并静音已发布的麦克风。
pub async fn revoke_hand(
&self,
caller: i64,
room_id: Uuid,
target: i64,
) -> Result<VoiceModerationDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
if !can_moderate_voice(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
let existing = read_hand(self, room_id, target).await?;
let requested_at = existing
.as_ref()
.map(|h| h.requested_at)
.unwrap_or_else(now_secs);
write_hand(
self,
room_id,
target,
&HandEntry {
state: HandState::Revoked,
requested_at,
decided_by: Some(caller),
},
)
.await?;
let _ = self.apply_publish_permission(room_id, target, false).await;
self.audience
.realtime
.publish_room(&RoomEvent::VoiceHands { room_id })
.await;
self.audience
.realtime
.publish_room(&RoomEvent::VoiceModeration {
room_id,
user_id: target,
action: "speak_revoked".to_string(),
})
.await;
Ok(VoiceModerationDto {
ok: true,
action: "speak_revoked".to_string(),
user_id: target,
})
}
/// 服务端翻转某人的发布权限（对 LiveKit 与 Redis 同时生效）。
pub async fn apply_publish_permission(
&self,
room_id: Uuid,
target: i64,
can_publish: bool,
) -> Result<(), ChatError> {
let api = livekit(self)?;
let room_name = room_id.to_string();
let identity = target.to_string();
let sources: Vec<i32> = if can_publish {
vec![
proto::TrackSource::Camera as i32,
proto::TrackSource::Microphone as i32,
proto::TrackSource::ScreenShare as i32,
proto::TrackSource::ScreenShareAudio as i32,
]
} else {
Vec::new()
};
let _ = api
.room()
.update_participant(
&room_name,
&identity,
UpdateParticipantOptions {
permission: Some(proto::ParticipantPermission {
can_subscribe: true,
can_publish,
can_publish_data: true,
can_publish_sources: sources,
..Default::default()
}),
..Default::default()
},
)
.await;
if let Some(mut member) = read_members(self, room_id)
.await?
.into_iter()
.find(|m| m.user_id == target)
{
member.can_publish = can_publish;
write_member(self, room_id, &member).await?;
}
Ok(())
}
/// 主持人强制静音/解除静音某人的麦克风。
pub async fn moderate_mute(
&self,
caller: i64,
room_id: Uuid,
target: i64,
muted: bool,
) -> Result<VoiceModerationDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
if !can_moderate_voice(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
let api = livekit(self)?;
let room_name = room_id.to_string();
let identity = target.to_string();
if muted {
let _ = api
.room()
.update_participant(
&room_name,
&identity,
UpdateParticipantOptions {
permission: Some(proto::ParticipantPermission {
can_subscribe: true,
can_publish: false,
can_publish_data: true,
can_publish_sources: Vec::new(),
..Default::default()
}),
..Default::default()
},
)
.await;
if let Ok(participants) = api.room().list_participants(&room_name).await
&& let Some(participant) = participants.iter().find(|p| p.identity == identity)
{
for track in &participant.tracks {
if track.source == proto::TrackSource::Microphone as i32 && !track.muted {
let _ = api
.room()
.mute_published_track(&room_name, &identity, &track.sid, true)
.await;
}
}
}
} else {
let _ = self.apply_publish_permission(room_id, target, true).await;
}
if let Some(mut member) = read_members(self, room_id)
.await?
.into_iter()
.find(|m| m.user_id == target)
{
member.moderator_muted = muted;
write_member(self, room_id, &member).await?;
}
let action = if muted { "muted" } else { "unmuted" }.to_string();
self.audience
.realtime
.publish_room(&RoomEvent::VoiceModeration {
room_id,
user_id: target,
action: action.clone(),
})
.await;
Ok(VoiceModerationDto {
ok: true,
action,
user_id: target,
})
}
/// 强制移出通话（可再次加入）。
pub async fn kick_participant(
&self,
caller: i64,
room_id: Uuid,
target: i64,
) -> Result<VoiceModerationDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
if !can_moderate_voice(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
if caller == target {
return Err(ChatError::BadRequest("cannot remove yourself".to_string()));
}
let api = livekit(self)?;
let room_name = room_id.to_string();
let identity = target.to_string();
let _ = api.room().remove_participant(&room_name, &identity).await;
remove_member(self, room_id, target).await?;
self.audience
.realtime
.publish_room(&RoomEvent::VoiceModeration {
room_id,
user_id: target,
action: "removed".to_string(),
})
.await;
Ok(VoiceModerationDto {
ok: true,
action: "removed".to_string(),
user_id: target,
})
}
/// 录制预留：需要 egress 组件，本次只做接口占位。
pub async fn start_recording(
&self,
caller: i64,
room_id: Uuid,
) -> Result<VoiceRecordingDto, ChatError> {
self.recording_unavailable(caller, room_id).await
}
pub async fn stop_recording(
&self,
caller: i64,
room_id: Uuid,
) -> Result<VoiceRecordingDto, ChatError> {
self.recording_unavailable(caller, room_id).await
}
async fn recording_unavailable(
&self,
caller: i64,
room_id: Uuid,
) -> Result<VoiceRecordingDto, ChatError> {
require_voice_room(self, room_id, caller).await?;
if !can_moderate_voice(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
if !self.settings.livekit.recording_enabled {
return Err(ChatError::BadRequest(
"recording is not enabled".to_string(),
));
}
Err(ChatError::BadRequest(
"recording requires the livekit egress component".to_string(),
))
}
}