feat: publish events over amqp

Signed-off-by: Zomatree <me@zomatree.live>

crates/core/config/Revolt.toml

@@ -34,6 +34,7 @@ default_exchange = "revolt.default"
 
 [rabbit.queues]
 acks = "internal.ack"
+events = "internal.event"
 
 [api]
 

crates/core/config/src/lib.rs

@@ -125,6 +125,7 @@ pub struct Database {
 #[derive(Deserialize, Debug, Clone)]
 pub struct RabbitQueues {
     pub acks: String,
+    pub events: String,
 }
 
 #[derive(Deserialize, Debug, Clone)]

crates/core/database/src/amqp/amqp.rs

@@ -1,20 +1,29 @@
-use std::collections::HashSet;
-use std::sync::Arc;
+use std::{
+    collections::HashSet,
+    sync::{Arc, OnceLock},
+};
 
-use crate::events::rabbit::*;
+use crate::events::{client::EventV1, rabbit::*};
 use crate::User;
 use lapin::{
     options::BasicPublishOptions,
     protocol::basic::AMQPProperties,
     types::{AMQPValue, FieldTable},
-    Channel, Connection, ConnectionProperties, Error as AMQPError,
+    BasicProperties, Channel, Connection, ConnectionProperties, Error as AMQPError,
 };
+use revolt_config::config;
 use revolt_models::v0::PushNotification;
 use revolt_presence::filter_online;
 use revolt_result::Result;
 
 use serde_json::to_string;
 
+static AMQP_INSTANCE: OnceLock<AMQP> = OnceLock::new();
+
+pub fn get_amqp() -> &'static AMQP {
+    AMQP_INSTANCE.get().expect("No AMQP instance set.")
+}
+
 #[derive(Clone)]
 pub struct AMQP {
     friend_request_accepted: Arc<Channel>,
@@ -25,13 +34,14 @@ pub struct AMQP {
     ack_notification_message: Arc<Channel>,
     dm_call_updated: Arc<Channel>,
     process_ack: Arc<Channel>,
+    publish_event: Arc<Channel>,
     #[allow(unused)]
     connection: Arc<Connection>,
 }
 
 impl AMQP {
     pub async fn new(connection: Arc<Connection>) -> Self {
-        Self {
+        let this = Self {
             friend_request_accepted: Self::create_channel(&connection).await,
             friend_request_received: Self::create_channel(&connection).await,
             generic_message: Self::create_channel(&connection).await,
@@ -40,8 +50,13 @@ impl AMQP {
             ack_notification_message: Self::create_channel(&connection).await,
             dm_call_updated: Self::create_channel(&connection).await,
             process_ack: Self::create_channel(&connection).await,
+            publish_event: Self::create_channel(&connection).await,
             connection,
-        }
+        };
+
+        let _ = AMQP_INSTANCE.set(this.clone());
+
+        this
     }
 
     pub async fn new_auto() -> Self {
@@ -379,4 +394,23 @@ impl AMQP {
 
         Ok(())
     }
+
+    pub async fn publish_event(&self, channel: String, event: &EventV1) -> Result<(), AMQPError> {
+        let mut headers = FieldTable::default();
+        headers.insert("c".into(), AMQPValue::LongString(channel.into()));
+
+        let config = config().await;
+
+        self.publish_event
+            .basic_publish(
+                config.rabbit.default_exchange.clone().into(),
+                config.rabbit.queues.events.into(),
+                BasicPublishOptions::default(),
+                &serde_json::to_vec(event).unwrap(),
+                BasicProperties::default().with_headers(headers),
+            )
+            .await?;
+
+        Ok(())
+    }
 }

crates/core/database/src/amqp/mod.rs

@@ -1,2 +1,4 @@
 #[allow(clippy::module_inception)]
 pub mod amqp;
+
+pub use amqp::{AMQP, get_amqp};

crates/core/database/src/events/client.rs

@@ -11,7 +11,7 @@ use revolt_models::v0::{
     UserVoiceState, Webhook,
 };
 
-use crate::Database;
+use crate::{Database, amqp::get_amqp};
 
 /// Ping Packet
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -372,8 +372,16 @@ impl EventV1 {
         #[cfg(debug_assertions)]
         info!("Publishing event to {channel}: {self:?}");
 
-        #[cfg(debug_assertions)]
-        redis_kiss::publish(channel, self).await.unwrap();
+        // #[cfg(debug_assertions)]
+        // redis_kiss::publish(channel, self).await.unwrap();
+
+        if let Err(e) = get_amqp().publish_event(channel, &self).await {
+            if cfg!(debug_assertions) {
+                panic!("{e:?}");
+            } else {
+                log::error!("{e:?}");
+            };
+        };
     }
 
     /// Publish user event