From 997d1fffc032719bf42632af2805d1f69de8e428 Mon Sep 17 00:00:00 2001
From: Paul <paulmakles@gmail.com>
Date: Mon, 1 Nov 2021 22:04:11 +0000
Subject: [PATCH] feat(task_queue): process embeds task

closes #112
---
 src/database/entities/message.rs       | 74 ++++----------------------
 src/task_queue/mod.rs                  |  2 +
 src/task_queue/task_last_message_id.rs |  1 +
 src/task_queue/task_process_embeds.rs  | 47 ++++++++++++++++
 4 files changed, 61 insertions(+), 63 deletions(-)
 create mode 100644 src/task_queue/task_process_embeds.rs

diff --git a/src/database/entities/message.rs b/src/database/entities/message.rs
index 8b972005..988e91c0 100644
--- a/src/database/entities/message.rs
+++ b/src/database/entities/message.rs
@@ -194,6 +194,11 @@ impl Message {
                 with: "message",
             })?;
 
+        // spawn task_queue ( process embeds )
+        if process_embeds {
+            self.process_embed().await;
+        }
+
         // spawn task_queue ( update last_message_id )
         match channel {
             Channel::DirectMessage { id, .. } =>
@@ -203,40 +208,18 @@ impl Message {
             _ => {}
         }
 
-        // spawn task_queue ( process embeds )
-
         // if mentions {
         //  spawn task_queue ( update channel_unreads )
         // }
+        /*if let Some(mentions) = &self.mentions {
+
+        }*/
 
         // if (channel => DM | Group) | mentions {
         //  spawn task_queue ( web push )
         // }
 
         /*
-        // ! FIXME: all this code is legitimately crap
-        // ! rewrite when can be asked
-
-        let ss = self.clone();
-        let c_clone = channel.clone();
-        async_std::task::spawn(async move {
-
-            let last_message_id = ss.id.clone();
-            let mut set = doc! { "last_message_id": last_message_id };
-
-            let channels = get_collection("channels");
-            match &c_clone {
-                Channel::DirectMessage { id, .. } => {
-                    // ! MARK AS ACTIVE
-                    set.insert("active", true);
-                    update_channels_last_message(&channels, id, &set).await;
-                },
-                Channel::Group { id, .. } | Channel::TextChannel { id, .. } => {
-                    update_channels_last_message(&channels, id, &set).await;
-                }
-                _ => {}
-            }
-        });
 
         // ! FIXME: also temp code
         // ! THIS ADDS ANY MENTIONS
@@ -269,10 +252,6 @@ impl Message {
             });
         }
 
-        if process_embeds {
-            self.process_embed();
-        }
-
         let mentions = self.mentions.clone();
 
         /*
@@ -356,49 +335,18 @@ impl Message {
             data,
         }
         .publish(channel);
-        self.process_embed();
 
+        self.process_embed().await;
         Ok(())
     }
 
-    pub fn process_embed(&self) {
+    pub async fn process_embed(&self) {
         if !*USE_JANUARY {
             return;
         }
 
         if let Content::Text(text) = &self.content {
-            // ! FIXME: re-write this at some point,
-            // ! or just before we allow user generated embeds
-            let id = self.id.clone();
-            let content = text.clone();
-            let channel = self.channel.clone();
-            async_std::task::spawn(async move {
-                if let Ok(embeds) = Embed::generate(content).await {
-                    if let Ok(bson) = to_bson(&embeds) {
-                        if let Ok(_) = get_collection("messages")
-                            .update_one(
-                                doc! {
-                                    "_id": &id
-                                },
-                                doc! {
-                                    "$set": {
-                                        "embeds": bson
-                                    }
-                                },
-                                None,
-                            )
-                            .await
-                        {
-                            ClientboundNotification::MessageUpdate {
-                                id,
-                                channel: channel.clone(),
-                                data: json!({ "embeds": embeds }),
-                            }
-                            .publish(channel);
-                        }
-                    }
-                }
-            });
+            crate::task_queue::task_process_embeds::queue(self.channel.clone(), self.id.clone(), text.clone()).await;
         }
     }
 
diff --git a/src/task_queue/mod.rs b/src/task_queue/mod.rs
index d013d5a8..10f0b5f8 100644
--- a/src/task_queue/mod.rs
+++ b/src/task_queue/mod.rs
@@ -1,5 +1,7 @@
 pub mod task_last_message_id;
+pub mod task_process_embeds;
 
 pub async fn start_queues() {
     async_std::task::spawn(task_last_message_id::run());
+    async_std::task::spawn(task_process_embeds::run());
 }
diff --git a/src/task_queue/task_last_message_id.rs b/src/task_queue/task_last_message_id.rs
index db13fbb3..31d1b67d 100644
--- a/src/task_queue/task_last_message_id.rs
+++ b/src/task_queue/task_last_message_id.rs
@@ -1,3 +1,4 @@
+// Queue Type: Debounced
 use std::{collections::HashMap, time::{Duration, Instant}};
 
 use async_channel::{ Sender, Receiver, bounded };
diff --git a/src/task_queue/task_process_embeds.rs b/src/task_queue/task_process_embeds.rs
new file mode 100644
index 00000000..ff40d097
--- /dev/null
+++ b/src/task_queue/task_process_embeds.rs
@@ -0,0 +1,47 @@
+// Queue Type: Linear
+use async_channel::{ Sender, Receiver, bounded };
+use mongodb::bson::{doc, to_bson};
+
+use crate::{database::*, notifications::events::ClientboundNotification};
+
+type Message = (String, String, String);
+
+lazy_static! {
+    static ref CHANNEL: (Sender<Message>, Receiver<Message>) = bounded(100);
+}
+
+pub async fn queue(channel: String, id: String, content: String) {
+    CHANNEL.0.send((channel, id, content)).await.ok();
+}
+
+pub async fn run() {
+    let messages = get_collection("messages");
+
+    while let Ok((channel, id, content)) = CHANNEL.1.recv().await {
+        if let Ok(embeds) = Embed::generate(content).await {
+            if let Ok(bson) = to_bson(&embeds) {
+                if let Ok(_) = messages
+                    .update_one(
+                        doc! {
+                            "_id": &id
+                        },
+                        doc! {
+                            "$set": {
+                                "embeds": bson
+                            }
+                        },
+                        None,
+                    )
+                    .await
+                {
+                    ClientboundNotification::MessageUpdate {
+                        id,
+                        channel: channel.clone(),
+                        data: json!({ "embeds": embeds }),
+                    }
+                    .publish(channel);
+                }
+            }
+        }
+    }
+}