Fix serialization.

src/database/message.rs

@@ -1,9 +1,9 @@
 use super::get_collection;
 use crate::guards::channel::ChannelRef;
-use crate::routes::channel::ChannelType;
 use crate::notifications;
-use crate::notifications::events::Notification::MessageCreate;
 use crate::notifications::events::message::Create;
+use crate::notifications::events::Notification::MessageCreate;
+use crate::routes::channel::ChannelType;
 
 use bson::{doc, to_bson, UtcDateTime};
 use serde::{Deserialize, Serialize};
@@ -37,20 +37,18 @@ impl Message {
             .insert_one(to_bson(&self).unwrap().as_document().unwrap().clone(), None)
             .is_ok()
         {
-            let data = MessageCreate(
+            let data = MessageCreate(Create {
-                Create {
+                id: self.id.clone(),
-                    id: self.id.clone(),
+                nonce: self.nonce.clone(),
-                    nonce: self.nonce.clone(),
+                channel: self.channel.clone(),
-                    channel: self.channel.clone(),
+                author: self.author.clone(),
-                    author: self.author.clone(),
+                content: self.content.clone(),
-                    content: self.content.clone(),
+            });
-                }
-            );
 
             match target.channel_type {
-                0..=1 => notifications::send_message(target.recipients.clone(), None, data),
+                0..=1 => notifications::send_message_threaded(target.recipients.clone(), None, data),
-                2 => notifications::send_message(target.recipients.clone(), None, data),
+                2 => notifications::send_message_threaded(None, target.guild.clone(), data),
-                _ => unreachable!()
+                _ => unreachable!(),
             };
 
             let short_content: String = self.content.chars().take(24).collect();

src/main.rs

@@ -20,6 +20,7 @@ use std::thread;
 fn main() {
     dotenv::dotenv().ok();
     database::connect();
+    notifications::start_worker();
 
     thread::spawn(|| {
         notifications::pubsub::launch_subscriber();

src/notifications/events/message.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Create {
     pub id: String,
     pub nonce: Option<String>,

src/notifications/events/mod.rs

@@ -1,8 +1,27 @@
 use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
 
 pub mod message;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum Notification {
     MessageCreate(message::Create),
 }
+
+impl Notification {
+    pub fn serialize(self) -> String {
+        if let Value::Object(obj) = json!(self) {
+            let (key, value) = obj.iter().next().unwrap();
+
+            if let Value::Object(data) = value {
+                let mut data = data.clone();
+                data.insert("type".to_string(), Value::String(key.to_string()));
+                json!(data).to_string()
+            } else {
+                unreachable!()
+            }
+        } else {
+            unreachable!()
+        }
+    }
+}

src/notifications/mod.rs

@@ -1,3 +1,7 @@
+use once_cell::sync::OnceCell;
+use std::sync::mpsc::{channel, Sender};
+use std::thread;
+
 pub mod events;
 pub mod pubsub;
 pub mod state;
@@ -11,11 +15,50 @@ pub fn send_message<U: Into<Option<Vec<String>>>, G: Into<Option<String>>>(
     let users = users.into();
     let guild = guild.into();
 
-    if pubsub::send_message(users.clone(), guild.clone(), data) {
+    if pubsub::send_message(users.clone(), guild.clone(), data.clone()) {
-        state::send_message(users, guild, "bruh".to_string());
+        state::send_message(users, guild, data.serialize());
 
         true
     } else {
         false
     }
 }
+
+struct NotificationArguments {
+    users: Option<Vec<String>>,
+    guild: Option<String>,
+    data: events::Notification,
+}
+
+static mut SENDER: OnceCell<Sender<NotificationArguments>> = OnceCell::new();
+
+pub fn start_worker() {
+    let (sender, receiver) = channel();
+    unsafe {
+        SENDER.set(sender).unwrap();
+    }
+
+    thread::spawn(move || {
+        while let Ok(data) = receiver.recv() {
+            send_message(data.users, data.guild, data.data);
+        }
+    });
+}
+
+pub fn send_message_threaded<U: Into<Option<Vec<String>>>, G: Into<Option<String>>>(
+    users: U,
+    guild: G,
+    data: events::Notification,
+) -> bool {
+    unsafe {
+        SENDER
+            .get()
+            .unwrap()
+            .send(NotificationArguments {
+                users: users.into(),
+                guild: guild.into(),
+                data,
+            })
+            .is_ok()
+    }
+}

src/notifications/pubsub.rs

@@ -23,11 +23,7 @@ pub struct PubSubMessage {
     data: Notification,
 }
 
-pub fn send_message(
+pub fn send_message(users: Option<Vec<String>>, guild: Option<String>, data: Notification) -> bool {
-    users: Option<Vec<String>>,
-    guild: Option<String>,
-    data: Notification,
-) -> bool {
     let message = PubSubMessage {
         id: Ulid::new().to_string(),
         source: SOURCEID.get().unwrap().to_string(),
@@ -92,7 +88,7 @@ pub fn launch_subscriber() {
                                 super::state::send_message(
                                     message.user_recipients,
                                     message.target_guild,
-                                    json!(message.data).to_string(),
+                                    message.data.serialize(),
                                 );
                             }
                         } else {

src/notifications/state.rs

@@ -1,4 +1,3 @@
-use super::events::Notification;
 use crate::database;
 use crate::util::vec_to_set;
 
@@ -174,14 +173,10 @@ pub fn init() {
     }
 }
 
-pub fn send_message(
+pub fn send_message(users: Option<Vec<String>>, guild: Option<String>, data: String) {
-    users: Option<Vec<String>>,
-    guild: Option<String>,
-    data: String,
-) {
     let state = unsafe { DATA.get().unwrap().read().unwrap() };
     let mut connections = HashSet::new();
-    
+
     let mut users = vec_to_set(&users.unwrap_or(vec![]));
     if let Some(guild) = guild {
         if let Some(entry) = state.guilds.get(&guild) {