fix: voice system messages and call notifs by fetching participant list

Signed-off-by: Zomatree <me@zomatree.live>
(cherry picked from commit 512a512def)

crates/core/database/src/voice/voice_client.rs

@@ -159,4 +159,17 @@ impl VoiceClient {
             .await
             .to_internal_error()
     }
+
+    pub async fn get_room_participants(
+        &self,
+        node: &str,
+        channel_id: &str,
+    ) -> Result<Vec<ParticipantInfo>> {
+        let room = self.get_node(node)?;
+
+        room.client
+            .list_participants(channel_id)
+            .await
+            .to_internal_error()
+    }
 }

crates/daemons/voice-ingress/src/api.rs

@@ -1,19 +1,15 @@
 use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver};
 use livekit_protocol::TrackType;
 use revolt_database::{
-    events::client::EventV1,
-    iso8601_timestamp::{Duration, Timestamp},
-    util::reference::Reference,
-    voice::{
-        create_voice_state, delete_channel_voice_state, delete_voice_state,
-        get_user_moved_from_voice, get_user_moved_to_voice, update_voice_state_tracks,
-        RoomMetadata, UserVoiceChannel, VoiceClient,
-    },
-    Database, AMQP,
+    AMQP, Database, PartialMessage, SystemMessage, events::client::EventV1, iso8601_timestamp::{Duration, Timestamp}, util::reference::Reference, voice::{
+        RoomMetadata, UserVoiceChannel, VoiceClient, create_voice_state, delete_channel_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
+    }
 };
+use revolt_models::v0;
 use revolt_result::{Result, ToRevoltError};
 use rocket::{post, State};
 use rocket_empty::EmptyResponse;
+use ulid::Ulid;
 
 use crate::guard::AuthHeader;
 
@@ -21,12 +17,12 @@ use crate::guard::AuthHeader;
 pub async fn ingress(
     db: &State<Database>,
     voice_client: &State<VoiceClient>,
-    _amqp: &State<AMQP>,
+    amqp: &State<AMQP>,
     node: &str,
     auth_header: AuthHeader<'_>,
     body: &str,
 ) -> Result<EmptyResponse> {
-    log::debug!("received event: {body:?}");
+    log::debug!("received event: {body}");
 
     let config = revolt_config::config().await;
 
@@ -63,16 +59,18 @@ pub async fn ingress(
             let channel_id = channel_id.to_internal_error()?;
             let user_id = user_id.to_internal_error()?;
             let server_id = room_metadata.to_internal_error()?.server;
-            let channel = UserVoiceChannel {
+            let voice_channel = UserVoiceChannel {
                 id: channel_id.clone(),
                 server_id: server_id.clone(),
             };
 
+            let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
+
             let joined_at = Timestamp::UNIX_EPOCH
                 .checked_add(Duration::seconds(event.created_at))
                 .unwrap();
 
-            let voice_state = create_voice_state(&channel, user_id, joined_at).await?;
+            let voice_state = create_voice_state(&voice_channel, user_id, joined_at).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? {
@@ -93,63 +91,66 @@ pub async fn ingress(
                 .await;
             };
 
-            // TODO: fix `num_participants` being incorrect sometimes see (#457)
-            // First user who joined - send call started system message.
-            // if event.room.as_ref().unwrap().num_participants == 1 {
-            //     let user = Reference::from_unchecked(user_id).as_user(db).await?;
+            let participants = voice_client.get_room_participants(node, channel_id).await?;
 
-            //     let message_id =
-            //         Ulid::from_datetime(DateTime::from_timestamp_secs(event.created_at).unwrap())
-            //             .to_string();
+            if participants.len() == 1 {
+                let user = Reference::from_unchecked(user_id).as_user(db).await?;
+                let message_id = Ulid::from_datetime(
+                    Timestamp::UNIX_EPOCH
+                        .checked_add(Duration::seconds(event.created_at))
+                        .unwrap()
+                        .into(),
+                )
+                .to_string();
 
-            //     let mut call_started_message = SystemMessage::CallStarted {
-            //         by: user_id.to_string(),
-            //         finished_at: None,
-            //     }
-            //     .into_message(channel.id().to_string());
+                let mut call_started_message = SystemMessage::CallStarted {
+                    by: user_id.to_string(),
+                    finished_at: None,
+                }
+                .into_message(channel_id.clone());
 
-            //     call_started_message.id = message_id;
+                call_started_message.id = message_id;
 
-            //     set_channel_call_started_system_message(channel.id(), &call_started_message.id)
-            //         .await?;
+                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?;
+                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?;
 
-            //     let recipients = get_call_notification_recipients(&channel_id, &user_id).await?;
-            //     let now = joined_at.format_short().to_string();
+                let recipients = get_call_notification_recipients(channel_id, user_id).await?;
+                let now = joined_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);
-            //     }
-            // }
+                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
         "participant_left" => {
             let channel_id = channel_id.to_internal_error()?;
             let user_id = user_id.to_internal_error()?;
             let server_id = room_metadata.to_internal_error()?.server;
-            let channel = UserVoiceChannel {
+            let voice_channel = UserVoiceChannel {
                 id: channel_id.clone(),
                 server_id: server_id.clone(),
             };
 
-            delete_voice_state(&channel, user_id).await?;
+            delete_voice_state(&voice_channel, user_id).await?;
 
             // Dont send leave event when a user is moved
             if get_user_moved_from_voice(channel_id, user_id)
@@ -164,49 +165,47 @@ pub async fn ingress(
                 .await;
             };
 
-            // See above for why this is commented out
-
             // // Update CallStarted system message if everyone has left with the end time
-            // let members = get_voice_channel_members(channel_id).await?;
+            let members = get_voice_channel_members(&voice_channel).await?;
 
-            // if members.is_none_or(|m| m.is_empty()) {
-            //     // The channel is empty so send out an "end" message for ringing
-            //     if let Err(e) = amqp
-            //         .dm_call_updated(user_id, channel_id, None, true, None)
-            //         .await
-            //     {
-            //         revolt_config::capture_internal_error!(&e);
-            //     }
+            if members.is_none_or(|m| m.is_empty()) {
+                // The channel is empty so send out an "end" message for ringing
+                if let Err(e) = amqp
+                    .dm_call_updated(user_id, channel_id, None, true, None)
+                    .await
+                {
+                    revolt_config::capture_internal_error!(&e);
+                }
 
-            //     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());
+                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)
-            //             }
-            //         };
-            //     };
-            // }
+                            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/unmuted/muted
         "track_published" | "track_unpublished" | "track_unmuted" | "track_muted" => {