feat: allow users to specify who is notified when starting a call

Cargo.lock

@@ -6772,6 +6772,7 @@ version = "0.7.1"
 dependencies = [
  "amqprs",
  "async-std",
+ "chrono",
  "futures",
  "livekit-api",
  "livekit-protocol",

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)
+    conn.get(&unique_key).await.to_internal_error()
-        .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<Option<Vec<String>>> {
+    get_connection()
+        .await?
+        .get_del(format!(
+            "call_notification_recipients:{channel_id}-{user_id}"
+        ))
+        .await
+        .to_internal_error()
+}

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<String>,
-        /// 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<bool>,
+        /// Users which should be notified of the call starting
+        ///
+        /// Only used when the user is the first one connected.
+        pub recipients: Option<Vec<String>>,
     }
 );
 

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<String> = if let Some(recipients) = _p.recipients {
+        let (revolt_database::Channel::DirectMessage { recipients, .. }
-            recipients
+        | revolt_database::Channel::Group { recipients, .. }) =
-        } else {
+            self.db.fetch_channel(&payload.channel_id).await?
-            match self.db.fetch_channel(&payload.channel_id).await? {
+        else {
-                revolt_database::Channel::DirectMessage { recipients, .. }
+            warn!(
-                | revolt_database::Channel::Group { recipients, .. } => recipients
+                "Discarding dm call start/stop event for non-dm/group channel {}",
-                    .into_iter()
+                payload.channel_id
-                    .filter(|r| *r != payload.initiator_id)
+            );
-                    .collect(),
+
-                _ => {
+            return Ok(());
-                    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::<Vec<_>>()
+        };
+
         let config = revolt_config::config().await;
 
         for user_id in call_recipients {

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"

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,
+        create_voice_state, delete_voice_state, get_call_notification_recipients,
-        get_voice_channel_members, set_channel_call_started_system_message,
+        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,
+        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

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("/<target>/join_call", data = "<data>")]
 pub async fn call(
     db: &State<Database>,
-    amqp: &State<AMQP>,
     voice_client: &State<VoiceClient>,
     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 {
+    if let Some(recipients) = recipients {
-        Channel::DirectMessage { .. } | Channel::Group { .. } => {
+        if room.num_participants == 0 {
-            let mut msg = SystemMessage::CallStarted {
+            set_call_notification_recipients(channel.id(), &user.id, &recipients).await?;
-                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);
-            }
         }
-        _ => {}
+    }
-    };
 
     Ok(Json(v0::CreateVoiceUserResponse {
         token,