feat(messaging): add masquerade to messages

closes #111 and closes #86

.github/workflows/docker.yml

@@ -53,7 +53,7 @@ jobs:
                 rm -rf /tmp/.buildx-cache/${{ matrix.architecture }}
                 mv /tmp/.buildx-cache-new/${{ matrix.architecture }} /tmp/.buildx-cache/${{ matrix.architecture }}
 
-    publish:
+    publish_amd64:
         needs: [test]
         runs-on: ubuntu-latest
         if: github.event_name != 'pull_request'
@@ -92,7 +92,7 @@ jobs:
               with:
                   context: .
                   push: true
-                  platforms: linux/amd64,linux/arm64
+                  platforms: linux/amd64
                   tags: ${{ steps.meta.outputs.tags }}
                   labels: ${{ steps.meta.outputs.labels }}
                   cache-from: type=local,src=/tmp/.buildx-cache/linux/amd64
@@ -101,3 +101,41 @@ jobs:
               run: |
                rm -rf /tmp/.buildx-cache
                mv /tmp/.buildx-cache-new /tmp/.buildx-cache
+
+    publish_arm64:
+        needs: [test]
+        runs-on: ubuntu-latest
+        if: github.event_name != 'pull_request' && startsWith(github.ref, 'refs/tags')
+        steps:
+            - name: Checkout
+              uses: actions/checkout@v2
+              with:
+                  submodules: "recursive"
+            - name: Set up QEMU
+              uses: docker/setup-qemu-action@v1
+            - name: Set up Docker Buildx
+              uses: docker/setup-buildx-action@v1
+            - name: Docker meta
+              id: meta
+              uses: docker/metadata-action@v3
+              with:
+                  images: revoltchat/server, ghcr.io/revoltchat/server
+            - name: Login to DockerHub
+              uses: docker/login-action@v1
+              with:
+                  username: ${{ secrets.DOCKERHUB_USERNAME }}
+                  password: ${{ secrets.DOCKERHUB_TOKEN }}
+            - name: Login to Github Container Registry
+              uses: docker/login-action@v1
+              with:
+                  registry: ghcr.io
+                  username: ${{ github.actor }}
+                  password: ${{ secrets.GITHUB_TOKEN }}
+            - name: Build and publish
+              uses: docker/build-push-action@v2
+              with:
+                  context: .
+                  push: true
+                  platforms: linux/arm64
+                  tags: ${{ steps.meta.outputs.tags }}
+                  labels: ${{ steps.meta.outputs.labels }}

Cargo.lock

@@ -27,6 +27,17 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
 
+[[package]]
+name = "ahash"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fcb51a0695d8f838b1ee009b3fbf66bda078cd64590202a864a8f3e8c4315c47"
+dependencies = [
+ "getrandom 0.2.3",
+ "once_cell",
+ "version_check",
+]
+
 [[package]]
 name = "aho-corasick"
 version = "0.7.18"
@@ -1180,6 +1191,9 @@ name = "hashbrown"
 version = "0.11.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ab5ef0d4909ef3724cc8cce6ccc8572c5c817592e9285f5464f8e86f8bd3726e"
+dependencies = [
+ "ahash",
+]
 
 [[package]]
 name = "heck"
@@ -1600,6 +1614,15 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "lru"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c748cfe47cb8da225c37595b3108bea1c198c84aaae8ea0ba76d01dda9fc803"
+dependencies = [
+ "hashbrown",
+]
+
 [[package]]
 name = "lru-cache"
 version = "0.1.2"
