From 1632e3dacf51603da6559d5c955e0a390483e3b2 Mon Sep 17 00:00:00 2001 From: Zomatree Date: Sun, 7 Sep 2025 05:01:14 +0100 Subject: [PATCH] feat: allow users to specify who is notified when starting a call --- Cargo.lock | 1 + crates/core/database/src/voice/mod.rs | 33 ++++++++++++-- crates/core/models/src/v0/channels.rs | 8 +++- .../pushd/src/consumers/inbound/dm_call.rs | 40 ++++++++++------- crates/daemons/voice-ingress/Cargo.toml | 1 + crates/daemons/voice-ingress/src/api.rs | 32 +++++++++++-- .../delta/src/routes/channels/voice_join.rs | 45 ++++--------------- 7 files changed, 98 insertions(+), 62 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2e9d3b77..c794a75e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6772,6 +6772,7 @@ version = "0.7.1" dependencies = [ "amqprs", "async-std", + "chrono", "futures", "livekit-api", "livekit-protocol", diff --git a/crates/core/database/src/voice/mod.rs b/crates/core/database/src/voice/mod.rs index 39705ebd..f064410a 100644 --- a/crates/core/database/src/voice/mod.rs +++ b/crates/core/database/src/voice/mod.rs @@ -124,9 +124,7 @@ pub async fn get_user_voice_channel_in_server( let unique_key = format!("{user_id}:{server_id}"); - conn.get(&unique_key) - .await - .to_internal_error() + conn.get(&unique_key).await.to_internal_error() } pub fn get_allowed_sources( @@ -479,3 +477,32 @@ pub async fn take_channel_call_started_system_message(channel_id: &str) -> Resul .await .to_internal_error() } + +pub async fn set_call_notification_recipients( + channel_id: &str, + user_id: &str, + recipients: &[String], +) -> Result<()> { + get_connection() + .await? + .set_ex( + format!("call_notification_recipients:{channel_id}-{user_id}"), + recipients, + 10, + ) + .await + .to_internal_error() +} + +pub async fn get_call_notification_recipients( + channel_id: &str, + user_id: &str, +) -> Result>> { + get_connection() + .await? + .get_del(format!( + "call_notification_recipients:{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 7281bd9a..f35a33bf 100644 --- a/crates/core/models/src/v0/channels.rs +++ b/crates/core/models/src/v0/channels.rs @@ -294,10 +294,14 @@ auto_derived!( pub struct DataJoinCall { /// Name of the node to join pub node: Option, - /// Whether to force disconnect any other existing voice connections. + /// Whether to force disconnect any other existing voice connections /// - /// useful for disconnecting on another device and joining on a new. + /// Useful for disconnecting on another device and joining on a new. pub force_disconnect: Option, + /// Users which should be notified of the call starting + /// + /// Only used when the user is the first one connected. + pub recipients: Option>, } ); diff --git a/crates/daemons/pushd/src/consumers/inbound/dm_call.rs b/crates/daemons/pushd/src/consumers/inbound/dm_call.rs index 99e208d1..b5b89dc1 100644 --- a/crates/daemons/pushd/src/consumers/inbound/dm_call.rs +++ b/crates/daemons/pushd/src/consumers/inbound/dm_call.rs @@ -69,24 +69,30 @@ impl DmCallConsumer { debug!("Received dm call start/stop event"); - let call_recipients: Vec = if let Some(recipients) = _p.recipients { - recipients - } else { - match self.db.fetch_channel(&payload.channel_id).await? { - revolt_database::Channel::DirectMessage { recipients, .. } - | revolt_database::Channel::Group { recipients, .. } => recipients - .into_iter() - .filter(|r| *r != payload.initiator_id) - .collect(), - _ => { - warn!( - "Discarding dm call start/stop event for non-dm/group channel {}", - payload.channel_id - ); - return Ok(()); - } - } + let (revolt_database::Channel::DirectMessage { recipients, .. } + | revolt_database::Channel::Group { recipients, .. }) = + self.db.fetch_channel(&payload.channel_id).await? + else { + warn!( + "Discarding dm call start/stop event for non-dm/group channel {}", + payload.channel_id + ); + + return Ok(()); }; + + let call_recipients = if let Some(user_recipients) = _p.recipients { + user_recipients + .into_iter() + .filter(|user_id| recipients.contains(user_id) && user_id != &payload.initiator_id) + .collect() + } else { + recipients + .into_iter() + .filter(|user_id| user_id != &payload.initiator_id) + .collect::>() + }; + let config = revolt_config::config().await; for user_id in call_recipients { diff --git a/crates/daemons/voice-ingress/Cargo.toml b/crates/daemons/voice-ingress/Cargo.toml index 58cc49e0..34a8ad1a 100644 --- a/crates/daemons/voice-ingress/Cargo.toml +++ b/crates/daemons/voice-ingress/Cargo.toml @@ -13,6 +13,7 @@ sentry = "0.31.5" lru = "0.7.6" ulid = "0.5.0" redis-kiss = "0.1.4" +chrono = "0.4.15" # Serde serde_json = "1.0.79" diff --git a/crates/daemons/voice-ingress/src/api.rs b/crates/daemons/voice-ingress/src/api.rs index f470b936..6c6fabc1 100644 --- a/crates/daemons/voice-ingress/src/api.rs +++ b/crates/daemons/voice-ingress/src/api.rs @@ -1,3 +1,4 @@ +use chrono::DateTime; use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver}; use livekit_protocol::TrackType; use revolt_database::{ @@ -5,9 +6,10 @@ use revolt_database::{ 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, set_channel_call_started_system_message, - take_channel_call_started_system_message, update_voice_state_tracks, VoiceClient, + create_voice_state, delete_voice_state, get_call_notification_recipients, + get_user_moved_from_voice, get_user_moved_to_voice, get_voice_channel_members, + set_channel_call_started_system_message, take_channel_call_started_system_message, + update_voice_state_tracks, VoiceClient, }, Database, PartialMessage, SystemMessage, AMQP, }; @@ -15,6 +17,7 @@ use revolt_models::v0; use revolt_result::{Result, ToRevoltError}; use rocket::{post, State}; use rocket_empty::EmptyResponse; +use ulid::Ulid; use crate::guard::AuthHeader; @@ -86,12 +89,25 @@ pub async fn ingress( if event.room.as_ref().unwrap().num_participants == 1 { let user = Reference::from_unchecked(user_id).as_user(db).await?; + let started_at = Timestamp::now_utc(); + let message_id = Ulid::from_datetime( + DateTime::from_timestamp_millis( + started_at + .duration_since(Timestamp::UNIX_EPOCH) + .whole_milliseconds() as i64, + ) + .unwrap(), + ) + .to_string(); + let mut call_started_message = SystemMessage::CallStarted { by: user_id.to_string(), finished_at: None, } .into_message(channel.id().to_string()); + call_started_message.id = message_id; + set_channel_call_started_system_message(channel.id(), &call_started_message.id) .await?; @@ -109,6 +125,16 @@ pub async fn ingress( false, ) .await?; + + let recipients = get_call_notification_recipients(&channel_id, &user_id).await?; + let now = started_at.format_short().to_string(); + + if let Err(e) = amqp + .dm_call_updated(&user.id, channel.id(), Some(&now), false, recipients) + .await + { + revolt_config::capture_error(&e); + } } } // User left a channel diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index 30f06d1f..c15fabfe 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -3,9 +3,9 @@ use revolt_database::{ util::{permissions::perms, reference::Reference}, voice::{ delete_voice_state, get_channel_node, get_user_voice_channels, get_voice_channel_members, - raise_if_in_voice, VoiceClient, + raise_if_in_voice, set_call_notification_recipients, VoiceClient, }, - Channel, Database, SystemMessage, User, AMQP, + Database, User, }; use revolt_models::v0; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; @@ -20,7 +20,6 @@ use rocket::{serde::json::Json, State}; #[post("//join_call", data = "")] pub async fn call( db: &State, - amqp: &State, voice_client: &State, user: User, target: Reference<'_>, @@ -33,6 +32,7 @@ pub async fn call( let v0::DataJoinCall { node, force_disconnect, + recipients, } = data.into_inner(); if user.bot.is_some() && force_disconnect == Some(true) { @@ -96,45 +96,16 @@ pub async fn call( let token = voice_client .create_token(&node, db, &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 { .. } => { - let mut msg = SystemMessage::CallStarted { - by: user.id.clone(), - finished_at: None, - } - .into_message(channel.id().to_string()); - - msg.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 now = iso8601_timestamp::Timestamp::now_utc() - .format_short() - .to_string(); - - if let Err(e) = amqp - .dm_call_updated(&user.id, channel.id(), Some(&now), false, None) - .await - { - revolt_config::capture_error(&e); - } + if let Some(recipients) = recipients { + if room.num_participants == 0 { + set_call_notification_recipients(channel.id(), &user.id, &recipients).await?; } - _ => {} - }; + } Ok(Json(v0::CreateVoiceUserResponse { token,