track current state

crates/bonfire/src/events/impl.rs

@@ -133,12 +133,8 @@ impl State {
         // Append known user IDs from DMs.
         for channel in &channels {
             match channel {
-                Channel::DirectMessage { id, recipients, .. } | Channel::Group { id, recipients, .. } => {
+                Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
                     user_ids.extend(&mut recipients.clone().into_iter());
-
-                    if let Ok(Some(voice_state)) = self.fetch_voice_state(&mut conn, id).await {
-                        voice_states.push(voice_state)
-                    }
                 }
                 _ => {}
             }
@@ -206,11 +202,9 @@ impl State {
             self.insert_subscription(channel.id().to_string());
         }
 
-        for server in &servers {
+        for channel in &channels {
-            for channel in &server.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(&mut conn, channel).await {
+                voice_states.push(voice_state)
-                    voice_states.push(voice_state)
-                }
             }
         }
 
@@ -220,7 +214,7 @@ impl State {
             channels: channels.into_iter().map(Into::into).collect(),
             members: members.into_iter().map(Into::into).collect(),
             emojis: emojis.into_iter().map(Into::into).collect(),
-            voice_states
+            voice_states,
         })
     }
 
@@ -568,15 +562,44 @@ impl State {
         true
     }
 
-    async fn fetch_voice_state(&self, conn: &mut Conn, channel: &str) -> Result<Option<v0::ChannelVoiceState>> {
+    async fn fetch_voice_state(
-        let members = conn.smembers::<_, Vec<String>>(format!("vc-members-{channel}"))
+        &self,
+        conn: &mut Conn,
+        channel: &Channel,
+    ) -> Result<Option<v0::ChannelVoiceState>> {
+        let members = conn
+            .smembers::<_, Vec<String>>(format!("vc-members-{}", channel.id()))
             .await
             .map_err(|_| create_error!(InternalError))?;
 
+        let channel_or_server_id = channel.server().unwrap_or_else(|| channel.id());
+
+
         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}");
+
+                let (audio, deafened, screensharing, camera) = conn
+                    .mget::<_, (bool, bool, bool, bool)>(&[
+                        format!("audio-{unique_key}"),
+                        format!("deafened-{unique_key}"),
+                        format!("screensharing-{unique_key}"),
+                        format!("camera-{unique_key}")
+
+                    ])
+                    .await
+                    .map_err(|_| create_error!(InternalError))?;
+
+                let voice_state = v0::UserVoiceState { id, audio, deafened, screensharing, camera };
+
+                participants.push(voice_state);
+            }
+
             Ok(Some(v0::ChannelVoiceState {
-                id: channel.to_string(),
+                id: channel.id(),
-                participants: members.into_iter().map(|id| v0::UserVoiceState { id }).collect()
+                participants,
             }))
         } else {
             Ok(None)

crates/core/database/src/events/client.rs

@@ -2,7 +2,7 @@ use authifier::AuthifierEvent;
 use serde::{Deserialize, Serialize};
 
 use revolt_models::v0::{
-    AppendMessage, Channel, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialWebhook, Report, Server, User, UserSettings, UserVoiceState, Webhook
+    AppendMessage, Channel, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, Report, Server, User, UserSettings, UserVoiceState, Webhook
 };
 use revolt_result::Error;
 
@@ -232,6 +232,11 @@ pub enum EventV1 {
     VoiceChannelLeave {
         id: String,
         user: String,
+    },
+    UserVoiceStateUpdate {
+        id: String,
+        channel_id: String,
+        data: PartialUserVoiceState
     }
 }
 

crates/core/database/src/models/channels/model.rs

@@ -430,6 +430,14 @@ impl Channel {
         }
     }
 
