diff --git a/Cargo.lock b/Cargo.lock index 6df07905..69e98842 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4016,6 +4016,7 @@ dependencies = [ "revolt-permissions", "revolt-presence", "revolt-result", + "revolt-voice", "rmp-serde", "sentry", "serde", @@ -4118,6 +4119,7 @@ dependencies = [ "revolt-models", "revolt-permissions", "revolt-result", + "revolt-voice", "revolt_rocket_okapi", "rocket", "rocket_authifier", @@ -4188,6 +4190,24 @@ dependencies = [ "serde_json", ] +[[package]] +name = "revolt-voice" +version = "0.1.0" +dependencies = [ + "async-std", + "futures", + "livekit-api", + "livekit-protocol", + "redis-kiss", + "revolt-config", + "revolt-database", + "revolt-models", + "revolt-permissions", + "revolt-result", + "serde", + "serde_json", +] + [[package]] name = "revolt-voice-ingress" version = "0.7.1" @@ -4203,8 +4223,8 @@ dependencies = [ "revolt-database", "revolt-models", "revolt-permissions", - "revolt-presence", "revolt-result", + "revolt-voice", "rmp-serde", "rocket", "rocket_empty", diff --git a/crates/bonfire/Cargo.toml b/crates/bonfire/Cargo.toml index 485ec527..4e5a2f0a 100644 --- a/crates/bonfire/Cargo.toml +++ b/crates/bonfire/Cargo.toml @@ -40,6 +40,7 @@ revolt-config = { path = "../core/config" } revolt-database = { path = "../core/database" } revolt-permissions = { version = "0.7.1", path = "../core/permissions" } revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] } +revolt-voice = { path = "../core/voice" } # redis fred = { version = "8.0.1", features = ["subscriber-client"] } diff --git a/crates/bonfire/src/events/impl.rs b/crates/bonfire/src/events/impl.rs index 744a99f8..2f014165 100644 --- a/crates/bonfire/src/events/impl.rs +++ b/crates/bonfire/src/events/impl.rs @@ -7,9 +7,9 @@ use revolt_database::{ use revolt_models::v0; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_presence::filter_online; -use revolt_result::{create_error, Result}; +use revolt_result::Result; +use revolt_voice::{delete_voice_state, get_voice_channel_members, get_voice_state}; -use redis_kiss::{get_connection, AsyncCommands, Conn}; use super::state::{Cache, State}; @@ -127,9 +127,6 @@ impl State { // Filter server channels by permission. let channels = self.cache.filter_accessible_channels(db, channels).await; - let mut voice_states = Vec::new(); - let mut conn = get_connection().await.unwrap(); - // Append known user IDs from DMs. for channel in &channels { match channel { @@ -202,8 +199,11 @@ impl State { self.insert_subscription(channel.id().to_string()); } + // fetch voice states for all the channels we can see + let mut voice_states = Vec::new(); + for channel in &channels { - if let Ok(Some(voice_state)) = self.fetch_voice_state(&mut conn, channel).await { + if let Ok(Some(voice_state)) = self.fetch_voice_state(channel).await { voice_states.push(voice_state) } } @@ -564,41 +564,21 @@ impl State { async fn fetch_voice_state( &self, - conn: &mut Conn, channel: &Channel, ) -> Result> { - let members = conn - .smembers::<_, Vec>(format!("vc-members-{}", channel.id())) - .await - .map_err(|_| create_error!(InternalError))?; - - let channel_or_server_id = channel.server().unwrap_or_else(|| channel.id()); + let members = get_voice_channel_members(&channel.id()).await?; if !members.is_empty() { let mut participants = Vec::with_capacity(members.len()); - for id in members { - let unique_key = format!("{channel_or_server_id}-{id}"); + for user_id in members { + if let Some(voice_state) = get_voice_state(&channel.id(), channel.server().as_deref(), &user_id).await? { + participants.push(voice_state); + } else { + log::info!("Voice state not found but member in voice channel members, removing."); - let (can_publish, can_receive, screensharing, camera) = conn - .mget::<_, (bool, bool, bool, bool)>(&[ - format!("can_publish-{unique_key}"), - format!("can_receive-{unique_key}"), - format!("screensharing-{unique_key}"), - format!("camera-{unique_key}"), - ]) - .await - .map_err(|_| create_error!(InternalError))?; - - let voice_state = v0::UserVoiceState { - id, - can_receive, - can_publish, - screensharing, - camera, - }; - - participants.push(voice_state); + delete_voice_state(&channel.id(), channel.server().as_deref(), &user_id).await?; + } } Ok(Some(v0::ChannelVoiceState { diff --git a/crates/bonfire/src/main.rs b/crates/bonfire/src/main.rs index 856eb243..54a2b56e 100644 --- a/crates/bonfire/src/main.rs +++ b/crates/bonfire/src/main.rs @@ -1,7 +1,9 @@ -use std::env; +use std::{env, sync::Arc}; use async_std::net::TcpListener; use revolt_presence::clear_region; +use once_cell::sync::OnceCell; +use revolt_voice::VoiceClient; #[macro_use] extern crate log; @@ -12,6 +14,15 @@ pub mod events; mod database; mod websocket; +pub static VOICE_CLIENT: OnceCell> = OnceCell::new(); + +pub fn get_voice_client() -> Arc { + VOICE_CLIENT + .get() + .expect("get_voice_client called before set") + .clone() +} + #[async_std::main] async fn main() { // Configure requirements for Bonfire. @@ -21,6 +32,8 @@ async fn main() { // Clean up the current region information. clear_region(None).await; + VOICE_CLIENT.set(Arc::new(VoiceClient::from_revolt_config().await)).unwrap(); + // Setup a TCP listener to accept WebSocket connections on. // By default, we bind to port 9000 on all interfaces. let bind = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9000".into()); diff --git a/crates/core/database/src/models/channels/model.rs b/crates/core/database/src/models/channels/model.rs index 77599993..c96151fb 100644 --- a/crates/core/database/src/models/channels/model.rs +++ b/crates/core/database/src/models/channels/model.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::{borrow::Cow, collections::HashMap}; use revolt_config::config; use revolt_models::v0::{self, MessageAuthor, VoiceInformation}; @@ -438,6 +438,15 @@ impl Channel { } } + /// Gets this channel's voice information + pub fn voice(&self) -> Option> { + match self { + Self::DirectMessage { .. } | Channel::VoiceChannel { .. } => Some(Cow::Owned(v0::VoiceInformation::default())), + Self::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)), + _ => None + } + } + /// Set role permission on a channel pub async fn set_role_permission( &mut self, diff --git a/crates/core/models/src/v0/server_members.rs b/crates/core/models/src/v0/server_members.rs index a1319143..0ed31fe0 100644 --- a/crates/core/models/src/v0/server_members.rs +++ b/crates/core/models/src/v0/server_members.rs @@ -131,12 +131,15 @@ auto_derived!( pub roles: Option>, /// Timestamp this member is timed out until pub timeout: Option, - /// Fields to remove from channel object - #[cfg_attr(feature = "validator", validate(length(min = 1)))] - pub remove: Option>, /// server-wide voice muted pub can_publish: Option, /// server-wide voice deafened pub can_receive: Option, + /// voice channel to move to if already in a voice channel + pub voice_channel: Option, + /// Fields to remove from channel object + #[cfg_attr(feature = "validator", validate(length(min = 1)))] + pub remove: Option>, + } ); diff --git a/crates/core/permissions/src/models/mod.rs b/crates/core/permissions/src/models/mod.rs index 330fe756..b8788230 100644 --- a/crates/core/permissions/src/models/mod.rs +++ b/crates/core/permissions/src/models/mod.rs @@ -8,7 +8,7 @@ pub use server::*; pub use user::*; /// Holds a permission value to manipulate. -#[derive(Clone, Debug)] +#[derive(Copy, Clone, Debug)] pub struct PermissionValue(u64); impl PermissionValue { diff --git a/crates/core/result/src/lib.rs b/crates/core/result/src/lib.rs index 3733a640..e3bdb4f2 100644 --- a/crates/core/result/src/lib.rs +++ b/crates/core/result/src/lib.rs @@ -134,7 +134,9 @@ pub enum ErrorType { // ? Voice errors LiveKitUnavailable, - AlreadyInVoiceChannel + AlreadyInVoiceChannel, + NotAVoiceChannel, + AlreadyConnected } #[macro_export] diff --git a/crates/core/result/src/rocket.rs b/crates/core/result/src/rocket.rs index 756da38e..fd2466dc 100644 --- a/crates/core/result/src/rocket.rs +++ b/crates/core/result/src/rocket.rs @@ -77,6 +77,8 @@ impl<'r> Responder<'r, 'static> for Error { ErrorType::LiveKitUnavailable => Status::BadRequest, ErrorType::AlreadyInVoiceChannel => Status::BadRequest, + ErrorType::NotAVoiceChannel => Status::BadRequest, + ErrorType::AlreadyConnected => Status::BadRequest }; // Serialize the error data structure into JSON. diff --git a/crates/core/voice/Cargo.toml b/crates/core/voice/Cargo.toml new file mode 100644 index 00000000..029b6193 --- /dev/null +++ b/crates/core/voice/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "revolt-voice" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +# voice +livekit-api = "0.3.2" +livekit-protocol = "0.3.2" + +# core +revolt-result = { path = "../result" } +revolt-models = { path = "../models" } +revolt-config = { path = "../config" } +revolt-database = { path = "../database" } +revolt-permissions = { version = "0.7.1", path = "../permissions" } + +# async +futures = "0.3.21" +async-std = { version = "1.8.0", features = [ + "tokio1", + "tokio02", + "attributes", +] } + +# util +redis-kiss = "0.1.4" + +# serde +serde_json = "1.0.79" +serde = "1.0.136" diff --git a/crates/core/voice/src/lib.rs b/crates/core/voice/src/lib.rs new file mode 100644 index 00000000..226364d5 --- /dev/null +++ b/crates/core/voice/src/lib.rs @@ -0,0 +1,266 @@ +use livekit_api::{access_token::{AccessToken, VideoGrants}, services::room::{CreateRoomOptions, RoomClient}}; +use livekit_protocol::Room; +use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands}; +use revolt_database::{Channel, User}; +use revolt_models::v0::{self, PartialUserVoiceState, UserVoiceState}; +use revolt_permissions::{ChannelPermission, PermissionValue}; +use revolt_result::{Result, ToRevoltError, create_error}; +use std::time::Duration; +use revolt_config::config; + +pub async fn raise_if_in_voice(user: &User, target: &str) -> Result<()> { + let mut conn = get_connection() + .await + .to_internal_error()?; + + if user.bot.is_some() + // bots can be in as many voice channels as it wants so we just check if its already connected to the one its trying to connect to + && conn.sismember(format!("vc-{}", &user.id), target) + .await + .to_internal_error()? + { + Err(create_error!(AlreadyConnected)) + + } else if conn.scard::<_, u32>(format!("vc-{}", &user.id)) // check if the current vc set is empty + .await + .to_internal_error()? > 0 + { + Err(create_error!(AlreadyInVoiceChannel)) + } else { + Ok(()) + } +} + +pub fn get_allowed_sources(permissions: PermissionValue) -> Vec { + let mut allowed_sources = Vec::new(); + + if permissions.has(ChannelPermission::Speak as u64) { + allowed_sources.push("MICROPHONE".to_string()) + }; + + if permissions.has(ChannelPermission::Video as u64) { + allowed_sources.extend([ + "CAMERA".to_string(), + "SCREEN_SHARE".to_string(), + "SCREEN_SHARE_AUDIO".to_string() + ]); + }; + + allowed_sources +} + +pub async fn create_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str) -> Result { + let unique_key = format!( + "{}-{}", + &user_id, + server_id.unwrap_or(channel_id) + ); + + let voice_state = UserVoiceState { + id: user_id.to_string(), + can_receive: true, + can_publish: false, + screensharing: false, + camera: false, + }; + + Pipeline::new() + .sadd(format!("vc-members-{channel_id}"), user_id) + .sadd(format!("vc-{user_id}"), channel_id) + .set(format!("can_publish-{unique_key}"), voice_state.can_publish) + .set(format!("can_receive-{unique_key}"), voice_state.can_receive) + .set(format!("screensharing-{unique_key}"), voice_state.screensharing) + .set(format!("camera-{unique_key}"), voice_state.camera) + .query_async(&mut get_connection() + .await + .to_internal_error()? + .into_inner()) + .await + .to_internal_error()?; + + Ok(voice_state) +} + +pub async fn delete_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str) -> Result<()> { + let unique_key = format!( + "{}-{}", + &user_id, + server_id.unwrap_or(channel_id) + ); + + Pipeline::new() + .srem(format!("vc-members-{channel_id}"), user_id) + .srem(format!("vc-{user_id}"), channel_id) + .del(&[ + format!("can_publish-{unique_key}"), + format!("can_receive-{unique_key}"), + format!("screensharing-{unique_key}"), + format!("camera-{unique_key}"), + ]) + .query_async(&mut get_connection() + .await + .to_internal_error()? + .into_inner()) + .await + .to_internal_error()?; + + Ok(()) +} + +pub async fn update_voice_state_tracks(channel_id: &str, server_id: Option<&str>, user_id: &str, added: bool, track: i32) -> Result { + let partial = match track { + /* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(), + /* TrackSource::Camera */ 1 => { + PartialUserVoiceState { + camera: Some(added), + ..Default::default() + } + } + /* TrackSource::Microphone */ 2 => { + PartialUserVoiceState { + can_publish: Some(added), + ..Default::default() + } + } + /* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */ 3 | 4 => { + PartialUserVoiceState { + screensharing: Some(added), + ..Default::default() + } + } + _ => unreachable!(), + }; + + update_voice_state(channel_id, server_id, user_id, &partial).await?; + + Ok(partial) +} + +pub async fn update_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str, partial: &PartialUserVoiceState) -> Result<()> { + let unique_key = format!( + "{}-{}", + &user_id, + server_id.unwrap_or(channel_id) + ); + + let mut pipeline = Pipeline::new(); + + if let Some(camera) = &partial.camera { + pipeline.set(format!("camera-{unique_key}"), camera); + }; + + if let Some(can_publish) = &partial.can_publish { + pipeline.set(format!("can_publish-{unique_key}"), can_publish); + } + + if let Some(can_receive) = &partial.can_receive { + pipeline.set(format!("can_receive-{unique_key}"), can_receive); + } + + if let Some(screensharing) = &partial.screensharing { + pipeline.set(format!("screensharing-{unique_key}"), screensharing); + } + + pipeline.query_async(&mut get_connection() + .await + .to_internal_error()? + .into_inner() + ) + .await + .to_internal_error()?; + + Ok(()) +} + +pub async fn get_voice_channel_members(channel_id: &str) -> Result> { + get_connection() + .await + .to_internal_error()? + .smembers::<_, Vec>(format!("vc-members-{}", channel_id)) + .await + .to_internal_error() +} + +pub async fn get_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str) -> Result> { + let unique_key = format!("{}-{user_id}", server_id.unwrap_or(channel_id)); + + let (can_publish, can_receive, screensharing, camera) = get_connection() + .await + .to_internal_error()? + .mget::<_, (Option, Option, Option, Option)>(&[ + format!("can_publish-{unique_key}"), + format!("can_receive-{unique_key}"), + format!("screensharing-{unique_key}"), + format!("camera-{unique_key}"), + ]) + .await + .to_internal_error()?; + + match (can_publish, can_receive, screensharing, camera) { + (Some(can_publish), Some(can_receive), Some(screensharing), Some(camera)) => { + Ok(Some(v0::UserVoiceState { + id: user_id.to_string(), + can_receive, + can_publish, + screensharing, + camera, + })) + }, + _ => Ok(None) + } +} + +#[derive(Debug)] +pub struct VoiceClient { + rooms: RoomClient, + api_key: String, + api_secret: String +} + +impl VoiceClient { + pub fn new(url: String, api_key: String, api_secret: String) -> Self { + Self { + rooms: RoomClient::with_api_key(&url, &api_key, &api_secret), + api_key, + api_secret + } + } + + pub async fn from_revolt_config() -> Self { + let config = config().await; + + Self::new(config.hosts.livekit, config.api.livekit.key, config.api.livekit.secret) + } + + pub fn create_token(&self, user: &User, permissions: PermissionValue, channel: &Channel) -> Result { + let allowed_sources = get_allowed_sources(permissions); + + AccessToken::with_api_key(&self.api_key, &self.api_secret) + .with_name(&format!("{}#{}", user.username, user.discriminator)) + .with_identity(&user.id) + .with_metadata(&serde_json::to_string(&user).to_internal_error()?) + .with_ttl(Duration::from_secs(10)) + .with_grants(VideoGrants { + room_join: true, + can_publish_sources: allowed_sources, + room: channel.id().to_string(), + ..Default::default() + }) + .to_jwt() + .to_internal_error() + } + + pub async fn create_room(&self, channel: &Channel) -> Result { + let voice = channel + .voice() + .ok_or_else(|| create_error!(NotAVoiceChannel))?; + + self.rooms.create_room(&channel.id(), CreateRoomOptions { + max_participants: voice.max_users.unwrap_or(u32::MAX), + empty_timeout: 5 * 60, // 5 minutes + ..Default::default() + }) + .await + .map_err(|_| create_error!(InternalError)) + } +} \ No newline at end of file diff --git a/crates/delta/Cargo.toml b/crates/delta/Cargo.toml index e8a4b684..a69c3fcb 100644 --- a/crates/delta/Cargo.toml +++ b/crates/delta/Cargo.toml @@ -79,6 +79,7 @@ revolt-models = { path = "../core/models", features = [ ] } revolt-result = { path = "../core/result", features = ["rocket", "okapi"] } revolt-permissions = { path = "../core/permissions", features = ["schemas"] } +revolt-voice = { path = "../core/voice" } # voice livekit-api = "0.3.2" diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index f910e660..7d7b760e 100644 --- a/crates/delta/src/main.rs +++ b/crates/delta/src/main.rs @@ -99,10 +99,8 @@ pub async fn web() -> Rocket { ) .into(); - - log::info!("{:?}", &config.api.livekit); - - let room_client = RoomClient::with_api_key(&config.api.livekit.url, &config.api.livekit.key, &config.api.livekit.secret); + // Voice handler + let voice_client = revolt_voice::VoiceClient::new(config.api.livekit.url.clone(), config.api.livekit.key.clone(), config.api.livekit.secret.clone()); // Configure Rocket let rocket = rocket::build(); @@ -117,7 +115,7 @@ pub async fn web() -> Rocket { .manage(authifier) .manage(db) .manage(cors.clone()) - .manage(room_client) + .manage(voice_client) .attach(util::ratelimiter::RatelimitFairing) .attach(cors) .configure(rocket::Config { diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index 08330b80..460b362d 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -1,12 +1,9 @@ -use std::borrow::Cow; - -use revolt_config::config; use revolt_models::v0; -use revolt_database::{util::{permissions::perms, reference::Reference}, Channel, Database, User}; +use revolt_database::{util::{permissions::perms, reference::Reference}, Database, User}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; -use revolt_result::{create_error, Result}; +use revolt_result::Result; +use revolt_voice::{VoiceClient, raise_if_in_voice}; -use livekit_api::{access_token::{AccessToken, VideoGrants}, services::room::{CreateRoomOptions, RoomClient}}; use rocket::{serde::json::Json, State}; /// # Join Call @@ -14,69 +11,20 @@ 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, rooms: &State, user: User, target: Reference) -> Result> { +pub async fn call(db: &State, voice: &State, user: User, target: Reference) -> Result> { let channel = target.as_channel(db).await?; + raise_if_in_voice(&user, &channel.id()).await?; + 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)?; - // TODO - decide if we should allow being in multiple voice channels for users + let token = voice.create_token(&user, current_permissions, &channel)?; + let room = voice.create_room(&channel).await?; - // if user.current_voice_channel(&channel.server().unwrap_or_else(|| channel.id())) - // .await? - // .is_some() - // { - // return Err(create_error!(AlreadyInVoiceChannel)) - // } - - let config = config().await; - - if config.api.livekit.url.is_empty() { - return Err(create_error!(LiveKitUnavailable)); - }; - - let voice = match &channel { - Channel::DirectMessage { .. } | Channel::VoiceChannel { .. } => Cow::Owned(v0::VoiceInformation::default()), - Channel::TextChannel { voice: Some(voice), .. } => Cow::Borrowed(voice), - _ => return Err(create_error!(CannotJoinCall)) - }; - - let mut allowed_sources = Vec::new(); - - if current_permissions.has(ChannelPermission::Speak as u64) { - allowed_sources.push("MICROPHONE".to_string()) - }; - - if current_permissions.has(ChannelPermission::Video as u64) { - allowed_sources.extend(["CAMERA".to_string(), "SCREEN_SHARE".to_string(), "SCREEN_SHARE_AUDIO".to_string()]); - }; - - let token = AccessToken::with_api_key(&config.api.livekit.key, &config.api.livekit.secret) - .with_name(&format!("{}#{}", user.username, user.discriminator)) - .with_identity(&user.id) - .with_metadata(&serde_json::to_string(&user).map_err(|_| create_error!(InternalError))?) - .with_grants(VideoGrants { - room_join: true, - can_publish_sources: allowed_sources, - room: channel.id().to_string(), - ..Default::default() - }) - .to_jwt() - .inspect_err(|e| log::error!("{e:?}")) - .map_err(|_| create_error!(InternalError))?; - - let room = rooms.create_room(&channel.id(), CreateRoomOptions { - max_participants: voice.max_users.unwrap_or(u32::MAX), - empty_timeout: 5 * 60, // 5 minutes - ..Default::default() - }) - .await - .inspect_err(|e| log::error!("{e:?}")) - .map_err(|_| create_error!(InternalError))?; - - log::info!("created room {room:?}"); + log::debug!("created room {}", room.name); Ok(Json(v0::CreateVoiceUserResponse { token })) } diff --git a/crates/delta/src/routes/servers/member_edit.rs b/crates/delta/src/routes/servers/member_edit.rs index 941a36ff..3506e955 100644 --- a/crates/delta/src/routes/servers/member_edit.rs +++ b/crates/delta/src/routes/servers/member_edit.rs @@ -1,17 +1,17 @@ use std::collections::HashSet; +use futures::TryFutureExt; use livekit_api::services::room::{RoomClient, UpdateParticipantOptions}; use livekit_protocol::ParticipantPermission; use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands}; use revolt_database::{ - util::{permissions::DatabasePermissionQuery, reference::Reference}, - Database, File, PartialMember, User, + events::client::EventV1, util::{permissions::DatabasePermissionQuery, reference::Reference}, Database, File, PartialMember, User }; -use revolt_models::v0; +use revolt_models::v0::{self, PartialUserVoiceState}; use revolt_permissions::{calculate_server_permissions, ChannelPermission}; use revolt_result::{create_error, Result, ToRevoltError}; -use rocket::{serde::json::Json, State}; +use rocket::{form::validate::Contains, serde::json::Json, State}; use validator::Validate; /// # Edit Member @@ -99,6 +99,25 @@ pub async fn edit( permissions.throw_if_lacking_channel_permission(ChannelPermission::DeafenMembers)?; } + if let Some(new_channel) = &data.voice_channel { + permissions.throw_if_lacking_channel_permission(ChannelPermission::MoveMembers)?; + + // ensure the channel we are moving them to is in the server and is a voice channel + + let channel = Reference::from_unchecked(new_channel.clone()) + .as_channel(db) + .await + .map_err(|_| create_error!(UnknownChannel))?; + + if !channel.server().is_some_and(|v| v == member.id.server) { + Err(create_error!(UnknownChannel))? + } + + if channel.voice().is_none() { + Err(create_error!(NotAVoiceChannel))? + } + } + // Resolve our ranking let our_ranking = query.get_member_rank().unwrap_or(i64::MIN); @@ -136,6 +155,7 @@ pub async fn edit( remove, can_publish, can_receive, + voice_channel } = data; let mut partial = PartialMember { @@ -171,13 +191,15 @@ pub async fn edit( ) .await?; - if can_publish.is_some() || can_receive.is_some() { + if can_publish.is_some() || can_receive.is_some() || voice_channel.is_some() { let mut conn = get_connection().await.to_internal_error()?; + let unique_key = format!("{}-{}", &member.id.user, &member.id.server); + // if we edit the member while they are in a voice channel we need to also update the perms // otherwise it wont take place until they leave and rejoin if let Some(channel) = conn - .get::<_, Option>(format!("vc-{}-{}", &member.id.user, &member.id.server)) + .get::<_, Option>(format!("vc-{}", &unique_key)) .await .to_internal_error()? { @@ -186,7 +208,7 @@ pub async fn edit( if let Some(can_publish) = can_publish { pipeline.set( - format!("can_publish-{}-{}", &channel, &member.id.user), + format!("can_publish-{}", unique_key), can_publish, ); @@ -196,13 +218,18 @@ pub async fn edit( if let Some(can_receive) = can_receive { pipeline.set( - format!("can_receive-{}-{}", &channel, &member.id.user), + format!("can_receive-{}", unique_key), can_receive, ); new_perms.can_subscribe = can_receive; }; + if let Some(new_channel) = voice_channel { + pipeline + .smove(format!("vc-members-{channel}"), format!("vc-members-{new_channel}"), &member.id.user); + }; + pipeline .query_async(&mut conn.into_inner()) .await @@ -220,6 +247,18 @@ pub async fn edit( ) .await .to_internal_error()?; + + EventV1::UserVoiceStateUpdate { + id: member.id.user.clone(), + channel_id: channel.clone(), + data: PartialUserVoiceState { + can_publish, + can_receive, + ..Default::default() + } + } + .p(channel) + .await; }; }; diff --git a/crates/voice-ingress/Cargo.toml b/crates/voice-ingress/Cargo.toml index b220968f..70fba32c 100644 --- a/crates/voice-ingress/Cargo.toml +++ b/crates/voice-ingress/Cargo.toml @@ -36,8 +36,8 @@ revolt-result = { path = "../core/result" } revolt-models = { path = "../core/models" } revolt-config = { path = "../core/config" } revolt-database = { path = "../core/database" } -revolt-permissions = { version = "0.7.1", path = "../core/permissions" } -revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] } +revolt-permissions = { path = "../core/permissions" } +revolt-voice = { path = "../core/voice" } # voice livekit-api = "0.3.2" diff --git a/crates/voice-ingress/src/main.rs b/crates/voice-ingress/src/main.rs index 28147efc..4ff6d360 100644 --- a/crates/voice-ingress/src/main.rs +++ b/crates/voice-ingress/src/main.rs @@ -1,12 +1,11 @@ use std::env; -use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands}; use livekit_protocol::WebhookEvent; use revolt_database::{ events::client::EventV1, util::reference::Reference, Database, DatabaseInfo, }; -use revolt_models::v0::{PartialUserVoiceState, UserVoiceState}; +use revolt_voice::{create_voice_state, delete_voice_state, update_voice_state_tracks, VoiceClient}; use rocket::{build, post, routes, serde::json::Json, Config, State}; use rocket_empty::EmptyResponse; @@ -17,9 +16,11 @@ async fn main() -> Result<(), rocket::Error> { revolt_config::configure!(voice_ingress); 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, @@ -34,13 +35,10 @@ async fn main() -> Result<(), rocket::Error> { } #[post("/", data = "")] -async fn ingress(db: &State, body: Json) -> Result { - let mut conn = get_connection().await.to_internal_error()?; - +async fn ingress(db: &State, voice_client: &State, body: Json) -> 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); match body.event.as_str() { @@ -52,29 +50,7 @@ async fn ingress(db: &State, body: Json) -> Result, body: Json) -> Result(format!("vc-members-{channel_id}"), user_id) - .await - .to_internal_error()?; - - let unique_key = format!( - "{}-{user_id}", - channel.server().unwrap_or_else(|| channel.id()) - ); - - conn.del::<_, Vec>(&[ - format!("vc-{unique_key}"), - format!("can_publish-{unique_key}"), - format!("can_receive-{unique_key}"), - format!("screensharing-{unique_key}"), - format!("camera-{unique_key}"), - ]) - .await - .to_internal_error()?; + delete_voice_state(channel_id, channel.server().as_deref(), user_id).await?; EventV1::VoiceChannelLeave { id: channel_id.clone(), @@ -118,66 +77,21 @@ async fn ingress(db: &State, body: Json) -> Result { - let value = body.event == "track_published"; // to avoid duplicating this entire case twice - 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 user = Reference::from_unchecked(user_id.clone()) - .as_user(db) - .await?; - let channel = Reference::from_unchecked(channel_id.clone()) .as_channel(db) .await?; - let unique_key = if user.bot.is_some() { - format!( - "{}-{user_id}", - channel.server().unwrap_or_else(|| channel.id()) - ) - } else { - user_id.to_string() - }; - - let partial = match track.source { - /* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(), - /* TrackSource::Camera */ - 1 => { - conn.set::<_, _, String>(format!("camera-{unique_key}"), value) - .await - .to_internal_error()?; - - PartialUserVoiceState { - camera: Some(value), - ..Default::default() - } - } - /* TrackSource::Microphone */ - 2 => { - conn.set::<_, _, String>(format!("can_receive-{unique_key}"), value) - .await - .to_internal_error()?; - - PartialUserVoiceState { - can_publish: Some(value), - ..Default::default() - } - } - /* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */ - 3 | 4 => { - conn.set::<_, _, String>(format!("screensharing-{unique_key}"), value) - .await - .to_internal_error()?; - - PartialUserVoiceState { - screensharing: Some(value), - ..Default::default() - } - } - _ => unreachable!(), - }; + let partial = update_voice_state_tracks( + channel_id, + channel.server().as_deref(), + user_id, + body.event == "track_published", // to avoid duplicating this entire case twice + track.source + ).await?; EventV1::UserVoiceStateUpdate { id: user_id.clone(),