From 512a512defec7a99da580db4c1b5d99b727e7b6a Mon Sep 17 00:00:00 2001 From: Zomatree Date: Thu, 2 Jul 2026 17:54:34 +0100 Subject: [PATCH] fix: voice system messages and call notifs by fetching participant list Signed-off-by: Zomatree --- .../core/database/src/voice/voice_client.rs | 13 ++ crates/daemons/voice-ingress/src/api.rs | 183 +++++++++--------- 2 files changed, 104 insertions(+), 92 deletions(-) diff --git a/crates/core/database/src/voice/voice_client.rs b/crates/core/database/src/voice/voice_client.rs index 94d1f22b..a82b6037 100644 --- a/crates/core/database/src/voice/voice_client.rs +++ b/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> { + let room = self.get_node(node)?; + + room.client + .list_participants(channel_id) + .await + .to_internal_error() + } } diff --git a/crates/daemons/voice-ingress/src/api.rs b/crates/daemons/voice-ingress/src/api.rs index dbe8934e..70bd9f67 100644 --- a/crates/daemons/voice-ingress/src/api.rs +++ b/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, voice_client: &State, - _amqp: &State, + amqp: &State, node: &str, auth_header: AuthHeader<'_>, body: &str, ) -> Result { - 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" => {