diff --git a/src/routes/channels/message_send.rs b/src/routes/channels/message_send.rs index d4328512..be1fbb3b 100644 --- a/src/routes/channels/message_send.rs +++ b/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)) } diff --git a/src/tasks/last_message_id.rs b/src/tasks/last_message_id.rs new file mode 100644 index 00000000..6845f42e --- /dev/null +++ b/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 = 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::>::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; + } +} diff --git a/src/tasks/mod.rs b/src/tasks/mod.rs index a71b25ee..6e92b5f7 100644 --- a/src/tasks/mod.rs +++ b/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 { + 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 DelayedTask { + /// 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 } }