diff --git a/crates/core/config/Revolt.toml b/crates/core/config/Revolt.toml index e3e8ec97..4c7ca17d 100644 --- a/crates/core/config/Revolt.toml +++ b/crates/core/config/Revolt.toml @@ -233,6 +233,18 @@ message_attachments = 5 # Maximum number of servers the user can create/join servers = 50 +# Maximum audio frequency (Hz) in voice calls +voice_quality = 16000 + +# Whether the user can use video streams in voice calls +video = true + +# Mamimum resolution (width, height) of video streams in voice calls +video_resolution = [1080, 720] + +# Minimum and maximum aspect ratio of video streams in voice calls +video_aspect_ratio = [0.3, 2.5] + [features.limits.new_user.file_upload_size_limit] # Maximum file size limits (in bytes) attachments = 20_000_000 @@ -260,6 +272,18 @@ message_attachments = 5 # Maximum number of servers the user can create/join servers = 100 +# Maximum audio frequency (Hz) in voice calls +voice_quality = 16000 + +# Whether the user can use video streams in voice calls +video = true + +# Mamimum resolution (width, height) of video streams in voice calls +video_resolution = [1080, 720] + +# Minimum and maximum aspect ratio of video streams in voice calls +video_aspect_ratio = [0.3, 2.5] + [features.limits.default.file_upload_size_limit] # Maximum file size limits (in bytes) attachments = 20_000_000 diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 2371aa7e..5d8c0998 100644 --- a/crates/core/config/src/lib.rs +++ b/crates/core/config/src/lib.rs @@ -333,6 +333,10 @@ pub struct FeaturesLimits { pub message_length: usize, pub message_attachments: usize, pub servers: usize, + pub voice_quality: u32, + pub video: bool, + pub video_resolution: [u32; 2], + pub video_aspect_ratio: [f32; 2], pub file_upload_size_limit: HashMap, } diff --git a/crates/core/database/src/models/channels/model.rs b/crates/core/database/src/models/channels/model.rs index 1c1edafd..9772d429 100644 --- a/crates/core/database/src/models/channels/model.rs +++ b/crates/core/database/src/models/channels/model.rs @@ -2,7 +2,7 @@ use std::{borrow::Cow, collections::HashMap}; use revolt_config::config; -use revolt_models::v0::{self, MessageAuthor, VoiceInformation}; +use revolt_models::v0::{self, MessageAuthor}; use revolt_permissions::OverrideField; use revolt_result::Result; use serde::{Deserialize, Serialize}; @@ -141,6 +141,13 @@ auto_derived!( nsfw: bool, }, } + + #[derive(Default)] + pub struct VoiceInformation { + /// Maximium amount of users allowed in the voice channel at once + #[serde(skip_serializing_if = "Option::is_none")] + pub max_users: Option + } ); auto_derived!( @@ -167,7 +174,7 @@ auto_derived!( #[serde(skip_serializing_if = "Option::is_none")] pub last_message_id: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub voice: Option + pub voice: Option, } /// Optional fields on channel object @@ -226,7 +233,7 @@ impl Channel { default_permissions: None, role_permissions: HashMap::new(), nsfw: data.nsfw.unwrap_or(false), - voice: data.voice + voice: data.voice.map(|voice| voice.into()) }, v0::LegacyServerChannelType::Voice => Channel::TextChannel { id: id.clone(), @@ -238,7 +245,7 @@ impl Channel { default_permissions: None, role_permissions: HashMap::new(), nsfw: data.nsfw.unwrap_or(false), - voice: Some(data.voice.unwrap_or_default()) + voice: Some(data.voice.unwrap_or_default().into()) }, }; @@ -455,7 +462,7 @@ 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::DirectMessage { .. } | Channel::VoiceChannel { .. } => Some(Cow::Owned(VoiceInformation::default())), Self::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)), _ => None } diff --git a/crates/core/database/src/models/messages/model.rs b/crates/core/database/src/models/messages/model.rs index 458eb927..33190def 100644 --- a/crates/core/database/src/models/messages/model.rs +++ b/crates/core/database/src/models/messages/model.rs @@ -113,7 +113,7 @@ auto_derived!( #[serde(rename = "message_unpinned")] MessageUnpinned { id: String, by: String }, #[serde(rename = "call_started")] - CallStarted { by: String }, + CallStarted { by: String, finished_at: Option }, } /// Name and / or avatar override information @@ -833,7 +833,7 @@ impl Message { v0::SystemMessage::MessageUnpinned { by, .. } => { users.push(by.clone()); } - v0::SystemMessage::CallStarted { by } => { + v0::SystemMessage::CallStarted { by, .. } => { users.push(by.clone()) } } diff --git a/crates/core/database/src/util/bridge/v0.rs b/crates/core/database/src/util/bridge/v0.rs index a4ab2044..6e60fe62 100644 --- a/crates/core/database/src/util/bridge/v0.rs +++ b/crates/core/database/src/util/bridge/v0.rs @@ -200,7 +200,7 @@ impl From for Channel { default_permissions, role_permissions, nsfw, - voice, + voice: voice.map(|voice| voice.into()), }, crate::Channel::VoiceChannel { id, @@ -283,7 +283,7 @@ impl From for crate::Channel { default_permissions, role_permissions, nsfw, - voice, + voice: voice.map(|voice| voice.into()), }, Channel::VoiceChannel { id, @@ -321,7 +321,7 @@ impl From for PartialChannel { role_permissions: value.role_permissions, default_permissions: value.default_permissions, last_message_id: value.last_message_id, - voice: value.voice + voice: value.voice.map(|voice| voice.into()) } } } @@ -339,7 +339,7 @@ impl From for crate::PartialChannel { role_permissions: value.role_permissions, default_permissions: value.default_permissions, last_message_id: value.last_message_id, - voice: value.voice + voice: value.voice.map(|voice| voice.into()) } } } @@ -551,7 +551,7 @@ impl From for SystemMessage { crate::SystemMessage::UserRemove { id, by } => Self::UserRemove { id, by }, crate::SystemMessage::MessagePinned { id, by } => Self::MessagePinned { id, by }, crate::SystemMessage::MessageUnpinned { id, by } => Self::MessageUnpinned { id, by }, - crate::SystemMessage::CallStarted { by } => Self::CallStarted { by } + crate::SystemMessage::CallStarted { by, finished_at } => Self::CallStarted { by, finished_at } } } } @@ -1406,3 +1406,19 @@ impl From for crate::FieldsMessage { } } } + +impl From for crate::VoiceInformation { + fn from(value: VoiceInformation) -> Self { + crate::VoiceInformation { + max_users: value.max_users + } + } +} + +impl From for VoiceInformation { + fn from(value: crate::VoiceInformation) -> Self { + VoiceInformation { + max_users: value.max_users + } + } +} \ No newline at end of file diff --git a/crates/core/database/src/voice/mod.rs b/crates/core/database/src/voice/mod.rs index ea41a3fd..37a42dc4 100644 --- a/crates/core/database/src/voice/mod.rs +++ b/crates/core/database/src/voice/mod.rs @@ -1,7 +1,12 @@ - +use crate::{ + events::client::EventV1, + models::{Channel, User}, + util::{permissions::DatabasePermissionQuery, reference::Reference}, + Database, Server, +}; use livekit_protocol::ParticipantPermission; -use redis_kiss::{get_connection as _get_connection, Conn, redis::Pipeline, AsyncCommands}; -use crate::{events::client::EventV1, models::{Channel, User}, util::{permissions::DatabasePermissionQuery, reference::Reference}, Database, Server}; +use redis_kiss::{get_connection as _get_connection, redis::Pipeline, AsyncCommands, Conn}; +use revolt_config::FeaturesLimits; use revolt_models::v0::{self, PartialUserVoiceState, UserVoiceState}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue}; use revolt_result::{create_error, Result, ToRevoltError}; @@ -59,7 +64,35 @@ pub async fn get_user_voice_channels(user_id: &str) -> Result> { .to_internal_error() } -pub async fn set_user_moved_from_voice(old_channel: &str, new_channel: &str, user_id: &str) -> Result<()> { +pub async fn set_user_moved_from_voice( + old_channel: &str, + new_channel: &str, + user_id: &str, +) -> Result<()> { + get_connection() + .await? + .set_ex( + format!("moved_from:{user_id}:{old_channel}"), + new_channel, + 10, + ) + .await + .to_internal_error() +} + +pub async fn get_user_moved_from_voice(channel_id: &str, user_id: &str) -> Result> { + get_connection() + .await? + .get(format!("moved_from:{user_id}:{channel_id}")) + .await + .to_internal_error() +} + +pub async fn set_user_moved_to_voice( + new_channel: &str, + old_channel: &str, + user_id: &str, +) -> Result<()> { get_connection() .await? .set_ex(format!("moved_to:{user_id}:{new_channel}"), old_channel, 10) @@ -67,7 +100,7 @@ pub async fn set_user_moved_from_voice(old_channel: &str, new_channel: &str, use .to_internal_error() } -pub async fn get_user_moved_from_voice(channel_id: &str, user_id: &str) -> Result> { +pub async fn get_user_moved_to_voice(channel_id: &str, user_id: &str) -> Result> { get_connection() .await? .get(format!("moved_to:{user_id}:{channel_id}")) @@ -78,7 +111,7 @@ pub async fn get_user_moved_from_voice(channel_id: &str, user_id: &str) -> Resul pub async fn is_in_voice_channel(user_id: &str, channel_id: &str) -> Result { get_connection() .await? - .sismember(format!("vc:{}", user_id), channel_id) + .sismember(format!("vc:{user_id}"), channel_id) .await .to_internal_error() } @@ -89,21 +122,21 @@ pub async fn get_user_voice_channel_in_server( ) -> Result> { let mut conn = get_connection().await?; - let unique_key = format!("{}:{}", user_id, server_id); + let unique_key = format!("{user_id}:{server_id}"); conn.get::<&str, Option>(&unique_key) .await .to_internal_error() } -pub fn get_allowed_sources(permissions: PermissionValue) -> Vec<&'static str> { +pub fn get_allowed_sources(limits: &FeaturesLimits, permissions: PermissionValue) -> Vec<&'static str> { let mut allowed_sources = Vec::new(); if permissions.has(ChannelPermission::Speak as u64) { allowed_sources.push("microphone") }; - if permissions.has(ChannelPermission::Video as u64) { + if permissions.has(ChannelPermission::Video as u64) && limits.video { allowed_sources.extend(["camera", "screen_share", "screen_share_audio"]); }; @@ -238,14 +271,10 @@ pub async fn update_voice_state( pub async fn get_voice_channel_members(channel_id: &str) -> Result>> { get_connection() .await? - .smembers::<_, Option>>(format!("vc_members:{}", channel_id)) + .smembers::<_, Option>>(format!("vc_members:{channel_id}")) .await .to_internal_error() - .map(|opt| opt.and_then(|v| if v.is_empty() { - None - } else { - Some(v) - })) + .map(|opt| opt.and_then(|v| if v.is_empty() { None } else { Some(v) })) } pub async fn get_voice_state( @@ -294,13 +323,13 @@ pub async fn get_channel_voice_state(channel: &Channel) -> Result Result<()> { .to_internal_error() } -pub async fn sync_voice_permissions(db: &Database, voice_client: &VoiceClient, channel: &Channel, server: Option<&Server>, role_id: Option<&str>) -> Result<()> { +pub async fn sync_voice_permissions( + db: &Database, + voice_client: &VoiceClient, + channel: &Channel, + server: Option<&Server>, + role_id: Option<&str>, +) -> Result<()> { let node = get_channel_node(channel.id()).await?.unwrap(); - for user_id in get_voice_channel_members(channel.id()).await?.iter().flatten() { - let user = Reference::from_unchecked(user_id.clone()).as_user(db).await?; + for user_id in get_voice_channel_members(channel.id()) + .await? + .iter() + .flatten() + { + let user = Reference::from_unchecked(user_id.clone()) + .as_user(db) + .await?; - sync_user_voice_permissions(db, voice_client, &node, &user, channel, server, role_id).await?; - }; + sync_user_voice_permissions(db, voice_client, &node, &user, channel, server, role_id) + .await?; + } Ok(()) } -pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClient, node: &str, user: &User, channel: &Channel, server: Option<&Server>, role_id: Option<&str>) -> Result<()> { +pub async fn sync_user_voice_permissions( + db: &Database, + voice_client: &VoiceClient, + node: &str, + user: &User, + channel: &Channel, + server: Option<&Server>, + role_id: Option<&str>, +) -> Result<()> { let channel_id = channel.id(); let server_id: Option<&str> = server.as_ref().map(|s| s.id.as_str()); let member = match server_id { - Some(server_id) => Some(Reference::from_unchecked(user.id.clone()).as_member(db, server_id).await?), - None => None + Some(server_id) => Some( + Reference::from_unchecked(user.id.clone()) + .as_member(db, server_id) + .await?, + ), + None => None, }; - if role_id.is_none_or(|role_id| member.as_ref().is_none_or(|member| member.roles.iter().any(|r| r == role_id))) { - let voice_state = get_voice_state(channel_id, server_id, &user.id).await?.unwrap(); + if role_id.is_none_or(|role_id| { + member + .as_ref() + .is_none_or(|member| member.roles.iter().any(|r| r == role_id)) + }) { + let voice_state = get_voice_state(channel_id, server_id, &user.id) + .await? + .unwrap(); let mut query = DatabasePermissionQuery::new(db, user) .channel(channel) .user(user); if let (Some(server), Some(member)) = (server, member.as_ref()) { - query = query.member(member).server(server) + query = query.member(member).server(server) } let permissions = calculate_channel_permissions(&mut query).await; + let limits = user.limits().await; let mut update_event = PartialUserVoiceState { id: Some(user.id.clone()), @@ -360,7 +421,7 @@ pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClie let before = update_event.clone(); - let can_video = permissions.has_channel_permission(ChannelPermission::Video); + let can_video = limits.video && permissions.has_channel_permission(ChannelPermission::Video); let can_speak = permissions.has_channel_permission(ChannelPermission::Speak); let can_listen = permissions.has_channel_permission(ChannelPermission::Listen); @@ -370,21 +431,49 @@ pub async fn sync_user_voice_permissions(db: &Database, voice_client: &VoiceClie update_voice_state(channel_id, server_id, &user.id, &update_event).await?; - voice_client.update_permissions(node, user, channel_id, ParticipantPermission { - can_subscribe: can_listen, - can_publish: can_speak, - can_publish_data: can_speak, - ..Default::default() - }).await?; + voice_client + .update_permissions( + node, + user, + channel_id, + ParticipantPermission { + can_subscribe: can_listen, + can_publish: can_speak, + can_publish_data: can_speak, + ..Default::default() + }, + ) + .await?; if update_event != before { EventV1::UserVoiceStateUpdate { id: user.id.clone(), channel_id: channel_id.to_string(), - data: update_event - }.p(channel_id.to_string()).await; + data: update_event, + } + .p(channel_id.to_string()) + .await; }; }; Ok(()) -} \ No newline at end of file +} + +pub async fn set_channel_call_started_system_message( + channel_id: &str, + message_id: &str, +) -> Result<()> { + get_connection() + .await? + .set(format!("call_started_message:{channel_id}"), message_id) + .await + .to_internal_error() +} + +pub async fn take_channel_call_started_system_message(channel_id: &str) -> Result> { + get_connection() + .await? + .get_del(format!("call_started_message:{channel_id}")) + .await + .to_internal_error() +} diff --git a/crates/core/database/src/voice/voice_client.rs b/crates/core/database/src/voice/voice_client.rs index d06fe1c3..6a7308df 100644 --- a/crates/core/database/src/voice/voice_client.rs +++ b/crates/core/database/src/voice/voice_client.rs @@ -1,27 +1,25 @@ +use crate::models::{Channel, User}; use livekit_api::{ access_token::{AccessToken, VideoGrants}, services::room::{CreateRoomOptions, RoomClient as InnerRoomClient, UpdateParticipantOptions}, }; use livekit_protocol::{ParticipantInfo, ParticipantPermission, Room}; use revolt_config::{config, LiveKitNode}; -use crate::models::{Channel, User}; -use revolt_models::v0; use revolt_permissions::{ChannelPermission, PermissionValue}; use revolt_result::{create_error, Result, ToRevoltError}; -use std::{borrow::Cow, collections::HashMap, time::Duration}; +use std::{collections::HashMap, time::Duration}; use super::get_allowed_sources; - #[derive(Debug)] pub struct RoomClient { pub client: InnerRoomClient, - pub node: LiveKitNode + pub node: LiveKitNode, } #[derive(Debug)] pub struct VoiceClient { - pub rooms: HashMap + pub rooms: HashMap, } impl VoiceClient { @@ -29,16 +27,20 @@ impl VoiceClient { Self { rooms: nodes .into_iter() - .map(|(name, node)| + .map(|(name, node)| { ( name, RoomClient { - client: InnerRoomClient::with_api_key(&node.url, &node.key, &node.secret), - node - } + client: InnerRoomClient::with_api_key( + &node.url, + &node.key, + &node.secret, + ), + node, + }, ) - ) - .collect() + }) + .collect(), } } @@ -53,11 +55,12 @@ impl VoiceClient { } pub fn get_node(&self, name: &str) -> Result<&RoomClient> { - self.rooms.get(name) + self.rooms + .get(name) .ok_or_else(|| create_error!(UnknownNode)) } - pub fn create_token( + pub async fn create_token( &self, node: &str, user: &User, @@ -66,7 +69,8 @@ impl VoiceClient { ) -> Result { let room = self.get_node(node)?; - let allowed_sources = get_allowed_sources(permissions); + let limits = user.limits().await; + let allowed_sources = get_allowed_sources(&limits, permissions); AccessToken::with_api_key(&room.node.key, &room.node.secret) .with_name(&format!("{}#{}", user.username, user.discriminator)) @@ -76,7 +80,10 @@ impl VoiceClient { .with_grants(VideoGrants { room_join: true, can_publish: true, - can_publish_sources: allowed_sources.into_iter().map(ToString::to_string).collect(), + can_publish_sources: allowed_sources + .into_iter() + .map(ToString::to_string) + .collect(), can_subscribe: permissions.has_channel_permission(ChannelPermission::Listen), room: channel.id().to_string(), ..Default::default() @@ -88,12 +95,9 @@ impl VoiceClient { pub async fn create_room(&self, node: &str, channel: &Channel) -> Result { let room = self.get_node(node)?; - let voice = match channel { - Channel::DirectMessage { .. } | Channel::Group { .. } => Some(Cow::Owned(v0::VoiceInformation::default())), - Channel::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)), - _ => None - } - .ok_or_else(|| create_error!(NotAVoiceChannel))?; + let voice = channel + .voice() + .ok_or_else(|| create_error!(NotAVoiceChannel))?; room.client .create_room( @@ -135,7 +139,8 @@ impl VoiceClient { pub async fn remove_user(&self, node: &str, user_id: &str, channel_id: &str) -> Result<()> { let room = self.get_node(node)?; - room.client.remove_participant(channel_id, user_id) + room.client + .remove_participant(channel_id, user_id) .await .to_internal_error() } diff --git a/crates/core/models/src/v0/channels.rs b/crates/core/models/src/v0/channels.rs index 0d99f956..28895bfe 100644 --- a/crates/core/models/src/v0/channels.rs +++ b/crates/core/models/src/v0/channels.rs @@ -158,6 +158,7 @@ auto_derived!( pub struct VoiceInformation { /// Maximium amount of users allowed in the voice channel at once #[cfg_attr(feature = "validator", validate(range(min = 1)))] + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub max_users: Option } diff --git a/crates/core/models/src/v0/messages.rs b/crates/core/models/src/v0/messages.rs index 442d4e5b..78d9cbc6 100644 --- a/crates/core/models/src/v0/messages.rs +++ b/crates/core/models/src/v0/messages.rs @@ -133,7 +133,7 @@ auto_derived!( #[serde(rename = "message_unpinned")] MessageUnpinned { id: String, by: String }, #[serde(rename = "call_started")] - CallStarted { by: String }, + CallStarted { by: String, finished_at: Option }, } /// Name and / or avatar override information diff --git a/crates/daemons/voice-ingress/src/api.rs b/crates/daemons/voice-ingress/src/api.rs new file mode 100644 index 00000000..5a7ed2f8 --- /dev/null +++ b/crates/daemons/voice-ingress/src/api.rs @@ -0,0 +1,201 @@ +use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver}; +use livekit_protocol::TrackType; +use revolt_database::{ + events::client::EventV1, + iso8601_timestamp::Timestamp, + util::reference::Reference, + voice::{ + create_voice_state, delete_voice_state, get_user_moved_from_voice, get_user_moved_to_voice, + get_voice_channel_members, take_channel_call_started_system_message, + update_voice_state_tracks, VoiceClient, + }, + Database, PartialMessage, SystemMessage, +}; +use revolt_result::{Result, ToRevoltError}; +use rocket::{post, State}; +use rocket_empty::EmptyResponse; + +use crate::guard::AuthHeader; + +#[post("/", data = "")] +pub async fn ingress( + db: &State, + node: &str, + voice_client: &State, + auth_header: AuthHeader<'_>, + body: &str, +) -> Result { + log::debug!("received event: {body:?}"); + + let config = revolt_config::config().await; + + let node_info = config + .api + .livekit + .nodes + .get(node) + .to_internal_error() + .inspect_err(|_| { + log::error!("Unknown node {node}, make sure livekit has the correct node name set and matches `hosts.livekit` and `api.livekit.nodes` in the Revolt config.") + })?; + + let webhook_receiver = WebhookReceiver::new(TokenVerifier::with_api_key( + &node_info.key, + &node_info.secret, + )); + + let event = webhook_receiver + .receive(body, &auth_header) + .to_internal_error()?; + + let channel_id = event.room.as_ref().map(|r| &r.name); + let user_id = event.participant.as_ref().map(|r| &r.identity); + + match event.event.as_str() { + // User joined a channel + "participant_joined" => { + 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 voice_state = create_voice_state(channel_id, channel.server(), user_id).await?; + + // Only publish one event when a user is moved from one channel to another. + if let Some(moved_from) = get_user_moved_to_voice(channel_id, user_id).await? { + EventV1::VoiceChannelMove { + user: user_id.clone(), + from: moved_from, + to: channel_id.clone(), + state: voice_state, + } + .p(channel_id.clone()) + .await; + } else { + EventV1::VoiceChannelJoin { + id: channel_id.clone(), + state: voice_state, + } + .p(channel_id.clone()) + .await; + }; + } + // User left a channel + "participant_left" => { + 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?; + + delete_voice_state(channel_id, channel.server(), user_id).await?; + + // Dont send leave event when a user is moved + if get_user_moved_from_voice(channel_id, user_id) + .await? + .is_none() + { + EventV1::VoiceChannelLeave { + id: channel_id.clone(), + user: user_id.clone(), + } + .p(channel_id.clone()) + .await; + }; + + // Update CallStarted system message if everyone has left with the end time + let members = get_voice_channel_members(channel_id).await?; + + if members.is_none_or(|m| m.is_empty()) { + if let Some(system_message_id) = + take_channel_call_started_system_message(channel_id).await? + { + // Could have been deleted + if let Ok(mut message) = Reference::from_unchecked(system_message_id) + .as_message(db) + .await + { + if let Some(SystemMessage::CallStarted { finished_at, .. }) = + &mut message.system + { + *finished_at = Some(Timestamp::now_utc()); + + message + .update( + db, + PartialMessage { + system: message.system.clone(), + ..Default::default() + }, + Vec::new(), + ) + .await?; + } else { + log::error!("Broken State: Call started message ID ({}) does not contain a CallStarted system message.", &message.id) + } + }; + }; + } + } + // Audio/video track was started/stopped + "track_published" | "track_unpublished" => { + let channel_id = channel_id.to_internal_error()?; + let user_id = user_id.to_internal_error()?; + let track = event.track.as_ref().to_internal_error()?; + + let channel = Reference::from_unchecked(channel_id.clone()) + .as_channel(db) + .await?; + + let user = Reference::from_unchecked(user_id.clone()) + .as_user(db) + .await?; + + let user_limits = user.limits().await; + + // forbid any size which goes over the limit and also limit the aspect ratio to stop people from making too tall or too wide and bypassing the limit. + // TODO: figure out how to track audio stream quality + + if event.event == "track_published" + && (track.r#type == TrackType::Data as i32 + || (track.r#type == TrackType::Video as i32 + && (user_limits.video_resolution[0] != 0 + && user_limits.video_resolution[1] != 0 + && track.width * track.height + > user_limits.video_resolution[0] + * user_limits.video_resolution[1]) + || (user_limits.video_aspect_ratio[0] + != user_limits.video_aspect_ratio[1] + && !(user_limits.video_aspect_ratio[0] + ..=user_limits.video_aspect_ratio[1]) + .contains(&(track.width as f32 / track.height as f32))))) + { + voice_client.remove_user(node, user_id, channel_id).await?; + delete_voice_state(channel_id, channel.server(), user_id).await?; + } else { + let partial = update_voice_state_tracks( + channel_id, + channel.server(), + user_id, + event.event == "track_published", // to avoid duplicating this entire case twice + track.source, + ) + .await?; + + EventV1::UserVoiceStateUpdate { + id: user_id.clone(), + channel_id: channel_id.clone(), + data: partial, + } + .p(channel_id.clone()) + .await; + } + } + _ => {} + }; + + Ok(EmptyResponse) +} diff --git a/crates/daemons/voice-ingress/src/guard.rs b/crates/daemons/voice-ingress/src/guard.rs new file mode 100644 index 00000000..f6ce11dd --- /dev/null +++ b/crates/daemons/voice-ingress/src/guard.rs @@ -0,0 +1,28 @@ +use revolt_result::{create_error, Error}; +use rocket::{ + http::Status, + request::{FromRequest, Outcome}, + Request, +}; + +pub struct AuthHeader<'a>(&'a str); + +#[rocket::async_trait] +impl<'r> FromRequest<'r> for AuthHeader<'r> { + type Error = Error; + + async fn from_request(request: &'r Request<'_>) -> Outcome { + match request.headers().get("Authorization").next() { + Some(token) => Outcome::Success(Self(token)), + None => Outcome::Error((Status::Unauthorized, create_error!(NotAuthenticated))), + } + } +} + +impl std::ops::Deref for AuthHeader<'_> { + type Target = str; + + fn deref(&self) -> &Self::Target { + self.0 + } +} diff --git a/crates/daemons/voice-ingress/src/main.rs b/crates/daemons/voice-ingress/src/main.rs index 395c6ba7..0058566c 100644 --- a/crates/daemons/voice-ingress/src/main.rs +++ b/crates/daemons/voice-ingress/src/main.rs @@ -1,15 +1,13 @@ use std::env; - -use livekit_protocol::TrackType; -use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver}; -use revolt_database::{ - events::client::EventV1, util::reference::Reference, voice::{create_voice_state, delete_voice_state, get_user_moved_from_voice, update_voice_state_tracks, VoiceClient}, Database, DatabaseInfo -}; -use rocket::{build, http::Status, post, request::{FromRequest, Outcome}, routes, Config, Request, State}; -use rocket_empty::EmptyResponse; +use revolt_database::voice::VoiceClient; +use revolt_database::DatabaseInfo; +use revolt_result::Result; +use rocket::{build, routes, Config}; use std::net::Ipv4Addr; -use revolt_result::{create_error, Error, Result, ToRevoltError}; + +mod api; +mod guard; #[rocket::main] async fn main() -> Result<(), rocket::Error> { @@ -20,7 +18,7 @@ async fn main() -> Result<(), rocket::Error> { let _rocket = build() .manage(database) .manage(voice_client) - .mount("/", routes![ingress]) + .mount("/", routes![api::ingress]) .configure(Config { port: 8500, address: Ipv4Addr::new(0, 0, 0, 0).into(), @@ -33,131 +31,3 @@ async fn main() -> Result<(), rocket::Error> { Ok(()) } - -struct AuthHeader<'a>(&'a str); - -#[rocket::async_trait] -impl<'r> FromRequest<'r> for AuthHeader<'r> { - type Error = Error; - - async fn from_request(request: &'r Request<'_>) -> Outcome { - match request.headers().get("Authorization").next() { - Some(token) => Outcome::Success(Self(token)), - None => Outcome::Error((Status::Unauthorized, create_error!(NotAuthenticated))) - } - } -} - -impl std::ops::Deref for AuthHeader<'_> { - type Target = str; - - fn deref(&self) -> &Self::Target { - self.0 - } -} - -#[post("/", data = "")] -async fn ingress(db: &State, node: &str, voice_client: &State, auth_header: AuthHeader<'_>, body: &str) -> Result { - log::debug!("received event: {body:?}"); - - let config = revolt_config::config().await; - - let node_info = config.api.livekit.nodes.get(node) - .to_internal_error() - .inspect_err(|_| log::error!("Unknown node {node}, make sure livekit has the correct node name set."))?; - - let webhook_receiver = WebhookReceiver::new(TokenVerifier::with_api_key(&node_info.key, &node_info.secret)); - let event = webhook_receiver.receive(body, &auth_header).to_internal_error()?; - - let channel_id = event.room.as_ref().map(|r| &r.name); - let user_id = event.participant.as_ref().map(|r| &r.identity); - - match event.event.as_str() { - "participant_joined" => { - 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 voice_state = create_voice_state(channel_id, channel.server(), user_id).await?; - - if let Some(moved_from) = get_user_moved_from_voice(channel_id, user_id).await? { - EventV1::VoiceChannelMove { - user: user_id.clone(), - from: moved_from, - to: channel_id.clone(), - state: voice_state - } - .p(channel_id.clone()) - .await; - } else { - EventV1::VoiceChannelJoin { - id: channel_id.clone(), - state: voice_state, - } - .p(channel_id.clone()) - .await; - }; - } - "participant_left" => { - 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?; - - delete_voice_state(channel_id, channel.server(), user_id).await?; - - EventV1::VoiceChannelLeave { - id: channel_id.clone(), - user: user_id.clone(), - } - .p(channel_id.clone()) - .await - } - "track_published" | "track_unpublished" => { - let channel_id = channel_id.to_internal_error()?; - let user_id = user_id.to_internal_error()?; - let track = event.track.as_ref().to_internal_error()?; - - let channel = Reference::from_unchecked(channel_id.clone()) - .as_channel(db) - .await?; - - // remove the user if they try publish a video larger than 1080x720 or they publish data - // TODO: move to config - if event.event == "track_published" - && ( - // handle any size which goes over the limit of "1080x720" to stop people from making too tall or too wide and bypassing the limit - (track.r#type == TrackType::Video as i32 && (track.width * track.height) >= (1080 * 720)) - | (track.r#type == TrackType::Data as i32) - ) - { - voice_client.remove_user(node, user_id, channel_id).await?; - delete_voice_state(channel_id, channel.server(), user_id).await?; - } - - let partial = update_voice_state_tracks( - channel_id, - channel.server(), - user_id, - event.event == "track_published", // to avoid duplicating this entire case twice - track.source - ).await?; - - EventV1::UserVoiceStateUpdate { - id: user_id.clone(), - channel_id: channel_id.clone(), - data: partial - } - .p(channel_id.clone()) - .await; - } - _ => {} - }; - - Ok(EmptyResponse) -} diff --git a/crates/delta/src/routes/channels/channel_edit.rs b/crates/delta/src/routes/channels/channel_edit.rs index d69fb8b3..b74c176c 100644 --- a/crates/delta/src/routes/channels/channel_edit.rs +++ b/crates/delta/src/routes/channels/channel_edit.rs @@ -38,6 +38,7 @@ pub async fn edit( && data.icon.is_none() && data.nsfw.is_none() && data.owner.is_none() + && data.voice.is_none() && data.remove.is_none() { return Ok(Json(channel.into())); @@ -242,8 +243,8 @@ pub async fn edit( } if let Some(new_voice) = data.voice { - *voice = Some(new_voice.clone()); - partial.voice = Some(new_voice); + *voice = Some(new_voice.clone().into()); + partial.voice = Some(new_voice.into()); } } _ => return Err(create_error!(InvalidOperation)), diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index c83e1609..0b3e747f 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -1,6 +1,6 @@ use revolt_config::config; use revolt_models::v0; -use revolt_database::{util::{permissions::perms, reference::Reference}, voice::{delete_voice_state, get_channel_node, get_user_voice_channels, raise_if_in_voice, VoiceClient}, Channel, Database, SystemMessage, User, AMQP}; +use revolt_database::{util::{permissions::perms, reference::Reference}, voice::{delete_voice_state, get_channel_node, get_user_voice_channels, raise_if_in_voice, set_channel_call_started_system_message, VoiceClient}, Channel, Database, SystemMessage, User, AMQP}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; @@ -65,31 +65,30 @@ pub async fn call( raise_if_in_voice(&user, channel.id()).await?; } - let token = voice_client.create_token(&node, &user, current_permissions, &channel)?; + let token = voice_client.create_token(&node, &user, current_permissions, &channel).await?; let room = voice_client.create_room(&node, &channel).await?; log::debug!("Created room {}", room.name); - match &channel { - Channel::DirectMessage { .. } | Channel::Group { .. } => { - SystemMessage::CallStarted { - by: user.id.clone() - } - .into_message(channel.id().to_string()) - .send( - db, - Some(amqp), - v0::MessageAuthor::System { - username: &user.username, - avatar: user.avatar.as_ref().map(|file| file.id.as_ref()), - }, - None, - None, - &channel, false - ).await?; - } - _ => {} - }; + let mut call_started_message = SystemMessage::CallStarted { + by: user.id.clone(), + finished_at: None + } + .into_message(channel.id().to_string()); + + set_channel_call_started_system_message(channel.id(), &call_started_message.id).await?; + + call_started_message.send( + db, + Some(amqp), + v0::MessageAuthor::System { + username: &user.username, + avatar: user.avatar.as_ref().map(|file| file.id.as_ref()), + }, + None, + None, + &channel, false + ).await?; Ok(Json(v0::CreateVoiceUserResponse { token, url: node_host.clone() })) } diff --git a/crates/delta/src/routes/servers/member_edit.rs b/crates/delta/src/routes/servers/member_edit.rs index 89a74b0e..1ba00b73 100644 --- a/crates/delta/src/routes/servers/member_edit.rs +++ b/crates/delta/src/routes/servers/member_edit.rs @@ -1,19 +1,24 @@ 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::{ events::client::EventV1, - util::{permissions::{perms, DatabasePermissionQuery}, reference::Reference}, - voice::{get_channel_node, get_user_voice_channel_in_server, move_user, set_channel_node, set_user_moved_from_voice, sync_user_voice_permissions, VoiceClient}, + util::{ + permissions::{perms, DatabasePermissionQuery}, + reference::Reference, + }, + voice::{ + get_channel_node, get_user_voice_channel_in_server, set_channel_node, + set_user_moved_from_voice, set_user_moved_to_voice, sync_user_voice_permissions, + VoiceClient, + }, Database, File, PartialMember, User, }; -use revolt_models::v0::{self, FieldsMember, PartialUserVoiceState}; +use revolt_models::v0::{self, FieldsMember}; -use revolt_permissions::{calculate_channel_permissions, calculate_server_permissions, ChannelPermission}; -use revolt_result::{create_error, Result, ToRevoltError}; +use revolt_permissions::{ + calculate_channel_permissions, calculate_server_permissions, ChannelPermission, +}; +use revolt_result::{create_error, Result}; use rocket::{form::validate::Contains, serde::json::Json, State}; use validator::Validate; @@ -109,7 +114,7 @@ pub async fn edit( let new_voice_channel = if let Some(new_channel) = &data.voice_channel { if !voice_client.is_enabled() { - return Err(create_error!(LiveKitUnavailable)) + return Err(create_error!(LiveKitUnavailable)); }; permissions.throw_if_lacking_channel_permission(ChannelPermission::MoveMembers)?; @@ -125,7 +130,10 @@ pub async fn edit( Err(create_error!(UnknownChannel))? } - if get_user_voice_channel_in_server(&target_user.id, &server.id).await?.is_none() { + if get_user_voice_channel_in_server(&target_user.id, &server.id) + .await? + .is_none() + { Err(create_error!(NotConnected))? }; @@ -213,7 +221,8 @@ pub async fn edit( .await?; if let Some(new_voice_channel) = new_voice_channel { - if let Some(channel) = get_user_voice_channel_in_server(&target_user.id, &server.id).await? { + if let Some(channel) = get_user_voice_channel_in_server(&target_user.id, &server.id).await? + { let old_node = get_channel_node(&channel).await?.unwrap(); let new_node = match get_channel_node(new_voice_channel.id()).await? { @@ -225,28 +234,45 @@ pub async fn edit( }; set_user_moved_from_voice(&channel, new_voice_channel.id(), &target_user.id).await?; + set_user_moved_to_voice(new_voice_channel.id(), &channel, &target_user.id).await?; let mut query = perms(db, &target_user).channel(&new_voice_channel); let permissions = calculate_channel_permissions(&mut query).await; - voice_client.create_room(&new_node, &new_voice_channel).await?; - let token = voice_client.create_token(&new_node, &target_user, permissions, &new_voice_channel)?; + voice_client + .create_room(&new_node, &new_voice_channel) + .await?; + let token = voice_client + .create_token(&new_node, &target_user, permissions, &new_voice_channel) + .await?; - voice_client.remove_user(&old_node, &target_user.id, &channel).await?; + voice_client + .remove_user(&old_node, &target_user.id, &channel) + .await?; EventV1::UserMoveVoiceChannel { node: new_node, - token + token, } .p_user(target_user.id.clone(), db) .await; }; } else if can_publish.is_some() || can_receive.is_some() || remove_contains_voice { - if let Some(channel) = get_user_voice_channel_in_server(&target_user.id, &server.id).await? { + if let Some(channel) = get_user_voice_channel_in_server(&target_user.id, &server.id).await? + { let node = get_channel_node(&channel).await?.unwrap(); let channel = Reference::from_unchecked(channel).as_channel(db).await?; - sync_user_voice_permissions(db, voice_client, &node, &user, &channel, Some(&server), None).await?; + sync_user_voice_permissions( + db, + voice_client, + &node, + &user, + &channel, + Some(&server), + None, + ) + .await?; }; }; diff --git a/scripts/build-image-layer.sh b/scripts/build-image-layer.sh index 0a0fa32b..5897180a 100644 --- a/scripts/build-image-layer.sh +++ b/scripts/build-image-layer.sh @@ -35,14 +35,16 @@ deps() { crates/services/autumn/src \ crates/services/january/src \ crates/daemons/crond/src \ - crates/daemons/pushd/src + crates/daemons/pushd/src \ + crates/daemons/voice-ingress/src echo 'fn main() { panic!("stub"); }' | tee crates/bonfire/src/main.rs | tee crates/delta/src/main.rs | tee crates/services/autumn/src/main.rs | tee crates/services/january/src/main.rs | tee crates/daemons/crond/src/main.rs | - tee crates/daemons/pushd/src/main.rs + tee crates/daemons/pushd/src/main.rs | + tee crates/daemons/voice-ingress/src/main.rs echo '' | tee crates/core/config/src/lib.rs | tee crates/core/database/src/lib.rs | @@ -66,6 +68,7 @@ apps() { crates/delta/src/main.rs \ crates/daemons/crond/src/main.rs \ crates/daemons/pushd/src/main.rs \ + crates/daemons/voice-ingress/src/main.rs \ crates/core/config/src/lib.rs \ crates/core/database/src/lib.rs \ crates/core/models/src/lib.rs \ diff --git a/scripts/publish-debug-image.sh b/scripts/publish-debug-image.sh index 6686344c..89fe6012 100755 --- a/scripts/publish-debug-image.sh +++ b/scripts/publish-debug-image.sh @@ -27,6 +27,7 @@ docker build -t ghcr.io/revoltchat/autumn:$TAG - < crates/services/autumn/Docker docker build -t ghcr.io/revoltchat/january:$TAG - < crates/services/january/Dockerfile docker build -t ghcr.io/revoltchat/crond:$TAG - < crates/daemons/crond/Dockerfile docker build -t ghcr.io/revoltchat/pushd:$TAG - < crates/daemons/pushd/Dockerfile +docker build -t ghcr.io/revoltchat/voice-ingress:$TAG - < crates/daemons/voice-ingress/Dockerfile if [ "$DEBUG" = "true" ]; then git restore Cargo.toml @@ -38,3 +39,4 @@ docker push ghcr.io/revoltchat/autumn:$TAG docker push ghcr.io/revoltchat/january:$TAG docker push ghcr.io/revoltchat/crond:$TAG docker push ghcr.io/revoltchat/pushd:$TAG +docker push ghcr.io/revoltchat/voice-ingress:$TAG \ No newline at end of file