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" => {

crates/services/gifbox/src/tenor/mod.rs

@@ -10,7 +10,7 @@ use tokio::sync::RwLock;
 
 pub mod types;
 
-const TENOR_API_BASE_URL: &str = "https://tenor.googleapis.com/v2";
+const TENOR_API_BASE_URL: &str = "https://compat.gifbox.me/v2";
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum TenorError {
@@ -59,7 +59,11 @@ impl Tenor {
         }
     }
 
-    pub async fn request<T: DeserializeOwned>(&self, path: &str, query: &[Option<(&str, &str)>]) -> Result<Arc<T>, TenorError> {
+    pub async fn request<T: DeserializeOwned>(
+        &self,
+        path: &str,
+        query: &[Option<(&str, &str)>],
+    ) -> Result<Arc<T>, TenorError> {
         let response = self
             .client
             .get(format!("{TENOR_API_BASE_URL}{path}"))
@@ -95,24 +99,27 @@ impl Tenor {
             }
         }
 
-        let res = self.coalescion.execute(unique_key.clone(), || async move {
-            self.request::<types::PaginatedMediaResponse>(
-                "/search",
-                &[
-                    Some(("key", &self.key)),
-                    Some(("q", query)),
-                    Some(("client_key", "Gifbox")),
-                    Some(("media_filter", "webm,tinywebm")),
-                    Some(("locale", locale)),
-                    Some(("contentfilter", "high")),
-                    Some(("limit", &limit.to_string())),
-                    position.is_empty().then_some(("pos", position)),
-                    is_category.then_some(("component", "categories"))
-                ]
-            ).await
-        })
-        .await
-        .unwrap();
+        let res = self
+            .coalescion
+            .execute(unique_key.clone(), || async move {
+                self.request::<types::PaginatedMediaResponse>(
+                    "/search",
+                    &[
+                        Some(("key", &self.key)),
+                        Some(("q", query)),
+                        Some(("client_key", "Gifbox")),
+                        Some(("media_filter", "webm,tinywebm")),
+                        Some(("locale", locale)),
+                        Some(("contentfilter", "high")),
+                        Some(("limit", &limit.to_string())),
+                        position.is_empty().then_some(("pos", position)),
+                        is_category.then_some(("component", "categories")),
+                    ],
+                )
+                .await
+            })
+            .await
+            .unwrap();
 
         if let Ok(resp) = &*res {
             self.cache.write().await.insert(unique_key, resp.clone());
@@ -143,8 +150,9 @@ impl Tenor {
                         Some(("client_key", "Gifbox")),
                         Some(("locale", locale)),
                         Some(("contentfilter", "high")),
-                    ]
-                ).await
+                    ],
+                )
+                .await
             })
             .await
             .unwrap();
@@ -173,22 +181,25 @@ impl Tenor {
             }
         }
 
-        let res = self.coalescion.execute(unique_key.clone(), || async move {
-            self.request::<types::PaginatedMediaResponse>(
-                "/featured",
-                &[
-                    Some(("key", &self.key)),
-                    Some(("client_key", "Gifbox")),
-                    Some(("media_filter", "webm,tinywebm")),
-                    Some(("locale", locale)),
-                    Some(("contentfilter", "high")),
-                    Some(("limit", &limit.to_string())),
-                    position.is_empty().then_some(("pos", position)),
-                ]
-            ).await
-        })
-        .await
-        .unwrap();
+        let res = self
+            .coalescion
+            .execute(unique_key.clone(), || async move {
+                self.request::<types::PaginatedMediaResponse>(
+                    "/featured",
+                    &[
+                        Some(("key", &self.key)),
+                        Some(("client_key", "Gifbox")),
+                        Some(("media_filter", "webm,tinywebm")),
+                        Some(("locale", locale)),
+                        Some(("contentfilter", "high")),
+                        Some(("limit", &limit.to_string())),
+                        position.is_empty().then_some(("pos", position)),
+                    ],
+                )
+                .await
+            })
+            .await
+            .unwrap();
 
         if let Ok(resp) = &*res {
             self.featured.write().await.insert(unique_key, resp.clone());