fix/hosepipe

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

@@ -11,6 +11,8 @@ use revolt_presence::filter_online;
 
 use serde_json::to_string;
 
+// TODO: move away from storing Connection and Channel to using thread local singletons - #659
+
 #[derive(Clone)]
 pub struct AMQP {
     #[allow(unused)]

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

@@ -1,12 +1,20 @@
+use amqprs::{
+    channel::{BasicPublishArguments},
+    BasicProperties, FieldTable,
+};
 use authifier::AuthifierEvent;
 use revolt_result::Error;
 use serde::{Deserialize, Serialize};
 
 use revolt_models::v0::{
-    AppendMessage, Channel, ChannelUnread, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, UserVoiceState, Webhook
+    AppendMessage, Channel, ChannelUnread, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember,
+    FieldsMessage, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey,
+    Message, PartialChannel, PartialMember, PartialMessage, PartialRole, PartialServer,
+    PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report,
+    Server, User, UserSettings, UserVoiceState, Webhook,
 };
 
-use crate::Database;
+use crate::{util::rabbit::get_channel, Database};
 
 /// Ping Packet
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -304,14 +312,23 @@ pub enum EventV1 {
 impl EventV1 {
     /// Publish helper wrapper
     pub async fn p(self, channel: String) {
-        #[cfg(not(debug_assertions))]
-        redis_kiss::p(channel, self).await;
-
         #[cfg(debug_assertions)]
         info!("Publishing event to {channel}: {self:?}");
 
-        #[cfg(debug_assertions)]
-        redis_kiss::publish(channel, self).await.unwrap();
+        let rmq = get_channel().await;
+
+        let mut headers = FieldTable::new();
+        headers.insert("c".try_into().unwrap(), channel.clone().into());
+        let mut properties = BasicProperties::default();
+        properties.with_headers(headers);
+
+        rmq.basic_publish(
+            properties,
+            serde_json::to_string(&self).unwrap().into_bytes(),
+            BasicPublishArguments::new("events", "events"),
+        )
+        .await
+        .unwrap();
     }
 
     /// Publish user event

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

@@ -5,5 +5,6 @@ pub mod idempotency;
 pub mod permissions;
 pub mod reference;
 pub mod test_fixtures;
+pub mod rabbit;
 
 pub use funcs::*;

crates/core/database/src/util/rabbit.rs

@@ -0,0 +1,48 @@
+use amqprs::{channel::Channel, connection::Connection};
+use once_cell::sync::OnceCell;
+use std::{
+    collections::HashMap, future::ready, sync::{LazyLock, RwLock}, thread::{ThreadId, current}
+};
+
+static RABBIT_CONNECTION: OnceCell<Connection> = OnceCell::new();
+static RABBIT_CHANNELS: LazyLock<RwLock<HashMap<ThreadId, Channel>>> =
+    LazyLock::new(|| RwLock::new(HashMap::new()));
+
+pub async fn get_channel_with_init<F: AsyncFnOnce(Channel) -> Channel>(init: F) -> Channel {
+    let conn = RABBIT_CONNECTION
+        .get()
+        .expect("Rabbit connection is not initialised.");
+
+    let thread_id = current().id();
+
+    let channel = RABBIT_CHANNELS
+        .read()
+        .expect("Channels poisioned")
+        .get(&thread_id)
+        .cloned();
+
+    if let Some(channel) = channel {
+        channel
+    } else {
+        let mut channel =
+            conn.open_channel(None)
+                .await
+                .expect("Failed to open rabbitmq channel");
+
+        channel = init(channel).await;
+
+        RABBIT_CHANNELS
+            .write()
+            .expect("Channels poisioned")
+            .insert(thread_id, channel.clone());
+        channel
+    }
+}
+
+pub async fn get_channel() -> Channel {
+    get_channel_with_init(ready).await
+}
+
+pub fn set_rabbitmq_connection(connection: Connection) -> bool {
+    RABBIT_CONNECTION.set(connection).is_ok()
+}