feat: Send push notifications for dm call start/end

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

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
 

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<String, String>
+    pub livekit: HashMap<String, String>,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -198,7 +198,8 @@ pub struct ApiWorkers {
 
 #[derive(Deserialize, Debug, Clone)]
 pub struct ApiLiveKit {
-    pub nodes: HashMap<String, LiveKitNode>
+    pub call_ring_duration: usize,
+    pub nodes: HashMap<String, LiveKitNode>,
 }
 
 #[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())
     }

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<Vec<String>>,
+    ) -> 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
+    }
 }

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<String>,
+    pub ended: bool,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct InternalDmCallPayload {
+    pub payload: DmCallPayload,
+    pub recipients: Option<Vec<String>>,
+}
+
 #[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)]

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<Option<String>> {
-        let mut conn = get_connection()
-            .await
-            .map_err(|_| create_error!(InternalError))?;
-
-        conn.get::<_, Option<String>>(format!("vc-{}-{current_context}", &self.id))
-            .await
-            .map_err(|_| create_error!(InternalError))
-    }
-
     /// Update user data
     pub async fn update(
         &mut self,

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<String> {
+    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))
+}

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::*;

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);
 

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<Connection>,
+    channel: Option<Channel>,
+}
+
+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<u8>,
+    ) -> 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<String> = 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<u8>,
+    ) {
+        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:?}");
+        }
+    }
+}

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;

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 {

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<String, Value> = 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.");
             }

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.");
             }

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,
+        ack::AckConsumer, dm_call::DmCallConsumer, fr_accepted::FRAcceptedConsumer,
-        generic::GenericConsumer, mass_mention::MassMessageConsumer, message::MessageConsumer,
+        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(

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)
+            let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
-                .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)
+                let user = Reference::from_unchecked(user_id).as_user(db).await?;
-                    .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)
+            let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
-                .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)
+            let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
-                .as_channel(db)
-                .await?;
 
-            let user = Reference::from_unchecked(user_id)
+            let user = Reference::from_unchecked(user_id).as_user(db).await?;
-                .as_user(db)
-                .await?;
 
             let user_limits = user.limits().await;
 

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)

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<Route>, 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,

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("/<target>/join_call", data = "<data>")]
 pub async fn call(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     voice_client: &State<VoiceClient>,
     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(),

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("/<target>/end_ring/<target_user>")]
+pub async fn stop_ring(
+    db: &State<Database>,
+    amqp: &State<AMQP>,
+    voice: &State<VoiceClient>,
+    user: User,
+    target: Reference<'_>,
+    target_user: Reference<'_>,
+) -> Result<EmptyResponse> {
+    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))
+    }
+}

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<Json<RevoltConfig>> {
             },
             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)| {
+                    .map(|(name, value)| VoiceNode {
-                        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()
+                        public_url: config
-                        }
+                            .hosts
+                            .livekit
+                            .get(name)
+                            .expect("Missing corresponding host for voice node")
+                            .clone(),
                     })
-                    .collect()
+                    .collect(),
             },
         },
         ws: config.hosts.events,