chore: merge local branch 'feat/discoverability'

feat: add new index

.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

.github/workflows/triage_issue.yml

@@ -0,0 +1,49 @@
+name: Add Issue to Board
+
+on:
+  issues:
+    types: [opened]
+
+jobs:
+  track_issue:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Get project data
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+        run: |
+          gh api graphql -f query='
+            query {
+              organization(login: "revoltchat"){
+                projectNext(number: 3) {
+                  id
+                  fields(first:20) {
+                    nodes {
+                      id
+                      name
+                      settings
+                    }
+                  }
+                }
+              }
+            }' > project_data.json
+
+          echo 'PROJECT_ID='$(jq '.data.organization.projectNext.id' project_data.json) >> $GITHUB_ENV
+          echo 'STATUS_FIELD_ID='$(jq '.data.organization.projectNext.fields.nodes[] | select(.name== "Status") | .id' project_data.json) >> $GITHUB_ENV
+          echo 'TODO_OPTION_ID='$(jq '.data.organization.projectNext.fields.nodes[] | select(.name== "Status") |.settings | fromjson.options[] | select(.name=="Todo") |.id' project_data.json) >> $GITHUB_ENV
+
+      - name: Add issue to project
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+          ISSUE_ID: ${{ github.event.issue.node_id }}
+        run: |
+          item_id="$( gh api graphql -f query='
+            mutation($project:ID!, $issue:ID!) {
+              addProjectNextItem(input: {projectId: $project, contentId: $issue}) {
+                projectNextItem {
+                  id
+                }
+              }
+            }' -f project=$PROJECT_ID -f issue=$ISSUE_ID --jq '.data.addProjectNextItem.projectNextItem.id')"
+          
+          echo 'ITEM_ID='$item_id >> $GITHUB_ENV

.github/workflows/triage_pr.yml

