From a36b138978a7f1a909b3dcb9e7afa1b1a581601f Mon Sep 17 00:00:00 2001 From: Zomatree Date: Mon, 18 Aug 2025 04:58:06 +0100 Subject: [PATCH] fix: move call system message to daemon, check max participants when creating a token to avoid giving tokens but erroring when attempting to join, check if the channel actually supports voice --- Cargo.lock | 1 + .../database/src/models/channels/model.rs | 32 ++++---- .../core/database/src/voice/voice_client.rs | 2 +- crates/core/models/src/v0/channels.rs | 9 +- crates/daemons/voice-ingress/Cargo.toml | 13 +-- crates/daemons/voice-ingress/src/api.rs | 41 +++++++++- crates/daemons/voice-ingress/src/main.rs | 36 +++++++- .../delta/src/routes/channels/voice_join.rs | 82 +++++++++++-------- 8 files changed, 151 insertions(+), 65 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index da5066d7..256b947f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6773,6 +6773,7 @@ dependencies = [ name = "revolt-voice-ingress" version = "0.7.1" dependencies = [ + "amqprs", "async-std", "futures", "livekit-api", diff --git a/crates/core/database/src/models/channels/model.rs b/crates/core/database/src/models/channels/model.rs index 9772d429..35d2c833 100644 --- a/crates/core/database/src/models/channels/model.rs +++ b/crates/core/database/src/models/channels/model.rs @@ -107,7 +107,7 @@ auto_derived!( /// Voice Information for when this channel is also a voice channel #[serde(skip_serializing_if = "Option::is_none")] - voice: Option + voice: Option, }, #[deprecated = "Use TextChannel { voice } instead"] VoiceChannel { @@ -146,7 +146,7 @@ auto_derived!( pub struct VoiceInformation { /// Maximium amount of users allowed in the voice channel at once #[serde(skip_serializing_if = "Option::is_none")] - pub max_users: Option + pub max_users: Option, } ); @@ -233,7 +233,7 @@ impl Channel { default_permissions: None, role_permissions: HashMap::new(), nsfw: data.nsfw.unwrap_or(false), - voice: data.voice.map(|voice| voice.into()) + voice: data.voice.map(|voice| voice.into()), }, v0::LegacyServerChannelType::Voice => Channel::TextChannel { id: id.clone(), @@ -245,7 +245,7 @@ impl Channel { default_permissions: None, role_permissions: HashMap::new(), nsfw: data.nsfw.unwrap_or(false), - voice: Some(data.voice.unwrap_or_default().into()) + voice: Some(data.voice.unwrap_or_default().into()), }, }; @@ -454,17 +454,23 @@ impl Channel { /// Clone this channel's server id pub fn server(&self) -> Option<&str> { match self { - Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => Some(server), - _ => None + Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => { + Some(server) + } + _ => None, } } /// Gets this channel's voice information pub fn voice(&self) -> Option> { match self { - Self::DirectMessage { .. } | Channel::VoiceChannel { .. } => Some(Cow::Owned(VoiceInformation::default())), - Self::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)), - _ => None + Self::DirectMessage { .. } | Channel::VoiceChannel { .. } => { + Some(Cow::Owned(VoiceInformation::default())) + } + Self::TextChannel { + voice: Some(voice), .. + } => Some(Cow::Borrowed(voice)), + _ => None, } } @@ -539,15 +545,13 @@ impl Channel { pub fn remove_field(&mut self, field: &FieldsChannel) { match field { FieldsChannel::Description => match self { - Self::Group { description, .. } - | Self::TextChannel { description, .. } => { + Self::Group { description, .. } | Self::TextChannel { description, .. } => { description.take(); } _ => {} }, FieldsChannel::Icon => match self { - Self::Group { icon, .. } - | Self::TextChannel { icon, .. } => { + Self::Group { icon, .. } | Self::TextChannel { icon, .. } => { icon.take(); } _ => {} @@ -651,7 +655,7 @@ impl Channel { if let Some(v) = partial.voice { voice.replace(v); } - }, + } Self::VoiceChannel { .. } => {} } } diff --git a/crates/core/database/src/voice/voice_client.rs b/crates/core/database/src/voice/voice_client.rs index 6a7308df..1321468f 100644 --- a/crates/core/database/src/voice/voice_client.rs +++ b/crates/core/database/src/voice/voice_client.rs @@ -103,7 +103,7 @@ impl VoiceClient { .create_room( channel.id(), CreateRoomOptions { - max_participants: voice.max_users.unwrap_or(0), + max_participants: voice.max_users.unwrap_or(0) as u32, empty_timeout: 5 * 60, // 5 minutes, ..Default::default() }, diff --git a/crates/core/models/src/v0/channels.rs b/crates/core/models/src/v0/channels.rs index 28895bfe..630bfde5 100644 --- a/crates/core/models/src/v0/channels.rs +++ b/crates/core/models/src/v0/channels.rs @@ -111,7 +111,7 @@ auto_derived!( /// Voice Information for when this channel is also a voice channel #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] - voice: Option + voice: Option, }, #[deprecated = "Use TextChannel { voice } instead"] VoiceChannel { @@ -159,7 +159,7 @@ auto_derived!( /// Maximium amount of users allowed in the voice channel at once #[cfg_attr(feature = "validator", validate(range(min = 1)))] #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] - pub max_users: Option + pub max_users: Option, } /// Partial representation of a channel @@ -283,7 +283,7 @@ auto_derived!( /// Voice Information for when this channel is also a voice channel #[serde(skip_serializing_if = "Option::is_none")] - pub voice: Option + pub voice: Option, } /// New default permissions @@ -334,9 +334,8 @@ auto_derived!( /// Whether to force disconnect any other existing voice connections. /// /// useful for disconnecting on another device and joining on a new. - pub force_disconnect: Option + pub force_disconnect: Option, } - ); impl Channel { diff --git a/crates/daemons/voice-ingress/Cargo.toml b/crates/daemons/voice-ingress/Cargo.toml index 21393e26..58cc49e0 100644 --- a/crates/daemons/voice-ingress/Cargo.toml +++ b/crates/daemons/voice-ingress/Cargo.toml @@ -14,16 +14,16 @@ lru = "0.7.6" ulid = "0.5.0" redis-kiss = "0.1.4" -# serde +# Serde serde_json = "1.0.79" rmp-serde = "1.0.0" serde = "1.0.136" -# http +# Http rocket = { version = "0.5.0-rc.2", features = ["json"] } rocket_empty = "0.1.1" -# async +# Async futures = "0.3.21" async-std = { version = "1.8.0", features = [ "tokio1", @@ -31,14 +31,17 @@ async-std = { version = "1.8.0", features = [ "attributes", ] } -# core +# Core revolt-result = { path = "../../core/result" } revolt-models = { path = "../../core/models" } revolt-config = { path = "../../core/config" } revolt-database = { path = "../../core/database" } revolt-permissions = { path = "../../core/permissions" } -# voice +# Voice livekit-api = "0.4.4" livekit-protocol = "0.4.0" livekit-runtime = { version = "0.3.1", features = ["tokio"] } + +# RabbitMQ +amqprs = { version = "1.7.0" } diff --git a/crates/daemons/voice-ingress/src/api.rs b/crates/daemons/voice-ingress/src/api.rs index 5a7ed2f8..086ea573 100644 --- a/crates/daemons/voice-ingress/src/api.rs +++ b/crates/daemons/voice-ingress/src/api.rs @@ -6,11 +6,12 @@ use revolt_database::{ util::reference::Reference, voice::{ create_voice_state, delete_voice_state, get_user_moved_from_voice, get_user_moved_to_voice, - get_voice_channel_members, take_channel_call_started_system_message, - update_voice_state_tracks, VoiceClient, + get_voice_channel_members, set_channel_call_started_system_message, + take_channel_call_started_system_message, update_voice_state_tracks, VoiceClient, }, - Database, PartialMessage, SystemMessage, + Database, PartialMessage, SystemMessage, AMQP, }; +use revolt_models::v0; use revolt_result::{Result, ToRevoltError}; use rocket::{post, State}; use rocket_empty::EmptyResponse; @@ -20,8 +21,9 @@ use crate::guard::AuthHeader; #[post("/", data = "")] pub async fn ingress( db: &State, - node: &str, voice_client: &State, + amqp: &State, + node: &str, auth_header: AuthHeader<'_>, body: &str, ) -> Result { @@ -81,6 +83,37 @@ pub async fn ingress( .p(channel_id.clone()) .await; }; + + // First user who joined - send call started system message. + if event.room.as_ref().unwrap().num_participants == 1 { + let user = Reference::from_unchecked(user_id.clone()) + .as_user(db) + .await?; + + let mut call_started_message = SystemMessage::CallStarted { + by: user_id.clone(), + finished_at: None, + } + .into_message(channel.id().to_string()); + + set_channel_call_started_system_message(channel.id(), &call_started_message.id) + .await?; + + call_started_message + .send( + db, + Some(amqp), + v0::MessageAuthor::System { + username: &user.username, + avatar: user.avatar.as_ref().map(|file| file.id.as_ref()), + }, + None, + None, + &channel, + false, + ) + .await?; + } } // User left a channel "participant_left" => { diff --git a/crates/daemons/voice-ingress/src/main.rs b/crates/daemons/voice-ingress/src/main.rs index 0058566c..8c599394 100644 --- a/crates/daemons/voice-ingress/src/main.rs +++ b/crates/daemons/voice-ingress/src/main.rs @@ -1,7 +1,12 @@ use std::env; -use revolt_database::voice::VoiceClient; +use amqprs::{ + channel::ExchangeDeclareArguments, + connection::{Connection, OpenConnectionArguments}, +}; +use revolt_config::config; use revolt_database::DatabaseInfo; +use revolt_database::{voice::VoiceClient, AMQP}; use revolt_result::Result; use rocket::{build, routes, Config}; use std::net::Ipv4Addr; @@ -13,11 +18,40 @@ mod guard; async fn main() -> Result<(), rocket::Error> { revolt_config::configure!(voice_ingress); + let config = config().await; + let database = DatabaseInfo::Auto.connect().await.unwrap(); let voice_client = VoiceClient::from_revolt_config().await; + + let connection = Connection::open(&OpenConnectionArguments::new( + &config.rabbit.host, + config.rabbit.port, + &config.rabbit.username, + &config.rabbit.password, + )) + .await + .expect("Failed to connect to RabbitMQ"); + + let channel = connection + .open_channel(None) + .await + .expect("Failed to open RabbitMQ channel"); + + channel + .exchange_declare( + ExchangeDeclareArguments::new(&config.pushd.exchange, "direct") + .durable(true) + .finish(), + ) + .await + .expect("Failed to declare exchange"); + + let amqp = AMQP::new(connection, channel); + let _rocket = build() .manage(database) .manage(voice_client) + .manage(amqp) .mount("/", routes![api::ingress]) .configure(Config { port: 8500, diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index 0b3e747f..c7c041a9 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -1,6 +1,13 @@ use revolt_config::config; +use revolt_database::{ + util::{permissions::perms, reference::Reference}, + voice::{ + delete_voice_state, get_channel_node, get_user_voice_channels, get_voice_channel_members, + raise_if_in_voice, VoiceClient, + }, + Database, User, +}; use revolt_models::v0; -use revolt_database::{util::{permissions::perms, reference::Reference}, voice::{delete_voice_state, get_channel_node, get_user_voice_channels, raise_if_in_voice, set_channel_call_started_system_message, VoiceClient}, Channel, Database, SystemMessage, User, AMQP}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; @@ -10,43 +17,59 @@ use rocket::{serde::json::Json, State}; /// /// Asks the voice server for a token to join the call. #[openapi(tag = "Voice")] -#[post("//join_call", data="")] +#[post("//join_call", data = "")] pub async fn call( db: &State, - amqp: &State, voice_client: &State, user: User, target: Reference, - data: Json + data: Json, ) -> Result> { if !voice_client.is_enabled() { - return Err(create_error!(LiveKitUnavailable)) + return Err(create_error!(LiveKitUnavailable)); } let v0::DataJoinCall { node, - force_disconnect + force_disconnect, } = data.into_inner(); if user.bot.is_some() && force_disconnect == Some(true) { - return Err(create_error!(IsBot)) + return Err(create_error!(IsBot)); } - let config = config().await; - let channel = target.as_channel(db).await?; + let Some(voice_info) = channel.voice() else { + return Err(create_error!(NotAVoiceChannel)); + }; + let mut permissions = perms(db, &user).channel(&channel); let current_permissions = calculate_channel_permissions(&mut permissions).await; current_permissions.throw_if_lacking_channel_permission(ChannelPermission::Connect)?; + if get_voice_channel_members(channel.id()) + .await? + .zip(voice_info.max_users) + .is_some_and(|(ms, max_users)| ms.len() >= max_users) + && !current_permissions.has(ChannelPermission::ManageChannel as u64) + { + return Err(create_error!(CannotJoinCall)); + } + let existing_node = get_channel_node(channel.id()).await?; - let node = existing_node.or(node) + let node = existing_node + .or(node) .ok_or_else(|| create_error!(UnknownNode))?; - let node_host = config.hosts.livekit.get(&node) + let config = config().await; + + let node_host = config + .hosts + .livekit + .get(&node) .ok_or_else(|| create_error!(UnknownNode))? .clone(); @@ -56,39 +79,28 @@ pub async fn call( for channel_id in get_user_voice_channels(&user.id).await? { let node = get_channel_node(&channel_id).await?.unwrap(); - let channel = Reference::from_unchecked(channel_id.clone()).as_channel(db).await?; + let channel = Reference::from_unchecked(channel_id.clone()) + .as_channel(db) + .await?; - voice_client.remove_user(&node, &user.id, &channel_id).await?; + voice_client + .remove_user(&node, &user.id, &channel_id) + .await?; delete_voice_state(&channel_id, channel.server(), &user.id).await?; } } else { raise_if_in_voice(&user, channel.id()).await?; } - let token = voice_client.create_token(&node, &user, current_permissions, &channel).await?; + let token = voice_client + .create_token(&node, &user, current_permissions, &channel) + .await?; let room = voice_client.create_room(&node, &channel).await?; log::debug!("Created room {}", room.name); - let mut call_started_message = SystemMessage::CallStarted { - by: user.id.clone(), - finished_at: None - } - .into_message(channel.id().to_string()); - - set_channel_call_started_system_message(channel.id(), &call_started_message.id).await?; - - call_started_message.send( - db, - Some(amqp), - v0::MessageAuthor::System { - username: &user.username, - avatar: user.avatar.as_ref().map(|file| file.id.as_ref()), - }, - None, - None, - &channel, false - ).await?; - - Ok(Json(v0::CreateVoiceUserResponse { token, url: node_host.clone() })) + Ok(Json(v0::CreateVoiceUserResponse { + token, + url: node_host.clone(), + })) }