feat: implement DelayedTask and last_id task

src/routes/channels/message_send.rs

@@ -220,5 +220,13 @@ pub async fn message_send(
     )
     .await;
 
+    // ! ANOTHER ONE
+    crate::tasks::last_message_id::queue(
+        channel.id().to_string(),
+        message.id.to_string(),
+        channel.is_direct_dm(),
+    )
+    .await;
+
     Ok(Json(message))
 }

src/tasks/last_message_id.rs

@@ -0,0 +1,79 @@
+// Queue Type: Debounced
+use deadqueue::limited::Queue;
+use log::info;
+use mongodb::bson::doc;
+use revolt_quark::{models::channel::PartialChannel, Database};
+use std::{collections::HashMap, time::Duration};
+
+use super::DelayedTask;
+
+struct Data {
+    channel: String,
+    id: String,
+    is_dm: bool,
+}
+
+#[derive(Debug)]
+struct Task {
+    id: String,
+    is_dm: bool,
+}
+
+lazy_static! {
+    static ref Q: Queue<Data> = Queue::new(10_000);
+}
+
+pub async fn queue(channel: String, id: String, is_dm: bool) {
+    Q.push(Data { channel, id, is_dm }).await;
+}
+
+pub async fn worker(db: Database) {
+    let mut tasks = HashMap::<String, DelayedTask<Task>>::new();
+    let mut keys = vec![];
+
+    loop {
+        // Find due tasks.
+        for (key, task) in &tasks {
+            if task.should_run() {
+                keys.push(key.clone());
+            }
+        }
+
+        // Commit any due tasks to the database.
+        for key in &keys {
+            if let Some(task) = tasks.remove(key) {
+                let Task { id, is_dm, .. } = task.data;
+
+                let mut channel = PartialChannel {
+                    last_message_id: Some(id.to_string()),
+                    ..Default::default()
+                };
+
+                if is_dm {
+                    channel.active = Some(true);
+                }
+
+                match db.update_channel(key, &channel, vec![]).await {
+                    Ok(_) => info!("Updated last_message_id for {key} to {id}."),
+                    Err(err) => error!("Failed to update last_message_id with {err:?}!"),
+                }
+            }
+        }
+
+        // Clear keys
+        keys.clear();
+
+        // Queue incoming tasks.
+        while let Some(Data { channel, id, is_dm }) = Q.try_pop() {
+            if let Some(mut task) = tasks.get_mut(&channel) {
+                task.data.id = id;
+                task.delay();
+            } else {
+                tasks.insert(channel, DelayedTask::new(Task { id, is_dm }));
+            }
+        }
+
+        // Sleep for an arbitrary amount of time.
+        async_std::task::sleep(Duration::from_secs(1)).await;
+    }
+}

src/tasks/mod.rs

@@ -1,10 +1,13 @@
 //! Semi-important background task management
 
+use std::time::Instant;
+
 use async_std::task;
 use revolt_quark::Database;
 
 const WORKER_COUNT: usize = 5;
 
+pub mod last_message_id;
 pub mod process_embeds;
 pub mod web_push;
 
@@ -13,5 +16,41 @@ pub async fn start_workers(db: Database) {
     for _ in 0..WORKER_COUNT {
         task::spawn(process_embeds::worker(db.clone()));
         task::spawn(web_push::worker(db.clone()));
+        task::spawn(last_message_id::worker(db.clone()));
+    }
+}
+
+/// Task with additional information on when it should run
+pub struct DelayedTask<T> {
+    pub data: T,
+    last_updated: Instant,
+    first_seen: Instant,
+}
+
+/// Commit to database every 30 seconds if the task is particularly active.
+static EXPIRE_CONSTANT: u64 = 30;
+
+/// Otherwise, commit to database after 5 seconds.
+static SAVE_CONSTANT: u64 = 5;
+
+impl<T> DelayedTask<T> {
+    /// Create a new delayed task
+    pub fn new(data: T) -> Self {
+        DelayedTask {
+            data,
+            last_updated: Instant::now(),
+            first_seen: Instant::now(),
+        }
+    }
+
+    /// Push a task further back in time
+    pub fn delay(&mut self) {
+        self.last_updated = Instant::now()
+    }
+
+    /// Check if a task should run yet
+    pub fn should_run(&self) -> bool {
+        self.first_seen.elapsed().as_secs() > EXPIRE_CONSTANT
+            || self.last_updated.elapsed().as_secs() > SAVE_CONSTANT
     }
 }