@@ -2771,6 +2794,7 @@ dependencies = [
 name = "revolt"
 version = "0.0.0"
 dependencies = [
+ "async-channel",
  "async-std",
  "async-tungstenite",
  "base64 0.13.0",
@@ -2786,6 +2810,7 @@ dependencies = [
  "lettre",
  "linkify",
  "log",
+ "lru",
  "many-to-many",
  "mobc",
  "mobc-redis",

Cargo.toml

@@ -11,6 +11,7 @@ edition = "2018"
 
 [dependencies]
 # Utility
+lru = "0.7.0"
 url = "2.2.2"
 log = "0.4.11"
 dotenv = "0.15.0"
@@ -41,6 +42,7 @@ rmp-serde = { git = "https://github.com/3Hren/msgpack-rust", rev = "5bf2c24203ad
 # async
 futures = "0.3.8"
 chrono = "0.4.15"
+async-channel = "1.6.1"
 reqwest = { version = "0.11.4", features = ["json"] }
 async-std = { version = "1.8.0", features = ["tokio1", "tokio02", "attributes"] }
 

set_version.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
-export version=0.5.3-alpha.6
+export version=0.5.3-alpha.8
 echo "pub const VERSION: &str = \"${version}\";" > src/version.rs

src/database/entities/channel.rs

@@ -33,8 +33,6 @@ pub enum Channel {
     Group {
         #[serde(rename = "_id")]
         id: String,
-        #[serde(skip_serializing_if = "Option::is_none")]
-        nonce: Option<String>,
 
         name: String,
         owner: String,
@@ -57,8 +55,6 @@ pub enum Channel {
         #[serde(rename = "_id")]
         id: String,
         server: String,
-        #[serde(skip_serializing_if = "Option::is_none")]
-        nonce: Option<String>,
 
         name: String,
         #[serde(skip_serializing_if = "Option::is_none")]
@@ -81,8 +77,6 @@ pub enum Channel {
         #[serde(rename = "_id")]
         id: String,
         server: String,
-        #[serde(skip_serializing_if = "Option::is_none")]
-        nonce: Option<String>,
 
         name: String,
         #[serde(skip_serializing_if = "Option::is_none")]

src/database/entities/message.rs

@@ -1,20 +1,15 @@
-use crate::util::variables::{USE_JANUARY, VAPID_PRIVATE_KEY, PUBLIC_URL};
+use crate::util::variables::{USE_JANUARY, PUBLIC_URL};
 use crate::{
     database::*,
     notifications::{events::ClientboundNotification, websocket::is_online},
     util::result::{Error, Result},
 };
 
-use futures::StreamExt;
-use mongodb::options::UpdateOptions;
-use mongodb::{
-    bson::{doc, to_bson, DateTime, Document},
-};
-use rauth::entities::{Model, Session};
+use mongodb::bson::{doc, to_bson, DateTime};
 use rocket::serde::json::Value;
 use serde::{Deserialize, Serialize};
 use ulid::Ulid;
-use web_push::{ContentEncoding, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder};
+use validator::Validate;
 use std::time::SystemTime;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -129,6 +124,7 @@ impl Content {
             target.id().to_string(),
             self,
             None,
+            None,
             None
         )
         .publish(&target, false)
@@ -136,6 +132,16 @@ impl Content {
     }
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, Validate)]
+pub struct Masquerade {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[validate(length(min = 1, max = 32))]
+    name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[validate(length(min = 1, max = 128))]
+    avatar: Option<String>
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Message {
     #[serde(rename = "_id")]
@@ -155,7 +161,9 @@ pub struct Message {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub mentions: Option<Vec<String>>,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub replies: Option<Vec<String>>
+    pub replies: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub masquerade: Option<Masquerade>
 }
 
 impl Message {
@@ -165,6 +173,7 @@ impl Message {
         content: Content,
         mentions: Option<Vec<String>>,
         replies: Option<Vec<String>>,
+        masquerade: Option<Masquerade>,
     ) -> Message {
         Message {
             id: Ulid::new().to_string(),
@@ -176,11 +185,17 @@ impl Message {
             edited: None,
             embeds: None,
             mentions,
-            replies
+            replies,
+            masquerade
         }
     }
 
     pub async fn publish(self, channel: &Channel, process_embeds: bool) -> Result<()> {
+        // Publish message event
+        ClientboundNotification::Message(self.clone())
+            .publish(channel.id().to_string());
+
+        // Commit message to database
         get_collection("messages")
             .insert_one(to_bson(&self).unwrap().as_document().unwrap().clone(), None)
             .await
@@ -189,137 +204,60 @@ impl Message {
                 with: "message",
             })?;
 
-        // ! 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
-        if let Some(mentions) = &self.mentions {
-            let message = self.id.clone();
-            let channel = self.channel.clone();
-            let mentions = mentions.clone();
-            async_std::task::spawn(async move {
-                get_collection("channel_unreads")
-                    .update_many(
-                        doc! {
-                            "_id.channel": channel,
-                            "_id.user": {
-                                "$in": mentions
-                            }
-                        },
-                        doc! {
-                            "$push": {
-                                "mentions": message
-                            }
-                        },
-                        UpdateOptions::builder().upsert(true).build(),
-                    )
-                    .await
-                    /*.map_err(|_| Error::DatabaseError {
-                        operation: "update_many",
-                        with: "channel_unreads",
-                    })?;*/
-                    .unwrap();
-            });
-        }
-
+        // spawn task_queue ( process embeds )
         if process_embeds {
-            self.process_embed();
+            self.process_embed().await;
         }
 
-        let mentions = self.mentions.clone();
-        ClientboundNotification::Message(self.clone()).publish(channel.id().to_string());
+        // spawn task_queue ( update last_message_id )
+        match channel {
+            Channel::DirectMessage { id, .. } =>
+                crate::task_queue::task_last_message_id::queue(id.clone(), self.id.clone(), true).await,
+            Channel::Group { id, .. } | Channel::TextChannel { id, .. } =>
+                crate::task_queue::task_last_message_id::queue(id.clone(), self.id.clone(), false).await,
+            _ => {}
+        }
 
-        /*
-           Web Push Test Code
-        */
-        let c_clone = channel.clone();
-        async_std::task::spawn(async move {
-            // Find all offline users.
-            let mut target_ids = vec![];
-            match &c_clone {
-                Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
-                    for recipient in recipients {
-                        if !is_online(recipient) {
-                            target_ids.push(recipient.clone());
-                        }
+        // if mentions {
+        //  spawn task_queue ( update channel_unreads )
+        // }
+        if let Some(mentions) = &self.mentions {
+            for user in mentions {
+                crate::task_queue::task_ack::queue(
+                    channel.id().into(),
+                    user.clone(),
+                    crate::task_queue::task_ack::AckEvent::AddMention {
+                        ids: vec![ self.id.clone() ]
                     }
-                }
-                Channel::TextChannel { .. } => {
-                    if let Some(mut mentions) = mentions {
-                        target_ids.append(&mut mentions);
-                    }
-                }
-                _ => {}
+                ).await;
             }
+        }
 
-            // Fetch their corresponding sessions.
-            if target_ids.len() > 0 {
-                if let Ok(mut cursor) = Session::find(
-                        &get_db(),
-                        doc! {
-                            "_id": {
-                                "$in": target_ids
-                            },
-                            "subscription": {
-                                "$exists": true
-                            }
-                        },
-                        None
-                    )
-                    .await {
-                    let enc = serde_json::to_string(&PushNotification::new(self, &c_clone).await).unwrap();
-                    let client = WebPushClient::new();
-                    let key =
-                        base64::decode_config(VAPID_PRIVATE_KEY.clone(), base64::URL_SAFE).unwrap();
-
-                    while let Some(Ok(session)) = cursor.next().await {
-                        if let Some(sub) = session.subscription {
-                            let subscription = SubscriptionInfo {
-                                endpoint: sub.endpoint,
-                                keys: SubscriptionKeys {
-                                    auth: sub.auth,
-                                    p256dh: sub.p256dh
-                                }
-                            };
-
-                            let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
-                            let sig_builder = VapidSignatureBuilder::from_pem(
-                                std::io::Cursor::new(&key),
-                                &subscription,
-                            )
-                            .unwrap();
-                            let signature = sig_builder.build().unwrap();
-                            builder.set_vapid_signature(signature);
-                            builder.set_payload(ContentEncoding::AesGcm, enc.as_bytes());
-                            let m = builder.build().unwrap();
-                            client.send(m).await.ok();
-                        }
+        // if (channel => DM | Group) | mentions {
+        //  spawn task_queue ( web push )
+        // }
+        let mut target_ids = vec![];
+        match &channel {
+            Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
+                for recipient in recipients {
+                    if !is_online(recipient) {
+                        target_ids.push(recipient.clone());
                     }
                 }
             }
-        });
+            Channel::TextChannel { .. } => {
+                if let Some(mentions) = &self.mentions {
+                    target_ids.append(&mut mentions.clone());
+                }
+            }
+            _ => {}
+        }
+
+        if target_ids.len() > 0 {
+            if let Ok(payload) = serde_json::to_string(&PushNotification::new(self, &channel).await) {
+                crate::task_queue::task_web_push::queue(target_ids, payload).await;
+            }
+        }
 
         Ok(())
     }
@@ -332,49 +270,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;
         }
     }
 
@@ -432,15 +339,3 @@ impl Message {
         Ok(())
     }
 }
-
-async fn update_channels_last_message(channels: &Collection, channel_id: &String, set: &Document) {
-    channels
-        .update_one(
-            doc! { "_id": channel_id },
-            doc! { "$set": set },
-            None,
-        )
-        .await
-        .expect("Server should not run with no, or a corrupted db");
-
-}

src/database/entities/server.rs

@@ -85,8 +85,6 @@ pub enum RemoveMember {
 pub struct Server {
     #[serde(rename = "_id")]
     pub id: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub nonce: Option<String>,
     pub owner: String,
 
     pub name: String,

src/database/migrations/scripts.rs

@@ -11,7 +11,7 @@ struct MigrationInfo {
     revision: i32,
 }
 
-pub const LATEST_REVISION: i32 = 10;
+pub const LATEST_REVISION: i32 = 11;
 
 pub async fn migrate_database() {
     let migrations = get_collection("migrations");
@@ -342,6 +342,36 @@ pub async fn run_migrations(revision: i32) -> i32 {
         }
     }
 
+    if revision <= 10 {
+        info!("Running migration [revision 10 / 2021-11-01]: Remove nonce values on channels and servers.");
+
+        get_collection("servers")
+            .update_many(
+                doc! {},
+                doc! {
+                    "$unset": {
+                        "nonce": 1,
+                    }
+                },
+                None
+            )
+            .await
+            .unwrap();
+
+        get_collection("channels")
+            .update_many(
+                doc! {},
+                doc! {
+                    "$unset": {
+                        "nonce": 1,
+                    }
+                },
+                None
+            )
+            .await
+            .unwrap();
+    }
+
     // Need to migrate fields on attachments, change `user_id`, `object_id`, etc to `parent`.
 
     // Reminder to update LATEST_REVISION when adding new migrations.

src/database/permissions/channel.rs

@@ -17,6 +17,7 @@ pub enum ChannelPermission {
     InviteOthers = 0b00000000000000000000000000100000,   // 32
     EmbedLinks = 0b00000000000000000000000001000000,     // 64
     UploadFiles = 0b00000000000000000000000010000000,   // 128
+    Masquerade = 0b00000000000000000000000100000000,   // 256
 }
 
 lazy_static! {
@@ -27,7 +28,8 @@ lazy_static! {
         + ChannelPermission::VoiceCall
         + ChannelPermission::InviteOthers
         + ChannelPermission::EmbedLinks
-        + ChannelPermission::UploadFiles;
+        + ChannelPermission::UploadFiles
+        + ChannelPermission::Masquerade;
 
     pub static ref DEFAULT_PERMISSION_SERVER: u32 =
         ChannelPermission::View
@@ -52,6 +54,7 @@ bitfield! {
     pub get_invite_others, _: 26;
     pub get_embed_links, _: 25;
     pub get_upload_files, _: 24;
+    pub get_masquerade, _: 23;
 }
 
 impl<'a> PermissionCalculator<'a> {

src/main.rs

@@ -20,6 +20,7 @@ pub mod routes;
 pub mod redis;
 pub mod util;
 pub mod version;
+pub mod task_queue;
 
 use async_std::task;
 use futures::join;
@@ -50,6 +51,7 @@ async fn main() {
     database::connect().await;
     redis::connect().await;
     notifications::hive::init_hive().await;
+    task_queue::start_queues().await;
 
     ctrlc::set_handler(move || {
         // Force ungraceful exit to avoid hang.

src/routes/channels/channel_ack.rs

@@ -2,9 +2,6 @@ use crate::database::*;
 use crate::notifications::events::ClientboundNotification;
 use crate::util::result::{Error, Result, EmptyResponse};
 
-use mongodb::bson::doc;
-use mongodb::options::UpdateOptions;
-
 #[put("/<target>/ack/<message>")]
 pub async fn req(user: User, target: Ref, message: Ref) -> Result<EmptyResponse> {
     if user.bot.is_some() {
@@ -22,31 +19,16 @@ pub async fn req(user: User, target: Ref, message: Ref) -> Result<EmptyResponse>
         Err(Error::MissingPermission)?
     }
 
-    let id = target.id();
-    get_collection("channel_unreads")
-        .update_one(
-            doc! {
-                "_id.channel": id,
-                "_id.user": &user.id
-            },
-            doc! {
-                "$unset": {
-                    "mentions": 1
-                },
-                "$set": {
-                    "last_id": &message.id
-                }
-            },
-            UpdateOptions::builder().upsert(true).build(),
-        )
-        .await
-        .map_err(|_| Error::DatabaseError {
-            operation: "update_one",
-            with: "channel_unreads",
-        })?;
+    crate::task_queue::task_ack::queue(
+        target.id().into(),
+        user.id.clone(),
+        crate::task_queue::task_ack::AckEvent::AckMessage {
+            id: message.id.clone()
+        }
+    ).await;
 
     ClientboundNotification::ChannelAck {
-        id: id.to_string(),
+        id: target.id().into(),
         user: user.id.clone(),
         message_id: message.id,
     }

src/routes/channels/group_create.rs

@@ -1,4 +1,5 @@
 use crate::database::*;
+use crate::util::idempotency::IdempotencyKey;
 use crate::util::result::{Error, Result};
 use crate::util::variables::MAX_GROUP_SIZE;
 
@@ -16,16 +17,13 @@ pub struct Data {
     name: String,
     #[validate(length(min = 0, max = 1024))]
     description: Option<String>,
-    // Maximum length of 36 allows both ULIDs and UUIDs.
-    #[validate(length(min = 1, max = 36))]
-    nonce: String,
     users: Vec<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     nsfw: Option<bool>
 }
 
 #[post("/create", data = "<info>")]
-pub async fn req(user: User, info: Json<Data>) -> Result<Value> {
+pub async fn req(_idempotency: IdempotencyKey, user: User, info: Json<Data>) -> Result<Value> {
     if user.bot.is_some() {
         return Err(Error::IsBot)
     }
@@ -52,27 +50,9 @@ pub async fn req(user: User, info: Json<Data>) -> Result<Value> {
         }
     }
 
-    if get_collection("channels")
-        .find_one(
-            doc! {
-                "nonce": &info.nonce
-            },
-            None,
-        )
-        .await
-        .map_err(|_| Error::DatabaseError {
-            operation: "find_one",
-            with: "channel",
-        })?
-        .is_some()
-    {
-        Err(Error::DuplicateNonce)?
-    }
-
     let id = Ulid::new().to_string();
     let channel = Channel::Group {
         id,
-        nonce: Some(info.nonce),
         name: info.name,
         description: info.description,
         owner: user.id,

src/routes/channels/message_send.rs

@@ -1,10 +1,11 @@
 use std::collections::HashSet;
 
 use crate::database::*;
+use crate::util::idempotency::IdempotencyKey;
 use crate::util::ratelimit::{Ratelimiter, RatelimitResponse};
 use crate::util::result::{Error, Result};
 
-use mongodb::{bson::doc, options::FindOneOptions};
+use mongodb::bson::doc;
 use regex::Regex;
 use rocket::serde::json::{Json, Value};
 use serde::{Deserialize, Serialize};
@@ -21,22 +22,24 @@ pub struct Reply {
 pub struct Data {
     #[validate(length(min = 0, max = 2000))]
     content: String,
-    // Maximum length of 36 allows both ULIDs and UUIDs.
-    #[validate(length(min = 1, max = 36))]
-    nonce: String,
     #[validate(length(min = 1, max = 128))]
     attachments: Option<Vec<String>>,
+    nonce: Option<String>,
     replies: Option<Vec<Reply>>,
+    #[validate]
+    masquerade: Option<Masquerade>
 }
 
 lazy_static! {
     // ignoring I L O and U is intentional
-    static ref RE_ULID: Regex = Regex::new(r"<@([0-9A-HJKMNP-TV-Z]{26})>").unwrap();
+    static ref RE_MENTION: Regex = Regex::new(r"<@([0-9A-HJKMNP-TV-Z]{26})>").unwrap();
 }
 
 #[post("/<target>/messages", data = "<message>")]
-pub async fn message_send(user: User, _r: Ratelimiter, target: Ref, message: Json<Data>) -> Result<RatelimitResponse<Value>> {
+pub async fn message_send(user: User, _r: Ratelimiter, mut idempotency: IdempotencyKey, target: Ref, message: Json<Data>) -> Result<RatelimitResponse<Value>> {
     let message = message.into_inner();
+    idempotency.consume_nonce(message.nonce.clone());
+
     message
         .validate()
         .map_err(|error| Error::FailedValidation { error })?;
@@ -59,38 +62,23 @@ pub async fn message_send(user: User, _r: Ratelimiter, target: Ref, message: Jso
         return Err(Error::MissingPermission)
     }
 
-    if get_collection("messages")
-        .find_one(
-            doc! {
-                "nonce": &message.nonce
-            },
-            FindOneOptions::builder()
-                .projection(doc! { "_id": 1 })
-                .build(),
-        )
-        .await
-        .map_err(|_| Error::DatabaseError {
-            operation: "find_one",
-            with: "message",
-        })?
-        .is_some()
-    {
-        Err(Error::DuplicateNonce)?
+    let mut mentions = HashSet::new();
+    for capture in RE_MENTION.captures_iter(&message.content) {
+        if let Some(mention) = capture.get(1) {
+            mentions.insert(mention.as_str().to_string());
+        }
     }
 
-    let id = Ulid::new().to_string();
-
-    let mut mentions = HashSet::new();
-    for capture in RE_ULID.captures_iter(&message.content) {
-        if let Some(mention) =  capture.get(1) {
-            mentions.insert(mention.as_str().to_string());
+    if let Some(_) = &message.masquerade {
+        if !perm.get_masquerade() {
+            return Err(Error::MissingPermission)
         }
     }
 
     let mut replies = HashSet::new();
     if let Some(entries) = message.replies {
         // ! FIXME: move this to app config
-        if entries.len() >= 5 {
+        if entries.len() > 5 {
             return Err(Error::TooManyReplies)
         }
 
@@ -107,14 +95,16 @@ pub async fn message_send(user: User, _r: Ratelimiter, target: Ref, message: Jso
         }
     }
 
+    let id = Ulid::new().to_string();
     let mut attachments = vec![];
+
     if let Some(ids) = &message.attachments {
         if ids.len() > 0 && !perm.get_upload_files() {
             return Err(Error::MissingPermission)
         }
 
         // ! FIXME: move this to app config
-        if ids.len() >= 5 {
+        if ids.len() > 5 {
             return Err(Error::TooManyAttachments)
         }
 
@@ -130,7 +120,7 @@ pub async fn message_send(user: User, _r: Ratelimiter, target: Ref, message: Jso
         author: user.id,
 
         content: Content::Text(message.content.clone()),
-        nonce: Some(message.nonce.clone()),
+        nonce: Some(idempotency.key),
         edited: None,
         embeds: None,
 
@@ -145,9 +135,9 @@ pub async fn message_send(user: User, _r: Ratelimiter, target: Ref, message: Jso
         } else {
             None
         },
+        masquerade: message.masquerade
     };
 
     msg.clone().publish(&target, perm.get_embed_links()).await?;
-
     Ok(RatelimitResponse(json!(msg)))
 }

src/routes/servers/channel_create.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 
 use crate::database::*;
+use crate::util::idempotency::IdempotencyKey;
 use crate::util::result::{Error, Result};
 
 use mongodb::bson::doc;
@@ -29,15 +30,12 @@ pub struct Data {
     name: String,
     #[validate(length(min = 0, max = 1024))]
     description: Option<String>,
-    // Maximum length of 36 allows both ULIDs and UUIDs.
-    #[validate(length(min = 1, max = 36))]
-    nonce: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     nsfw: Option<bool>,
 }
 
 #[post("/<target>/channels", data = "<info>")]
-pub async fn req(user: User, target: Ref, info: Json<Data>) -> Result<Value> {
+pub async fn req(_idempotency: IdempotencyKey, user: User, target: Ref, info: Json<Data>) -> Result<Value> {
     let info = info.into_inner();
     info.validate()
         .map_err(|error| Error::FailedValidation { error })?;
@@ -52,29 +50,11 @@ pub async fn req(user: User, target: Ref, info: Json<Data>) -> Result<Value> {
         Err(Error::MissingPermission)?
     }
 
-    if get_collection("channels")
-        .find_one(
-            doc! {
-                "nonce": &info.nonce
-            },
-            None,
-        )
-        .await
-        .map_err(|_| Error::DatabaseError {
-            operation: "find_one",
-            with: "channel",
-        })?
-        .is_some()
-    {
-        Err(Error::DuplicateNonce)?
-    }
-
     let id = Ulid::new().to_string();
     let channel = match info.channel_type {
         ChannelType::Text => Channel::TextChannel {
             id: id.clone(),
             server: target.id.clone(),
-            nonce: Some(info.nonce),
 
             name: info.name,
             description: info.description,
@@ -89,7 +69,6 @@ pub async fn req(user: User, target: Ref, info: Json<Data>) -> Result<Value> {
         ChannelType::Voice => Channel::VoiceChannel {
             id: id.clone(),
             server: target.id.clone(),
-            nonce: Some(info.nonce),
 
             name: info.name,
             description: info.description,

src/routes/servers/member_remove.rs

@@ -21,7 +21,7 @@ pub async fn req(user: User, target: Ref, member: String) -> Result<EmptyRespons
         return Err(Error::InvalidOperation);
     }
 
-    if target.id == target.owner {
+    if member.id.user == target.owner {
         return Err(Error::MissingPermission);
     }
 

src/routes/servers/server_create.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 
 use crate::database::*;
+use crate::util::idempotency::IdempotencyKey;
 use crate::util::result::{Error, Result};
 use crate::util::variables::MAX_SERVER_COUNT;
 
@@ -16,15 +17,12 @@ pub struct Data {
     name: String,
     #[validate(length(min = 0, max = 1024))]
     description: Option<String>,
-    // Maximum length of 36 allows both ULIDs and UUIDs.
-    #[validate(length(min = 1, max = 36))]
-    nonce: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     nsfw: Option<bool>
 }
 
 #[post("/create", data = "<info>")]
-pub async fn req(user: User, info: Json<Data>) -> Result<Value> {
+pub async fn req(_idempotency: IdempotencyKey, user: User, info: Json<Data>) -> Result<Value> {
     if user.bot.is_some() {
         return Err(Error::IsBot)
     }
@@ -39,29 +37,11 @@ pub async fn req(user: User, info: Json<Data>) -> Result<Value> {
     info.validate()
         .map_err(|error| Error::FailedValidation { error })?;
 
-    if get_collection("servers")
-        .find_one(
-            doc! {
-                "nonce": &info.nonce
-            },
-            None,
-        )
-        .await
-        .map_err(|_| Error::DatabaseError {
-            operation: "find_one",
-            with: "server",
-        })?
-        .is_some()
-    {
-        Err(Error::DuplicateNonce)?
-    }
-
     let id = Ulid::new().to_string();
     let cid = Ulid::new().to_string();
 
     let server = Server {
         id: id.clone(),
-        nonce: Some(info.nonce.clone()),
         owner: user.id.clone(),
 
         name: info.name,
@@ -92,7 +72,6 @@ pub async fn req(user: User, info: Json<Data>) -> Result<Value> {
     Channel::TextChannel {
         id: cid,
         server: id,
-        nonce: Some(info.nonce),
         name: "general".to_string(),
         description: None,
 

src/task_queue/mod.rs

@@ -0,0 +1,11 @@
+pub mod task_last_message_id;
+pub mod task_process_embeds;
+pub mod task_web_push;
+pub mod task_ack;
+
+pub async fn start_queues() {
+    async_std::task::spawn(task_last_message_id::run());
+    async_std::task::spawn(task_process_embeds::run());
+    async_std::task::spawn(task_web_push::run());
+    async_std::task::spawn(task_ack::run());
+}

src/task_queue/task_ack.rs

@@ -0,0 +1,143 @@
+// Queue Type: Debounced
+// TODO: Group similar events together with $in.
+
+use std::{collections::HashMap, time::{Duration, Instant}};
+
+use async_channel::{ Sender, Receiver, bounded };
+use log::info;
+use mongodb::{bson::doc, options::UpdateOptions};
+
+use crate::database::*;
+
+// 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;
+
+struct Message {
+    channel: String,
+    user: String,
+    event: AckEvent,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub enum AckEvent {
+    AddMention { ids: Vec<String> },
+    AckMessage { id: String }
+}
+
+#[derive(Debug)]
+struct Task {
+    event: AckEvent,
+    last_updated: Instant,
+    first_seen: Instant,
+}
+
+lazy_static! {
+    static ref CHANNEL: (Sender<Message>, Receiver<Message>) = bounded(100);
+}
+
+pub async fn queue(channel: String, user: String, event: AckEvent) {
+    CHANNEL.0.send(Message { channel, user, event }).await.ok();
+}
+
+pub async fn run() {
+    let unreads = get_collection("channel_unreads");
+    let mut tasks = HashMap::<(String, String), Task>::new();
+    let mut keys = vec![];
+
+    loop {
+        // Find due tasks.
+        for (key, Task { first_seen, last_updated, .. }) in &tasks {
+            if first_seen.elapsed().as_secs() > EXPIRE_CONSTANT ||
+                last_updated.elapsed().as_secs() > SAVE_CONSTANT {
+                keys.push(key.clone());
+            }
+        }
+
+        // Commit any due tasks to the database.
+        for key in &keys {
+            if let Some(Task { event, .. }) = tasks.remove(key) {
+                let (user, channel) = key;
+
+                match event {
+                    AckEvent::AddMention { ids } => {
+                        unreads.update_one(
+                            doc! {
+                                "_id.channel": channel,
+                                "_id.user": user,
+                            },
+                            doc! {
+                                "$push": {
+                                    "mentions": {
+                                        "$each": ids
+                                    }
+                                }
+                            },
+                            UpdateOptions::builder().upsert(true).build(),
+                        )
+                        .await
+                        .ok();
+
+                        info!("Added mentions for {} in {}.", user, channel);
+                    },
+                    AckEvent::AckMessage { id } => {
+                        unreads.update_one(
+                            doc! {
+                                "_id.channel": channel,
+                                "_id.user": user,
+                            },
+                            doc! {
+                                "$unset": {
+                                    "mentions": 1
+                                },
+                                "$set": {
+                                    "last_id": id
+                                }
+                            },
+                            UpdateOptions::builder().upsert(true).build(),
+                        )
+                        .await
+                        .ok();
+
+                        info!("User {} acknowledged {}.", user, channel);
+                    }
+                }
+            }
+        }
+
+        // Clear keys
+        keys.clear();
+
+        // Queue incoming tasks.
+        while let Ok(Message { channel, user, mut event }) = CHANNEL.1.try_recv() {
+            let key = (user, channel);
+            if let Some(mut existing_task) = tasks.get_mut(&key) {
+                existing_task.last_updated = Instant::now();
+
+                match &mut event {
+                    AckEvent::AddMention { ids } => {
+                        if let AckEvent::AddMention { ids: existing } = &mut existing_task.event {
+                            existing.append(ids);
+                        } else {
+                            existing_task.event = event;
+                        }
+                    }
+                    AckEvent::AckMessage { .. } => {
+                        existing_task.event = event;
+                    }
+                }
+            } else {
+                tasks.insert(key, Task {
+                    event,
+                    last_updated: Instant::now(),
+                    first_seen: Instant::now()
+                });
+            }
+        }
+
+        // Sleep for an arbitrary amount of time.
+        async_std::task::sleep(Duration::from_secs(1)).await;
+    }
+}

src/task_queue/task_last_message_id.rs

@@ -0,0 +1,95 @@
+// Queue Type: Debounced
+use std::{collections::HashMap, time::{Duration, Instant}};
+
+use async_channel::{ Sender, Receiver, bounded };
+use log::info;
+use mongodb::bson::doc;
+
+use crate::database::*;
+
+// 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;
+
+struct Message {
+    channel: String,
+    id: String,
+    is_dm: bool
+}
+
+#[derive(Debug)]
+struct Task {
+    id: String,
+    is_dm: bool,
+    last_updated: Instant,
+    first_seen: Instant,
+}
+
+lazy_static! {
+    static ref CHANNEL: (Sender<Message>, Receiver<Message>) = bounded(100);
+}
+
+pub async fn queue(channel: String, id: String, is_dm: bool) {
+    CHANNEL.0.send(Message { channel, id, is_dm }).await.ok();
+}
+
+pub async fn run() {
+    let channels = get_collection("channels");
+    let mut tasks = HashMap::<String, Task>::new();
+    let mut keys = vec![];
+
+    loop {
+        // Find due tasks.
+        for (key, Task { first_seen, last_updated, .. }) in &tasks {
+            if first_seen.elapsed().as_secs() > EXPIRE_CONSTANT ||
+                last_updated.elapsed().as_secs() > SAVE_CONSTANT {
+                keys.push(key.clone());
+            }
+        }
+
+        // Commit any due tasks to the database.
+        for key in &keys {
+            if let Some(Task { id, is_dm, .. }) = tasks.remove(key) {
+                let mut set = doc! { "last_message_id": &id };
+
+                if is_dm {
+                    set.insert("active", true);
+                }
+
+                channels
+                    .update_one(
+                        doc! { "_id": key },
+                        doc! { "$set": set },
+                        None,
+                    )
+                    .await
+                    .ok();
+
+                info!("Updated last_message_id for {} to {}.", key, id);
+            }
+        }
+
+        // Clear keys
+        keys.clear();
+
+        // Queue incoming tasks.
+        while let Ok(Message { channel, id, is_dm }) = CHANNEL.1.try_recv() {
+            if let Some(mut existing_task) = tasks.get_mut(&channel) {
+                existing_task.id = id;
+                existing_task.last_updated = Instant::now();
+            } else {
+                tasks.insert(channel, Task {
+                    id,
+                    is_dm,
+                    last_updated: Instant::now(),
+                    first_seen: Instant::now()
+                });
+            }
+        }
+
+        // Sleep for an arbitrary amount of time.
+        async_std::task::sleep(Duration::from_secs(1)).await;
+    }
+}

src/task_queue/task_process_embeds.rs

@@ -0,0 +1,53 @@
+// Queue Type: Linear
+use async_channel::{ Sender, Receiver, bounded };
+use log::info;
+use mongodb::bson::{doc, to_bson};
+
+use crate::{database::*, notifications::events::ClientboundNotification};
+
+struct Message {
+    channel: String,
+    id: String,
+    content: 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(Message { channel, id, content }).await.ok();
+}
+
+pub async fn run() {
+    let messages = get_collection("messages");
+
+    while let Ok(Message { 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
+                {
+                    info!("Generated embeds for {}.", &id);
+                    ClientboundNotification::MessageUpdate {
+                        id,
+                        channel: channel.clone(),
+                        data: json!({ "embeds": embeds }),
+                    }
+                    .publish(channel);
+                }
+            }
+        }
+    }
+}

src/task_queue/task_web_push.rs

@@ -0,0 +1,72 @@
+// Queue Type: Linear
+use async_channel::{ Sender, Receiver, bounded };
+use mongodb::bson::{doc};
+use web_push::{ContentEncoding, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder};
+use rauth::entities::{Model, Session};
+use futures::StreamExt;
+
+use crate::util::variables::VAPID_PRIVATE_KEY;
+use crate::database::*;
+
+struct Message {
+    recipients: Vec<String>,
+    payload: String
+}
+
+lazy_static! {
+    static ref CHANNEL: (Sender<Message>, Receiver<Message>) = bounded(100);
+}
+
+pub async fn queue(recipients: Vec<String>, payload: String) {
+    CHANNEL.0.send(Message { recipients, payload }).await.ok();
+}
+
+pub async fn run() {
+    let client = WebPushClient::new();
+    let key = base64::decode_config(VAPID_PRIVATE_KEY.clone(), base64::URL_SAFE)
+        .expect("valid `VAPID_PRIVATE_KEY`");
+
+    while let Ok(Message { recipients, payload }) = CHANNEL.1.recv().await {
+        if let Ok(mut cursor) = Session::find(
+            &get_db(),
+            doc! {
+                "_id": {
+                    "$in": recipients
+                },
+                "subscription": {
+                    "$exists": true
+                }
+            },
+            None
+        )
+        .await {
+            while let Some(Ok(session)) = cursor.next().await {
+                if let Some(sub) = session.subscription {
+                    let subscription = SubscriptionInfo {
+                        endpoint: sub.endpoint,
+                        keys: SubscriptionKeys {
+                            auth: sub.auth,
+                            p256dh: sub.p256dh
+                        }
+                    };
+
+                    let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
+                    let sig_builder = VapidSignatureBuilder::from_pem(
+                        std::io::Cursor::new(&key),
+                        &subscription,
+                    )
+                    .unwrap();
+
+                    let signature = sig_builder.build().unwrap();
+                    builder.set_vapid_signature(signature);
+                    builder.set_payload(ContentEncoding::AesGcm, payload.as_bytes());
+
+                    let msg = builder.build().unwrap();
+                    client.send(msg).await.ok();
+
+                    info!("Sent Web Push notification to {:?}.", session.id);
+                }
+            }
+        }
+    }
+}

src/util/idempotency.rs

@@ -0,0 +1,56 @@
+use crate::util::result::Error;
+
+use mongodb::bson::doc;
+use rocket::http::Status;
+use rocket::request::{FromRequest, Outcome};
+use serde::{Deserialize, Serialize};
+use validator::Validate;
+
+#[derive(Validate, Serialize, Deserialize)]
+pub struct IdempotencyKey {
+    #[validate(length(min = 1, max = 64))]
+    pub key: String,
+}
+
+impl IdempotencyKey {
+    // Backwards compatibility.
+    // Issue #109
+    pub fn consume_nonce(&mut self, v: Option<String>) {
+        if let Some(v) = v {
+            self.key = v;
+        }
+    }
+}
+
+lazy_static! {
+    static ref TOKEN_CACHE: std::sync::Mutex<lru::LruCache<String, ()>> = std::sync::Mutex::new(lru::LruCache::new(100));
+}
+
+#[async_trait]
+impl<'r> FromRequest<'r> for IdempotencyKey {
+    type Error = Error;
+
+    async fn from_request(request: &'r rocket::Request<'_>) -> Outcome<Self, Self::Error> {
+        if let Some(key) = request
+            .headers()
+            .get("Idempotency-Key")
+            .next()
+            .map(|k| k.to_string()) {
+            let idempotency = IdempotencyKey { key };
+            if let Err(error) = idempotency.validate() {
+                return Outcome::Failure((Status::BadRequest, Error::FailedValidation { error }));
+            }
+
+            if let Ok(mut cache) = TOKEN_CACHE.lock() {
+                if cache.get(&idempotency.key).is_some() {
+                    return Outcome::Failure((Status::Conflict, Error::DuplicateNonce));
+                }
+        
+                cache.put(idempotency.key.clone(), ());
+                return Outcome::Success(idempotency);
+            }
+        }
+        
+        Outcome::Success(IdempotencyKey { key: ulid::Ulid::new().to_string() })
+    }
+}

src/util/mod.rs

@@ -1,4 +1,5 @@
 pub mod result;
 pub mod variables;
 pub mod ratelimit;
-pub mod regex;
+pub mod regex;
+pub mod idempotency;

src/version.rs

@@ -1 +1 @@
-pub const VERSION: &str = "0.5.3-alpha.6";
+pub const VERSION: &str = "0.5.3-alpha.8";