From 67569f932ed0d61dfe048c50d0b3c03d2be74cb1 Mon Sep 17 00:00:00 2001 From: Zomatree Date: Wed, 24 Apr 2024 02:23:52 +0100 Subject: [PATCH] more stuff --- Cargo.lock | 1 + crates/bonfire/src/events/impl.rs | 12 +- .../admin_migrations/ops/mongodb/scripts.rs | 20 ++- .../src/models/server_members/model.rs | 13 +- crates/core/database/src/util/bridge/v0.rs | 16 +- crates/core/models/src/v0/server_members.rs | 12 ++ crates/core/models/src/v0/users.rs | 6 +- crates/core/result/src/lib.rs | 33 ++++ crates/delta/Cargo.toml | 1 + .../delta/src/routes/servers/member_edit.rs | 70 +++++++- crates/voice-ingress/src/main.rs | 149 ++++++++++-------- 11 files changed, 254 insertions(+), 79 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 10f7a1c2..6df07905 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4103,6 +4103,7 @@ dependencies = [ "lettre", "linkify 0.6.0", "livekit-api", + "livekit-protocol", "log", "lru 0.7.8", "nanoid", diff --git a/crates/bonfire/src/events/impl.rs b/crates/bonfire/src/events/impl.rs index afe5877b..3f6eb954 100644 --- a/crates/bonfire/src/events/impl.rs +++ b/crates/bonfire/src/events/impl.rs @@ -574,7 +574,6 @@ impl State { let channel_or_server_id = channel.server().unwrap_or_else(|| channel.id()); - if !members.is_empty() { let mut participants = Vec::with_capacity(members.len()); @@ -586,13 +585,18 @@ impl State { format!("audio-{unique_key}"), format!("deafened-{unique_key}"), format!("screensharing-{unique_key}"), - format!("camera-{unique_key}") - + format!("camera-{unique_key}"), ]) .await .map_err(|_| create_error!(InternalError))?; - let voice_state = v0::UserVoiceState { id, audio, deafened, screensharing, camera }; + let voice_state = v0::UserVoiceState { + id, + can_receive: audio, + can_publish: deafened, + screensharing, + camera, + }; participants.push(voice_state); } diff --git a/crates/core/database/src/models/admin_migrations/ops/mongodb/scripts.rs b/crates/core/database/src/models/admin_migrations/ops/mongodb/scripts.rs index ecc14672..bf36fed2 100644 --- a/crates/core/database/src/models/admin_migrations/ops/mongodb/scripts.rs +++ b/crates/core/database/src/models/admin_migrations/ops/mongodb/scripts.rs @@ -19,7 +19,7 @@ struct MigrationInfo { revision: i32, } -pub const LATEST_REVISION: i32 = 26; +pub const LATEST_REVISION: i32 = 27; pub async fn migrate_database(db: &MongoDb) { let migrations = db.col::("migrations"); @@ -983,6 +983,24 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 { .expect("Failed to create ratelimit_events index."); } + if revision <= 26 { + info!("Running migration [revision 26 / 17-04-2024]: Add `can_publish` and `can_receive` to members"); + + db.col::("server_members") + .update_many( + doc! {}, + doc! { + "$set": { + "can_publish": true, + "can_receive": true + } + }, + None + ) + .await + .expect("Failed to update members"); + } + // Need to migrate fields on attachments, change `user_id`, `object_id`, etc to `parent`. // Reminder to update LATEST_REVISION when adding new migrations. diff --git a/crates/core/database/src/models/server_members/model.rs b/crates/core/database/src/models/server_members/model.rs index 63a35ec3..fd068aee 100644 --- a/crates/core/database/src/models/server_members/model.rs +++ b/crates/core/database/src/models/server_members/model.rs @@ -3,8 +3,8 @@ use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; use crate::{ - events::client::EventV1, util::permissions::DatabasePermissionQuery, Channel, Database, File, - Server, SystemMessage, User, + events::client::EventV1, if_false, util::permissions::DatabasePermissionQuery, Channel, + Database, File, Server, SystemMessage, User, }; auto_derived_partial!( @@ -30,6 +30,13 @@ auto_derived_partial!( /// Timestamp this member is timed out until #[serde(skip_serializing_if = "Option::is_none")] pub timeout: Option, + + /// Whether the member is server-wide voice muted + #[serde(skip_serializing_if = "if_false")] + pub can_publish: bool, + /// Whether the member is server-wide voice deafened + #[serde(skip_serializing_if = "if_false")] + pub can_receive: bool, }, "PartialMember" ); @@ -69,6 +76,8 @@ impl Default for Member { avatar: None, roles: vec![], timeout: None, + can_publish: false, + can_receive: false, } } } diff --git a/crates/core/database/src/util/bridge/v0.rs b/crates/core/database/src/util/bridge/v0.rs index 2d49fd02..042f10c8 100644 --- a/crates/core/database/src/util/bridge/v0.rs +++ b/crates/core/database/src/util/bridge/v0.rs @@ -186,7 +186,7 @@ impl From for Channel { default_permissions, role_permissions, nsfw, - voice + voice, } => Channel::TextChannel { id, server, @@ -197,7 +197,7 @@ impl From for Channel { default_permissions, role_permissions, nsfw, - voice + voice, }, crate::Channel::VoiceChannel { id, @@ -268,7 +268,7 @@ impl From for crate::Channel { default_permissions, role_permissions, nsfw, - voice + voice, } => crate::Channel::TextChannel { id, server, @@ -279,7 +279,7 @@ impl From for crate::Channel { default_permissions, role_permissions, nsfw, - voice + voice, }, Channel::VoiceChannel { id, @@ -614,6 +614,8 @@ impl From for Member { avatar: value.avatar.map(|f| f.into()), roles: value.roles, timeout: value.timeout, + can_publish: value.can_publish, + can_receive: value.can_receive, } } } @@ -627,6 +629,8 @@ impl From for crate::Member { avatar: value.avatar.map(|f| f.into()), roles: value.roles, timeout: value.timeout, + can_publish: value.can_publish, + can_receive: value.can_receive, } } } @@ -640,6 +644,8 @@ impl From for PartialMember { avatar: value.avatar.map(|f| f.into()), roles: value.roles, timeout: value.timeout, + can_publish: value.can_publish, + can_receive: value.can_receive, } } } @@ -653,6 +659,8 @@ impl From for crate::PartialMember { avatar: value.avatar.map(|f| f.into()), roles: value.roles, timeout: value.timeout, + can_publish: value.can_publish, + can_receive: value.can_receive, } } } diff --git a/crates/core/models/src/v0/server_members.rs b/crates/core/models/src/v0/server_members.rs index 5b26b88a..a1319143 100644 --- a/crates/core/models/src/v0/server_members.rs +++ b/crates/core/models/src/v0/server_members.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use super::{File, Role, User}; +use crate::if_false; use iso8601_timestamp::Timestamp; use once_cell::sync::Lazy; @@ -57,6 +58,13 @@ auto_derived_partial!( /// Timestamp this member is timed out until #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub timeout: Option, + + /// Whether the member is server-wide voice muted + #[serde(skip_serializing_if = "if_false")] + pub can_publish: bool, + /// Whether the member is server-wide voice deafened + #[serde(skip_serializing_if = "if_false")] + pub can_receive: bool, }, "PartialMember" ); @@ -126,5 +134,9 @@ auto_derived!( /// 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, } ); diff --git a/crates/core/models/src/v0/users.rs b/crates/core/models/src/v0/users.rs index 50d8163c..7b959d9a 100644 --- a/crates/core/models/src/v0/users.rs +++ b/crates/core/models/src/v0/users.rs @@ -276,10 +276,10 @@ auto_derived_partial!( /// Voice State information for a user pub struct UserVoiceState { pub id: String, - pub audio: bool, - pub deafened: bool, + pub can_receive: bool, + pub can_publish: bool, pub screensharing: bool, - pub camera: bool + pub camera: bool, }, "PartialUserVoiceState" ); diff --git a/crates/core/result/src/lib.rs b/crates/core/result/src/lib.rs index c8e6c1c9..3733a640 100644 --- a/crates/core/result/src/lib.rs +++ b/crates/core/result/src/lib.rs @@ -1,3 +1,5 @@ +use std::panic::Location; + #[cfg(feature = "serde")] #[macro_use] extern crate serde; @@ -172,6 +174,37 @@ macro_rules! query { }; } +pub trait ToRevoltError { + #[track_caller] + fn to_internal_error(self) -> Result; +} + +impl ToRevoltError for Result { + fn to_internal_error(self) -> Result { + self.map_err(|_| { + let loc = Location::caller(); + + Error { + error_type: ErrorType::InternalError, + location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column()) + } + }) + } +} + +impl ToRevoltError for Option { + fn to_internal_error(self) -> Result { + self.ok_or_else(|| { + let loc = Location::caller(); + + Error { + error_type: ErrorType::InternalError, + location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column()) + } + }) + } +} + #[cfg(test)] mod tests { use crate::ErrorType; diff --git a/crates/delta/Cargo.toml b/crates/delta/Cargo.toml index e5227d04..e8a4b684 100644 --- a/crates/delta/Cargo.toml +++ b/crates/delta/Cargo.toml @@ -82,6 +82,7 @@ revolt-permissions = { path = "../core/permissions", features = ["schemas"] } # voice livekit-api = "0.3.2" +livekit-protocol = "0.3.2" [build-dependencies] vergen = "7.5.0" diff --git a/crates/delta/src/routes/servers/member_edit.rs b/crates/delta/src/routes/servers/member_edit.rs index 8d478812..941a36ff 100644 --- a/crates/delta/src/routes/servers/member_edit.rs +++ b/crates/delta/src/routes/servers/member_edit.rs @@ -1,5 +1,8 @@ use std::collections::HashSet; +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, @@ -7,7 +10,7 @@ use revolt_database::{ use revolt_models::v0; use revolt_permissions::{calculate_server_permissions, ChannelPermission}; -use revolt_result::{create_error, Result}; +use revolt_result::{create_error, Result, ToRevoltError}; use rocket::{serde::json::Json, State}; use validator::Validate; @@ -17,6 +20,7 @@ use validator::Validate; #[openapi(tag = "Server Members")] #[patch("//members/", data = "")] pub async fn edit( + room_client: &State, db: &State, user: User, server: Reference, @@ -87,6 +91,14 @@ pub async fn edit( permissions.throw_if_lacking_channel_permission(ChannelPermission::TimeoutMembers)?; } + if data.can_publish.is_some() { + permissions.throw_if_lacking_channel_permission(ChannelPermission::MuteMembers)?; + } + + if data.can_receive.is_some() { + permissions.throw_if_lacking_channel_permission(ChannelPermission::DeafenMembers)?; + } + // Resolve our ranking let our_ranking = query.get_member_rank().unwrap_or(i64::MIN); @@ -122,12 +134,16 @@ pub async fn edit( roles, timeout, remove, + can_publish, + can_receive, } = data; let mut partial = PartialMember { nickname, roles, timeout, + can_publish, + can_receive, ..Default::default() }; @@ -155,5 +171,57 @@ pub async fn edit( ) .await?; + if can_publish.is_some() || can_receive.is_some() { + let mut conn = get_connection().await.to_internal_error()?; + + // 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)) + .await + .to_internal_error()? + { + let mut pipeline = Pipeline::new(); + let mut new_perms = ParticipantPermission::default(); + + if let Some(can_publish) = can_publish { + pipeline.set( + format!("can_publish-{}-{}", &channel, &member.id.user), + can_publish, + ); + + new_perms.can_publish = can_publish; + new_perms.can_publish_data = can_publish; + }; + + if let Some(can_receive) = can_receive { + pipeline.set( + format!("can_receive-{}-{}", &channel, &member.id.user), + can_receive, + ); + + new_perms.can_subscribe = can_receive; + }; + + pipeline + .query_async(&mut conn.into_inner()) + .await + .to_internal_error()?; + + room_client + .update_participant( + &channel, + &member.id.user, + UpdateParticipantOptions { + permission: Some(new_perms), + name: "".to_string(), + metadata: "".to_string(), + }, + ) + .await + .to_internal_error()?; + }; + }; + Ok(Json(member.into())) } diff --git a/crates/voice-ingress/src/main.rs b/crates/voice-ingress/src/main.rs index dbe9bbfc..d6b0a813 100644 --- a/crates/voice-ingress/src/main.rs +++ b/crates/voice-ingress/src/main.rs @@ -1,14 +1,16 @@ use std::env; -use redis_kiss::{get_connection, AsyncCommands}; +use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands}; -use revolt_database::{events::client::EventV1, util::reference::Reference, Database, DatabaseInfo}; +use livekit_protocol::WebhookEvent; +use revolt_database::{ + events::client::EventV1, util::reference::Reference, Database, DatabaseInfo, +}; use revolt_models::v0::{PartialUserVoiceState, UserVoiceState}; use rocket::{build, post, routes, serde::json::Json, Config, State}; use rocket_empty::EmptyResponse; -use livekit_protocol::WebhookEvent; -use revolt_result::Result; +use revolt_result::{Result, ToRevoltError}; #[rocket::main] async fn main() -> Result<(), rocket::Error> { @@ -19,7 +21,10 @@ async fn main() -> Result<(), rocket::Error> { let _rocket = build() .manage(database) .mount("/", routes![ingress]) - .configure(Config { port: 8500, ..Default::default() }) + .configure(Config { + port: 8500, + ..Default::default() + }) .ignite() .await? .launch() @@ -28,90 +33,96 @@ async fn main() -> Result<(), rocket::Error> { Ok(()) } - -#[post("/", data="")] +#[post("/", data = "")] async fn ingress(db: &State, body: Json) -> Result { - let mut conn = get_connection().await.unwrap(); + let mut conn = get_connection().await.to_internal_error()?; log::debug!("received event: {body:?}"); - let channel_id = body - .room - .as_ref() - .map(|r| &r.name); + let channel_id = body.room.as_ref().map(|r| &r.name); - let user_id = body - .participant - .as_ref() - .map(|r| &r.identity); + let user_id = body.participant.as_ref().map(|r| &r.identity); match body.event.as_str() { "participant_joined" => { - let channel_id = channel_id.unwrap(); - let user_id = user_id.unwrap(); + let channel_id = channel_id.to_internal_error()?; + let user_id = user_id.to_internal_error()?; let channel = Reference::from_unchecked(channel_id.clone()) .as_channel(db) .await?; + let unique_key = format!( + "{}-{user_id}", + channel.server().unwrap_or_else(|| channel.id()) + ); - let unique_key = format!("{}-{user_id}", channel.server().unwrap_or_else(|| channel.id())); - - conn.sadd::<_, _, u64>(format!("vc-members-{channel_id}"), user_id).await.unwrap(); - - conn.set::<_, _, String>(format!("vc-{unique_key}"), &channel_id).await.unwrap(); - conn.set::<_, _, String>(format!("audio-{unique_key}"), true).await.unwrap(); - conn.set::<_, _, String>(format!("deafened-{unique_key}"), false).await.unwrap(); - conn.set::<_, _, String>(format!("screensharing-{unique_key}"), false).await.unwrap(); - conn.set::<_, _, String>(format!("camera-{unique_key}"), false).await.unwrap(); + Pipeline::new() + .sadd(format!("vc-members-{channel_id}"), user_id) + .set(format!("vc-{unique_key}"), channel_id) + .set(format!("publish-{unique_key}"), true) + .set(format!("subscribe-{unique_key}"), false) + .set(format!("screensharing-{unique_key}"), false) + .set(format!("camera-{unique_key}"), false) + .query_async(&mut conn.into_inner()) + .await + .to_internal_error()?; let voice_state = UserVoiceState { id: user_id.clone(), - audio: false, - deafened: false, + can_receive: false, + can_publish: false, screensharing: false, - camera: false + camera: false, }; EventV1::VoiceChannelJoin { id: channel_id.clone(), - state: voice_state + state: voice_state, } .p(channel_id.clone()) .await - }, + } "participant_left" => { - let channel_id = channel_id.unwrap(); - let user_id = user_id.unwrap(); + let channel_id = channel_id.to_internal_error()?; + let user_id = user_id.to_internal_error()?; let channel = Reference::from_unchecked(channel_id.clone()) .as_channel(db) .await?; - conn.srem::<_, _, u64>(format!("vc-members-{channel_id}"), user_id).await.unwrap(); + conn.srem::<_, _, u64>(format!("vc-members-{channel_id}"), user_id) + .await + .to_internal_error()?; - let unique_key = format!("{}-{user_id}", channel.server().unwrap_or_else(|| channel.id())); + let unique_key = format!( + "{}-{user_id}", + channel.server().unwrap_or_else(|| channel.id()) + ); - conn.del::<_, u64>(format!("vc-{unique_key}")).await.unwrap(); - - conn.del::<_, u64>(format!("audio-{unique_key}")).await.unwrap(); - conn.del::<_, u64>(format!("deafened-{unique_key}")).await.unwrap(); - conn.del::<_, u64>(format!("screensharing-{unique_key}")).await.unwrap(); - conn.del::<_, u64>(format!("camera-{unique_key}")).await.unwrap(); + conn.del::<_, Vec>(&[ + format!("vc-{unique_key}"), + format!("audio-{unique_key}"), + format!("deafened-{unique_key}"), + format!("screensharing-{unique_key}"), + format!("camera-{unique_key}"), + ]) + .await + .to_internal_error()?; EventV1::VoiceChannelLeave { id: channel_id.clone(), - user: user_id.clone() + user: user_id.clone(), } .p(channel_id.clone()) .await - }, + } "track_published" | "track_unpublished" => { let value = body.event == "track_published"; // to avoid duplicating this entire case twice - let channel_id = channel_id.unwrap(); - let user_id = user_id.unwrap(); - let track = body.track.as_ref().unwrap(); + 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) @@ -122,52 +133,62 @@ async fn ingress(db: &State, body: Json) -> Result { - PartialUserVoiceState::default() - } - /* TrackSource::Camera */ 1 => { - conn.set::<_, _, String>(format!("camera-{unique_key}"), value).await.unwrap(); + /* 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!("audio-{unique_key}"), value).await.unwrap(); + /* TrackSource::Microphone */ + 2 => { + conn.set::<_, _, String>(format!("audio-{unique_key}"), value) + .await + .to_internal_error()?; PartialUserVoiceState { - audio: Some(value), + can_publish: Some(value), ..Default::default() } - }, - /* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */ 3 | 4 => { - conn.set::<_, _, String>(format!("screensharing-{unique_key}"), value).await.unwrap(); + } + /* 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!() + } + _ => unreachable!(), }; EventV1::UserVoiceStateUpdate { id: user_id.clone(), channel_id: channel_id.clone(), - data: partial + data: partial, } .p(channel_id.clone()) .await; - }, + } _ => {} }; Ok(EmptyResponse) -} \ No newline at end of file +}