@@ -0,0 +1,72 @@
+name: Add PR to Board
+
+on:
+  pull_request_target:
+    types: [opened, synchronize, ready_for_review, review_requested]
+
+jobs:
+  track_pr:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Get project data
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+        run: |
+          gh api graphql -f query='
+            query {
+              organization(login: "revoltchat"){
+                projectNext(number: 3) {
+                  id
+                  fields(first:20) {
+                    nodes {
+                      id
+                      name
+                      settings
+                    }
+                  }
+                }
+              }
+            }' > project_data.json
+
+          echo 'PROJECT_ID='$(jq '.data.organization.projectNext.id' project_data.json) >> $GITHUB_ENV
+          echo 'STATUS_FIELD_ID='$(jq '.data.organization.projectNext.fields.nodes[] | select(.name== "Status") | .id' project_data.json) >> $GITHUB_ENV
+          echo 'INCOMING_OPTION_ID='$(jq '.data.organization.projectNext.fields.nodes[] | select(.name== "Status") |.settings | fromjson.options[] | select(.name=="Incoming PRs") |.id' project_data.json) >> $GITHUB_ENV
+
+      - name: Add PR to project
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+          PR_ID: ${{ github.event.pull_request.node_id }}
+        run: |
+          item_id="$( gh api graphql -f query='
+            mutation($project:ID!, $pr:ID!) {
+              addProjectNextItem(input: {projectId: $project, contentId: $pr}) {
+                projectNextItem {
+                  id
+                }
+              }
+            }' -f project=$PROJECT_ID -f pr=$PR_ID --jq '.data.addProjectNextItem.projectNextItem.id')"
+          
+          echo 'ITEM_ID='$item_id >> $GITHUB_ENV
+
+      - name: Set fields
+        env:
+          GITHUB_TOKEN: ${{ secrets.PAT }}
+        run: |
+          gh api graphql -f query='
+            mutation (
+              $project: ID!
+              $item: ID!
+              $status_field: ID!
+              $status_value: String!
+            ) {
+              set_status: updateProjectNextItemField(input: {
+                projectId: $project
+                itemId: $item
+                fieldId: $status_field
+                value: $status_value
+              }) {
+                projectNextItem {
+                  id
+                  }
+              }
+            }' -f project=$PROJECT_ID -f item=$ITEM_ID -f status_field=$STATUS_FIELD_ID -f status_value=${{ env.INCOMING_OPTION_ID }} --silent

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"
@@ -2554,7 +2577,7 @@ dependencies = [
 [[package]]
 name = "rauth"
 version = "1.0.0"
-source = "git+https://github.com/insertish/rauth?rev=775a6f3ffb9ffc0e382af32fac2924e9a707e890#775a6f3ffb9ffc0e382af32fac2924e9a707e890"
+source = "git+https://github.com/insertish/rauth?rev=157263ffcbd6cb1073e288b215db227634c4c29a#157263ffcbd6cb1073e288b215db227634c4c29a"
 dependencies = [
  "async-std",
  "base32",
@@ -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",
@@ -2827,7 +2852,7 @@ dependencies = [
 [[package]]
 name = "rmp"
 version = "0.8.10"
-source = "git+https://github.com/3Hren/msgpack-rust?rev=5bf2c24203ad422233cf35b7b7bfad9f7e811814#5bf2c24203ad422233cf35b7b7bfad9f7e811814"
+source = "git+https://github.com/insertish/msgpack-rust?rev=5bf2c24203ad422233cf35b7b7bfad9f7e811814#5bf2c24203ad422233cf35b7b7bfad9f7e811814"
 dependencies = [
  "byteorder",
  "num-traits",
@@ -2836,7 +2861,7 @@ dependencies = [
 [[package]]
 name = "rmp-serde"
 version = "0.15.5"
-source = "git+https://github.com/3Hren/msgpack-rust?rev=5bf2c24203ad422233cf35b7b7bfad9f7e811814#5bf2c24203ad422233cf35b7b7bfad9f7e811814"
+source = "git+https://github.com/insertish/msgpack-rust?rev=5bf2c24203ad422233cf35b7b7bfad9f7e811814#5bf2c24203ad422233cf35b7b7bfad9f7e811814"
 dependencies = [
  "byteorder",
  "rmp",

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"
@@ -36,11 +37,12 @@ base64 = "0.13.0"
 serde_json = "1.0.57"
 serde = { version = "1.0.115", features = ["derive"] }
 validator = { version = "0.11", features = ["derive"] }
-rmp-serde = { git = "https://github.com/3Hren/msgpack-rust", rev = "5bf2c24203ad422233cf35b7b7bfad9f7e811814" }
+rmp-serde = { git = "https://github.com/insertish/msgpack-rust", rev = "5bf2c24203ad422233cf35b7b7bfad9f7e811814" }
 
 # 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"] }
 
@@ -48,7 +50,7 @@ async-std = { version = "1.8.0", features = ["tokio1", "tokio02", "attributes"]
 web-push = "0.7.2"
 many-to-many = "0.1.2"
 lettre = "0.10.0-alpha.4"
-rauth = { git = "https://github.com/insertish/rauth", rev = "775a6f3ffb9ffc0e382af32fac2924e9a707e890" }
+rauth = { git = "https://github.com/insertish/rauth", rev = "157263ffcbd6cb1073e288b215db227634c4c29a" }
 hive_pubsub = { git = "https://gitlab.insrt.uk/insert/hive", rev = "b0f3db9d33990530d7640d4bbb309c8d9eb5c0cf", features = ["redis-backend"] }
 
 # redis

set_version.sh

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

src/database/entities/bots.rs

@@ -1,5 +1,9 @@
 use serde::{Serialize, Deserialize};
 
+pub fn if_false(t: &bool) -> bool {
+    !t
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Bot {
     #[serde(rename = "_id")]
@@ -7,6 +11,10 @@ pub struct Bot {
     pub owner: String,
     pub token: String,
     pub public: bool,
+    #[serde(skip_serializing_if = "if_false", default)]
+    pub analytics: bool,
+    #[serde(skip_serializing_if = "if_false", default)]
+    pub discoverable: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub interactions_url: Option<String>,
 }

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,16 @@
-use crate::util::variables::{USE_JANUARY, VAPID_PRIVATE_KEY, PUBLIC_URL};
+use crate::util::variables::{USE_JANUARY, PUBLIC_URL, APP_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::collections::HashSet;
 use std::time::SystemTime;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -26,6 +22,7 @@ pub struct PushNotification {
     pub body: String,
     pub tag: String,
     pub timestamp: u64,
+    pub url: String,
 }
 
 impl PushNotification {
@@ -69,6 +66,7 @@ impl PushNotification {
             body,
             tag: channel.id().to_string(),
             timestamp,
+            url: format!("{}/channel/{}/{}", *APP_URL, channel.id(), msg.id),
         }
     }
 }
@@ -129,6 +127,7 @@ impl Content {
             target.id().to_string(),
             self,
             None,
+            None,
             None
         )
         .publish(&target, false)
@@ -136,6 +135,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 +164,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 +176,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 +188,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 +207,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 +273,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;
         }
     }
 
@@ -431,16 +341,23 @@ 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");
+    pub fn add_associated_user_ids(&self, ids: &mut HashSet<String>) {
+        ids.insert(self.author.clone());
 
+        if let Content::SystemMessage(sys) = &self.content {
+            match sys {
+                SystemMessage::Text { .. } => {},
+                SystemMessage::UserAdded { id, by } => { ids.insert(id.clone()); ids.insert(by.clone()); },
+                SystemMessage::UserRemove { id, by } => { ids.insert(id.clone()); ids.insert(by.clone()); },
+                SystemMessage::UserJoined { id } => { ids.insert(id.clone()); },
+                SystemMessage::UserLeft { id } => { ids.insert(id.clone()); },
+                SystemMessage::UserKicked { id } => { ids.insert(id.clone()); },
+                SystemMessage::UserBanned { id } => { ids.insert(id.clone()); },
+                SystemMessage::ChannelRenamed { by, .. } => { ids.insert(by.clone()); },
+                SystemMessage::ChannelDescriptionChanged { by } => { ids.insert(by.clone()); },
+                SystemMessage::ChannelIconChanged { by } => { ids.insert(by.clone()); }
+            }
+        }
+    }
 }

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,
@@ -112,7 +110,11 @@ pub struct Server {
     pub flags: Option<i32>,
 
     #[serde(skip_serializing_if = "if_false", default)]
-    pub nsfw: bool
+    pub nsfw: bool,
+    #[serde(skip_serializing_if = "if_false", default)]
+    pub analytics: bool,
+    #[serde(skip_serializing_if = "if_false", default)]
+    pub discoverable: bool
 }
 
 impl Server {
@@ -234,6 +236,30 @@ impl Server {
             .collect::<Vec<Member>>())
     }
 
+    pub async fn fetch_members_with_ids(id: &str, ids: &Vec<String>) -> Result<Vec<Member>> {
+        Ok(get_collection("server_members")
+            .find(
+                doc! {
+                    "_id.server": id,
+                    "_id.user": {
+                        "$in": ids
+                    }
+                },
+                None,
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "find",
+                with: "server_members",
+            })?
+            .filter_map(async move |s| s.ok())
+            .collect::<Vec<Document>>()
+            .await
+            .into_iter()
+            .filter_map(|x| from_document(x).ok())
+            .collect::<Vec<Member>>())
+    }
+
     pub async fn fetch_member_ids(id: &str) -> Result<Vec<String>> {
         Ok(get_collection("server_members")
             .find(

src/database/entities/user.rs

@@ -68,8 +68,12 @@ pub enum Badges {
     Translator = 2,
     Supporter = 4,
     ResponsibleDisclosure = 8,
-    RevoltTeam = 16,
+    Founder = 16,
+    PlatformModeration = 32,
+    ActiveSupporter = 64,
+    Paw = 128,
     EarlyAdopter = 256,
+    ReservedRelevantJokeBadge1 = 512
 }
 
 impl_op_ex_commutative!(+ |a: &i32, b: &Badges| -> i32 { *a | *b as i32 });
@@ -214,7 +218,7 @@ impl User {
 
     /// Utility function for fetching multiple users from the perspective of one.
     /// Assumes user has a mutual connection with others.
-    pub async fn fetch_multiple_users(&self, user_ids: Vec<String>) -> Result<Vec<User>> {
+    pub async fn fetch_multiple_users(&self, user_ids: &Vec<String>) -> Result<Vec<User>> {
         let mut users = vec![];
         let mut cursor = get_collection("users")
             .find(

src/database/guards/reference.rs

@@ -58,7 +58,25 @@ impl Ref {
     }
 
     pub async fn fetch_invite(&self) -> Result<Invite> {
-        self.fetch("channel_invites").await
+        match self.fetch("channel_invites").await {
+            Ok(invite) => Ok(invite),
+            Err(err) => {
+                if let Ok(server) = self.fetch::<Server>("servers").await {
+                    if server.discoverable {
+                        return Ok(
+                            Invite::Server {
+                                code: server.id.clone(),
+                                server: server.id,
+                                creator: server.owner,
+                                channel: server.channels[0].clone(),
+                            }
+                        )
+                    }
+                }
+
+                Err(err)
+            }
+        }
     }
 
     pub async fn fetch_bot(&self) -> Result<Bot> {

src/database/migrations/init.rs

@@ -102,6 +102,19 @@ pub async fn create_database() {
                         "content": "text"
                     },
                     "name": "content"
+                },
+                {
+                    "key": {
+                        "channel": 1
+                    },
+                    "name": "channel"
+                },
+                {
+                    "key": {
+                        "channel": 1,
+                        "_id": 1
+                    },
+                    "name": "channel_id_compound"
                 }
             ]
         },
@@ -110,6 +123,56 @@ pub async fn create_database() {
     .await
     .expect("Failed to create message index.");
 
+    get_db()
+    .run_command(
+        doc! {
+            "createIndexes": "channel_unreads",
+            "indexes": [
+                {
+                    "key": {
+                        "_id.channel": 1,
+                        "_id.user": 1,
+                    },
+                    "name": "compound_id"
+                },
+                {
+                    "key": {
+                        "_id.user": 1,
+                    },
+                    "name": "user_id"
+                }
+            ]
+        },
+        None,
+    )
+    .await
+    .expect("Failed to create channel_unreads index.");
+
+    get_db()
+    .run_command(
+        doc! {
+            "createIndexes": "server_members",
+            "indexes": [
+                {
+                    "key": {
+                        "_id.server": 1,
+                        "_id.user": 1,
+                    },
+                    "name": "compound_id"
+                },
+                {
+                    "key": {
+                        "_id.user": 1,
+                    },
+                    "name": "user_id"
+                }
+            ]
+        },
+        None,
+    )
+    .await
+    .expect("Failed to create server_members index.");
+
     db.collection("migrations")
         .insert_one(
             doc! {

src/database/migrations/scripts.rs

@@ -11,7 +11,7 @@ struct MigrationInfo {
     revision: i32,
 }
 
-pub const LATEST_REVISION: i32 = 10;
+pub const LATEST_REVISION: i32 = 13;
 
 pub async fn migrate_database() {
     let migrations = get_collection("migrations");
@@ -342,6 +342,131 @@ 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();
+    }
+
+    if revision <= 11 {
+        info!("Running migration [revision 11 / 2021-11-14]: Add indexes to database.");
+
+        get_db()
+        .run_command(
+            doc! {
+                "createIndexes": "messages",
+                "indexes": [
+                    {
+                        "key": {
+                            "channel": 1
+                        },
+                        "name": "channel"
+                    }
+                ]
+            },
+            None,
+        )
+        .await
+        .expect("Failed to create message index.");
+
+        get_db()
+        .run_command(
+            doc! {
+                "createIndexes": "channel_unreads",
+                "indexes": [
+                    {
+                        "key": {
+                            "_id.channel": 1,
+                            "_id.user": 1,
+                        },
+                        "name": "compound_id"
+                    },
+                    {
+                        "key": {
+                            "_id.user": 1,
+                        },
+                        "name": "user_id"
+                    }
+                ]
+            },
+            None,
+        )
+        .await
+        .expect("Failed to create channel_unreads index.");
+
+        get_db()
+        .run_command(
+            doc! {
+                "createIndexes": "server_members",
+                "indexes": [
+                    {
+                        "key": {
+                            "_id.server": 1,
+                            "_id.user": 1,
+                        },
+                        "name": "compound_id"
+                    },
+                    {
+                        "key": {
+                            "_id.user": 1,
+                        },
+                        "name": "user_id"
+                    }
+                ]
+            },
+            None,
+        )
+        .await
+        .expect("Failed to create server_members index.");
+    }
+
+    if revision <= 12 {
+        info!("Running migration [revision 12 / 2021-11-21]: Add indexes to database.");
+
+        get_db()
+        .run_command(
+            doc! {
+                "createIndexes": "messages",
+                "indexes": [
+                    {
+                        "key": {
+                            "channel": 1,
+                            "_id": 1
+                        },
+                        "name": "channel_id_compound"
+                    }
+                ]
+            },
+            None,
+        )
+        .await
+        .expect("Failed to create message index.");
+    }
+
     // 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/notifications/payload.rs

@@ -104,7 +104,7 @@ pub async fn generate_ready(mut user: User) -> Result<ClientboundNotification> {
 
     user_ids.remove(&user.id);
     let mut users = if user_ids.len() > 0 {
-        user.fetch_multiple_users(user_ids.into_iter().collect::<Vec<String>>())
+        user.fetch_multiple_users(&user_ids.into_iter().collect::<Vec<String>>())
             .await?
     } else {
         vec![]

src/routes/bots/create.rs

@@ -48,6 +48,8 @@ pub async fn create_bot(user: User, info: Json<Data>) -> Result<Value> {
         owner: user.id.clone(),
         token,
         public: false,
+        analytics: false,
+        discoverable: false,
         interactions_url: None
     };
 

src/routes/bots/edit.rs

@@ -14,6 +14,7 @@ pub struct Data {
     #[serde(skip_serializing_if = "Option::is_none")]
     name: Option<String>,
     public: Option<bool>,
+    analytics: Option<bool>,
     interactions_url: Option<String>,
     remove: Option<RemoveBotField>,
 }
@@ -30,6 +31,7 @@ pub async fn edit_bot(user: User, target: Ref, data: Json<Data>) -> Result<Empty
 
     if data.name.is_none()
         && data.public.is_none()
+        && data.analytics.is_none()
         && data.interactions_url.is_none()
         && data.remove.is_none()
     {
@@ -87,6 +89,10 @@ pub async fn edit_bot(user: User, target: Ref, data: Json<Data>) -> Result<Empty
         set.insert("public", public);
     }
 
+    if let Some(analytics) = &data.analytics {
+        set.insert("analytics", analytics);
+    }
+
     if let Some(interactions_url) = &data.interactions_url {
         set.insert("interactions_url", interactions_url);
     }

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/members_fetch.rs

@@ -16,7 +16,7 @@ pub async fn req(user: User, target: Ref) -> Result<Value> {
     }
 
     if let Channel::Group { recipients, .. } = target {
-        Ok(json!(user.fetch_multiple_users(recipients).await?))
+        Ok(json!(user.fetch_multiple_users(&recipients).await?))
     } else {
         Err(Error::InvalidOperation)
     }

src/routes/channels/message_query.rs

@@ -161,18 +161,18 @@ pub async fn req(user: User, target: Ref, options: Options) -> Result<Value> {
     if options.include_users.unwrap_or_else(|| false) {
         let mut ids = HashSet::new();
         for message in &messages {
-            ids.insert(message.author.clone());
+            message.add_associated_user_ids(&mut ids);
         }
 
         ids.remove(&user.id);
         let user_ids = ids.into_iter().collect();
-        let users = user.fetch_multiple_users(user_ids).await?;
+        let users = user.fetch_multiple_users(&user_ids).await?;
 
         if let Channel::TextChannel { server, .. } = target {
             Ok(json!({
                 "messages": messages,
                 "users": users,
-                "members": Server::fetch_members(&server).await?
+                "members": Server::fetch_members_with_ids(&server, &user_ids).await?
             }))
         } else {
             Ok(json!({

src/routes/channels/message_search.rs

@@ -108,10 +108,10 @@ pub async fn req(user: User, target: Ref, options: Json<Options>) -> Result<Valu
                             }
                         },
                         Sort::Latest => doc! {
-                            "_id": -1
+                            "_id": -1 as i32
                         },
                         Sort::Oldest => doc! {
-                            "_id": 1
+                            "_id": 1 as i32
                         }
                     }
                 )
@@ -137,18 +137,18 @@ pub async fn req(user: User, target: Ref, options: Json<Options>) -> Result<Valu
     if options.include_users.unwrap_or_else(|| false) {
         let mut ids = HashSet::new();
         for message in &messages {
-            ids.insert(message.author.clone());
+            message.add_associated_user_ids(&mut ids);
         }
 
         ids.remove(&user.id);
         let user_ids = ids.into_iter().collect();
-        let users = user.fetch_multiple_users(user_ids).await?;
+        let users = user.fetch_multiple_users(&user_ids).await?;
 
         if let Channel::TextChannel { server, .. } = target {
             Ok(json!({
                 "messages": messages,
                 "users": users,
-                "members": Server::fetch_members(&server).await?
+                "members": Server::fetch_members_with_ids(&server, &user_ids).await?
             }))
         } else {
             Ok(json!({

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_fetch_all.rs

@@ -46,6 +46,6 @@ pub async fn req(user: User, target: Ref) -> Result<Value> {
 
     Ok(json!({
         "members": members,
-        "users": user.fetch_multiple_users(member_ids).await?
+        "users": user.fetch_multiple_users(&member_ids).await?
     }))
 }

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,
@@ -86,13 +66,14 @@ pub async fn req(user: User, info: Json<Data>) -> Result<Value> {
         banner: None,
 
         flags: None,
-        nsfw: info.nsfw.unwrap_or_default()
+        nsfw: info.nsfw.unwrap_or_default(),
+        analytics: false,
+        discoverable: false,
     };
 
     Channel::TextChannel {
         id: cid,
         server: id,
-        nonce: Some(info.nonce),
         name: "general".to_string(),
         description: None,
 

src/routes/servers/server_edit.rs

@@ -18,8 +18,8 @@ pub struct Data {
     categories: Option<Vec<Category>>,
     system_messages: Option<SystemMessageChannels>,
     remove: Option<RemoveServerField>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    nsfw: Option<bool>
+    nsfw: Option<bool>,
+    analytics: Option<bool>,
 }
 
 #[patch("/<target>", data = "<data>")]
@@ -28,7 +28,7 @@ pub async fn req(user: User, target: Ref, data: Json<Data>) -> Result<EmptyRespo
     data.validate()
         .map_err(|error| Error::FailedValidation { error })?;
 
-    if data.name.is_none() && data.description.is_none() && data.icon.is_none() && data.banner.is_none() && data.remove.is_none() && data.categories.is_none() && data.system_messages.is_none()
+    if data.name.is_none() && data.description.is_none() && data.icon.is_none() && data.banner.is_none() && data.remove.is_none() && data.categories.is_none() && data.system_messages.is_none() && data.nsfw.is_none() && data.analytics.is_none()
     {
         return Ok(EmptyResponse {});
     }
@@ -111,6 +111,10 @@ pub async fn req(user: User, target: Ref, data: Json<Data>) -> Result<EmptyRespo
         set.insert("nsfw", nsfw);
     }
 
+    if let Some(analytics) = &data.analytics {
+        set.insert("analytics", analytics);
+    }
+
     let mut operations = doc! {};
     if set.len() > 0 {
         operations.insert("$set", &set);

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! {
+                "user_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.13";