From 77a23239eaf52db9a5ec840b15a0caff3a412ef9 Mon Sep 17 00:00:00 2001 From: IAmTomahawkx Date: Sat, 6 Sep 2025 15:14:48 -0700 Subject: [PATCH] feat: Send push notifications for dm call start/end --- compose.yml | 2 +- crates/core/config/Revolt.toml | 3 + crates/core/config/src/lib.rs | 10 +- crates/core/database/src/amqp/amqp.rs | 78 ++++++++- crates/core/database/src/events/rabbit.rs | 15 ++ .../core/database/src/models/users/model.rs | 14 -- crates/core/database/src/util/funcs.rs | 24 +++ crates/core/database/src/util/mod.rs | 3 + crates/core/database/src/voice/mod.rs | 3 +- .../pushd/src/consumers/inbound/dm_call.rs | 162 ++++++++++++++++++ .../pushd/src/consumers/inbound/mod.rs | 1 + .../pushd/src/consumers/outbound/apn.rs | 56 ++++++ .../pushd/src/consumers/outbound/fcm.rs | 34 +++- .../pushd/src/consumers/outbound/vapid.rs | 29 +++- crates/daemons/pushd/src/main.rs | 18 +- crates/daemons/voice-ingress/src/api.rs | 28 ++- crates/daemons/voice-ingress/src/main.rs | 32 +--- crates/delta/src/routes/channels/mod.rs | 2 + .../delta/src/routes/channels/voice_join.rs | 39 ++++- .../src/routes/channels/voice_stop_ring.rs | 69 ++++++++ crates/delta/src/routes/root.rs | 26 +-- 21 files changed, 568 insertions(+), 80 deletions(-) create mode 100644 crates/core/database/src/util/funcs.rs create mode 100644 crates/daemons/pushd/src/consumers/inbound/dm_call.rs create mode 100644 crates/delta/src/routes/channels/voice_stop_ring.rs diff --git a/compose.yml b/compose.yml index 89b29594..028969aa 100644 --- a/compose.yml +++ b/compose.yml @@ -40,7 +40,7 @@ services: # Rabbit rabbit: - image: rabbitmq:3-management + image: rabbitmq:4-management environment: RABBITMQ_DEFAULT_USER: rabbituser RABBITMQ_DEFAULT_PASS: rabbitpass diff --git a/crates/core/config/Revolt.toml b/crates/core/config/Revolt.toml index 4c7ca17d..e2addd9a 100644 --- a/crates/core/config/Revolt.toml +++ b/crates/core/config/Revolt.toml @@ -69,6 +69,8 @@ hcaptcha_sitekey = "" max_concurrent_connections = 50 [api.livekit] +# How long to ring devices for when calling in dms/groups, in seconds +call_ring_duration = 30 [api.livekit.nodes] @@ -90,6 +92,7 @@ message_queue = "notifications.origin.message" mass_mention_queue = "notifications.origin.mass_mention" # handles messages that contain role or everyone mentions fr_accepted_queue = "notifications.ingest.fr_accepted" # friend request accepted fr_received_queue = "notifications.ingest.fr_received" # friend request received +dm_call_queue = "notifications.ingest.dm_call" # friend request received generic_queue = "notifications.ingest.generic" # generic messages (title + body) ack_queue = "notifications.process.ack" # updates badges for apple devices diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 93df9c11..8dc1b316 100644 --- a/crates/core/config/src/lib.rs +++ b/crates/core/config/src/lib.rs @@ -125,7 +125,7 @@ pub struct Hosts { pub events: String, pub autumn: String, pub january: String, - pub livekit: HashMap + pub livekit: HashMap, } #[derive(Deserialize, Debug, Clone)] @@ -198,7 +198,8 @@ pub struct ApiWorkers { #[derive(Deserialize, Debug, Clone)] pub struct ApiLiveKit { - pub nodes: HashMap + pub call_ring_duration: usize, + pub nodes: HashMap, } #[derive(Deserialize, Debug, Clone)] @@ -238,6 +239,7 @@ pub struct Pushd { // Queues pub message_queue: String, pub mass_mention_queue: String, + pub dm_call_queue: String, pub fr_accepted_queue: String, pub fr_received_queue: String, pub generic_queue: String, @@ -268,6 +270,10 @@ impl Pushd { self.get_routing_key(self.mass_mention_queue.clone()) } + pub fn get_dm_call_routing_key(&self) -> String { + self.get_routing_key(self.dm_call_queue.clone()) + } + pub fn get_fr_accepted_routing_key(&self) -> String { self.get_routing_key(self.fr_accepted_queue.clone()) } diff --git a/crates/core/database/src/amqp/amqp.rs b/crates/core/database/src/amqp/amqp.rs index b567c7f7..b29105ed 100644 --- a/crates/core/database/src/amqp/amqp.rs +++ b/crates/core/database/src/amqp/amqp.rs @@ -2,7 +2,8 @@ use std::collections::HashSet; use crate::events::rabbit::*; use crate::User; -use amqprs::channel::BasicPublishArguments; +use amqprs::channel::{BasicPublishArguments, ExchangeDeclareArguments}; +use amqprs::connection::OpenConnectionArguments; use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError}; use amqprs::{BasicProperties, FieldTable}; use revolt_models::v0::PushNotification; @@ -25,6 +26,35 @@ impl AMQP { } } + pub async fn new_auto() -> AMQP { + let config = revolt_config::config().await; + + let connection = Connection::open(&OpenConnectionArguments::new( + &config.rabbit.host, + config.rabbit.port, + &config.rabbit.username, + &config.rabbit.password, + )) + .await + .expect("Failed to connect to RabbitMQ"); + + let channel = connection + .open_channel(None) + .await + .expect("Failed to open RabbitMQ channel"); + + channel + .exchange_declare( + ExchangeDeclareArguments::new(&config.pushd.exchange, "direct") + .durable(true) + .finish(), + ) + .await + .expect("Failed to declare exchange"); + + AMQP::new(connection, channel) + } + pub async fn friend_request_accepted( &self, accepted_request_user: &User, @@ -240,4 +270,50 @@ impl AMQP { ) .await } + + /// # DM Call Update + /// Used to send an update about a DM call, eg. start or end of a call. + /// Recipients can be used to narrow the scope of recipients, otherwise all recipients will be notified. + /// `ended` refers to the ringing period, not necessarily the call itself. + pub async fn dm_call_updated( + &self, + initiator_id: &str, + channel_id: &str, + started_at: Option<&str>, + ended: bool, + recipients: Option>, + ) -> Result<(), AMQPError> { + let config = revolt_config::config().await; + + let payload = InternalDmCallPayload { + payload: DmCallPayload { + initiator_id: initiator_id.to_string(), + channel_id: channel_id.to_string(), + started_at: started_at.map(|f| f.to_string()), + ended, + }, + recipients, + }; + let payload = to_string(&payload).unwrap(); + + debug!( + "Sending dm call update payload on channel {}: {}", + config.pushd.get_dm_call_routing_key(), + payload + ); + + self.channel + .basic_publish( + BasicProperties::default() + .with_content_type("application/json") + .with_persistence(true) + .finish(), + payload.into(), + BasicPublishArguments::new( + &config.pushd.exchange, + &config.pushd.get_dm_call_routing_key(), + ), + ) + .await + } } diff --git a/crates/core/database/src/events/rabbit.rs b/crates/core/database/src/events/rabbit.rs index 0ae91be1..6f5c9ab3 100644 --- a/crates/core/database/src/events/rabbit.rs +++ b/crates/core/database/src/events/rabbit.rs @@ -37,6 +37,20 @@ pub struct GenericPayload { pub user: User, } +#[derive(Serialize, Deserialize, Clone)] +pub struct DmCallPayload { + pub initiator_id: String, + pub channel_id: String, + pub started_at: Option, + pub ended: bool, +} + +#[derive(Serialize, Deserialize, Clone)] +pub struct InternalDmCallPayload { + pub payload: DmCallPayload, + pub recipients: Option>, +} + #[derive(Serialize, Deserialize)] #[serde(tag = "type", content = "data")] #[allow(clippy::large_enum_variant)] @@ -46,6 +60,7 @@ pub enum PayloadKind { FRReceived(FRReceivedPayload), BadgeUpdate(usize), Generic(GenericPayload), + DmCallStartEnd(DmCallPayload), } #[derive(Serialize, Deserialize)] diff --git a/crates/core/database/src/models/users/model.rs b/crates/core/database/src/models/users/model.rs index 881cee30..0c2b6cee 100644 --- a/crates/core/database/src/models/users/model.rs +++ b/crates/core/database/src/models/users/model.rs @@ -7,7 +7,6 @@ use futures::future::join_all; use iso8601_timestamp::Timestamp; use once_cell::sync::Lazy; use rand::seq::SliceRandom; -use redis_kiss::{get_connection, AsyncCommands}; use revolt_config::{config, FeaturesLimits}; use revolt_models::v0::{self, UserBadges, UserFlags}; use revolt_presence::filter_online; @@ -640,19 +639,6 @@ impl User { } } - /// Gets current voice channel - /// - /// current_context: Either the channel id if a dm or group, or server_id if in a server - pub async fn current_voice_channel(&self, current_context: &str) -> Result> { - let mut conn = get_connection() - .await - .map_err(|_| create_error!(InternalError))?; - - conn.get::<_, Option>(format!("vc-{}-{current_context}", &self.id)) - .await - .map_err(|_| create_error!(InternalError)) - } - /// Update user data pub async fn update( &mut self, diff --git a/crates/core/database/src/util/funcs.rs b/crates/core/database/src/util/funcs.rs new file mode 100644 index 00000000..24f9906d --- /dev/null +++ b/crates/core/database/src/util/funcs.rs @@ -0,0 +1,24 @@ +use crate::Database; +use revolt_result::Result; + +/// Formats a user's name depending on their optional features and location. +/// Factors in server display names and user display names before falling back to username#discriminator. +/// Passing a server in which the user is not a member will result in an Err. +pub async fn format_display_name( + db: &Database, + user_id: &str, + server_id: Option<&str>, +) -> Result { + if let Some(server_id) = server_id { + let member = db.fetch_member(server_id, user_id).await?; + if let Some(nick) = member.nickname { + return Ok(nick); + } + } + + let user = db.fetch_user(user_id).await?; + if let Some(display) = user.display_name { + return Ok(display); + } + Ok(format!("{}#{}", user.username, user.discriminator)) +} diff --git a/crates/core/database/src/util/mod.rs b/crates/core/database/src/util/mod.rs index 1baf7d8b..1a03a176 100644 --- a/crates/core/database/src/util/mod.rs +++ b/crates/core/database/src/util/mod.rs @@ -1,6 +1,9 @@ pub mod bridge; pub mod bulk_permissions; +mod funcs; pub mod idempotency; pub mod permissions; pub mod reference; pub mod test_fixtures; + +pub use funcs::*; diff --git a/crates/core/database/src/voice/mod.rs b/crates/core/database/src/voice/mod.rs index 11be9020..d126c7fe 100644 --- a/crates/core/database/src/voice/mod.rs +++ b/crates/core/database/src/voice/mod.rs @@ -424,7 +424,8 @@ pub async fn sync_user_voice_permissions( let before = update_event.clone(); - let can_video = limits.video && permissions.has_channel_permission(ChannelPermission::Video); + let can_video = + limits.video && permissions.has_channel_permission(ChannelPermission::Video); let can_speak = permissions.has_channel_permission(ChannelPermission::Speak); let can_listen = permissions.has_channel_permission(ChannelPermission::Listen); diff --git a/crates/daemons/pushd/src/consumers/inbound/dm_call.rs b/crates/daemons/pushd/src/consumers/inbound/dm_call.rs new file mode 100644 index 00000000..99e208d1 --- /dev/null +++ b/crates/daemons/pushd/src/consumers/inbound/dm_call.rs @@ -0,0 +1,162 @@ +use std::collections::HashMap; + +use crate::consumers::inbound::internal::*; +use amqprs::{ + channel::{BasicPublishArguments, Channel}, + connection::Connection, + consumer::AsyncConsumer, + BasicProperties, Deliver, +}; +use anyhow::Result; +use async_trait::async_trait; +use log::debug; +use revolt_database::{events::rabbit::*, Database}; + +pub struct DmCallConsumer { + #[allow(dead_code)] + db: Database, + authifier_db: authifier::Database, + conn: Option, + channel: Option, +} + +impl Channeled for DmCallConsumer { + fn get_connection(&self) -> Option<&Connection> { + if self.conn.is_none() { + None + } else { + Some(self.conn.as_ref().unwrap()) + } + } + + fn get_channel(&self) -> Option<&Channel> { + if self.channel.is_none() { + None + } else { + Some(self.channel.as_ref().unwrap()) + } + } + + fn set_connection(&mut self, conn: Connection) { + self.conn = Some(conn); + } + + fn set_channel(&mut self, channel: Channel) { + self.channel = Some(channel) + } +} + +impl DmCallConsumer { + pub fn new(db: Database, authifier_db: authifier::Database) -> DmCallConsumer { + DmCallConsumer { + db, + authifier_db, + conn: None, + channel: None, + } + } + + async fn consume_event( + &mut self, + _channel: &Channel, + _deliver: Deliver, + _basic_properties: BasicProperties, + content: Vec, + ) -> Result<()> { + let content = String::from_utf8(content)?; + let _p: InternalDmCallPayload = serde_json::from_str(content.as_str())?; + let payload = _p.payload; + + debug!("Received dm call start/stop event"); + + let call_recipients: Vec = if let Some(recipients) = _p.recipients { + recipients + } else { + match self.db.fetch_channel(&payload.channel_id).await? { + revolt_database::Channel::DirectMessage { recipients, .. } + | revolt_database::Channel::Group { recipients, .. } => recipients + .into_iter() + .filter(|r| *r != payload.initiator_id) + .collect(), + _ => { + warn!( + "Discarding dm call start/stop event for non-dm/group channel {}", + payload.channel_id + ); + return Ok(()); + } + } + }; + let config = revolt_config::config().await; + + for user_id in call_recipients { + if let Ok(sessions) = self.authifier_db.find_sessions(&user_id).await { + for session in sessions { + if let Some(sub) = session.subscription { + let mut sendable = PayloadToService { + notification: PayloadKind::DmCallStartEnd(payload.clone()), + token: sub.auth, + user_id: session.user_id, + session_id: session.id, + extras: HashMap::new(), + }; + + let args: BasicPublishArguments; + + if sub.endpoint == "apn" { + args = BasicPublishArguments::new( + config.pushd.exchange.as_str(), + config.pushd.apn.queue.as_str(), + ) + .finish(); + } else if sub.endpoint == "fcm" { + args = BasicPublishArguments::new( + config.pushd.exchange.as_str(), + config.pushd.fcm.queue.as_str(), + ) + .finish(); + } else { + // web push (vapid) + args = BasicPublishArguments::new( + config.pushd.exchange.as_str(), + config.pushd.vapid.queue.as_str(), + ) + .finish(); + sendable.extras.insert("p265dh".to_string(), sub.p256dh); + sendable + .extras + .insert("endpoint".to_string(), sub.endpoint.clone()); + } + + let payload = serde_json::to_string(&sendable)?; + + publish_message(self, payload.into(), args).await; + } + } + } + } + + Ok(()) + } +} + +#[allow(unused_variables)] +#[async_trait] +impl AsyncConsumer for DmCallConsumer { + /// This consumer handles delegating messages into their respective platform queues. + async fn consume( + &mut self, + channel: &Channel, + deliver: Deliver, + basic_properties: BasicProperties, + content: Vec, + ) { + if let Err(err) = self + .consume_event(channel, deliver, basic_properties, content) + .await + { + revolt_config::capture_anyhow(&err); + warn!("Failed to process dm call start/stop event: {err:?}"); + } + } +} diff --git a/crates/daemons/pushd/src/consumers/inbound/mod.rs b/crates/daemons/pushd/src/consumers/inbound/mod.rs index 9c359336..4d6d36b2 100644 --- a/crates/daemons/pushd/src/consumers/inbound/mod.rs +++ b/crates/daemons/pushd/src/consumers/inbound/mod.rs @@ -1,4 +1,5 @@ pub mod ack; +pub mod dm_call; pub mod fr_accepted; pub mod fr_received; pub mod generic; diff --git a/crates/daemons/pushd/src/consumers/outbound/apn.rs b/crates/daemons/pushd/src/consumers/outbound/apn.rs index 02f9c0da..ec478c1f 100644 --- a/crates/daemons/pushd/src/consumers/outbound/apn.rs +++ b/crates/daemons/pushd/src/consumers/outbound/apn.rs @@ -47,6 +47,32 @@ impl<'a> PayloadLike for MessagePayload<'a> { } } +#[derive(Serialize, Debug)] +struct CallStartStopPayload<'a> { + aps: APS<'a>, + #[serde(skip_serializing)] + options: NotificationOptions<'a>, + #[serde(skip_serializing)] + device_token: &'a str, + + initiator_id: &'a str, + #[serde(rename = "camelCase")] + channel_id: &'a str, + #[serde(rename = "camelCase")] + started_at: &'a str, + #[serde(rename = "camelCase")] + ended: bool, +} + +impl<'a> PayloadLike for CallStartStopPayload<'a> { + fn get_device_token(&self) -> &'a str { + self.device_token + } + fn get_options(&self) -> &NotificationOptions { + &self.options + } +} + // region: consumer pub struct ApnsOutboundConsumer { @@ -310,6 +336,7 @@ impl ApnsOutboundConsumer { ); resp = self.client.send(apn_payload).await; } + PayloadKind::BadgeUpdate(badge) => { let apn_payload = Payload { aps: APS { @@ -324,6 +351,35 @@ impl ApnsOutboundConsumer { debug!("Sending badge update for user: {:}", &payload.user_id); resp = self.client.send(apn_payload).await; } + + PayloadKind::DmCallStartEnd(alert) => { + let started_at = alert.started_at.map_or(String::new(), |f| f.clone()); + + let apn_payload = CallStartStopPayload { + aps: APS { + alert: None, + badge: self.get_badge_count(&payload.user_id).await, + sound: None, + thread_id: None, + content_available: None, + category: None, + mutable_content: Some(1), + url_args: None, + }, + device_token: &payload.token, + options: payload_options.clone(), + initiator_id: &alert.initiator_id, + channel_id: &alert.channel_id, + started_at: &started_at, + ended: alert.ended, + }; + + debug!( + "Sending call start/stop notification for user: {:}", + &payload.user_id + ); + resp = self.client.send(apn_payload).await; + } } if let Err(err) = resp { diff --git a/crates/daemons/pushd/src/consumers/outbound/fcm.rs b/crates/daemons/pushd/src/consumers/outbound/fcm.rs index b13829c9..196a9eea 100644 --- a/crates/daemons/pushd/src/consumers/outbound/fcm.rs +++ b/crates/daemons/pushd/src/consumers/outbound/fcm.rs @@ -5,11 +5,12 @@ use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProp use anyhow::{anyhow, bail, Result}; use async_trait::async_trait; use fcm_v1::{ - android::AndroidConfig, + android::{AndroidConfig, AndroidMessagePriority}, auth::{Authenticator, ServiceAccountKey}, message::{Message, Notification}, Client, Error as FcmError, }; +use revolt_config::config; use revolt_database::{events::rabbit::*, Database}; use revolt_models::v0::{Channel, PushNotification}; use serde_json::Value; @@ -170,6 +171,37 @@ impl FcmOutboundConsumer { resp = self.client.send(&msg).await; } + PayloadKind::DmCallStartEnd(alert) => { + let mut data: HashMap = HashMap::new(); + data.insert( + "initiator_id".to_string(), + Value::String(alert.initiator_id), + ); + data.insert("channel_id".to_string(), Value::String(alert.channel_id)); + data.insert( + "started_at".to_string(), + Value::String(alert.started_at.unwrap_or_else(|| "".to_string())), + ); + data.insert("ended".to_string(), Value::Bool(alert.ended)); + + let msg = Message { + token: Some(payload.token), + notification: None, + data: Some(data), + android: Some(AndroidConfig { + priority: Some(AndroidMessagePriority::High), + ttl: Some(format!( + "{}s", + config().await.api.livekit.call_ring_duration + )), + ..Default::default() + }), + ..Default::default() + }; + + resp = self.client.send(&msg).await; + } + PayloadKind::BadgeUpdate(_) => { bail!("FCM cannot handle badge updates and they should not be sent here."); } diff --git a/crates/daemons/pushd/src/consumers/outbound/vapid.rs b/crates/daemons/pushd/src/consumers/outbound/vapid.rs index 92ce6e56..67ec31fb 100644 --- a/crates/daemons/pushd/src/consumers/outbound/vapid.rs +++ b/crates/daemons/pushd/src/consumers/outbound/vapid.rs @@ -8,7 +8,7 @@ use base64::{ engine::{self}, Engine as _, }; -use revolt_database::{events::rabbit::*, Database}; +use revolt_database::{events::rabbit::*, util::format_display_name, Database}; use web_push::{ ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder, WebPushClient, WebPushError, WebPushMessageBuilder, @@ -107,6 +107,33 @@ impl VapidOutboundConsumer { PayloadKind::MessageNotification(alert) => { payload_body = serde_json::to_string(&alert)?; } + PayloadKind::DmCallStartEnd(alert) => { + let initiator_name = if let Some(server_id) = + self.db.fetch_channel(&alert.channel_id).await?.server() + { + format_display_name(&self.db, &alert.initiator_id, Some(server_id)).await + } else { + format_display_name(&self.db, &alert.initiator_id, None).await + }?; + + let channel = self.db.fetch_channel(&alert.channel_id).await?; + let mut body = HashMap::new(); + + match channel { + revolt_database::Channel::DirectMessage { .. } => { + body.insert("body", format!("{} is calling you", initiator_name)); + } + revolt_database::Channel::Group { name, .. } => { + body.insert( + "body", + format!("{} is calling your group, {}", initiator_name, name), + ); + } + _ => bail!("Invalid DmCallStart/End channel type"), + } + + payload_body = serde_json::to_string(&body)?; + } PayloadKind::BadgeUpdate(_) => { bail!("Vapid cannot handle badge updates and they should not be sent here."); } diff --git a/crates/daemons/pushd/src/main.rs b/crates/daemons/pushd/src/main.rs index 78845a63..472caafe 100644 --- a/crates/daemons/pushd/src/main.rs +++ b/crates/daemons/pushd/src/main.rs @@ -16,8 +16,9 @@ use tokio::sync::Notify; mod consumers; use consumers::{ inbound::{ - ack::AckConsumer, fr_accepted::FRAcceptedConsumer, fr_received::FRReceivedConsumer, - generic::GenericConsumer, mass_mention::MassMessageConsumer, message::MessageConsumer, + ack::AckConsumer, dm_call::DmCallConsumer, fr_accepted::FRAcceptedConsumer, + fr_received::FRReceivedConsumer, generic::GenericConsumer, + mass_mention::MassMessageConsumer, message::MessageConsumer, }, outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer}, }; @@ -102,6 +103,7 @@ async fn main() { .await, ); + // inbound: Mass Mentions connections.push( make_queue_and_consume( &config, @@ -113,6 +115,18 @@ async fn main() { .await, ); + // inbound: Dm Calls + connections.push( + make_queue_and_consume( + &config, + &config.pushd.dm_call_queue, + config.pushd.get_dm_call_routing_key().as_str(), + None, + DmCallConsumer::new(db.clone(), authifier.clone()), + ) + .await, + ); + if !config.pushd.apn.pkcs8.is_empty() { connections.push( make_queue_and_consume( diff --git a/crates/daemons/voice-ingress/src/api.rs b/crates/daemons/voice-ingress/src/api.rs index 9004a17c..f470b936 100644 --- a/crates/daemons/voice-ingress/src/api.rs +++ b/crates/daemons/voice-ingress/src/api.rs @@ -59,9 +59,7 @@ pub async fn ingress( let channel_id = channel_id.to_internal_error()?; let user_id = user_id.to_internal_error()?; - let channel = Reference::from_unchecked(channel_id) - .as_channel(db) - .await?; + let channel = Reference::from_unchecked(channel_id).as_channel(db).await?; let voice_state = create_voice_state(channel_id, channel.server(), user_id).await?; @@ -86,9 +84,7 @@ pub async fn ingress( // 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 user = Reference::from_unchecked(user_id).as_user(db).await?; let mut call_started_message = SystemMessage::CallStarted { by: user_id.to_string(), @@ -120,9 +116,7 @@ pub async fn ingress( let channel_id = channel_id.to_internal_error()?; let user_id = user_id.to_internal_error()?; - let channel = Reference::from_unchecked(channel_id) - .as_channel(db) - .await?; + let channel = Reference::from_unchecked(channel_id).as_channel(db).await?; delete_voice_state(channel_id, channel.server(), user_id).await?; @@ -143,6 +137,14 @@ pub async fn ingress( let members = get_voice_channel_members(channel_id).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 let Some(system_message_id) = take_channel_call_started_system_message(channel_id).await? { @@ -179,13 +181,9 @@ pub async fn ingress( let user_id = user_id.to_internal_error()?; let track = event.track.as_ref().to_internal_error()?; - let channel = Reference::from_unchecked(channel_id) - .as_channel(db) - .await?; + let channel = Reference::from_unchecked(channel_id).as_channel(db).await?; - let user = Reference::from_unchecked(user_id) - .as_user(db) - .await?; + let user = Reference::from_unchecked(user_id).as_user(db).await?; let user_limits = user.limits().await; diff --git a/crates/daemons/voice-ingress/src/main.rs b/crates/daemons/voice-ingress/src/main.rs index 8c599394..45191c2c 100644 --- a/crates/daemons/voice-ingress/src/main.rs +++ b/crates/daemons/voice-ingress/src/main.rs @@ -1,10 +1,5 @@ use std::env; -use amqprs::{ - channel::ExchangeDeclareArguments, - connection::{Connection, OpenConnectionArguments}, -}; -use revolt_config::config; use revolt_database::DatabaseInfo; use revolt_database::{voice::VoiceClient, AMQP}; use revolt_result::Result; @@ -18,36 +13,11 @@ mod guard; async fn main() -> Result<(), rocket::Error> { revolt_config::configure!(voice_ingress); - let config = config().await; + let amqp = AMQP::new_auto().await; let database = DatabaseInfo::Auto.connect().await.unwrap(); let voice_client = VoiceClient::from_revolt_config().await; - let connection = Connection::open(&OpenConnectionArguments::new( - &config.rabbit.host, - config.rabbit.port, - &config.rabbit.username, - &config.rabbit.password, - )) - .await - .expect("Failed to connect to RabbitMQ"); - - let channel = connection - .open_channel(None) - .await - .expect("Failed to open RabbitMQ channel"); - - channel - .exchange_declare( - ExchangeDeclareArguments::new(&config.pushd.exchange, "direct") - .durable(true) - .finish(), - ) - .await - .expect("Failed to declare exchange"); - - let amqp = AMQP::new(connection, channel); - let _rocket = build() .manage(database) .manage(voice_client) diff --git a/crates/delta/src/routes/channels/mod.rs b/crates/delta/src/routes/channels/mod.rs index c57fbaa8..5d3fbc85 100644 --- a/crates/delta/src/routes/channels/mod.rs +++ b/crates/delta/src/routes/channels/mod.rs @@ -25,6 +25,7 @@ mod message_unreact; mod permissions_set; mod permissions_set_default; mod voice_join; +mod voice_stop_ring; mod webhook_create; mod webhook_fetch_all; @@ -49,6 +50,7 @@ pub fn routes() -> (Vec, OpenApi) { group_add_member::add_member, group_remove_member::remove_member, voice_join::call, + voice_stop_ring::stop_ring, permissions_set::set_role_permissions, permissions_set_default::set_default_channel_permissions, message_react::react_message, diff --git a/crates/delta/src/routes/channels/voice_join.rs b/crates/delta/src/routes/channels/voice_join.rs index 58d35fd7..30f06d1f 100644 --- a/crates/delta/src/routes/channels/voice_join.rs +++ b/crates/delta/src/routes/channels/voice_join.rs @@ -5,7 +5,7 @@ use revolt_database::{ delete_voice_state, get_channel_node, get_user_voice_channels, get_voice_channel_members, raise_if_in_voice, VoiceClient, }, - Database, User, + Channel, Database, SystemMessage, User, AMQP, }; use revolt_models::v0; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; @@ -20,6 +20,7 @@ use rocket::{serde::json::Json, State}; #[post("//join_call", data = "")] pub async fn call( db: &State, + amqp: &State, voice_client: &State, user: User, target: Reference<'_>, @@ -99,6 +100,42 @@ pub async fn call( log::debug!("Created room {}", room.name); + match &channel { + Channel::DirectMessage { .. } | Channel::Group { .. } => { + let mut msg = SystemMessage::CallStarted { + 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, url: node_host.clone(), diff --git a/crates/delta/src/routes/channels/voice_stop_ring.rs b/crates/delta/src/routes/channels/voice_stop_ring.rs new file mode 100644 index 00000000..f8db8fc4 --- /dev/null +++ b/crates/delta/src/routes/channels/voice_stop_ring.rs @@ -0,0 +1,69 @@ +use revolt_database::{ + util::reference::Reference, + voice::{get_voice_state, VoiceClient}, + Channel, Database, User, AMQP, +}; +use revolt_result::{create_error, Result, ToRevoltError}; + +use rocket::State; +use rocket_empty::EmptyResponse; + +/// # Stop Ring +/// Stops ringing a specific user in a dm call. +/// You must be in the call to use this endpoint, returns NotConnected otherwise. +/// Only valid in DM/Group channels, will return NoEffect in servers. +/// Returns NotFound if the user is not in the dm/group channel +#[openapi(tag = "Voice")] +#[put("//end_ring/")] +pub async fn stop_ring( + db: &State, + amqp: &State, + voice: &State, + user: User, + target: Reference<'_>, + target_user: Reference<'_>, +) -> Result { + if !voice.is_enabled() { + return Err(create_error!(LiveKitUnavailable)); + } + + let channel = target.as_channel(db).await?; + if channel.server().is_some() { + return Err(create_error!(NoEffect)); + } + + if get_voice_state(channel.id(), None, &user.id) + .await? + .is_none() + { + return Err(create_error!(NotConnected)); + } + + let members = match channel { + Channel::DirectMessage { ref recipients, .. } | Channel::Group { ref recipients, .. } => { + recipients + } + _ => return Err(create_error!(NoEffect)), + }; + + if members.iter().any(|m| &target_user.id == m) { + if let Err(e) = amqp + .dm_call_updated( + &user.id, + channel.id(), + None, + true, + Some(vec![target_user.id.to_string()]), + ) + .await + .to_internal_error() + { + revolt_config::capture_internal_error!(&e); + return Err(e); + } + + Ok(EmptyResponse) + } else { + Err(create_error!(NotFound)) + } +} diff --git a/crates/delta/src/routes/root.rs b/crates/delta/src/routes/root.rs index aa3293cd..3ddbfcc7 100644 --- a/crates/delta/src/routes/root.rs +++ b/crates/delta/src/routes/root.rs @@ -27,7 +27,7 @@ pub struct VoiceNode { pub name: String, pub lat: f64, pub lon: f64, - pub public_url: String + pub public_url: String, } /// # Voice Server Configuration @@ -115,18 +115,24 @@ pub async fn root() -> Result> { }, livekit: VoiceFeature { enabled: !config.hosts.livekit.is_empty(), - nodes: config.api.livekit.nodes + nodes: config + .api + .livekit + .nodes .iter() .filter(|(_, node)| !node.private) - .map(|(name, value)| { - VoiceNode { - name: name.clone(), - lat: value.lat, - lon: value.lon, - public_url: config.hosts.livekit.get(name).expect("Missing corresponding host for voice node").clone() - } + .map(|(name, value)| VoiceNode { + name: name.clone(), + lat: value.lat, + lon: value.lon, + public_url: config + .hosts + .livekit + .get(name) + .expect("Missing corresponding host for voice node") + .clone(), }) - .collect() + .collect(), }, }, ws: config.hosts.events,