feat(core): implement message tasks

includes:
- acknowledgements
- last_message_id
- embeds
- web push

crates/core/database/Cargo.toml

@@ -13,15 +13,17 @@ description = "Revolt Backend: Database Implementation"
 mongodb = ["dep:mongodb", "bson"]
 
 # ... Other
+tasks = ["isahc", "linkify", "url-escape"]
 async-std-runtime = ["async-std"]
 rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi"]
 redis-is-patched = ["revolt-presence/redis-is-patched"]
 
 # Default Features
-default = ["mongodb", "async-std-runtime"]
+default = ["mongodb", "async-std-runtime", "tasks"]
 
 [dependencies]
 # Core
+revolt-config = { version = "0.6.7", path = "../config" }
 revolt-result = { version = "0.6.7", path = "../result" }
 revolt-models = { version = "0.6.7", path = "../models" }
 revolt-presence = { version = "0.6.7", path = "../presence" }
@@ -36,9 +38,14 @@ lru = "0.11.0"
 rand = "0.8.5"
 ulid = "1.0.0"
 nanoid = "0.4.0"
+base64 = "0.21.3"
 once_cell = "1.17"
 indexmap = "1.9.1"
 decancer = "1.6.2"
+deadqueue = "0.2.4"
+linkify = { optional = true, version = "0.8.1" }
+url-escape = { optional = true, version = "0.1.1" }
+isahc = { optional = true, version = "1.7", features = ["json"] }
 
 # Serialisation
 serde_json = "1"
@@ -59,6 +66,7 @@ regex = "1"
 
 # Async Language Features
 futures = "0.3.19"
+async-lock = "2.8.0"
 async-trait = "0.1.51"
 async-recursion = "1.0.4"
 
@@ -73,5 +81,9 @@ rocket = { version = "0.5.0-rc.2", default-features = false, features = [
 revolt_okapi = { version = "0.9.1", optional = true }
 revolt_rocket_okapi = { version = "0.9.1", optional = true }
 
+# Notifications
+fcm = "0.9.2"
+web-push = "0.10.0"
+
 # Authifier
 authifier = { version = "1.0" }

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

@@ -65,3 +65,14 @@ impl DatabaseInfo {
         })
     }
 }
+
+impl From<Database> for authifier::Database {
+    fn from(value: Database) -> Self {
+        match value {
+            Database::Reference(_) => Default::default(),
+            Database::MongoDb(MongoDb(client, _)) => authifier::Database::MongoDb(
+                authifier::database::MongoDb(client.database("revolt")),
+            ),
+        }
+    }
+}

crates/core/database/src/lib.rs

@@ -81,6 +81,7 @@ pub mod util;
 pub use models::*;
 
 pub mod events;
