jmug/stoatchat
1
0
Fork 1
Code Issues Packages Projects Releases Wiki Activity

fix: voice system messages and call notifs by fetching participant list (#846)

Browse Source 
Signed-off-by: Zomatree <me@zomatree.live>
This commit is contained in:
Zomatree
2026-07-07 06:37:27 +01:00
committed by GitHub
parent 21daf3aec6
commit 0b53db9921
2 changed files with 104 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
crates/core/database/src/voice/voice_client.rs
View File

@@ -159,4 +159,17 @@ impl VoiceClient {
.await .await
.to_internal_error() .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()
}
} }

183
crates/daemons/voice-ingress/src/api.rs
View File

@@ -1,19 +1,15 @@
use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver}; use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver};
use livekit_protocol::TrackType; use livekit_protocol::TrackType;
use revolt_database::{ use revolt_database::{
events::client::EventV1, AMQP, Database, PartialMessage, SystemMessage, events::client::EventV1, iso8601_timestamp::{Duration, Timestamp}, util::reference::Reference, voice::{
iso8601_timestamp::{Duration, Timestamp}, 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
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,
}; };
use revolt_models::v0;
use revolt_result::{Result, ToRevoltError}; use revolt_result::{Result, ToRevoltError};
use rocket::{post, State}; use rocket::{post, State};
use rocket_empty::EmptyResponse; use rocket_empty::EmptyResponse;
use ulid::Ulid;
use crate::guard::AuthHeader; use crate::guard::AuthHeader;
@@ -21,12 +17,12 @@ use crate::guard::AuthHeader;
pub async fn ingress( pub async fn ingress(
db: &State<Database>, db: &State<Database>,
voice_client: &State<VoiceClient>, voice_client: &State<VoiceClient>,
_amqp: &State<AMQP>, amqp: &State<AMQP>,
node: &str, node: &str,
auth_header: AuthHeader<'_>, auth_header: AuthHeader<'_>,
body: &str, body: &str,
) -> Result<EmptyResponse> { ) -> Result<EmptyResponse> {
log::debug!("received event: {body:?}"); log::debug!("received event: {body}");
let config = revolt_config::config().await; let config = revolt_config::config().await;
@@ -63,16 +59,18 @@ pub async fn ingress(
let channel_id = channel_id.to_internal_error()?; let channel_id = channel_id.to_internal_error()?;
let user_id = user_id.to_internal_error()?; let user_id = user_id.to_internal_error()?;
let server_id = room_metadata.to_internal_error()?.server; let server_id = room_metadata.to_internal_error()?.server;
let channel = UserVoiceChannel { let voice_channel = UserVoiceChannel {
id: channel_id.clone(), id: channel_id.clone(),
server_id: server_id.clone(), server_id: server_id.clone(),
}; };
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
let joined_at = Timestamp::UNIX_EPOCH let joined_at = Timestamp::UNIX_EPOCH
.checked_add(Duration::seconds(event.created_at)) .checked_add(Duration::seconds(event.created_at))
.unwrap(); .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. // 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? { if let Some(moved_from) = get_user_moved_to_voice(channel_id, user_id).await? {
@@ -93,63 +91,66 @@ pub async fn ingress(
.await; .await;
}; };
// TODO: fix `num_participants` being incorrect sometimes see (#457) let participants = voice_client.get_room_participants(node, channel_id).await?;
// 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 message_id = if participants.len() == 1 {
// Ulid::from_datetime(DateTime::from_timestamp_secs(event.created_at).unwrap()) let user = Reference::from_unchecked(user_id).as_user(db).await?;
// .to_string(); 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 { let mut call_started_message = SystemMessage::CallStarted {
// by: user_id.to_string(), by: user_id.to_string(),
// finished_at: None, finished_at: None,
// } }
// .into_message(channel.id().to_string()); .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) set_channel_call_started_system_message(channel_id, &call_started_message.id)
// .await?; .await?;
// call_started_message call_started_message
// .send( .send(
// db, db,
// Some(amqp), Some(amqp),
// v0::MessageAuthor::System { v0::MessageAuthor::System {
// username: &user.username, username: &user.username,
// avatar: user.avatar.as_ref().map(|file| file.id.as_ref()), avatar: user.avatar.as_ref().map(|file| file.id.as_ref()),
// }, },
// None, None,
// None, None,
// &channel, &channel,
// false, false,
// ) )
// .await?; .await?;
// let recipients = get_call_notification_recipients(&channel_id, &user_id).await?; let recipients = get_call_notification_recipients(channel_id, user_id).await?;
// let now = joined_at.format_short().to_string(); let now = joined_at.format_short().to_string();
// if let Err(e) = amqp if let Err(e) = amqp
// .dm_call_updated(&user.id, channel.id(), Some(&now), false, recipients) .dm_call_updated(&user.id, channel_id, Some(&now), false, recipients)
// .await .await
// { {
// revolt_config::capture_error(&e); revolt_config::capture_error(&e);
// } }
// } }
} }
// User left a channel // User left a channel
"participant_left" => { "participant_left" => {
let channel_id = channel_id.to_internal_error()?; let channel_id = channel_id.to_internal_error()?;
let user_id = user_id.to_internal_error()?; let user_id = user_id.to_internal_error()?;
let server_id = room_metadata.to_internal_error()?.server; let server_id = room_metadata.to_internal_error()?.server;
let channel = UserVoiceChannel { let voice_channel = UserVoiceChannel {
id: channel_id.clone(), id: channel_id.clone(),
server_id: server_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 // Dont send leave event when a user is moved
if get_user_moved_from_voice(channel_id, user_id) if get_user_moved_from_voice(channel_id, user_id)
@@ -164,49 +165,47 @@ pub async fn ingress(
.await; .await;
}; };
// See above for why this is commented out
// // Update CallStarted system message if everyone has left with the end time // // 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()) { if members.is_none_or(|m| m.is_empty()) {
// // The channel is empty so send out an "end" message for ringing // The channel is empty so send out an "end" message for ringing
// if let Err(e) = amqp if let Err(e) = amqp
// .dm_call_updated(user_id, channel_id, None, true, None) .dm_call_updated(user_id, channel_id, None, true, None)
// .await .await
// { {
// revolt_config::capture_internal_error!(&e); revolt_config::capture_internal_error!(&e);
// } }
// if let Some(system_message_id) = if let Some(system_message_id) =
// take_channel_call_started_system_message(channel_id).await? take_channel_call_started_system_message(channel_id).await?
// { {
// // Could have been deleted // Could have been deleted
// if let Ok(mut message) = Reference::from_unchecked(&system_message_id) if let Ok(mut message) = Reference::from_unchecked(&system_message_id)
// .as_message(db) .as_message(db)
// .await .await
// { {
// if let Some(SystemMessage::CallStarted { finished_at, .. }) = if let Some(SystemMessage::CallStarted { finished_at, .. }) =
// &mut message.system &mut message.system
// { {
// *finished_at = Some(Timestamp::now_utc()); *finished_at = Some(Timestamp::now_utc());
// message message
// .update( .update(
// db, db,
// PartialMessage { PartialMessage {
// system: message.system.clone(), system: message.system.clone(),
// ..Default::default() ..Default::default()
// }, },
// Vec::new(), Vec::new(),
// ) )
// .await?; .await?;
// } else { } else {
// log::error!("Broken State: Call started message ID ({}) does not contain a CallStarted system message.", &message.id) 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 // Audio/video track was started/stopped/unmuted/muted
"track_published" | "track_unpublished" | "track_unmuted" | "track_muted" => { "track_published" | "track_unpublished" | "track_unmuted" | "track_muted" => {