more stuff

Cargo.lock

@@ -4103,6 +4103,7 @@ dependencies = [
  "lettre",
  "linkify 0.6.0",
  "livekit-api",
+ "livekit-protocol",
  "log",
  "lru 0.7.8",
  "nanoid",

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);
             }

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::<Document>("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::<Document>("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.

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,
+    events::client::EventV1, if_false, util::permissions::DatabasePermissionQuery, Channel,
-    Server, SystemMessage, User,
+    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<Timestamp>,
+
+        /// 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,
         }
     }
 }

crates/core/database/src/util/bridge/v0.rs

@@ -186,7 +186,7 @@ impl From<crate::Channel> for Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice
+                voice,
             } => Channel::TextChannel {
                 id,
                 server,
@@ -197,7 +197,7 @@ impl From<crate::Channel> for Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice
+                voice,
             },
             crate::Channel::VoiceChannel {
                 id,
@@ -268,7 +268,7 @@ impl From<Channel> for crate::Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice
+                voice,
             } => crate::Channel::TextChannel {
                 id,
                 server,
@@ -279,7 +279,7 @@ impl From<Channel> for crate::Channel {
                 default_permissions,
                 role_permissions,
                 nsfw,
-                voice
+                voice,
             },
             Channel::VoiceChannel {
                 id,
@@ -614,6 +614,8 @@ impl From<crate::Member> 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<Member> 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<crate::PartialMember> 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<PartialMember> 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,
         }
     }
 }

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<Timestamp>,
+
+        /// 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<Vec<FieldsMember>>,
+        /// server-wide voice muted
+        pub can_publish: Option<bool>,
+        /// server-wide voice deafened
+        pub can_receive: Option<bool>,
     }
 );

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 can_receive: bool,
-        pub deafened: bool,
+        pub can_publish: bool,
         pub screensharing: bool,
-        pub camera: bool
+        pub camera: bool,
     },
     "PartialUserVoiceState"
 );

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<T> {
+    #[track_caller]
+    fn to_internal_error(self) -> Result<T, Error>;
+}
+
+impl<T, E> ToRevoltError<T> for Result<T, E> {
+    fn to_internal_error(self) -> Result<T, Error> {
+        self.map_err(|_| {
+            let loc = Location::caller();
+
+            Error {
+                error_type: ErrorType::InternalError,
+                location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())
+            }
+        })
+    }
+}
+
+impl<T> ToRevoltError<T> for Option<T> {
+    fn to_internal_error(self) -> Result<T, Error> {
+        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;

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"

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("/<server>/members/<target>", data = "<data>")]
 pub async fn edit(
+    room_client: &State<RoomClient>,
     db: &State<Database>,
     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<String>>(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()))
 }

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 = "<body>")]
-#[post("/", data="<body>")]
 async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<EmptyResponse> {
-    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
+    let channel_id = body.room.as_ref().map(|r| &r.name);
-        .room
-        .as_ref()
-        .map(|r| &r.name);
 
-    let user_id = body
+    let user_id = body.participant.as_ref().map(|r| &r.identity);
-        .participant
-        .as_ref()
-        .map(|r| &r.identity);
 
     match body.event.as_str() {
         "participant_joined" => {
-            let channel_id = channel_id.unwrap();
+            let channel_id = channel_id.to_internal_error()?;
-            let user_id = user_id.unwrap();
+            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()));
+            Pipeline::new()
-
+                .sadd(format!("vc-members-{channel_id}"), user_id)
-            conn.sadd::<_, _, u64>(format!("vc-members-{channel_id}"), user_id).await.unwrap();
+                .set(format!("vc-{unique_key}"), channel_id)
-
+                .set(format!("publish-{unique_key}"), true)
-            conn.set::<_, _, String>(format!("vc-{unique_key}"), &channel_id).await.unwrap();
+                .set(format!("subscribe-{unique_key}"), false)
-            conn.set::<_, _, String>(format!("audio-{unique_key}"), true).await.unwrap();
+                .set(format!("screensharing-{unique_key}"), false)
-            conn.set::<_, _, String>(format!("deafened-{unique_key}"), false).await.unwrap();
+                .set(format!("camera-{unique_key}"), false)
-            conn.set::<_, _, String>(format!("screensharing-{unique_key}"), false).await.unwrap();
+                .query_async(&mut conn.into_inner())
-            conn.set::<_, _, String>(format!("camera-{unique_key}"), false).await.unwrap();
+                .await
+                .to_internal_error()?;
 
             let voice_state = UserVoiceState {
                 id: user_id.clone(),
-                audio: false,
+                can_receive: false,
-                deafened: 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 channel_id = channel_id.to_internal_error()?;
-            let user_id = user_id.unwrap();
+            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::<_, Vec<u64>>(&[
-
+                format!("vc-{unique_key}"),
-            conn.del::<_, u64>(format!("audio-{unique_key}")).await.unwrap();
+                format!("audio-{unique_key}"),
-            conn.del::<_, u64>(format!("deafened-{unique_key}")).await.unwrap();
+                format!("deafened-{unique_key}"),
-            conn.del::<_, u64>(format!("screensharing-{unique_key}")).await.unwrap();
+                format!("screensharing-{unique_key}"),
-            conn.del::<_, u64>(format!("camera-{unique_key}")).await.unwrap();
+                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 channel_id = channel_id.to_internal_error()?;
-            let user_id = user_id.unwrap();
+            let user_id = user_id.to_internal_error()?;
-            let track = body.track.as_ref().unwrap();
+            let track = body.track.as_ref().to_internal_error()?;
 
             let user = Reference::from_unchecked(user_id.clone())
                 .as_user(db)
@@ -122,50 +133,60 @@ async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<Empty
                 .await?;
 
             let unique_key = if user.bot.is_some() {
-                format!("{}-{user_id}", channel.server().unwrap_or_else(|| channel.id()))
+                format!(
+                    "{}-{user_id}",
+                    channel.server().unwrap_or_else(|| channel.id())
+                )
             } else {
                 user_id.to_string()
             };
 
             let partial = match track.source {
-                /* TrackSource::Unknown */ 0 => {
+                /* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(),
-                    PartialUserVoiceState::default()
+                /* TrackSource::Camera */
-                }
+                1 => {
-                /* TrackSource::Camera */ 1 => {
+                    conn.set::<_, _, String>(format!("camera-{unique_key}"), value)
-                    conn.set::<_, _, String>(format!("camera-{unique_key}"), value).await.unwrap();
+                        .await
+                        .to_internal_error()?;
 
                     PartialUserVoiceState {
                         camera: Some(value),
                         ..Default::default()
                     }
                 }
-                /* TrackSource::Microphone */ 2 => {
+                /* TrackSource::Microphone */
-                    conn.set::<_, _, String>(format!("audio-{unique_key}"), value).await.unwrap();
+                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 => {
+                /* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */
-                    conn.set::<_, _, String>(format!("screensharing-{unique_key}"), value).await.unwrap();
+                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;
-        },
+        }
         _ => {}
     };