From 7a4421089ce5a44f58d4a999a2d43b2b423395c7 Mon Sep 17 00:00:00 2001 From: Zomatree Date: Thu, 11 Apr 2024 01:07:03 +0100 Subject: [PATCH] track current state --- crates/bonfire/src/events/impl.rs | 53 +++++--- crates/core/database/src/events/client.rs | 7 +- .../database/src/models/channels/model.rs | 8 ++ .../core/database/src/models/users/model.rs | 10 +- crates/core/models/src/v0/users.rs | 13 +- .../delta/src/routes/channels/voice_join.rs | 27 +++- crates/voice-ingress/src/main.rs | 127 +++++++++++++++--- 7 files changed, 199 insertions(+), 46 deletions(-) diff --git a/crates/bonfire/src/events/impl.rs b/crates/bonfire/src/events/impl.rs index b7e90082..afe5877b 100644 --- a/crates/bonfire/src/events/impl.rs +++ b/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 &server.channels { - if let Ok(Some(voice_state)) = self.fetch_voice_state(&mut conn, channel).await { - voice_states.push(voice_state) - } + for channel in &channels { + if let Ok(Some(voice_state)) = self.fetch_voice_state(&mut conn, channel).await { + 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> { - let members = conn.smembers::<_, Vec>(format!("vc-members-{channel}")) + 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()); + + 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(), - participants: members.into_iter().map(|id| v0::UserVoiceState { id }).collect() + id: channel.id(), + participants, })) } else { Ok(None) diff --git a/crates/core/database/src/events/client.rs b/crates/core/database/src/events/client.rs index feab64d4..473c5cb3 100644 --- a/crates/core/database/src/events/client.rs +++ b/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 } } diff --git a/crates/core/database/src/models/channels/model.rs b/crates/core/database/src/models/channels/model.rs index 2b671c38..77599993 100644 --- a/crates/core/database/src/models/channels/model.rs +++ b/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 { + 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, diff --git a/crates/core/database/src/models/users/model.rs b/crates/core/database/src/models/users/model.rs index 624cbefc..7d52ed38 100644 --- a/crates/core/database/src/models/users/model.rs +++ b/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> { - 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> { + let mut conn = get_connection() + .await + .map_err(|_| create_error!(InternalError))?; - conn.get::<_, Option>(format!("vc-{}", &self.id)) + conn.get::<_, Option>(format!("vc-{}-{current_context}", &self.id)) .await .map_err(|_| create_error!(InternalError)) } diff --git a/crates/core/models/src/v0/users.rs b/crates/core/models/src/v0/users.rs index 1e6c4fff..50d8163c 100644 --- a/crates/core/models/src/v0/users.rs +++ b/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 - // TODO - muted, etc - } - + pub id: String, + pub audio: bool, + pub deafened: bool, + pub screensharing: bool, + pub camera: bool + }, + "PartialUserVoiceState" ); pub trait CheckRelationship { diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index e6afd62b..08330b80 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -19,13 +19,17 @@ pub async fn call(db: &State, rooms: &State, user: User, t let mut permissions = perms(db, &user).channel(&channel); - calculate_channel_permissions(&mut permissions) - .await - .throw_if_lacking_channel_permission(ChannelPermission::Connect)?; + let current_permissions = calculate_channel_permissions(&mut permissions).await; + current_permissions.throw_if_lacking_channel_permission(ChannelPermission::Connect)?; - if user.current_voice_channel().await?.is_some() { - return Err(create_error!(AlreadyInVoiceChannel)) - } + // TODO - decide if we should allow being in multiple voice channels for users + + // 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, rooms: &State, 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() }) diff --git a/crates/voice-ingress/src/main.rs b/crates/voice-ingress/src/main.rs index 6fc6c0d1..dbe9bbfc 100644 --- a/crates/voice-ingress/src/main.rs +++ b/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_models::v0::UserVoiceState; -use rocket::{build, post, routes, serde::json::Json, Config}; +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_config::Database; use revolt_result::Result; -#[async_std::main] -async fn main() { +#[rocket::main] +async fn main() -> Result<(), rocket::Error> { revolt_config::configure!(voice_ingress); - let rocket = build() - .manage(get_connection().await.unwrap()) + let database = DatabaseInfo::Auto.connect().await.unwrap(); + + let _rocket = build() + .manage(database) .mount("/", routes![ingress]) .configure(Config { port: 8500, ..Default::default() }) + .ignite() + .await? .launch() - .await - .unwrap(); + .await?; + + Ok(()) } + #[post("/", data="")] -async fn ingress(body: Json) -> Result { - let mut redis = get_connection().await.unwrap(); +async fn ingress(db: &State, body: Json) -> Result { + let mut conn = get_connection().await.unwrap(); log::debug!("received event: {body:?}"); @@ -45,11 +50,27 @@ async fn ingress(body: Json) -> Result { let channel_id = channel_id.unwrap(); let user_id = user_id.unwrap(); - redis.sadd::<_, _, u64>(format!("vc-members-{}", channel_id), user_id).await.unwrap(); - redis.set::<_, _, String>(format!("vc-{}", user_id), &channel_id).await.unwrap(); + 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())); + + 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) -> Result { let channel_id = channel_id.unwrap(); let user_id = user_id.unwrap(); - redis.srem::<_, _, u64>(format!("vc-members-{}", channel_id), user_id).await.unwrap(); - redis.del::<_, u64>(format!("vc-{}", user_id)).await.unwrap(); + let channel = Reference::from_unchecked(channel_id.clone()) + .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) -> Result { .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; + }, _ => {} };