+    /// Clone this channel's server id
+    pub fn server(&self) -> Option<String> {
+        match self {
+            Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => Some(server.clone()),
+            _ => None
+        }
+    }
+
     /// Set role permission on a channel
     pub async fn set_role_permission(
         &mut self,

crates/core/database/src/models/users/model.rs

@@ -569,10 +569,14 @@ impl User {
     }
 
     /// Gets current voice channel
-    pub async fn current_voice_channel(&self) -> Result<Option<String>> {
+    ///
-        let mut conn = get_connection().await.map_err(|_| create_error!(InternalError))?;
+    /// current_context: Either the channel id if a dm or group, or server_id if in a server
+    pub async fn current_voice_channel(&self, current_context: &str) -> Result<Option<String>> {
+        let mut conn = get_connection()
+            .await
+            .map_err(|_| create_error!(InternalError))?;
 
-        conn.get::<_, Option<String>>(format!("vc-{}", &self.id))
+        conn.get::<_, Option<String>>(format!("vc-{}-{current_context}", &self.id))
             .await
             .map_err(|_| create_error!(InternalError))
     }

crates/core/models/src/v0/users.rs

@@ -270,13 +270,18 @@ auto_derived!(
         /// Username and discriminator combo separated by #
         pub username: String,
     }
+);
 
+auto_derived_partial!(
     /// Voice State information for a user
     pub struct UserVoiceState {
-        pub id: String
+        pub id: String,
-        // TODO - muted, etc
+        pub audio: bool,
-    }
+        pub deafened: bool,
-
+        pub screensharing: bool,
+        pub camera: bool
+    },
+    "PartialUserVoiceState"
 );
 
 pub trait CheckRelationship {

crates/delta/src/routes/channels/voice_join.rs

@@ -19,13 +19,17 @@ pub async fn call(db: &State<Database>, rooms: &State<RoomClient>, user: User, t
 
     let mut permissions = perms(db, &user).channel(&channel);
 
-    calculate_channel_permissions(&mut permissions)
+    let current_permissions = calculate_channel_permissions(&mut permissions).await;
-        .await
+    current_permissions.throw_if_lacking_channel_permission(ChannelPermission::Connect)?;
-        .throw_if_lacking_channel_permission(ChannelPermission::Connect)?;
 
-    if user.current_voice_channel().await?.is_some() {
+    // TODO - decide if we should allow being in multiple voice channels for users
-        return Err(create_error!(AlreadyInVoiceChannel))
+
-    }
+    // if user.current_voice_channel(&channel.server().unwrap_or_else(|| channel.id()))
+    //     .await?
+    //     .is_some()
+    // {
+    //     return Err(create_error!(AlreadyInVoiceChannel))
+    // }
 
     let config = config().await;
 
@@ -39,12 +43,23 @@ pub async fn call(db: &State<Database>, rooms: &State<RoomClient>, user: User, t
         _ => 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()
         })

crates/voice-ingress/src/main.rs

@@ -2,31 +2,36 @@ use std::env;
 
 use redis_kiss::{get_connection, AsyncCommands};
 
-use revolt_database::events::client::EventV1;
+use revolt_database::{events::client::EventV1, util::reference::Reference, Database, DatabaseInfo};
-use revolt_models::v0::UserVoiceState;
+use revolt_models::v0::{PartialUserVoiceState, UserVoiceState};
-use rocket::{build, post, routes, serde::json::Json, Config};
+use rocket::{build, post, routes, serde::json::Json, Config, State};
 use rocket_empty::EmptyResponse;
 use livekit_protocol::WebhookEvent;
 
-use revolt_config::Database;
 use revolt_result::Result;
 
-#[async_std::main]
+#[rocket::main]
-async fn main() {
+async fn main() -> Result<(), rocket::Error> {
     revolt_config::configure!(voice_ingress);
 
-    let rocket = build()
+    let database = DatabaseInfo::Auto.connect().await.unwrap();
-        .manage(get_connection().await.unwrap())
+
+    let _rocket = build()
+        .manage(database)
         .mount("/", routes![ingress])
         .configure(Config { port: 8500, ..Default::default() })
+        .ignite()
+        .await?
         .launch()
-        .await
+        .await?;
-        .unwrap();
+
+    Ok(())
 }
 
+
 #[post("/", data="<body>")]
-async fn ingress(body: Json<WebhookEvent>) -> Result<EmptyResponse> {
+async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<EmptyResponse> {
-    let mut redis = get_connection().await.unwrap();
+    let mut conn = get_connection().await.unwrap();
 
     log::debug!("received event: {body:?}");
 
@@ -45,11 +50,27 @@ async fn ingress(body: Json<WebhookEvent>) -> Result<EmptyResponse> {
             let channel_id = channel_id.unwrap();
             let user_id = user_id.unwrap();
 
-            redis.sadd::<_, _, u64>(format!("vc-members-{}", channel_id), user_id).await.unwrap();
+            let channel = Reference::from_unchecked(channel_id.clone())
-            redis.set::<_, _, String>(format!("vc-{}", user_id), &channel_id).await.unwrap();
+                .as_channel(db)
+                .await?;
+
+
+            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();
 
             let voice_state = UserVoiceState {
-                id: user_id.clone()
+                id: user_id.clone(),
+                audio: false,
+                deafened: false,
+                screensharing: false,
+                camera: false
             };
 
             EventV1::VoiceChannelJoin {
@@ -63,8 +84,20 @@ async fn ingress(body: Json<WebhookEvent>) -> Result<EmptyResponse> {
             let channel_id = channel_id.unwrap();
             let user_id = user_id.unwrap();
 
-            redis.srem::<_, _, u64>(format!("vc-members-{}", channel_id), user_id).await.unwrap();
+            let channel = Reference::from_unchecked(channel_id.clone())
-            redis.del::<_, u64>(format!("vc-{}", user_id)).await.unwrap();
+                .as_channel(db)
+                .await?;
+
+            conn.srem::<_, _, u64>(format!("vc-members-{channel_id}"), user_id).await.unwrap();
+
+            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();
 
             EventV1::VoiceChannelLeave {
                 id: channel_id.clone(),
@@ -73,6 +106,66 @@ async fn ingress(body: Json<WebhookEvent>) -> Result<EmptyResponse> {
             .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 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.unwrap();
+
+                    PartialUserVoiceState {
+                        camera: Some(value),
+                        ..Default::default()
+                    }
+                }
+                /* TrackSource::Microphone */ 2 => {
+                    conn.set::<_, _, String>(format!("audio-{unique_key}"), value).await.unwrap();
+
+                    PartialUserVoiceState {
+                        audio: Some(value),
+                        ..Default::default()
+                    }
+                },
+                /* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */ 3 | 4 => {
+                    conn.set::<_, _, String>(format!("screensharing-{unique_key}"), value).await.unwrap();
+
+                    PartialUserVoiceState {
+                        screensharing: Some(value),
+                        ..Default::default()
+                    }
+                },
+                _ => unreachable!()
+            };
+
+            EventV1::UserVoiceStateUpdate {
+                id: user_id.clone(),
+                channel_id: channel_id.clone(),
+                data: partial
+            }
+            .p(channel_id.clone())
+            .await;
+        },
         _ => {}
     };