+pub mod tasks;
 
 /// Utility function to check if a boolean value is false
 pub fn if_false(t: &bool) -> bool {

crates/core/database/src/models/messages/model.rs

@@ -1,10 +1,15 @@
 use indexmap::{IndexMap, IndexSet};
 use iso8601_timestamp::Timestamp;
-use revolt_models::v0::{Embed, MessageSort, MessageWebhook};
+use revolt_models::v0::{Embed, MessageAuthor, MessageSort, MessageWebhook, PushNotification};
 use revolt_result::Result;
+use serde_json::json;
 use ulid::Ulid;
 
-use crate::{events::client::EventV1, Database, File};
+use crate::{
+    events::client::EventV1,
+    tasks::{self, ack::AckEvent},
+    Channel, Database, File,
+};
 
 auto_derived_partial!(
     /// Message
@@ -170,7 +175,12 @@ auto_derived!(
 #[allow(clippy::disallowed_methods)]
 impl Message {
     /// Send a message without any notifications
-    pub async fn send_without_notifications(&mut self, db: &Database) -> Result<()> {
+    pub async fn send_without_notifications(
+        &mut self,
+        db: &Database,
+        is_dm: bool,
+        generate_embeds: bool,
+    ) -> Result<()> {
         db.insert_message(self).await?;
 
         // Fan out events
@@ -178,21 +188,90 @@ impl Message {
             .p(self.channel.to_string())
             .await;
 
-        // TODO: update last_message_id
+        // Update last_message_id
-        // TODO: add mentions for affected users
+        tasks::last_message_id::queue(self.channel.to_string(), self.id.to_string(), is_dm).await;
-        // TODO: generate embeds
+
+        // Add mentions for affected users
+        if let Some(mentions) = &self.mentions {
+            for user in mentions {
+                tasks::ack::queue(
+                    self.channel.to_string(),
+                    user.to_string(),
+                    AckEvent::AddMention {
+                        ids: vec![self.id.to_string()],
+                    },
+                )
+                .await;
+            }
+        }
+
+        // Generate embeds
+        if generate_embeds {
+            if let Some(content) = &self.content {
+                tasks::process_embeds::queue(
+                    self.channel.to_string(),
+                    self.id.to_string(),
+                    content.clone(),
+                )
+                .await;
+            }
+        }
 
         Ok(())
     }
 
     /// Send a message
-    pub async fn send(&mut self, db: &Database) -> Result<()> {
+    pub async fn send(
-        self.send_without_notifications(db).await?;
+        &mut self,
+        db: &Database,
+        author: MessageAuthor<'_>,
+        channel: &Channel,
+        generate_embeds: bool,
+    ) -> Result<()> {
+        self.send_without_notifications(
+            db,
+            matches!(channel, Channel::DirectMessage { .. }),
+            generate_embeds,
+        )
+        .await?;
 
-        // TODO: web push / FCM
+        // Push out Web Push notifications
+        crate::tasks::web_push::queue(
+            {
+                match channel {
+                    Channel::DirectMessage { recipients, .. }
+                    | Channel::Group { recipients, .. } => recipients.clone(),
+                    Channel::TextChannel { .. } => self.mentions.clone().unwrap_or_default(),
+                    _ => vec![],
+                }
+            },
+            json!(PushNotification::from(self.clone().into(), Some(author), &channel.id()).await)
+                .to_string(),
+        )
+        .await;
 
         todo!()
     }
+
+    /// Append content to message
+    pub async fn append(
+        db: &Database,
+        id: String,
+        channel: String,
+        append: AppendMessage,
+    ) -> Result<()> {
+        db.append_message(&id, &append).await?;
+
+        EventV1::MessageAppend {
+            id,
+            channel: channel.to_string(),
+            append: append.into(),
+        }
+        .p(channel)
+        .await;
+
+        Ok(())
+    }
 }
 
 impl SystemMessage {
@@ -209,29 +288,6 @@ impl SystemMessage {
 }
 
 impl Interactions {
-    /// Validate interactions info is correct
-    /* pub async fn validate(
-        &self,
-        db: &Database,
-        permissions: &mut PermissionCalculator<'_>,
-    ) -> Result<()> {
-        if let Some(reactions) = &self.reactions {
-            permissions.throw_permission(db, Permission::React).await?;
-
-            if reactions.len() > 20 {
-                return Err(Error::InvalidOperation);
-            }
-
-            for reaction in reactions {
-                if !Emoji::can_use(db, reaction).await? {
-                    return Err(Error::InvalidOperation);
-                }
-            }
-        }
-
-        Ok(())
-    }*/
-
     /// Check if we can use a given emoji to react
     pub fn can_use(&self, emoji: &str) -> bool {
         if self.restrict_reactions {

crates/core/database/src/models/server_members/model.rs

@@ -131,7 +131,7 @@ impl Member {
                 id: user.id.clone(),
             }
             .into_message(id.to_string())
-            .send_without_notifications(db)
+            .send_without_notifications(db, false, false)
             .await
             .ok();
         }

crates/core/database/src/tasks/ack.rs

@@ -0,0 +1,122 @@
+// Queue Type: Debounced
+use crate::Database;
+
+use deadqueue::limited::Queue;
+use once_cell::sync::Lazy;
+use std::{collections::HashMap, time::Duration};
+
+use super::DelayedTask;
+
+/// Enumeration of possible events
+#[derive(Debug, Eq, PartialEq)]
+pub enum AckEvent {
+    /// Add mentions for a user in a channel
+    AddMention {
+        /// Message IDs
+        ids: Vec<String>,
+    },
+
+    /// Acknowledge message in a channel for a user
+    AckMessage {
+        /// Message ID
+        id: String,
+    },
+}
+
+/// Task information
+struct Data {
+    /// Channel to ack
+    channel: String,
+    /// User to ack for
+    user: String,
+    /// Event
+    event: AckEvent,
+}
+
+#[derive(Debug)]
+struct Task {
+    event: AckEvent,
+}
+
+static Q: Lazy<Queue<Data>> = Lazy::new(|| Queue::new(10_000));
+
+/// Queue a new task for a worker
+pub async fn queue(channel: String, user: String, event: AckEvent) {
+    Q.try_push(Data {
+        channel,
+        user,
+        event,
+    })
+    .ok();
+
+    info!("Queue is using {} slots from {}.", Q.len(), Q.capacity());
+}
+
+/// Start a new worker
+pub async fn worker(db: Database) {
+    let mut tasks = HashMap::<(String, 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 { event } = task.data;
+                let (user, channel) = key;
+
+                if let Err(err) = match &event {
+                    #[allow(clippy::disallowed_methods)] // event is sent by higher level function
+                    AckEvent::AckMessage { id } => db.acknowledge_message(channel, user, id).await,
+                    AckEvent::AddMention { ids } => {
+                        db.add_mention_to_unread(channel, user, ids).await
+                    }
+                } {
+                    error!("{err:?} for {event:?}. ({user}, {channel})");
+                } else {
+                    info!("User {user} ack in {channel} with {event:?}");
+                }
+            }
+        }
+
+        // Clear keys
+        keys.clear();
+
+        // Queue incoming tasks.
+        while let Some(Data {
+            channel,
+            user,
+            mut event,
+        }) = Q.try_pop()
+        {
+            let key = (user, channel);
+            if let Some(task) = tasks.get_mut(&key) {
+                task.delay();
+
+                match &mut event {
+                    AckEvent::AddMention { ids } => {
+                        if let AckEvent::AddMention { ids: existing } = &mut task.data.event {
+                            existing.append(ids);
+                        } else {
+                            task.data.event = event;
+                        }
+                    }
+                    AckEvent::AckMessage { .. } => {
+                        task.data.event = event;
+                    }
+                }
+            } else {
+                tasks.insert(key, DelayedTask::new(Task { event }));
+            }
+        }
+
+        // Sleep for an arbitrary amount of time.
+        async_std::task::sleep(Duration::from_secs(1)).await;
+    }
+}

crates/core/database/src/tasks/last_message_id.rs

@@ -0,0 +1,87 @@
+// Queue Type: Debounced
+use deadqueue::limited::Queue;
+use once_cell::sync::Lazy;
+use std::{collections::HashMap, time::Duration};
+
+use crate::{Database, PartialChannel};
+
+use super::DelayedTask;
+
+/// Task information
+struct Data {
+    /// Channel to update
+    channel: String,
+    /// Latest message ID
+    id: String,
+    /// Whether the channel is a DM
+    is_dm: bool,
+}
+
+/// Task information
+#[derive(Debug)]
+struct Task {
+    /// Latest message ID
+    id: String,
+    /// Whether the channel is a DM
+    is_dm: bool,
+}
+
+static Q: Lazy<Queue<Data>> = Lazy::new(|| Queue::new(10_000));
+
+/// Queue a new task for a worker
+pub async fn queue(channel: String, id: String, is_dm: bool) {
+    Q.try_push(Data { channel, id, is_dm }).ok();
+    info!("Queue is using {} slots from {}.", Q.len(), Q.capacity());
+}
+
+/// Start a new worker
+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(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;
+    }
+}

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

@@ -0,0 +1,58 @@
+//! Semi-important background task management
+
+use crate::Database;
+
+use async_std::task;
+use std::time::Instant;
+
+const WORKER_COUNT: usize = 5;
+
+pub mod ack;
+pub mod last_message_id;
+pub mod process_embeds;
+pub mod web_push;
+
+/// Spawn background workers
+pub async fn start_workers(db: Database) {
+    for _ in 0..WORKER_COUNT {
+        task::spawn(ack::worker(db.clone()));
+        task::spawn(last_message_id::worker(db.clone()));
+        task::spawn(process_embeds::worker(db.clone()));
+        task::spawn(web_push::worker(db.clone().into()));
+    }
+}
+
+/// 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
+    }
+}

crates/core/database/src/tasks/process_embeds.rs

@@ -0,0 +1,170 @@
+use crate::{models::Message, AppendMessage, Database};
+
+use futures::future::join_all;
+use linkify::{LinkFinder, LinkKind};
+use regex::Regex;
+use revolt_config::config;
+use revolt_result::Result;
+
+use async_lock::Semaphore;
+use async_std::task::spawn;
+use deadqueue::limited::Queue;
+use once_cell::sync::Lazy;
+use revolt_models::v0::Embed;
+use std::{collections::HashSet, sync::Arc};
+
+use isahc::prelude::*;
+
+/// Task information
+#[derive(Debug)]
+struct EmbedTask {
+    /// Channel we're processing the event in
+    channel: String,
+    /// ID of the message we're processing
+    id: String,
+    /// Content of the message
+    content: String,
+}
+
+static Q: Lazy<Queue<EmbedTask>> = Lazy::new(|| Queue::new(10_000));
+
+/// Queue a new task for a worker
+pub async fn queue(channel: String, id: String, content: String) {
+    Q.try_push(EmbedTask {
+        channel,
+        id,
+        content,
+    })
+    .ok();
+
+    info!("Queue is using {} slots from {}.", Q.len(), Q.capacity());
+}
+
+/// Start a new worker
+pub async fn worker(db: Database) {
+    let semaphore = Arc::new(Semaphore::new(
+        config().await.api.workers.max_concurrent_connections,
+    ));
+
+    loop {
+        let task = Q.pop().await;
+        let db = db.clone();
+        let semaphore = semaphore.clone();
+
+        spawn(async move {
+            let config = config().await;
+            let embeds = generate(
+                task.content,
+                &config.hosts.january,
+                config.features.limits.default.message_embeds,
+                semaphore,
+            )
+            .await;
+
+            if let Ok(embeds) = embeds {
+                if let Err(err) = Message::append(
+                    &db,
+                    task.id,
+                    task.channel,
+                    AppendMessage {
+                        embeds: Some(embeds),
+                    },
+                )
+                .await
+                {
+                    error!("Encountered an error appending to message: {:?}", err);
+                }
+            }
+        });
+    }
+}
+
+static RE_CODE: Lazy<Regex> = Lazy::new(|| Regex::new("```(?:.|\n)+?```|`(?:.|\n)+?`").unwrap());
+static RE_IGNORED: Lazy<Regex> = Lazy::new(|| Regex::new("(<http.+>)").unwrap());
+
+pub async fn generate(
+    content: String,
+    host: &str,
+    max_embeds: usize,
+    semaphore: Arc<Semaphore>,
+) -> Result<Vec<Embed>> {
+    // Ignore code blocks.
+    let content = RE_CODE.replace_all(&content, "");
+
+    // Ignore all content between angle brackets starting with http.
+    let content = RE_IGNORED.replace_all(&content, "");
+
+    let content = content
+        // Ignore quoted lines.
+        .split('\n')
+        .map(|v| {
+            if let Some(c) = v.chars().next() {
+                if c == '>' {
+                    return "";
+                }
+            }
+
+            v
+        })
+        .collect::<Vec<&str>>()
+        .join("\n");
+
+    let mut finder = LinkFinder::new();
+    finder.kinds(&[LinkKind::Url]);
+
+    // Process all links, stripping anchors and
+    // only taking up to `max_embeds` of links.
+    let links: Vec<String> = finder
+        .links(&content)
+        .map(|x| {
+            x.as_str()
+                .chars()
+                .take_while(|&ch| ch != '#')
+                .collect::<String>()
+        })
+        .collect::<HashSet<String>>()
+        .into_iter()
+        .take(max_embeds)
+        .collect();
+
+    // If no links, fail out.
+    if links.is_empty() {
+        return Err(create_error!(LabelMe));
+    }
+
+    // ! FIXME: batch request to january
+    let mut tasks = Vec::new();
+
+    for link in links {
+        let semaphore = semaphore.clone();
+        let host = host.to_string();
+        tasks.push(spawn(async move {
+            let guard = semaphore.acquire().await;
+
+            if let Ok(mut response) = isahc::get_async(format!(
+                "{host}/embed?url={}",
+                url_escape::encode_component(&link)
+            ))
+            .await
+            {
+                drop(guard);
+                response.json::<Embed>().await.ok()
+            } else {
+                None
+            }
+        }));
+    }
+
+    let embeds = join_all(tasks)
+        .await
+        .into_iter()
+        .flatten()
+        .collect::<Vec<Embed>>();
+
+    // Prevent database update when no embeds are found.
+    if !embeds.is_empty() {
+        Ok(embeds)
+    } else {
+        Err(create_error!(LabelMe))
+    }
+}

crates/core/database/src/tasks/web_push.rs

@@ -0,0 +1,122 @@
+use std::collections::HashSet;
+
+use authifier::Database;
+use base64::{
+    alphabet,
+    engine::{self},
+    Engine as _,
+};
+use deadqueue::limited::Queue;
+use once_cell::sync::Lazy;
+use revolt_presence::filter_online;
+use web_push::{
+    ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
+    WebPushClient, WebPushMessageBuilder,
+};
+
+/// Task information
+#[derive(Debug)]
+struct PushTask {
+    /// User IDs of the targets that are to receive this notification
+    recipients: Vec<String>,
+    /// Raw JSON payload to send to clients
+    payload: String,
+}
+
+static Q: Lazy<Queue<PushTask>> = Lazy::new(|| Queue::new(10_000));
+
+/// Queue a new task for a worker
+pub async fn queue(recipients: Vec<String>, payload: String) {
+    if recipients.is_empty() {
+        return;
+    }
+
+    let online_ids = filter_online(&recipients).await;
+    let recipients = (&recipients.into_iter().collect::<HashSet<String>>() - &online_ids)
+        .into_iter()
+        .collect::<Vec<String>>();
+
+    Q.try_push(PushTask {
+        recipients,
+        payload,
+    })
+    .ok();
+
+    info!("Queue is using {} slots from {}.", Q.len(), Q.capacity());
+}
+
+/// Start a new worker
+pub async fn worker(db: Database) {
+    let client = IsahcWebPushClient::new().unwrap();
+    let key = engine::GeneralPurpose::new(&alphabet::URL_SAFE, engine::general_purpose::NO_PAD)
+        .decode(std::env::var("REVOLT_VAPID_PRIVATE_KEY").unwrap())
+        .expect("valid `VAPID_PRIVATE_KEY`");
+
+    loop {
+        let task = Q.pop().await;
+
+        if let Ok(sessions) = db.find_sessions_with_subscription(&task.recipients).await {
+            for session in sessions {
+                if let Some(sub) = session.subscription {
+                    if sub.endpoint == "fcm" {
+                        // Use Firebase Cloud Messaging
+                        info!("Skipping FCM client, unimplemented...");
+                    } else {
+                        // Use Web Push Standard
+                        let subscription = SubscriptionInfo {
+                            endpoint: sub.endpoint,
+                            keys: SubscriptionKeys {
+                                auth: sub.auth,
+                                p256dh: sub.p256dh,
+                            },
+                        };
+
+                        match VapidSignatureBuilder::from_pem(
+                            std::io::Cursor::new(&key),
+                            &subscription,
+                        ) {
+                            Ok(sig_builder) => match sig_builder.build() {
+                                Ok(signature) => {
+                                    let mut builder = WebPushMessageBuilder::new(&subscription);
+                                    builder.set_vapid_signature(signature);
+                                    builder.set_payload(
+                                        ContentEncoding::AesGcm,
+                                        task.payload.as_bytes(),
+                                    );
+
+                                    match builder.build() {
+                                        Ok(msg) => match client.send(msg).await {
+                                            Ok(_) => {
+                                                info!(
+                                                    "Sent Web Push notification to {:?}.",
+                                                    session.id
+                                                )
+                                            }
+                                            Err(err) => {
+                                                error!("Hit error sending Web Push! {:?}", err)
+                                            }
+                                        },
+                                        Err(err) => {
+                                            error!(
+                                                "Failed to build message for {}! {:?}",
+                                                session.user_id, err
+                                            )
+                                        }
+                                    }
+                                }
+                                Err(err) => error!(
+                                    "Failed to build signature for {}! {:?}",
+                                    session.user_id, err
+                                ),
+                            },
+                            Err(err) => error!(
+                                "Failed to create signature builder for {}! {:?}",
+                                session.user_id, err
+                            ),
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

crates/core/database/src/util/bridge/v0.rs

@@ -391,6 +391,16 @@ impl From<crate::Interactions> for Interactions {
     }
 }
 
+impl From<crate::AppendMessage> for AppendMessage {
+    fn from(value: crate::AppendMessage) -> Self {
+        AppendMessage {
+            embeds: value
+                .embeds
+                .map(|embeds| embeds.into_iter().map(|embed| embed.into()).collect()),
+        }
+    }
+}
+
 impl From<crate::Masquerade> for Masquerade {
     fn from(value: crate::Masquerade) -> Self {
         Masquerade {

crates/core/models/Cargo.toml

@@ -18,6 +18,7 @@ default = ["serde", "partials"]
 
 [dependencies]
 # Core
+revolt-config = { version = "0.6.7", path = "../config" }
 revolt-permissions = { version = "0.6.7", path = "../permissions" }
 
 # Utility

crates/core/models/src/v0/messages.rs

@@ -1,7 +1,11 @@
+use std::time::SystemTime;
+
+use revolt_config::config;
+
 use indexmap::{IndexMap, IndexSet};
 use iso8601_timestamp::Timestamp;
 
-use super::{Embed, File, MessageWebhook};
+use super::{Embed, File, MessageWebhook, User, Webhook};
 
 auto_derived_partial!(
     /// Message
@@ -129,11 +133,129 @@ auto_derived!(
         /// Sort by the oldest messages first
         Oldest,
     }
+
+    /// Push Notification
+    pub struct PushNotification {
+        /// Known author name
+        pub author: String,
+        /// URL to author avatar
+        pub icon: String,
+        /// URL to first matching attachment
+        #[serde(skip_serializing_if = "Option::is_none")]
+        pub image: Option<String>,
+        /// Message content or system message information
+        pub body: String,
+        /// Unique tag, usually the channel ID
+        pub tag: String,
+        /// Timestamp at which this notification was created
+        pub timestamp: u64,
+        /// URL to open when clicking notification
+        pub url: String,
+    }
 );
 
+/// Message Author Abstraction
+pub enum MessageAuthor<'a> {
+    User(&'a User),
+    Webhook(&'a Webhook),
+}
+
 impl Interactions {
     /// Check if default initialisation of fields
     pub fn is_default(&self) -> bool {
         !self.restrict_reactions && self.reactions.is_none()
     }
 }
+
+impl<'a> MessageAuthor<'a> {
+    pub fn id(&self) -> &str {
+        match self {
+            MessageAuthor::User(user) => &user.id,
+            MessageAuthor::Webhook(webhook) => &webhook.id,
+        }
+    }
+
+    pub fn avatar(&self) -> Option<&str> {
+        match self {
+            MessageAuthor::User(user) => user.avatar.as_ref().map(|file| file.id.as_str()),
+            MessageAuthor::Webhook(webhook) => webhook.avatar.as_ref().map(|file| file.id.as_str()),
+        }
+    }
+
+    pub fn username(&self) -> &str {
+        match self {
+            MessageAuthor::User(user) => &user.username,
+            MessageAuthor::Webhook(webhook) => &webhook.name,
+        }
+    }
+}
+
+impl From<SystemMessage> for String {
+    fn from(s: SystemMessage) -> String {
+        match s {
+            SystemMessage::Text { content } => content,
+            SystemMessage::UserAdded { .. } => "User added to the channel.".to_string(),
+            SystemMessage::UserRemove { .. } => "User removed from the channel.".to_string(),
+            SystemMessage::UserJoined { .. } => "User joined the channel.".to_string(),
+            SystemMessage::UserLeft { .. } => "User left the channel.".to_string(),
+            SystemMessage::UserKicked { .. } => "User kicked from the channel.".to_string(),
+            SystemMessage::UserBanned { .. } => "User banned from the channel.".to_string(),
+            SystemMessage::ChannelRenamed { .. } => "Channel renamed.".to_string(),
+            SystemMessage::ChannelDescriptionChanged { .. } => {
+                "Channel description changed.".to_string()
+            }
+            SystemMessage::ChannelIconChanged { .. } => "Channel icon changed.".to_string(),
+            SystemMessage::ChannelOwnershipChanged { .. } => {
+                "Channel ownership changed.".to_string()
+            }
+        }
+    }
+}
+
+impl PushNotification {
+    /// Create a new notification from a given message, author and channel ID
+    pub async fn from(msg: Message, author: Option<MessageAuthor<'_>>, channel_id: &str) -> Self {
+        let config = config().await;
+
+        let icon = if let Some(author) = &author {
+            if let Some(avatar) = author.avatar() {
+                format!("{}/avatars/{}", config.hosts.autumn, avatar)
+            } else {
+                format!("{}/users/{}/default_avatar", config.hosts.api, author.id())
+            }
+        } else {
+            format!("{}/assets/logo.png", config.hosts.app)
+        };
+
+        let image = msg.attachments.and_then(|attachments| {
+            attachments
+                .first()
+                .map(|v| format!("{}/attachments/{}", config.hosts.autumn, v.id))
+        });
+
+        let body = if let Some(sys) = msg.system {
+            sys.into()
+        } else if let Some(text) = msg.content {
+            text
+        } else {
+            "Empty Message".to_string()
+        };
+
+        let timestamp = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .expect("Time went backwards")
+            .as_secs();
+
+        Self {
+            author: author
+                .map(|x| x.username().to_string())
+                .unwrap_or_else(|| "Revolt".to_string()),
+            icon,
+            image,
+            body,
+            tag: channel_id.to_string(),
+            timestamp,
+            url: format!("{}/channel/{}/{}", config.hosts.app, channel_id, msg.id),
+        }
+    }
+}

crates/delta/src/main.rs

@@ -39,7 +39,7 @@ async fn rocket() -> _ {
 
     // Setup Authifier
     let authifier = Authifier {
-        database: legacy_db.clone().into(),
+        database: db.clone().into(),
         config: revolt_quark::util::authifier::config(),
         event_channel: Some(sender),
     };
@@ -61,6 +61,7 @@ async fn rocket() -> _ {
     });
 
     // Launch background task workers
+    async_std::task::spawn(revolt_database::tasks::start_workers(db.clone()));
     async_std::task::spawn(revolt_quark::tasks::start_workers(legacy_db.clone()));
 
     // Configure CORS