diff --git a/Revolt.toml b/Revolt.toml index 97da148d..e5058c51 100644 --- a/Revolt.toml +++ b/Revolt.toml @@ -20,10 +20,14 @@ api = "http://local.revolt.chat:14702" events = "ws://local.revolt.chat:14703" autumn = "http://local.revolt.chat:14704" january = "http://local.revolt.chat:14705" -livekit = "ws://local.revolt.chat:14706" voso_legacy = "" voso_legacy_ws = "" +# Public urls for livekit nodes +# each entry here should have a corresponding entry under `api.livekit.nodes` +[hosts.livekit] +worldwide = "ws://local.revolt.chat:14706" + [api] [api.smtp] @@ -38,9 +42,16 @@ port = 14025 use_tls = false [api.livekit] -url = "ws://local.revolt.chat:14706" -key = "" -secret = "" + +# Config for livekit nodes +# Make sure to change the secret when deploying +# The key and secret should match the values livekit is using +[api.livekit.nodes.worldwide] +url = "http://livekit" +lat = 0.0 +lon = 0.0 +key = "worldwide_key" +secret = "ZjCofRlfm6GGtjlifmNpCDkcQbEIIVC0" [files.s3] # S3 protocol endpoint diff --git a/crates/core/config/Revolt.toml b/crates/core/config/Revolt.toml index d298a2d7..c2c2c2b3 100644 --- a/crates/core/config/Revolt.toml +++ b/crates/core/config/Revolt.toml @@ -17,10 +17,11 @@ api = "http://local.revolt.chat/api" events = "ws://local.revolt.chat/ws" autumn = "http://local.revolt.chat/autumn" january = "http://local.revolt.chat/january" -livekit = "ws://local.revolt.chat/livekit" voso_legacy = "" voso_legacy_ws = "" +[hosts.livekit] + [rabbit] host = "rabbit" port = 5672 @@ -64,13 +65,8 @@ hcaptcha_sitekey = "" max_concurrent_connections = 50 [api.livekit] -# Livekit server url -url = "ws://local.revolt.chat/livekit" -# Livekit security key name -key = "" -# Livekit security secret value -secret = "" +[api.livekit.nodes] [pushd] # this changes the names of the queues to not overlap diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 168788dc..5e640434 100644 --- a/crates/core/config/src/lib.rs +++ b/crates/core/config/src/lib.rs @@ -105,7 +105,7 @@ pub struct Hosts { pub events: String, pub autumn: String, pub january: String, - pub livekit: String + pub livekit: HashMap } #[derive(Deserialize, Debug, Clone)] @@ -176,7 +176,14 @@ pub struct ApiWorkers { #[derive(Deserialize, Debug, Clone)] pub struct ApiLiveKit { + pub nodes: HashMap +} + +#[derive(Deserialize, Debug, Clone)] +pub struct LiveKitNode { pub url: String, + pub lat: f64, + pub lon: f64, pub key: String, pub secret: String } diff --git a/crates/core/database/src/voice/mod.rs b/crates/core/database/src/voice/mod.rs index 18380660..10ee8b16 100644 --- a/crates/core/database/src/voice/mod.rs +++ b/crates/core/database/src/voice/mod.rs @@ -35,6 +35,24 @@ pub async fn raise_if_in_voice(user: &User, target: &str) -> Result<()> { } } +pub async fn set_channel_node(channel: &str, node: &str) -> Result<()> { + get_connection() + .await? + .set(format!("node:{channel}"), node) + .await + .to_internal_error()?; + + Ok(()) +} + +pub async fn get_channel_node(channel: &str) -> Result { + get_connection() + .await? + .get(format!("node:{channel}")) + .await + .to_internal_error() +} + pub async fn get_user_voice_channels(user_id: &str) -> Result> { get_connection() .await? @@ -207,10 +225,10 @@ pub async fn update_voice_state( Ok(()) } -pub async fn get_voice_channel_members(channel_id: &str) -> Result> { +pub async fn get_voice_channel_members(channel_id: &str) -> Result>> { get_connection() .await? - .smembers::<_, Vec>(format!("vc_members:{}", channel_id)) + .smembers(format!("vc_members:{}", channel_id)) .await .to_internal_error() } @@ -255,8 +273,9 @@ pub async fn get_channel_voice_state(channel: &Channel) -> Result None }; - if !members.is_empty() { + if let Some(members) = members { let mut participants = Vec::with_capacity(members.len()); + let node = get_channel_node(channel.id()).await?; for user_id in members { if let Some(voice_state) = get_voice_state(channel.id(), server, &user_id).await? { @@ -271,6 +290,7 @@ pub async fn get_channel_voice_state(channel: &Channel) -> Result Result<()> { } pub async fn sync_voice_permissions(db: &Database, voice_client: &VoiceClient, channel: &Channel, server: Option<&Server>, role_id: Option<&str>) -> Result<()> { + let node = get_channel_node(channel.id()).await?; - for user_id in get_voice_channel_members(channel.id()).await? { + for user_id in get_voice_channel_members(channel.id()).await?.iter().flatten() { let user = Reference::from_unchecked(user_id.clone()).as_user(db).await?; - sync_user_voice_permissions(db, voice_client, &user, channel, server, role_id).await?; + sync_user_voice_permissions(db, voice_client, &node, &user, channel, server, role_id).await?; }; Ok(()) } -pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClient, user: &User, channel: &Channel, server: Option<&Server>, role_id: Option<&str>) -> Result<()> { +pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClient, node: &str, user: &User, channel: &Channel, server: Option<&Server>, role_id: Option<&str>) -> Result<()> { let channel_id = channel.id(); let server_id: Option<&str> = server.as_ref().map(|s| s.id.as_str()); @@ -315,7 +336,8 @@ pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClie let voice_state = get_voice_state(channel_id, server_id, &user.id).await?.unwrap(); let mut query = DatabasePermissionQuery::new(db, user) - .channel(channel); + .channel(channel) + .user(user); if let (Some(server), Some(member)) = (server, member.as_ref()) { query = query.member(member).server(server) @@ -340,7 +362,7 @@ pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClie update_voice_state(channel_id, server_id, &user.id, &update_event).await?; - voice_client.update_permissions(user, channel_id, ParticipantPermission { + voice_client.update_permissions(node, user, channel_id, ParticipantPermission { can_subscribe: can_listen, can_publish: can_speak, can_publish_data: can_speak, diff --git a/crates/core/database/src/voice/voice_client.rs b/crates/core/database/src/voice/voice_client.rs index a6b2ba03..ea6f52e8 100644 --- a/crates/core/database/src/voice/voice_client.rs +++ b/crates/core/database/src/voice/voice_client.rs @@ -1,9 +1,9 @@ use livekit_api::{ access_token::{AccessToken, VideoGrants}, - services::room::{CreateRoomOptions, RoomClient, UpdateParticipantOptions}, + services::room::{CreateRoomOptions, RoomClient as InnerRoomClient, UpdateParticipantOptions}, }; use livekit_protocol::{ParticipantInfo, ParticipantPermission, Room}; -use revolt_config::config; +use revolt_config::{config, LiveKitNode}; use crate::models::{Channel, User}; use revolt_models::v0; use revolt_permissions::{ChannelPermission, PermissionValue}; @@ -12,41 +12,59 @@ use std::{borrow::Cow, collections::HashMap, time::Duration}; use super::get_allowed_sources; + +#[derive(Debug)] +pub struct RoomClient { + pub client: InnerRoomClient, + pub node: LiveKitNode +} + #[derive(Debug)] pub struct VoiceClient { - rooms: RoomClient, - api_key: String, - api_secret: String, + pub rooms: HashMap } impl VoiceClient { - pub fn new(url: String, api_key: String, api_secret: String) -> Self { + pub fn new(nodes: HashMap) -> Self { Self { - rooms: RoomClient::with_api_key(&url, &api_key, &api_secret), - api_key, - api_secret, + rooms: nodes + .into_iter() + .map(|(name, node)| + ( + name, + RoomClient { + client: InnerRoomClient::with_api_key(&node.url, &node.key, &node.secret), + node + } + ) + ) + .collect() } } pub async fn from_revolt_config() -> Self { let config = config().await; - Self::new( - config.hosts.livekit, - config.api.livekit.key, - config.api.livekit.secret, - ) + Self::new(config.api.livekit.nodes.clone()) + } + + pub fn get_node(&self, name: &str) -> Result<&RoomClient> { + self.rooms.get(name) + .ok_or_else(|| create_error!(UnknownNode)) } pub fn create_token( &self, + node: &str, user: &User, permissions: PermissionValue, channel: &Channel, ) -> Result { + let room = self.get_node(node)?; + let allowed_sources = get_allowed_sources(permissions); - AccessToken::with_api_key(&self.api_key, &self.api_secret) + AccessToken::with_api_key(&room.node.key, &room.node.secret) .with_name(&format!("{}#{}", user.username, user.discriminator)) .with_identity(&user.id) .with_metadata(&serde_json::to_string(&user).to_internal_error()?) @@ -63,7 +81,9 @@ impl VoiceClient { .to_internal_error() } - pub async fn create_room(&self, channel: &Channel) -> Result { + pub async fn create_room(&self, node: &str, channel: &Channel) -> Result { + let room = self.get_node(node)?; + let voice = match channel { Channel::DirectMessage { .. } | Channel::VoiceChannel { .. } => Some(Cow::Owned(v0::VoiceInformation::default())), Channel::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)), @@ -72,7 +92,7 @@ impl VoiceClient { .ok_or_else(|| create_error!(NotAVoiceChannel))?; - self.rooms + room.client .create_room( channel.id(), CreateRoomOptions { @@ -87,11 +107,14 @@ impl VoiceClient { pub async fn update_permissions( &self, + node: &str, user: &User, channel_id: &str, new_permissions: ParticipantPermission, ) -> Result { - self.rooms + let room = self.get_node(node)?; + + room.client .update_participant( channel_id, &user.id, @@ -106,8 +129,10 @@ impl VoiceClient { .to_internal_error() } - pub async fn remove_user(&self, user_id: &str, channel_id: &str) -> Result<()> { - self.rooms.remove_participant(channel_id, user_id) + pub async fn remove_user(&self, node: &str, user_id: &str, channel_id: &str) -> Result<()> { + let room = self.get_node(node)?; + + room.client.remove_participant(channel_id, user_id) .await .to_internal_error() } diff --git a/crates/core/models/src/v0/channels.rs b/crates/core/models/src/v0/channels.rs index 7947f1fa..e4dafbe0 100644 --- a/crates/core/models/src/v0/channels.rs +++ b/crates/core/models/src/v0/channels.rs @@ -315,7 +315,14 @@ auto_derived!( pub struct ChannelVoiceState { pub id: String, /// The states of the users who are connected to the channel - pub participants: Vec + pub participants: Vec, + /// The node's node which the channel is currently using + pub node: String + } + + /// Join a voice channel + pub struct DataJoinCall { + pub node: String } ); diff --git a/crates/core/permissions/src/models/channel.rs b/crates/core/permissions/src/models/channel.rs index 0b2867ae..252fb35a 100644 --- a/crates/core/permissions/src/models/channel.rs +++ b/crates/core/permissions/src/models/channel.rs @@ -128,6 +128,7 @@ pub static DEFAULT_PERMISSION: Lazy = Lazy::new(|| { + ChannelPermission::Connect + ChannelPermission::Speak + ChannelPermission::Listen + + ChannelPermission::Video ) }); diff --git a/crates/core/result/src/axum.rs b/crates/core/result/src/axum.rs index 78c35eb7..813ba76a 100644 --- a/crates/core/result/src/axum.rs +++ b/crates/core/result/src/axum.rs @@ -78,6 +78,7 @@ impl IntoResponse for Error { ErrorType::AlreadyInVoiceChannel => StatusCode::BAD_REQUEST, ErrorType::NotAVoiceChannel => StatusCode::BAD_REQUEST, ErrorType::AlreadyConnected => StatusCode::BAD_REQUEST, + ErrorType::UnknownNode => StatusCode::BAD_REQUEST, ErrorType::ProxyError => StatusCode::BAD_REQUEST, ErrorType::FileTooSmall => StatusCode::UNPROCESSABLE_ENTITY, ErrorType::FileTooLarge { .. } => StatusCode::UNPROCESSABLE_ENTITY, diff --git a/crates/core/result/src/lib.rs b/crates/core/result/src/lib.rs index e9d4f24b..9be2de7e 100644 --- a/crates/core/result/src/lib.rs +++ b/crates/core/result/src/lib.rs @@ -161,6 +161,7 @@ pub enum ErrorType { AlreadyInVoiceChannel, NotAVoiceChannel, AlreadyConnected, + UnknownNode, // ? Micro-service errors ProxyError, FileTooSmall, diff --git a/crates/core/result/src/rocket.rs b/crates/core/result/src/rocket.rs index 046b0684..78a7f844 100644 --- a/crates/core/result/src/rocket.rs +++ b/crates/core/result/src/rocket.rs @@ -82,6 +82,7 @@ impl<'r> Responder<'r, 'static> for Error { ErrorType::AlreadyInVoiceChannel => Status::BadRequest, ErrorType::NotAVoiceChannel => Status::BadRequest, ErrorType::AlreadyConnected => Status::BadRequest, + ErrorType::UnknownNode => Status::BadRequest, ErrorType::ProxyError => Status::BadRequest, ErrorType::FileTooSmall => Status::UnprocessableEntity, ErrorType::FileTooLarge { .. } => Status::UnprocessableEntity, diff --git a/crates/daemons/voice-ingress/src/main.rs b/crates/daemons/voice-ingress/src/main.rs index 04ec6e09..6bf98c22 100644 --- a/crates/daemons/voice-ingress/src/main.rs +++ b/crates/daemons/voice-ingress/src/main.rs @@ -1,15 +1,16 @@ use std::env; -use livekit_protocol::{TrackType, WebhookEvent}; +use livekit_protocol::TrackType; +use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver}; use revolt_database::{ events::client::EventV1, util::reference::Reference, Database, DatabaseInfo, voice::{create_voice_state, delete_voice_state, update_voice_state_tracks, VoiceClient} }; -use rocket::{build, post, routes, serde::json::Json, Config, State}; +use rocket::{build, http::Status, post, request::{FromRequest, Outcome}, routes, Config, Request, State}; use rocket_empty::EmptyResponse; - -use revolt_result::{Result, ToRevoltError}; +use std::net::Ipv4Addr; +use revolt_result::{create_error, Error, Result, ToRevoltError}; #[rocket::main] async fn main() -> Result<(), rocket::Error> { @@ -17,13 +18,13 @@ async fn main() -> Result<(), rocket::Error> { let database = DatabaseInfo::Auto.connect().await.unwrap(); let voice_client = VoiceClient::from_revolt_config().await; - let _rocket = build() .manage(database) .manage(voice_client) .mount("/", routes![ingress]) .configure(Config { port: 8500, + address: Ipv4Addr::new(0, 0, 0, 0).into(), ..Default::default() }) .ignite() @@ -34,14 +35,44 @@ async fn main() -> Result<(), rocket::Error> { Ok(()) } -#[post("/", data = "")] -async fn ingress(db: &State, voice_client: &State, body: Json) -> Result { +struct AuthHeader<'a>(&'a str); + +#[rocket::async_trait] +impl<'r> FromRequest<'r> for AuthHeader<'r> { + type Error = Error; + + async fn from_request(request: &'r Request<'_>) -> Outcome { + match request.headers().get("Authorization").next() { + Some(token) => Outcome::Success(Self(token)), + None => Outcome::Error((Status::Unauthorized, create_error!(NotAuthenticated))) + } + } +} + +impl std::ops::Deref for AuthHeader<'_> { + type Target = str; + + fn deref(&self) -> &Self::Target { + self.0 + } +} + +#[post("/", data = "")] +async fn ingress(db: &State, node: &str, voice_client: &State, auth_header: AuthHeader<'_>, body: &str) -> Result { log::debug!("received event: {body:?}"); - let channel_id = body.room.as_ref().map(|r| &r.name); - let user_id = body.participant.as_ref().map(|r| &r.identity); + let config = revolt_config::config().await; - match body.event.as_str() { + let node_info = config.api.livekit.nodes.get(node) + .ok_or_else(|| create_error!(NotAuthenticated))?; + + let webhook_receiver = WebhookReceiver::new(TokenVerifier::with_api_key(&node_info.key, &node_info.secret)); + let event = webhook_receiver.receive(body, &auth_header).to_internal_error()?; + + let channel_id = event.room.as_ref().map(|r| &r.name); + let user_id = event.participant.as_ref().map(|r| &r.identity); + + match event.event.as_str() { "participant_joined" => { let channel_id = channel_id.to_internal_error()?; let user_id = user_id.to_internal_error()?; @@ -79,20 +110,22 @@ async fn ingress(db: &State, voice_client: &State, body: "track_published" | "track_unpublished" => { let channel_id = channel_id.to_internal_error()?; let user_id = user_id.to_internal_error()?; - let track = body.track.as_ref().to_internal_error()?; + let track = event.track.as_ref().to_internal_error()?; let channel = Reference::from_unchecked(channel_id.clone()) .as_channel(db) .await?; // remove the user if they try publish a video larger than 1080x720 or they publish data - if body.event == "track_published" + // TODO: move to config + if event.event == "track_published" && ( - (track.r#type == TrackType::Video as i32 && (track.width > 1080 || track.height > 720)) + // handle any size which goes over the limit of "1080x720" to stop people from making too tall or too wide and bypassing the limit + (track.r#type == TrackType::Video as i32 && (track.width * track.height) >= (1080 * 720)) | (track.r#type == TrackType::Data as i32) ) { - voice_client.remove_user(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?; } @@ -100,7 +133,7 @@ async fn ingress(db: &State, voice_client: &State, body: channel_id, channel.server(), user_id, - body.event == "track_published", // to avoid duplicating this entire case twice + event.event == "track_published", // to avoid duplicating this entire case twice track.source ).await?; diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index f9feb40b..aaa76932 100644 --- a/crates/delta/src/main.rs +++ b/crates/delta/src/main.rs @@ -84,7 +84,7 @@ pub async fn web() -> Rocket { .into(); // Voice handler - let voice_client = VoiceClient::new(config.api.livekit.url.clone(), config.api.livekit.key.clone(), config.api.livekit.secret.clone()); + let voice_client = VoiceClient::new(config.api.livekit.nodes.clone()); // Configure Rabbit let connection = Connection::open(&OpenConnectionArguments::new( &config.rabbit.host, diff --git a/crates/delta/src/routes/channels/channel_delete.rs b/crates/delta/src/routes/channels/channel_delete.rs index f3dab86d..4264e47b 100644 --- a/crates/delta/src/routes/channels/channel_delete.rs +++ b/crates/delta/src/routes/channels/channel_delete.rs @@ -1,5 +1,5 @@ use revolt_database::{ - util::{permissions::DatabasePermissionQuery, reference::Reference}, voice::{delete_voice_state, get_voice_channel_members, VoiceClient}, Channel, Database, PartialChannel, User, AMQP + util::{permissions::DatabasePermissionQuery, reference::Reference}, voice::{delete_voice_state, get_channel_node, get_voice_channel_members, VoiceClient}, Channel, Database, PartialChannel, User, AMQP }; use revolt_models::v0; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; @@ -53,9 +53,13 @@ pub async fn delete( } }; - for user_id in get_voice_channel_members(channel.id()).await? { - voice_client.remove_user(&user_id, channel.id()).await?; - delete_voice_state(channel.id(), channel.server(), &user.id).await?; + if let Some(users) = get_voice_channel_members(channel.id()).await? { + let node = get_channel_node(channel.id()).await?; + + for user in users { + voice_client.remove_user(&node, &user, channel.id()).await?; + delete_voice_state(channel.id(), channel.server(), &user).await?; + } }; Ok(EmptyResponse) diff --git a/crates/delta/src/routes/channels/group_remove_member.rs b/crates/delta/src/routes/channels/group_remove_member.rs index 9836a7c3..0f6291bb 100644 --- a/crates/delta/src/routes/channels/group_remove_member.rs +++ b/crates/delta/src/routes/channels/group_remove_member.rs @@ -1,4 +1,4 @@ -use revolt_database::{util::reference::Reference, voice::{delete_voice_state, is_in_voice_channel, raise_if_in_voice, VoiceClient}, Channel, Database, User, AMQP}; +use revolt_database::{util::reference::Reference, voice::{delete_voice_state, get_channel_node, is_in_voice_channel, raise_if_in_voice, VoiceClient}, Channel, Database, User, AMQP}; use revolt_permissions::ChannelPermission; use revolt_result::{create_error, Result}; @@ -48,7 +48,9 @@ pub async fn remove_member( }; if is_in_voice_channel(&user.id, channel.id()).await? { - voice_client.remove_user(&user.id, channel.id()).await?; + let node = get_channel_node(channel.id()).await?; + + voice_client.remove_user(&node, &user.id, channel.id()).await?; delete_voice_state(channel.id(), None, &user.id).await?; }; diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index c2f9f309..31359088 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -1,7 +1,7 @@ use revolt_models::v0; use revolt_database::{util::{permissions::perms, reference::Reference}, voice::{raise_if_in_voice, VoiceClient}, Channel, Database, SystemMessage, User, AMQP}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; -use revolt_result::Result; +use revolt_result::{create_error, Result}; use rocket::{serde::json::Json, State}; @@ -9,8 +9,12 @@ use rocket::{serde::json::Json, State}; /// /// Asks the voice server for a token to join the call. #[openapi(tag = "Voice")] -#[post("//join_call")] -pub async fn call(db: &State, amqp: &State, voice: &State, user: User, target: Reference) -> Result> { +#[post("//join_call", data="")] +pub async fn call(db: &State, amqp: &State, voice: &State, user: User, target: Reference, data: Json) -> Result> { + if !voice.rooms.contains_key(&data.node) { + return Err(create_error!(UnknownNode)) + } + let channel = target.as_channel(db).await?; raise_if_in_voice(&user, channel.id()).await?; @@ -20,8 +24,8 @@ pub async fn call(db: &State, amqp: &State, voice: &State Result> { url: config.hosts.january, }, livekit: VoiceFeature { - enabled: !config.hosts.livekit.is_empty(), - url: config.hosts.livekit.to_string(), + enabled: !config.hosts.livekit.is_empty() }, }, ws: config.hosts.events, diff --git a/crates/delta/src/routes/servers/ban_create.rs b/crates/delta/src/routes/servers/ban_create.rs index 96a8d4c9..491347bb 100644 --- a/crates/delta/src/routes/servers/ban_create.rs +++ b/crates/delta/src/routes/servers/ban_create.rs @@ -1,6 +1,6 @@ use revolt_database::{ util::{permissions::DatabasePermissionQuery, reference::Reference}, - voice::{delete_voice_state, get_user_voice_channel_in_server, VoiceClient}, + voice::{delete_voice_state, get_channel_node, get_user_voice_channel_in_server, VoiceClient}, Database, RemovalIntention, ServerBan, User, }; use revolt_models::v0; @@ -61,7 +61,8 @@ pub async fn ban( // If the member is in a voice channel while banned kick them from the voice channel if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? { - voice_client.remove_user(&user.id, &channel_id).await?; + let node = get_channel_node(&channel_id).await?; + voice_client.remove_user(&node, &user.id, &channel_id).await?; delete_voice_state(&channel_id, Some(&server.id), &user.id).await? } } diff --git a/crates/delta/src/routes/servers/member_edit.rs b/crates/delta/src/routes/servers/member_edit.rs index 16ebbd96..e08bbfd5 100644 --- a/crates/delta/src/routes/servers/member_edit.rs +++ b/crates/delta/src/routes/servers/member_edit.rs @@ -7,7 +7,7 @@ use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands}; use revolt_database::{ events::client::EventV1, util::{permissions::DatabasePermissionQuery, reference::Reference}, - voice::{get_user_voice_channel_in_server, move_user, sync_user_voice_permissions, VoiceClient}, + voice::{get_channel_node, get_user_voice_channel_in_server, move_user, sync_user_voice_permissions, VoiceClient}, Database, File, PartialMember, User, }; use revolt_models::v0::{self, FieldsMember, PartialUserVoiceState}; @@ -185,6 +185,8 @@ pub async fn edit( partial.avatar = Some(File::use_user_avatar(db, &avatar, &user.id, &user.id).await?); } + // TODO: impelement moving users + let remove_contains_voice = remove .as_ref() .map(|r| r.contains(FieldsMember::CanPublish) || r.contains(FieldsMember::CanReceive)) @@ -196,9 +198,11 @@ pub async fn edit( || remove_contains_voice { if let Some(channel) = get_user_voice_channel_in_server(&target_user.id, &server.id).await? { + let node = get_channel_node(&channel).await?; let channel = Reference::from_unchecked(channel).as_channel(db).await?; - sync_user_voice_permissions(db, voice_client, &user, &channel, Some(&server), None).await?; + + sync_user_voice_permissions(db, voice_client, &node, &user, &channel, Some(&server), None).await?; }; }; diff --git a/crates/delta/src/routes/servers/member_remove.rs b/crates/delta/src/routes/servers/member_remove.rs index 3276add7..9315b1b4 100644 --- a/crates/delta/src/routes/servers/member_remove.rs +++ b/crates/delta/src/routes/servers/member_remove.rs @@ -1,6 +1,6 @@ use revolt_database::{ util::{permissions::DatabasePermissionQuery, reference::Reference}, - voice::{delete_voice_state, get_user_voice_channel_in_server, VoiceClient}, + voice::{delete_voice_state, get_channel_node, get_user_voice_channel_in_server, VoiceClient}, Database, RemovalIntention, User, }; use revolt_permissions::{calculate_server_permissions, ChannelPermission}; @@ -43,7 +43,9 @@ pub async fn kick( } if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? { - voice_client.remove_user(&user.id, &channel_id).await?; + let node = get_channel_node(&channel_id).await?; + + voice_client.remove_user(&node, &user.id, &channel_id).await?; delete_voice_state(&channel_id, Some(&server.id), &user.id).await?; } diff --git a/crates/delta/src/routes/servers/server_delete.rs b/crates/delta/src/routes/servers/server_delete.rs index 1cae301c..abf6724f 100644 --- a/crates/delta/src/routes/servers/server_delete.rs +++ b/crates/delta/src/routes/servers/server_delete.rs @@ -1,6 +1,6 @@ use revolt_database::{ util::reference::Reference, - voice::{delete_voice_state, get_user_voice_channel_in_server, VoiceClient}, + voice::{delete_voice_state, get_channel_node, get_user_voice_channel_in_server, get_voice_channel_members, VoiceClient}, Database, RemovalIntention, User, }; use revolt_models::v0; @@ -26,15 +26,22 @@ pub async fn delete( if server.owner == user.id { for channel_id in &server.channels { - voice_client.remove_user(&user.id, channel_id).await?; - delete_voice_state(channel_id, Some(&server.id), &user.id).await?; + if let Some(users) = get_voice_channel_members(channel_id).await? { + let node = get_channel_node(channel_id).await?; + + for user in users { + voice_client.remove_user(&node, &user, channel_id).await?; + delete_voice_state(channel_id, Some(&server.id), &user).await?; + } + } } server.delete(db).await } else { if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? { if server.channels.iter().any(|c| c == &channel_id) { - voice_client.remove_user(&user.id, &channel_id).await?; + let node = get_channel_node(&channel_id).await?; + voice_client.remove_user(&node, &user.id, &channel_id).await?; delete_voice_state(&channel_id, Some(&server.id), &user.id).await?; } };