Add utility function for including relationship on User.

Code cleanup/formatting

See merge request revolt/delta!2

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt"
-version = "0.3.0-alpha"
+version = "0.3.2"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"
 
@@ -12,14 +12,13 @@ many-to-many = "0.1.2"
 impl_ops = "0.1.1"
 ctrlc = { version = "3.0", features = ["termination"] }
 async-tungstenite = { version = "0.10.0", features = ["async-std-runtime"] }
-rauth = { git = "https://gitlab.insrt.uk/insert/rauth" }
-async-std = { version = "1.8.0", features = ["tokio02"] }
+rauth = { git = "https://gitlab.insrt.uk/insert/rauth", rev = "8f3ea627" }
+async-std = { version = "1.8.0", features = ["tokio02", "attributes"] }
 
 hive_pubsub = { version = "0.4.3", features = ["mongo"] }
-rocket_cors = { git = "https://github.com/lawliet89/rocket_cors", branch = "master" }
-rocket_contrib = { git = "https://github.com/SergioBenitez/Rocket", branch = "master" }
-rocket = { git = "https://github.com/SergioBenitez/Rocket", branch = "master", default-features = false }
-# ! FIXME: Switch to async-std runtime.
+rocket_cors = { git = "https://github.com/insertish/rocket_cors", branch = "master" }
+rocket_contrib = { git = "https://github.com/SergioBenitez/Rocket", rev = "031948c1daaa146128d8a435be116476f2adde00" }
+rocket = { git = "https://github.com/SergioBenitez/Rocket", rev = "031948c1daaa146128d8a435be116476f2adde00", default-features = false }
 mongodb = { version = "1.1.1", features = ["tokio-runtime"], default-features = false }
 
 once_cell = "1.4.1"

src/database/entities/channel.rs

@@ -1,42 +1,20 @@
-use crate::{
-    database::get_collection,
-    util::result::{Error, Result},
-};
-use mongodb::bson::to_document;
+use crate::database::*;
+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+use mongodb::bson::{doc, from_document, to_document};
+use rocket_contrib::json::JsonValue;
 use serde::{Deserialize, Serialize};
 
-/*#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct LastMessage {
+    #[serde(rename = "_id")]
     id: String,
-    user_id: String,
-    short_content: String,
+    author: String,
+    short: String,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
-pub struct Channel {
-    #[serde(rename = "_id")]
-    pub id: String,
-    #[serde(rename = "type")]
-    pub channel_type: u8,
-
-    // DM: whether the DM is active
-    pub active: Option<bool>,
-    // DM + GDM: last message in channel
-    pub last_message: Option<LastMessage>,
-    // DM + GDM: recipients for channel
-    pub recipients: Option<Vec<String>>,
-    // GDM: owner of group
-    pub owner: Option<String>,
-    // GUILD: channel parent
-    pub guild: Option<String>,
-    // GUILD + GDM: channel name
-    pub name: Option<String>,
-    // GUILD + GDM: channel description
-    pub description: Option<String>,
-}*/
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(tag = "type")]
+#[serde(tag = "channel_type")]
 pub enum Channel {
     SavedMessages {
         #[serde(rename = "_id")]
@@ -48,19 +26,49 @@ pub enum Channel {
         id: String,
         active: bool,
         recipients: Vec<String>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        last_message: Option<LastMessage>,
     },
     Group {
         #[serde(rename = "_id")]
         id: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        nonce: Option<String>,
         name: String,
         owner: String,
         description: String,
         recipients: Vec<String>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        last_message: Option<LastMessage>,
     },
 }
 
 impl Channel {
-    pub async fn save(&self) -> Result<()> {
+    pub fn id(&self) -> &str {
+        match self {
+            Channel::SavedMessages { id, .. } => id,
+            Channel::DirectMessage { id, .. } => id,
+            Channel::Group { id, .. } => id,
+        }
+    }
+
+    pub async fn get(id: &str) -> Result<Channel> {
+        let doc = get_collection("channels")
+            .find_one(doc! { "_id": id }, None)
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "find_one",
+                with: "channel",
+            })?
+            .ok_or_else(|| Error::UnknownChannel)?;
+
+        from_document::<Channel>(doc).map_err(|_| Error::DatabaseError {
+            operation: "from_document",
+            with: "channel",
+        })
+    }
+
+    pub async fn publish(self) -> Result<()> {
         get_collection("channels")
             .insert_one(
                 to_document(&self).map_err(|_| Error::DatabaseError {
@@ -75,6 +83,61 @@ impl Channel {
                 with: "channel",
             })?;
 
+        let channel_id = self.id().to_string();
+        ClientboundNotification::ChannelCreate(self)
+            .publish(channel_id)
+            .await
+            .ok();
+
+        Ok(())
+    }
+
+    pub async fn publish_update(&self, data: JsonValue) -> Result<()> {
+        let id = self.id().to_string();
+        ClientboundNotification::ChannelUpdate {
+            id: id.clone(),
+            data,
+        }
+        .publish(id)
+        .await
+        .ok();
+
+        Ok(())
+    }
+
+    pub async fn delete(&self) -> Result<()> {
+        let id = self.id();
+        get_collection("messages")
+            .delete_many(
+                doc! {
+                    "channel": id
+                },
+                None,
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "delete_many",
+                with: "messages",
+            })?;
+
+        get_collection("channels")
+            .delete_one(
+                doc! {
+                    "_id": id
+                },
+                None,
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "delete_one",
+                with: "channel",
+            })?;
+
+        ClientboundNotification::ChannelDelete { id: id.to_string() }
+            .publish(id.to_string())
+            .await
+            .ok();
+
         Ok(())
     }
 }

src/database/entities/message.rs

@@ -1,22 +1,141 @@
-// use mongodb::bson::DateTime;
-// use serde::{Deserialize, Serialize};
+use crate::{
+    database::*,
+    notifications::events::ClientboundNotification,
+    util::result::{Error, Result},
+};
+use mongodb::bson::{doc, to_bson, DateTime};
+use rocket_contrib::json::JsonValue;
+use serde::{Deserialize, Serialize};
+use ulid::Ulid;
 
-/*#[derive(Serialize, Deserialize, Debug)]
-pub struct PreviousEntry {
-    pub content: String,
-    pub time: DateTime,
-}
-
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Message {
     #[serde(rename = "_id")]
     pub id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub nonce: Option<String>,
     pub channel: String,
     pub author: String,
 
     pub content: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub edited: Option<DateTime>,
+}
 
-    pub previous_content: Vec<PreviousEntry>,
-}*/
+impl Message {
+    pub fn create(author: String, channel: String, content: String) -> Message {
+        Message {
+            id: Ulid::new().to_string(),
+            nonce: None,
+            channel,
+            author,
+
+            content,
+            edited: None,
+        }
+    }
+
+    pub async fn publish(self, channel: &Channel) -> Result<()> {
+        get_collection("messages")
+            .insert_one(to_bson(&self).unwrap().as_document().unwrap().clone(), None)
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "insert_one",
+                with: "message",
+            })?;
+
+        // ! FIXME: temp code
+        let channels = get_collection("channels");
+        match channel {
+            Channel::DirectMessage { id, .. } => {
+                channels
+                    .update_one(
+                        doc! { "_id": id },
+                        doc! {
+                            "$set": {
+                                "active": true,
+                                "last_message": {
+                                    "_id": self.id.clone(),
+                                    "author": self.author.clone(),
+                                    "short": self.content.chars().take(24).collect::<String>()
+                                }
+                            }
+                        },
+                        None,
+                    )
+                    .await
+                    .map_err(|_| Error::DatabaseError {
+                        operation: "update_one",
+                        with: "channel",
+                    })?;
+            }
+            Channel::Group { id, .. } => {
+                channels
+                    .update_one(
+                        doc! { "_id": id },
+                        doc! {
+                            "$set": {
+                                "last_message": {
+                                    "_id": self.id.clone(),
+                                    "author": self.author.clone(),
+                                    "short": self.content.chars().take(24).collect::<String>()
+                                }
+                            }
+                        },
+                        None,
+                    )
+                    .await
+                    .map_err(|_| Error::DatabaseError {
+                        operation: "update_one",
+                        with: "channel",
+                    })?;
+            }
+            _ => {}
+        }
+
+        ClientboundNotification::Message(self)
+            .publish(channel.id().to_string())
+            .await
+            .ok();
+
+        Ok(())
+    }
+
+    pub async fn publish_update(&self, data: JsonValue) -> Result<()> {
+        let channel = self.channel.clone();
+        ClientboundNotification::MessageUpdate {
+            id: self.id.clone(),
+            data,
+        }
+        .publish(channel)
+        .await
+        .ok();
+
+        Ok(())
+    }
+
+    pub async fn delete(&self) -> Result<()> {
+        get_collection("messages")
+            .delete_one(
+                doc! {
+                    "_id": &self.id
+                },
+                None,
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "delete_one",
+                with: "message",
+            })?;
+
+        let channel = self.channel.clone();
+        ClientboundNotification::MessageDelete {
+            id: self.id.clone(),
+        }
+        .publish(channel)
+        .await
+        .ok();
+
+        Ok(())
+    }
+}

src/database/entities/user.rs

@@ -1,12 +1,8 @@
-use crate::database::get_collection;
-use crate::database::guards::reference::Ref;
-use mongodb::bson::{doc, from_bson, Bson};
-use rauth::auth::Session;
-use rocket::http::Status;
-use rocket::request::{self, FromRequest, Outcome, Request};
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
+use crate::{database::permissions::user::UserPermissions, notifications::websocket::is_online};
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub enum RelationshipStatus {
     None,
     User,
@@ -31,42 +27,42 @@ pub struct User {
     pub username: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub relations: Option<Vec<Relationship>>,
-}
 
-#[rocket::async_trait]
-impl<'a, 'r> FromRequest<'a, 'r> for User {
-    type Error = rauth::util::Error;
-
-    async fn from_request(request: &'a Request<'r>) -> request::Outcome<Self, Self::Error> {
-        let session: Session = try_outcome!(request.guard::<Session>().await);
-
-        if let Ok(result) = get_collection("users")
-            .find_one(
-                doc! {
-                    "_id": &session.user_id
-                },
-                None,
-            )
-            .await
-        {
-            if let Some(doc) = result {
-                Outcome::Success(from_bson(Bson::Document(doc)).unwrap())
-            } else {
-                Outcome::Failure((Status::Forbidden, rauth::util::Error::InvalidSession))
-            }
-        } else {
-            Outcome::Failure((
-                Status::InternalServerError,
-                rauth::util::Error::DatabaseError,
-            ))
-        }
-    }
+    // ? This should never be pushed to the collection.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub relationship: Option<RelationshipStatus>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub online: Option<bool>,
 }
 
 impl User {
-    pub fn as_ref(&self) -> Ref {
-        Ref {
-            id: self.id.to_string(),
+    /// Mutate the user object to include relationship as seen by user.
+    pub fn from(mut self, user: &User) -> User {
+        if self.id == user.id {
+            self.relationship = Some(RelationshipStatus::User);
+            return self;
         }
+
+        if let Some(relations) = &user.relations {
+            if let Some(relationship) = relations.iter().find(|x| self.id == x.id) {
+                self.relationship = Some(relationship.status.clone());
+                return self;
+            }
+        }
+
+        self
+    }
+
+    /// Mutate the user object to appear as seen by user.
+    pub fn with(mut self, permissions: UserPermissions<[u32; 1]>) -> User {
+        if !permissions.get_view_all() {
+            self.relations = None;
+        }
+
+        if permissions.get_view_profile() {
+            self.online = Some(is_online(&self.id));
+        }
+
+        self
     }
 }

src/database/guards/mod.rs

@@ -1,2 +1,4 @@
 pub mod reference;
 pub mod user;
+
+pub use reference::Ref;

src/database/guards/reference.rs

@@ -1,10 +1,10 @@
-use crate::database::entities::*;
-use crate::database::get_collection;
+use crate::database::*;
 use crate::util::result::{Error, Result};
-use mongodb::bson::{doc, from_bson, Bson};
+
+use mongodb::bson::{doc, from_document};
 use rocket::http::RawStr;
 use rocket::request::FromParam;
-use serde::{Deserialize, Serialize};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use validator::Validate;
 
 #[derive(Validate, Serialize, Deserialize)]
@@ -18,8 +18,8 @@ impl Ref {
         Ok(Ref { id })
     }
 
-    pub async fn fetch_user(&self) -> Result<User> {
-        let doc = get_collection("users")
+    pub async fn fetch<T: DeserializeOwned>(&self, collection: &'static str) -> Result<T> {
+        let doc = get_collection(&collection)
             .find_one(
                 doc! {
                     "_id": &self.id
@@ -29,16 +29,39 @@ impl Ref {
             .await
             .map_err(|_| Error::DatabaseError {
                 operation: "find_one",
-                with: "user",
+                with: &collection,
             })?
             .ok_or_else(|| Error::UnknownUser)?;
 
-        Ok(
-            from_bson(Bson::Document(doc)).map_err(|_| Error::DatabaseError {
-                operation: "from_bson",
-                with: "user",
-            })?,
-        )
+        Ok(from_document::<T>(doc).map_err(|_| Error::DatabaseError {
+            operation: "from_document",
+            with: &collection,
+        })?)
+    }
+
+    pub async fn fetch_user(&self) -> Result<User> {
+        self.fetch("users").await
+    }
+
+    pub async fn fetch_channel(&self) -> Result<Channel> {
+        self.fetch("channels").await
+    }
+
+    pub async fn fetch_message(&self, channel: &Channel) -> Result<Message> {
+        let message: Message = self.fetch("messages").await?;
+        if &message.channel != channel.id() {
+            Err(Error::InvalidOperation)
+        } else {
+            Ok(message)
+        }
+    }
+}
+
+impl User {
+    pub fn as_ref(&self) -> Ref {
+        Ref {
+            id: self.id.to_string(),
+        }
     }
 }
 

src/database/guards/user.rs

@@ -1 +1,39 @@
+use crate::database::*;
 
+use mongodb::bson::{doc, from_document};
+use rauth::auth::Session;
+use rocket::http::Status;
+use rocket::request::{self, FromRequest, Outcome, Request};
+
+#[rocket::async_trait]
+impl<'a, 'r> FromRequest<'a, 'r> for User {
+    type Error = rauth::util::Error;
+
+    async fn from_request(request: &'a Request<'r>) -> request::Outcome<Self, Self::Error> {
+        let session: Session = try_outcome!(request.guard::<Session>().await);
+
+        if let Ok(result) = get_collection("users")
+            .find_one(
+                doc! {
+                    "_id": &session.user_id
+                },
+                None,
+            )
+            .await
+        {
+            if let Some(doc) = result {
+                Outcome::Success(from_document(doc).unwrap())
+            } else {
+                Outcome::Failure((Status::Forbidden, rauth::util::Error::InvalidSession))
+            }
+        } else {
+            Outcome::Failure((
+                Status::InternalServerError,
+                rauth::util::Error::DatabaseError {
+                    operation: "find_one",
+                    with: "user",
+                },
+            ))
+        }
+    }
+}

src/database/migrations/init.rs

@@ -9,6 +9,10 @@ pub async fn create_database() {
     info!("Creating database.");
     let db = get_db();
 
+    db.create_collection("accounts", None)
+        .await
+        .expect("Failed to create accounts collection.");
+
     db.create_collection("users", None)
         .await
         .expect("Failed to create users collection.");
@@ -17,14 +21,6 @@ pub async fn create_database() {
         .await
         .expect("Failed to create channels collection.");
 
-    db.create_collection("guilds", None)
-        .await
-        .expect("Failed to create guilds collection.");
-
-    db.create_collection("members", None)
-        .await
-        .expect("Failed to create members collection.");
-
     db.create_collection("messages", None)
         .await
         .expect("Failed to create messages collection.");
@@ -43,6 +39,39 @@ pub async fn create_database() {
     .await
     .expect("Failed to create pubsub collection.");
 
+    db.run_command(
+        doc! {
+            "createIndexes": "accounts",
+            "indexes": [
+                {
+                    "key": {
+                        "email": 1
+                    },
+                    "name": "email",
+                    "unique": true,
+                    "collation": {
+                        "locale": "en",
+                        "strength": 2
+                    }
+                },
+                {
+                    "key": {
+                        "email_normalised": 1
+                    },
+                    "name": "email_normalised",
+                    "unique": true,
+                    "collation": {
+                        "locale": "en",
+                        "strength": 2
+                    }
+                }
+            ]
+        },
+        None,
+    )
+    .await
+    .expect("Failed to create account index.");
+
     db.run_command(
         doc! {
             "createIndexes": "users",

src/database/migrations/scripts.rs

@@ -1,9 +1,7 @@
-use super::super::{get_collection, get_db};
+use crate::database::get_collection;
 
-use crate::rocket::futures::StreamExt;
 use log::info;
-use mongodb::bson::{doc, from_bson, Bson};
-use mongodb::options::FindOptions;
+use mongodb::bson::{doc, from_document};
 use serde::{Deserialize, Serialize};
 
 #[derive(Serialize, Deserialize)]
@@ -12,7 +10,7 @@ struct MigrationInfo {
     revision: i32,
 }
 
-pub const LATEST_REVISION: i32 = 3;
+pub const LATEST_REVISION: i32 = 0;
 
 pub async fn migrate_database() {
     let migrations = get_collection("migrations");
@@ -23,7 +21,7 @@ pub async fn migrate_database() {
 
     if let Some(doc) = data {
         let info: MigrationInfo =
-            from_bson(Bson::Document(doc)).expect("Failed to read migration information.");
+            from_document(doc).expect("Failed to read migration information.");
 
         let revision = run_migrations(info.revision).await;
 
@@ -55,99 +53,6 @@ pub async fn run_migrations(revision: i32) -> i32 {
         info!("Running migration [revision 0]: Test migration system.");
     }
 
-    if revision <= 1 {
-        info!("Running migration [revision 1]: Add channels to guild object.");
-
-        let col = get_collection("guilds");
-        let mut guilds = col
-            .find(
-                None,
-                FindOptions::builder().projection(doc! { "_id": 1 }).build(),
-            )
-            .await
-            .expect("Failed to fetch guilds.");
-
-        let mut result = get_collection("channels")
-            .find(
-                doc! {
-                    "type": 2
-                },
-                FindOptions::builder()
-                    .projection(doc! { "_id": 1, "guild": 1 })
-                    .build(),
-            )
-            .await
-            .expect("Failed to fetch channels.");
-
-        let mut channels = vec![];
-        while let Some(doc) = result.next().await {
-            let channel = doc.expect("Failed to fetch channel.");
-            let id = channel
-                .get_str("_id")
-                .expect("Failed to get channel id.")
-                .to_string();
-
-            let gid = channel
-                .get_str("guild")
-                .expect("Failed to get guild id.")
-                .to_string();
-
-            channels.push((id, gid));
-        }
-
-        while let Some(doc) = guilds.next().await {
-            let guild = doc.expect("Failed to fetch guild.");
-            let id = guild.get_str("_id").expect("Failed to get guild id.");
-
-            let list: Vec<String> = channels
-                .iter()
-                .filter(|x| x.1 == id)
-                .map(|x| x.0.clone())
-                .collect();
-
-            col.update_one(
-                doc! {
-                    "_id": id
-                },
-                doc! {
-                    "$set": {
-                        "channels": list
-                    }
-                },
-                None,
-            )
-            .await
-            .expect("Failed to update guild.");
-        }
-    }
-
-    if revision <= 2 {
-        info!("Running migration [revision 2]: Add username index to users.");
-
-        get_db()
-            .run_command(
-                doc! {
-                    "createIndexes": "users",
-                    "indexes": [
-                        {
-                            "key": {
-                                "username": 1
-                            },
-                            "name": "username",
-                            "unique": true,
-                            "collation": {
-                                "locale": "en",
-                                "strength": 2
-                            }
-                        }
-                    ]
-                },
-                None,
-            )
-            .await
-            .expect("Failed to create username index.");
-    }
-
     // Reminder to update LATEST_REVISION when adding new migrations.
     LATEST_REVISION
 }

src/database/mod.rs

@@ -30,3 +30,7 @@ pub mod entities;
 pub mod guards;
 pub mod migrations;
 pub mod permissions;
+
+pub use entities::*;
+pub use guards::*;
+pub use permissions::*;

src/database/permissions/channel.rs

@@ -0,0 +1,123 @@
+use crate::database::*;
+use crate::util::result::Result;
+
+use super::PermissionCalculator;
+
+use num_enum::TryFromPrimitive;
+use std::ops;
+
+#[derive(Debug, PartialEq, Eq, TryFromPrimitive, Copy, Clone)]
+#[repr(u32)]
+pub enum ChannelPermission {
+    View = 1,
+    SendMessage = 2,
+    ManageMessages = 4,
+}
+
+bitfield! {
+    pub struct ChannelPermissions(MSB0 [u32]);
+    u32;
+    pub get_view, _: 31;
+    pub get_send_message, _: 30;
+    pub get_manage_messages, _: 29;
+}
+
+impl_op_ex!(+ |a: &ChannelPermission, b: &ChannelPermission| -> u32 { *a as u32 | *b as u32 });
+impl_op_ex_commutative!(+ |a: &u32, b: &ChannelPermission| -> u32 { *a | *b as u32 });
+
+/*pub async fn calculate(user: &User, target: &Channel) -> Result<u32> {
+    match target {
+        Channel::SavedMessages { user: owner, .. } => {
+            if &user.id == owner {
+                Ok(ChannelPermission::View
+                    + ChannelPermission::SendMessage
+                    + ChannelPermission::ManageMessages)
+            } else {
+                Ok(0)
+            }
+        }
+        Channel::DirectMessage { recipients, .. } => {
+            if recipients.iter().find(|x| *x == &user.id).is_some() {
+                if let Some(recipient) = recipients.iter().find(|x| *x != &user.id) {
+                    let perms = super::user::get(&user, recipient).await?;
+
+                    if perms.get_send_message() {
+                        return Ok(ChannelPermission::View + ChannelPermission::SendMessage);
+                    }
+
+                    return Ok(ChannelPermission::View as u32);
+                }
+            }
+
+            Ok(0)
+        }
+        Channel::Group { recipients, .. } => {
+            if recipients.iter().find(|x| *x == &user.id).is_some() {
+                Ok(ChannelPermission::View + ChannelPermission::SendMessage)
+            } else {
+                Ok(0)
+            }
+        }
+    }
+}
+
+pub async fn get(user: &User, target: &Channel) -> Result<ChannelPermissions<[u32; 1]>> {
+    Ok(ChannelPermissions([calculate(&user, &target).await?]))
+}*/
+
+impl<'a> PermissionCalculator<'a> {
+    pub async fn calculate_channel(self) -> Result<u32> {
+        let channel = if let Some(channel) = self.channel {
+            channel
+        } else {
+            unreachable!()
+        };
+
+        match channel {
+            Channel::SavedMessages { user: owner, .. } => {
+                if &self.perspective.id == owner {
+                    Ok(ChannelPermission::View
+                        + ChannelPermission::SendMessage
+                        + ChannelPermission::ManageMessages)
+                } else {
+                    Ok(0)
+                }
+            }
+            Channel::DirectMessage { recipients, .. } => {
+                if recipients
+                    .iter()
+                    .find(|x| *x == &self.perspective.id)
+                    .is_some()
+                {
+                    if let Some(recipient) = recipients.iter().find(|x| *x != &self.perspective.id)
+                    {
+                        let perms = self.for_user(recipient).await?;
+
+                        if perms.get_send_message() {
+                            return Ok(ChannelPermission::View + ChannelPermission::SendMessage);
+                        }
+
+                        return Ok(ChannelPermission::View as u32);
+                    }
+                }
+
+                Ok(0)
+            }
+            Channel::Group { recipients, .. } => {
+                if recipients
+                    .iter()
+                    .find(|x| *x == &self.perspective.id)
+                    .is_some()
+                {
+                    Ok(ChannelPermission::View + ChannelPermission::SendMessage)
+                } else {
+                    Ok(0)
+                }
+            }
+        }
+    }
+
+    pub async fn for_channel(self) -> Result<ChannelPermissions<[u32; 1]>> {
+        Ok(ChannelPermissions([self.calculate_channel().await?]))
+    }
+}

src/database/permissions/mod.rs

@@ -1,49 +1,49 @@
-use crate::database::entities::User;
-use num_enum::TryFromPrimitive;
-use std::ops;
+pub use crate::database::*;
 
-#[derive(Debug, PartialEq, Eq, TryFromPrimitive, Copy, Clone)]
-#[repr(u32)]
-pub enum UserPermission {
-    Access = 1,
-    SendMessage = 2,
-    Invite = 4,
+pub mod channel;
+pub mod user;
+
+pub use user::get_relationship;
+
+pub struct PermissionCalculator<'a> {
+    perspective: &'a User,
+
+    user: Option<&'a User>,
+    channel: Option<&'a Channel>,
+
+    has_mutual_conncetion: bool,
 }
 
-bitfield! {
-    pub struct UserPermissions(MSB0 [u32]);
-    u32;
-    pub get_access, _: 31;
-    pub get_send_message, _: 30;
-    pub get_invite, _: 29;
-}
+impl<'a> PermissionCalculator<'a> {
+    pub fn new(perspective: &'a User) -> PermissionCalculator {
+        PermissionCalculator {
+            perspective,
 
-impl_op_ex!(+ |a: &UserPermission, b: &UserPermission| -> u32 { *a as u32 | *b as u32 });
-impl_op_ex_commutative!(+ |a: &u32, b: &UserPermission| -> u32 { *a | *b as u32 });
+            user: None,
+            channel: None,
 
-pub async fn temp_calc_perm(_user: &User, _target: &User) -> UserPermissions<[u32; 1]> {
-    // if friends; Access + Message + Invite
-    // if mutually know each other:
-    //    and has DMs from users enabled -> Access + Message
-    //    otherwise -> Access
-    // otherwise; None
-
-    UserPermissions([UserPermission::Access + UserPermission::SendMessage + UserPermission::Invite])
-}
-
-use crate::database::entities::RelationshipStatus;
-use crate::database::guards::reference::Ref;
-
-pub fn get_relationship(a: &User, b: &Ref) -> RelationshipStatus {
-    if a.id == b.id {
-        return RelationshipStatus::Friend;
-    }
-
-    if let Some(relations) = &a.relations {
-        if let Some(relationship) = relations.iter().find(|x| x.id == b.id) {
-            return relationship.status.clone();
+            has_mutual_conncetion: false,
         }
     }
 
-    RelationshipStatus::None
+    pub fn with_user(self, user: &'a User) -> PermissionCalculator {
+        PermissionCalculator {
+            user: Some(&user),
+            ..self
+        }
+    }
+
+    pub fn with_channel(self, channel: &'a Channel) -> PermissionCalculator {
+        PermissionCalculator {
+            channel: Some(&channel),
+            ..self
+        }
+    }
+
+    pub fn with_mutual_connection(self) -> PermissionCalculator<'a> {
+        PermissionCalculator {
+            has_mutual_conncetion: true,
+            ..self
+        }
+    }
 }

src/database/permissions/user.rs

@@ -0,0 +1,106 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use super::PermissionCalculator;
+
+use mongodb::bson::doc;
+use num_enum::TryFromPrimitive;
+use std::ops;
+
+#[derive(Debug, PartialEq, Eq, TryFromPrimitive, Copy, Clone)]
+#[repr(u32)]
+pub enum UserPermission {
+    Access = 1,
+    ViewProfile = 2,
+    SendMessage = 4,
+    Invite = 8,
+
+    ViewAll = 2147483648,
+}
+
+bitfield! {
+    pub struct UserPermissions(MSB0 [u32]);
+    u32;
+    pub get_access, _: 31;
+    pub get_view_profile, _: 30;
+    pub get_send_message, _: 29;
+    pub get_invite, _: 28;
+
+    pub get_view_all, _: 0;
+}
+
+impl_op_ex!(+ |a: &UserPermission, b: &UserPermission| -> u32 { *a as u32 | *b as u32 });
+impl_op_ex!(-|a: &UserPermission, b: &UserPermission| -> u32 { *a as u32 & !(*b as u32) });
+impl_op_ex!(-|a: &u32, b: &UserPermission| -> u32 { *a & !(*b as u32) });
+impl_op_ex_commutative!(+ |a: &u32, b: &UserPermission| -> u32 { *a | *b as u32 });
+
+pub fn get_relationship(a: &User, b: &str) -> RelationshipStatus {
+    if a.id == b {
+        return RelationshipStatus::Friend;
+    }
+
+    if let Some(relations) = &a.relations {
+        if let Some(relationship) = relations.iter().find(|x| x.id == b) {
+            return relationship.status.clone();
+        }
+    }
+
+    RelationshipStatus::None
+}
+
+impl<'a> PermissionCalculator<'a> {
+    pub async fn calculate_user(self, target: &str) -> Result<u32> {
+        if &self.perspective.id == target {
+            return Ok(u32::MAX);
+        }
+
+        let mut permissions: u32 = 0;
+        match get_relationship(&self.perspective, &target) {
+            RelationshipStatus::Friend => return Ok(u32::MAX - UserPermission::ViewAll),
+            RelationshipStatus::Blocked | RelationshipStatus::BlockedOther => {
+                return Ok(UserPermission::Access as u32)
+            }
+            RelationshipStatus::Incoming | RelationshipStatus::Outgoing => {
+                permissions = UserPermission::Access as u32;
+                // ! INFO: if we add boolean switch for permission to
+                // ! message people who have mutual, we need to get
+                // ! rid of this return statement.
+                return Ok(permissions);
+            }
+            _ => {}
+        }
+
+        if self.has_mutual_conncetion
+            || get_collection("channels")
+                .find_one(
+                    doc! {
+                        "type": "Group",
+                        "$and": {
+                            "recipients": &self.perspective.id,
+                            "recipients": target
+                        }
+                    },
+                    None,
+                )
+                .await
+                .map_err(|_| Error::DatabaseError {
+                    operation: "find",
+                    with: "channels",
+                })?
+                .is_some()
+        {
+            return Ok(UserPermission::Access as u32);
+        }
+
+        Ok(permissions)
+    }
+
+    pub async fn for_user(self, target: &str) -> Result<UserPermissions<[u32; 1]>> {
+        Ok(UserPermissions([self.calculate_user(&target).await?]))
+    }
+
+    pub async fn for_user_given(self) -> Result<UserPermissions<[u32; 1]>> {
+        let id = &self.user.unwrap().id;
+        Ok(UserPermissions([self.calculate_user(&id).await?]))
+    }
+}

src/main.rs

@@ -18,17 +18,16 @@ pub mod notifications;
 pub mod routes;
 pub mod util;
 
-use async_std::task;
+use chrono::Duration;
 use futures::join;
 use log::info;
-use rauth;
+use rauth::auth::Auth;
+use rauth::options::{EmailVerification, Options, SMTP};
 use rocket_cors::AllowedOrigins;
+use util::variables::{PUBLIC_URL, SMTP_FROM, SMTP_HOST, SMTP_PASSWORD, SMTP_USERNAME, USE_EMAIL};
 
-fn main() {
-    task::block_on(entry())
-}
-
-async fn entry() {
+#[async_std::main]
+async fn main() {
     dotenv::dotenv().ok();
     env_logger::init_from_env(env_logger::Env::default().filter_or("RUST_LOG", "info"));
 
@@ -59,10 +58,32 @@ async fn launch_web() {
     .to_cors()
     .expect("Failed to create CORS.");
 
-    let auth = rauth::auth::Auth::new(database::get_collection("accounts"));
+    let auth = Auth::new(
+        database::get_collection("accounts"),
+        Options::new()
+            .base_url(format!("{}/auth", *PUBLIC_URL))
+            .email_verification(if *USE_EMAIL {
+                EmailVerification::Enabled {
+                    success_redirect_uri: format!("{}/welcome", *PUBLIC_URL),
+                    verification_expiry: Duration::days(1),
+                    verification_ratelimit: Duration::minutes(1),
 
-    routes::mount(rauth::routes::mount(rocket::ignite(), "/auth", auth))
+                    smtp: SMTP {
+                        from: (*SMTP_FROM).to_string(),
+                        host: (*SMTP_HOST).to_string(),
+                        username: (*SMTP_USERNAME).to_string(),
+                        password: (*SMTP_PASSWORD).to_string(),
+                    },
+                }
+            } else {
+                EmailVerification::Disabled
+            }),
+    );
+
+    routes::mount(rocket::ignite())
         .mount("/", rocket_cors::catch_all_options_routes())
+        .mount("/auth", rauth::routes::routes())
+        .manage(auth)
         .manage(cors.clone())
         .attach(cors)
         .launch()

src/notifications/events.rs

@@ -1,10 +1,11 @@
+use hive_pubsub::PubSub;
 use rauth::auth::Session;
+use rocket_contrib::json::JsonValue;
 use serde::{Deserialize, Serialize};
 use snafu::Snafu;
 
-use crate::database::entities::{RelationshipStatus, User};
-
-use super::hive::get_hive;
+use super::hive::{get_hive, subscribe_if_exists};
+use crate::database::*;
 
 #[derive(Serialize, Deserialize, Debug, Snafu)]
 #[serde(tag = "error")]
@@ -12,7 +13,7 @@ pub enum WebSocketError {
     #[snafu(display("This error has not been labelled."))]
     LabelMe,
     #[snafu(display("Internal server error."))]
-    InternalError,
+    InternalError { at: String },
     #[snafu(display("Invalid session."))]
     InvalidSession,
     #[snafu(display("User hasn't completed onboarding."))]
@@ -33,73 +34,100 @@ pub enum ClientboundNotification {
     Error(WebSocketError),
     Authenticated,
     Ready {
-        user: User
+        users: Vec<User>,
+        channels: Vec<Channel>,
     },
 
-    /*MessageCreate {
+    Message(Message),
+    MessageUpdate {
         id: String,
-        nonce: Option<String>,
-        channel: String,
-        author: String,
-        content: String,
+        data: JsonValue,
     },
-
-    MessageEdit {
-        id: String,
-        channel: String,
-        author: String,
-        content: String,
-    },
-
     MessageDelete {
         id: String,
     },
 
-    GroupUserJoin {
+    ChannelCreate(Channel),
+    ChannelUpdate {
+        id: String,
+        data: JsonValue,
+    },
+    ChannelGroupJoin {
         id: String,
         user: String,
     },
-
-    GroupUserLeave {
+    ChannelGroupLeave {
         id: String,
         user: String,
     },
-
-    GuildUserJoin {
+    ChannelDelete {
         id: String,
-        user: String,
     },
 
-    GuildUserLeave {
-        id: String,
-        user: String,
-        banned: bool,
-    },
-
-    GuildChannelCreate {
-        id: String,
-        channel: String,
-        name: String,
-        description: String,
-    },
-
-    GuildChannelDelete {
-        id: String,
-        channel: String,
-    },
-
-    GuildDelete {
-        id: String,
-    },*/
     UserRelationship {
         id: String,
         user: String,
         status: RelationshipStatus,
     },
+    UserPresence {
+        id: String,
+        online: bool,
+    },
 }
 
 impl ClientboundNotification {
     pub async fn publish(self, topic: String) -> Result<(), String> {
+        prehandle_hook(&self); // ! TODO: this should be moved to pubsub
         hive_pubsub::backend::mongo::publish(get_hive(), &topic, self).await
     }
 }
+
+pub fn prehandle_hook(notification: &ClientboundNotification) {
+    match &notification {
+        ClientboundNotification::ChannelGroupJoin { id, user } => {
+            subscribe_if_exists(user.clone(), id.clone()).ok();
+        }
+        ClientboundNotification::ChannelCreate(channel) => {
+            let channel_id = channel.id();
+            match &channel {
+                Channel::SavedMessages { user, .. } => {
+                    subscribe_if_exists(user.clone(), channel_id.to_string()).ok();
+                }
+                Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
+                    for recipient in recipients {
+                        subscribe_if_exists(recipient.clone(), channel_id.to_string()).ok();
+                    }
+                }
+            }
+        }
+        ClientboundNotification::UserRelationship { id, user, status } => {
+            if status != &RelationshipStatus::None {
+                subscribe_if_exists(id.clone(), user.clone()).ok();
+            }
+        }
+        _ => {}
+    }
+}
+
+pub fn posthandle_hook(notification: &ClientboundNotification) {
+    match &notification {
+        ClientboundNotification::ChannelDelete { id } => {
+            get_hive().hive.drop_topic(&id).ok();
+        }
+        ClientboundNotification::UserRelationship { id, user, status } => {
+            if status == &RelationshipStatus::None {
+                get_hive()
+                    .hive
+                    .unsubscribe(&id.to_string(), &user.to_string())
+                    .ok();
+            }
+        }
+        ClientboundNotification::ChannelGroupLeave { id, user } => {
+            get_hive()
+                .hive
+                .unsubscribe(&user.to_string(), &id.to_string())
+                .ok();
+        }
+        _ => {}
+    }
+}

src/notifications/hive.rs

@@ -1,5 +1,5 @@
 use super::{events::ClientboundNotification, websocket};
-use crate::database::get_collection;
+use crate::database::*;
 
 use futures::FutureExt;
 use hive_pubsub::backend::mongo::MongodbPubSub;
@@ -14,6 +14,8 @@ static HIVE: OnceCell<Hive> = OnceCell::new();
 pub async fn init_hive() {
     let hive = MongodbPubSub::new(
         |ids, notification| {
+            super::events::posthandle_hook(&notification);
+
             if let Ok(data) = to_string(&notification) {
                 debug!("Pushing out notification. {}", data);
                 websocket::publish(ids, notification);
@@ -36,8 +38,6 @@ pub async fn listen() {
         .fuse()
         .await
         .expect("Hive hit an error");
-
-    dbg!("a");
 }
 
 pub fn subscribe_multiple(user: String, topics: Vec<String>) -> Result<(), String> {

src/notifications/mod.rs

@@ -1,4 +1,5 @@
 pub mod events;
 pub mod hive;
+pub mod payload;
 pub mod subscriptions;
 pub mod websocket;

src/notifications/payload.rs

@@ -0,0 +1,114 @@
+use std::collections::HashSet;
+
+use crate::{database::*, notifications::events::ClientboundNotification};
+use crate::{
+    database::{entities::User, get_collection},
+    util::result::{Error, Result},
+};
+use futures::StreamExt;
+use mongodb::{
+    bson::{doc, from_document},
+    options::FindOptions,
+};
+
+pub async fn generate_ready(mut user: User) -> Result<ClientboundNotification> {
+    let mut users = vec![];
+    let mut user_ids: HashSet<String> = HashSet::new();
+
+    if let Some(relationships) = &user.relations {
+        user_ids.extend(
+            relationships
+                .iter()
+                .map(|relationship| relationship.id.clone()),
+        );
+    }
+
+    let mut cursor = get_collection("channels")
+        .find(
+            doc! {
+                "$or": [
+                    {
+                        "channel_type": "SavedMessages",
+                        "user": &user.id
+                    },
+                    {
+                        "channel_type": "DirectMessage",
+                        "recipients": &user.id
+                    },
+                    {
+                        "channel_type": "Group",
+                        "recipients": &user.id
+                    }
+                ]
+            },
+            None,
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find",
+            with: "channels",
+        })?;
+
+    let mut channels = vec![];
+    while let Some(result) = cursor.next().await {
+        if let Ok(doc) = result {
+            let channel = from_document(doc).map_err(|_| Error::DatabaseError {
+                operation: "from_document",
+                with: "channel",
+            })?;
+
+            if let Channel::Group { recipients, .. } = &channel {
+                user_ids.extend(recipients.iter().cloned());
+            }
+
+            channels.push(channel);
+        }
+    }
+
+    user_ids.remove(&user.id);
+    if user_ids.len() > 0 {
+        let mut cursor = get_collection("users")
+            .find(
+                doc! {
+                    "_id": {
+                        "$in": user_ids.into_iter().collect::<Vec<String>>()
+                    }
+                },
+                FindOptions::builder()
+                    .projection(doc! { "_id": 1, "username": 1 })
+                    .build(),
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "find",
+                with: "users",
+            })?;
+
+        while let Some(result) = cursor.next().await {
+            if let Ok(doc) = result {
+                let other: User = from_document(doc).map_err(|_| Error::DatabaseError {
+                    operation: "from_document",
+                    with: "user",
+                })?;
+
+                let permissions = PermissionCalculator::new(&user)
+                    .with_mutual_connection()
+                    .with_user(&other)
+                    .for_user_given()
+                    .await?;
+
+                users.push(
+                    other
+                        .from(&user)
+                        .with(permissions)
+                );
+            }
+        }
+    }
+
+    user.relationship = Some(RelationshipStatus::User);
+    user.online = Some(true);
+    users.push(user);
+
+    Ok(ClientboundNotification::Ready { users, channels })
+}

src/notifications/subscriptions.rs

@@ -1,7 +1,10 @@
-use crate::database::entities::User;
+use crate::database::*;
 
 use super::hive::get_hive;
+use futures::StreamExt;
 use hive_pubsub::PubSub;
+use mongodb::bson::doc;
+use mongodb::options::FindOptions;
 
 pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
     let hive = get_hive();
@@ -13,5 +16,34 @@ pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
         }
     }
 
+    let mut cursor = get_collection("channels")
+        .find(
+            doc! {
+                "$or": [
+                    {
+                        "channel_type": "SavedMessages",
+                        "user": &user.id
+                    },
+                    {
+                        "channel_type": "DirectMessage",
+                        "recipients": &user.id
+                    },
+                    {
+                        "channel_type": "Group",
+                        "recipients": &user.id
+                    }
+                ]
+            },
+            FindOptions::builder().projection(doc! { "_id": 1 }).build(),
+        )
+        .await
+        .map_err(|_| "Failed to fetch channels.".to_string())?;
+
+    while let Some(result) = cursor.next().await {
+        if let Ok(doc) = result {
+            hive.subscribe(user.id.clone(), doc.get_str("_id").unwrap().to_string())?;
+        }
+    }
+
     Ok(())
 }

src/notifications/websocket.rs

@@ -1,5 +1,4 @@
-use crate::database::get_collection;
-use crate::database::guards::reference::Ref;
+use crate::database::*;
 use crate::util::variables::WS_HOST;
 
 use super::subscriptions;
@@ -13,7 +12,10 @@ use futures::{pin_mut, prelude::*};
 use hive_pubsub::PubSub;
 use log::{debug, info};
 use many_to_many::ManyToMany;
-use rauth::auth::{Auth, Session};
+use rauth::{
+    auth::{Auth, Session},
+    options::Options,
+};
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::{Arc, Mutex, RwLock};
@@ -64,69 +66,106 @@ async fn accept(stream: TcpStream) {
         }
     };
 
-    let mut session: Option<Session> = None;
+    let session: Arc<Mutex<Option<Session>>> = Arc::new(Mutex::new(None));
+    let mutex_generator = || session.clone();
     let fwd = rx.map(Ok).forward(write);
-    let incoming = read.try_for_each(|msg| {
+    let incoming = read.try_for_each(async move |msg| {
+        let mutex = mutex_generator();
+        //dbg!(&mutex.lock().unwrap());
+
         if let Message::Text(text) = msg {
             if let Ok(notification) = serde_json::from_str::<ServerboundNotification>(&text) {
                 match notification {
                     ServerboundNotification::Authenticate(new_session) => {
-                        if session.is_some() {
-                            send(ClientboundNotification::Error(
-                                WebSocketError::AlreadyAuthenticated,
-                            ));
-                            return future::ok(());
+                        {
+                            if mutex.lock().unwrap().is_some() {
+                                send(ClientboundNotification::Error(
+                                    WebSocketError::AlreadyAuthenticated,
+                                ));
+
+                                return Ok(());
+                            }
                         }
 
-                        match task::block_on(
-                            Auth::new(get_collection("accounts")).verify_session(new_session),
-                        ) {
-                            Ok(validated_session) => {
-                                match task::block_on(
-                                    Ref {
-                                        id: validated_session.user_id.clone(),
-                                    }
-                                    .fetch_user(),
-                                ) {
-                                    Ok(user) => {
-                                        if let Ok(mut map) = USERS.write() {
-                                            map.insert(validated_session.user_id.clone(), addr);
-                                            session = Some(validated_session);
-                                            if let Ok(_) = task::block_on(
-                                                subscriptions::generate_subscriptions(&user),
-                                            ) {
-                                                send(ClientboundNotification::Authenticated);
-                                                send(ClientboundNotification::Ready { user });
-                                            } else {
-                                                send(ClientboundNotification::Error(
-                                                    WebSocketError::InternalError,
-                                                ));
-                                            }
-                                        } else {
+                        if let Ok(validated_session) =
+                            Auth::new(get_collection("accounts"), Options::new())
+                                .verify_session(new_session)
+                                .await
+                        {
+                            let id = validated_session.user_id.clone();
+                            if let Ok(user) = (Ref { id: id.clone() }).fetch_user().await {
+                                let was_online = is_online(&id);
+                                {
+                                    match USERS.write() {
+                                        Ok(mut map) => {
+                                            map.insert(id.clone(), addr);
+                                        }
+                                        Err(_) => {
                                             send(ClientboundNotification::Error(
-                                                WebSocketError::InternalError,
+                                                WebSocketError::InternalError {
+                                                    at: "Writing users map.".to_string(),
+                                                },
                                             ));
+
+                                            return Ok(());
+                                        }
+                                    }
+                                }
+
+                                *mutex.lock().unwrap() = Some(validated_session);
+
+                                if let Err(_) = subscriptions::generate_subscriptions(&user).await {
+                                    send(ClientboundNotification::Error(
+                                        WebSocketError::InternalError {
+                                            at: "Generating subscriptions.".to_string(),
+                                        },
+                                    ));
+
+                                    return Ok(());
+                                }
+
+                                send(ClientboundNotification::Authenticated);
+
+                                match super::payload::generate_ready(user).await {
+                                    Ok(payload) => {
+                                        send(payload);
+
+                                        if !was_online {
+                                            ClientboundNotification::UserPresence {
+                                                id: id.clone(),
+                                                online: true,
+                                            }
+                                            .publish(id)
+                                            .await
+                                            .ok();
                                         }
                                     }
                                     Err(_) => {
                                         send(ClientboundNotification::Error(
-                                            WebSocketError::OnboardingNotFinished,
+                                            WebSocketError::InternalError {
+                                                at: "Generating payload.".to_string(),
+                                            },
                                         ));
+
+                                        return Ok(());
                                     }
                                 }
-                            }
-                            Err(_) => {
+                            } else {
                                 send(ClientboundNotification::Error(
-                                    WebSocketError::InvalidSession,
+                                    WebSocketError::OnboardingNotFinished,
                                 ));
                             }
+                        } else {
+                            send(ClientboundNotification::Error(
+                                WebSocketError::InvalidSession,
+                            ));
                         }
                     }
                 }
             }
         }
 
-        future::ok(())
+        Ok(())
     });
 
     pin_mut!(fwd, incoming);
@@ -135,7 +174,8 @@ async fn accept(stream: TcpStream) {
     info!("User {} disconnected.", &addr);
     CONNECTIONS.lock().unwrap().remove(&addr);
 
-    if let Some(session) = session {
+    let session = session.lock().unwrap();
+    if let Some(session) = session.as_ref() {
         let mut users = USERS.write().unwrap();
         users.remove(&session.user_id, &addr);
         if users.get_left(&session.user_id).is_none() {
@@ -173,3 +213,7 @@ pub fn publish(ids: Vec<String>, notification: ClientboundNotification) {
         }
     }
 }
+
+pub fn is_online(user: &String) -> bool {
+    USERS.read().unwrap().get_left(&user).is_some()
+}

src/routes/channels/delete_channel.rs

@@ -0,0 +1,114 @@
+use crate::util::result::{Error, Result};
+use crate::{database::*, notifications::events::ClientboundNotification};
+
+use mongodb::bson::doc;
+
+#[delete("/<target>")]
+pub async fn req(user: User, target: Ref) -> Result<()> {
+    let target = target.fetch_channel().await?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    match &target {
+        Channel::SavedMessages { .. } => Err(Error::NoEffect),
+        Channel::DirectMessage { .. } => {
+            get_collection("channels")
+                .update_one(
+                    doc! {
+                        "_id": target.id()
+                    },
+                    doc! {
+                        "$set": {
+                            "active": false
+                        }
+                    },
+                    None,
+                )
+                .await
+                .map_err(|_| Error::DatabaseError {
+                    operation: "update_one",
+                    with: "channel",
+                })?;
+
+            Ok(())
+        }
+        Channel::Group {
+            id,
+            owner,
+            recipients,
+            ..
+        } => {
+            if &user.id == owner {
+                if let Some(new_owner) = recipients.iter().find(|x| *x != &user.id) {
+                    get_collection("channels")
+                        .update_one(
+                            doc! {
+                                "_id": &id
+                            },
+                            doc! {
+                                "$set": {
+                                    "owner": new_owner
+                                },
+                                "$pull": {
+                                    "recipients": &user.id
+                                }
+                            },
+                            None,
+                        )
+                        .await
+                        .map_err(|_| Error::DatabaseError {
+                            operation: "update_one",
+                            with: "channel",
+                        })?;
+
+                    target.publish_update(json!({ "owner": new_owner })).await?;
+                } else {
+                    return target.delete().await;
+                }
+            } else {
+                get_collection("channels")
+                    .update_one(
+                        doc! {
+                            "_id": &id
+                        },
+                        doc! {
+                            "$pull": {
+                                "recipients": &user.id
+                            }
+                        },
+                        None,
+                    )
+                    .await
+                    .map_err(|_| Error::DatabaseError {
+                        operation: "update_one",
+                        with: "channel",
+                    })?;
+            }
+
+            ClientboundNotification::ChannelGroupLeave {
+                id: id.clone(),
+                user: user.id.clone(),
+            }
+            .publish(id.clone())
+            .await
+            .ok();
+
+            Message::create(
+                "00000000000000000000000000".to_string(),
+                id.clone(),
+                format!("<@{}> left the group.", user.id),
+            )
+            .publish(&target)
+            .await
+            .ok();
+
+            Ok(())
+        }
+    }
+}

src/routes/channels/fetch_channel.rs

@@ -0,0 +1,19 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use rocket_contrib::json::JsonValue;
+
+#[get("/<target>")]
+pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
+    let target = target.fetch_channel().await?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    Ok(json!(target))
+}

src/routes/channels/group_add_member.rs

@@ -0,0 +1,72 @@
+use crate::util::result::{Error, Result};
+use crate::util::variables::MAX_GROUP_SIZE;
+use crate::{database::*, notifications::events::ClientboundNotification};
+
+use mongodb::bson::doc;
+
+#[put("/<target>/recipients/<member>")]
+pub async fn req(user: User, target: Ref, member: Ref) -> Result<()> {
+    if get_relationship(&user, &member.id) != RelationshipStatus::Friend {
+        Err(Error::NotFriends)?
+    }
+
+    let channel = target.fetch_channel().await?;
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    if let Channel::Group { id, recipients, .. } = &channel {
+        if recipients.len() >= *MAX_GROUP_SIZE {
+            Err(Error::GroupTooLarge {
+                max: *MAX_GROUP_SIZE,
+            })?
+        }
+
+        if recipients.iter().find(|x| *x == &member.id).is_some() {
+            Err(Error::AlreadyInGroup)?
+        }
+
+        get_collection("channels")
+            .update_one(
+                doc! {
+                    "_id": &id
+                },
+                doc! {
+                    "$push": {
+                        "recipients": &member.id
+                    }
+                },
+                None,
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "update_one",
+                with: "channel",
+            })?;
+
+        ClientboundNotification::ChannelGroupJoin {
+            id: id.clone(),
+            user: member.id.clone(),
+        }
+        .publish(id.clone())
+        .await
+        .ok();
+
+        Message::create(
+            "00000000000000000000000000".to_string(),
+            id.clone(),
+            format!("<@{}> added <@{}> to the group.", user.id, member.id),
+        )
+        .publish(&channel)
+        .await
+        .ok();
+
+        Ok(())
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/group_create.rs

@@ -0,0 +1,79 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+use crate::util::variables::MAX_GROUP_SIZE;
+
+use mongodb::bson::doc;
+use rocket_contrib::json::{Json, JsonValue};
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+use std::iter::FromIterator;
+use ulid::Ulid;
+use validator::Validate;
+
+#[derive(Validate, Serialize, Deserialize)]
+pub struct Data {
+    #[validate(length(min = 1, max = 32))]
+    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>,
+}
+
+#[post("/create", data = "<info>")]
+pub async fn req(user: User, info: Json<Data>) -> Result<JsonValue> {
+    info.validate()
+        .map_err(|error| Error::FailedValidation { error })?;
+
+    let mut set: HashSet<String> = HashSet::from_iter(info.users.iter().cloned());
+    set.insert(user.id.clone());
+
+    if set.len() > *MAX_GROUP_SIZE {
+        Err(Error::GroupTooLarge {
+            max: *MAX_GROUP_SIZE,
+        })?
+    }
+
+    for target in &set {
+        if get_relationship(&user, target) != RelationshipStatus::Friend {
+            Err(Error::NotFriends)?
+        }
+    }
+
+    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.clone()),
+        name: info.name.clone(),
+        description: info
+            .description
+            .clone()
+            .unwrap_or_else(|| "A group.".to_string()),
+        owner: user.id,
+        recipients: set.into_iter().collect::<Vec<String>>(),
+        last_message: None,
+    };
+
+    channel.clone().publish().await?;
+
+    Ok(json!(channel))
+}

src/routes/channels/group_remove_member.rs

@@ -0,0 +1,69 @@
+use crate::util::result::{Error, Result};
+use crate::{database::*, notifications::events::ClientboundNotification};
+
+use mongodb::bson::doc;
+
+#[delete("/<target>/recipients/<member>")]
+pub async fn req(user: User, target: Ref, member: Ref) -> Result<()> {
+    if &user.id == &member.id {
+        Err(Error::CannotRemoveYourself)?
+    }
+
+    let channel = target.fetch_channel().await?;
+
+    if let Channel::Group {
+        id,
+        owner,
+        recipients,
+        ..
+    } = &channel
+    {
+        if &user.id != owner {
+            // figure out if we want to use perm system here
+            Err(Error::LabelMe)?
+        }
+
+        if recipients.iter().find(|x| *x == &member.id).is_none() {
+            Err(Error::NotInGroup)?
+        }
+
+        get_collection("channels")
+            .update_one(
+                doc! {
+                    "_id": &id
+                },
+                doc! {
+                    "$pull": {
+                        "recipients": &member.id
+                    }
+                },
+                None,
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "update_one",
+                with: "channel",
+            })?;
+
+        ClientboundNotification::ChannelGroupLeave {
+            id: id.clone(),
+            user: member.id.clone(),
+        }
+        .publish(id.clone())
+        .await
+        .ok();
+
+        Message::create(
+            "00000000000000000000000000".to_string(),
+            id.clone(),
+            format!("<@{}> removed <@{}> from the group.", user.id, member.id),
+        )
+        .publish(&channel)
+        .await
+        .ok();
+
+        Ok(())
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/message_delete.rs

@@ -0,0 +1,27 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use mongodb::bson::doc;
+
+#[delete("/<target>/messages/<msg>")]
+pub async fn req(user: User, target: Ref, msg: Ref) -> Result<()> {
+    let channel = target.fetch_channel().await?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let message = msg.fetch_message(&channel).await?;
+    if message.author != user.id && !perm.get_manage_messages() {
+        match channel {
+            Channel::SavedMessages { .. } => unreachable!(),
+            _ => Err(Error::CannotEditMessage)?,
+        }
+    }
+
+    message.delete().await
+}

src/routes/channels/message_edit.rs

@@ -0,0 +1,58 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use chrono::Utc;
+use mongodb::bson::{doc, Bson, DateTime};
+use rocket_contrib::json::Json;
+use serde::{Deserialize, Serialize};
+use validator::Validate;
+
+#[derive(Validate, Serialize, Deserialize)]
+pub struct Data {
+    #[validate(length(min = 1, max = 2000))]
+    content: String,
+}
+
+#[patch("/<target>/messages/<msg>", data = "<edit>")]
+pub async fn req(user: User, target: Ref, msg: Ref, edit: Json<Data>) -> Result<()> {
+    edit.validate()
+        .map_err(|error| Error::FailedValidation { error })?;
+
+    let channel = target.fetch_channel().await?;
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let message = msg.fetch_message(&channel).await?;
+    if message.author != user.id {
+        Err(Error::CannotEditMessage)?
+    }
+
+    let edited = Utc::now();
+    get_collection("messages")
+        .update_one(
+            doc! {
+                "_id": &message.id
+            },
+            doc! {
+                "$set": {
+                    "content": &edit.content,
+                    "edited": Bson::DateTime(edited)
+                }
+            },
+            None,
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "update_one",
+            with: "message",
+        })?;
+
+    message
+        .publish_update(json!({ "content": edit.content, "edited": DateTime(edited) }))
+        .await
+}

src/routes/channels/message_fetch.rs

@@ -0,0 +1,20 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use rocket_contrib::json::JsonValue;
+
+#[get("/<target>/messages/<msg>")]
+pub async fn req(user: User, target: Ref, msg: Ref) -> Result<JsonValue> {
+    let channel = target.fetch_channel().await?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let message = msg.fetch_message(&channel).await?;
+    Ok(json!(message))
+}

src/routes/channels/message_query.rs

@@ -0,0 +1,79 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use futures::StreamExt;
+use mongodb::{
+    bson::{doc, from_document},
+    options::FindOptions,
+};
+use rocket::request::Form;
+use rocket_contrib::json::JsonValue;
+use serde::{Deserialize, Serialize};
+use validator::Validate;
+
+#[derive(Validate, Serialize, Deserialize, FromForm)]
+pub struct Options {
+    #[validate(range(min = 1, max = 100))]
+    limit: Option<i64>,
+    #[validate(length(min = 26, max = 26))]
+    before: Option<String>,
+    #[validate(length(min = 26, max = 26))]
+    after: Option<String>,
+}
+
+#[get("/<target>/messages?<options..>")]
+pub async fn req(user: User, target: Ref, options: Form<Options>) -> Result<JsonValue> {
+    options
+        .validate()
+        .map_err(|error| Error::FailedValidation { error })?;
+
+    let target = target.fetch_channel().await?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let mut query = doc! { "channel": target.id() };
+
+    if let Some(before) = &options.before {
+        query.insert("_id", doc! { "$lt": before });
+    }
+
+    if let Some(after) = &options.after {
+        query.insert("_id", doc! { "$gt": after });
+    }
+
+    let mut cursor = get_collection("messages")
+        .find(
+            query,
+            FindOptions::builder()
+                .limit(options.limit.unwrap_or(50))
+                .sort(doc! {
+                    "_id": -1
+                })
+                .build(),
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find",
+            with: "messages",
+        })?;
+
+    let mut messages = vec![];
+    while let Some(result) = cursor.next().await {
+        if let Ok(doc) = result {
+            messages.push(
+                from_document::<Message>(doc).map_err(|_| Error::DatabaseError {
+                    operation: "from_document",
+                    with: "message",
+                })?,
+            );
+        }
+    }
+
+    Ok(json!(messages))
+}

src/routes/channels/message_send.rs

@@ -0,0 +1,65 @@
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use mongodb::bson::doc;
+use rocket_contrib::json::{Json, JsonValue};
+use serde::{Deserialize, Serialize};
+use ulid::Ulid;
+use validator::Validate;
+
+#[derive(Validate, Serialize, Deserialize)]
+pub struct Data {
+    #[validate(length(min = 1, max = 2000))]
+    content: String,
+    // Maximum length of 36 allows both ULIDs and UUIDs.
+    #[validate(length(min = 1, max = 36))]
+    nonce: String,
+}
+
+#[post("/<target>/messages", data = "<message>")]
+pub async fn req(user: User, target: Ref, message: Json<Data>) -> Result<JsonValue> {
+    message
+        .validate()
+        .map_err(|error| Error::FailedValidation { error })?;
+
+    let target = target.fetch_channel().await?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_send_message() {
+        Err(Error::LabelMe)?
+    }
+
+    if get_collection("messages")
+        .find_one(
+            doc! {
+                "nonce": &message.nonce
+            },
+            None,
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find_one",
+            with: "message",
+        })?
+        .is_some()
+    {
+        Err(Error::DuplicateNonce)?
+    }
+
+    let msg = Message {
+        id: Ulid::new().to_string(),
+        channel: target.id().to_string(),
+        author: user.id,
+
+        content: message.content.clone(),
+        nonce: Some(message.nonce.clone()),
+        edited: None,
+    };
+
+    msg.clone().publish(&target).await?;
+
+    Ok(json!(msg))
+}

src/routes/channels/mod.rs

@@ -1,5 +1,27 @@
 use rocket::Route;
 
+mod delete_channel;
+mod fetch_channel;
+mod group_add_member;
+mod group_create;
+mod group_remove_member;
+mod message_delete;
+mod message_edit;
+mod message_fetch;
+mod message_query;
+mod message_send;
+
 pub fn routes() -> Vec<Route> {
-    routes![]
+    routes![
+        fetch_channel::req,
+        delete_channel::req,
+        message_send::req,
+        message_query::req,
+        message_fetch::req,
+        message_edit::req,
+        message_delete::req,
+        group_create::req,
+        group_add_member::req,
+        group_remove_member::req
+    ]
 }

src/routes/onboard/complete.rs

@@ -1,6 +1,6 @@
-use crate::database::entities::User;
-use crate::database::get_collection;
+use crate::database::*;
 use crate::util::result::{Error, Result};
+
 use mongodb::bson::doc;
 use mongodb::options::{Collation, FindOneOptions};
 use rauth::auth::Session;
@@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize};
 use validator::Validate;
 
 lazy_static! {
-    static ref RE_USERNAME: Regex = Regex::new(r"^[a-zA-Z0-9-_]+$").unwrap();
+    static ref RE_USERNAME: Regex = Regex::new(r"^[a-zA-Z0-9_]+$").unwrap();
 }
 
 #[derive(Validate, Serialize, Deserialize)]
@@ -28,6 +28,10 @@ pub async fn req(session: Session, user: Option<User>, data: Json<Data>) -> Resu
     data.validate()
         .map_err(|error| Error::FailedValidation { error })?;
 
+    if data.username == "revolt" {
+        Err(Error::UsernameTaken)?
+    }
+
     let col = get_collection("users");
     if col
         .find_one(

src/routes/onboard/hello.rs

@@ -1,4 +1,5 @@
-use crate::database::entities::User;
+use crate::database::*;
+
 use rauth::auth::Session;
 use rocket_contrib::json::JsonValue;
 

src/routes/root.rs

@@ -1,4 +1,6 @@
-use crate::util::variables::{DISABLE_REGISTRATION, HCAPTCHA_SITEKEY, USE_EMAIL, USE_HCAPTCHA};
+use crate::util::variables::{
+    DISABLE_REGISTRATION, EXTERNAL_WS_URL, HCAPTCHA_SITEKEY, USE_EMAIL, USE_HCAPTCHA,
+};
 
 use mongodb::bson::doc;
 use rocket_contrib::json::JsonValue;
@@ -6,7 +8,7 @@ use rocket_contrib::json::JsonValue;
 #[get("/")]
 pub async fn root() -> JsonValue {
     json!({
-        "revolt": "0.3.0-alpha",
+        "revolt": "0.3.1",
         "features": {
             "registration": !*DISABLE_REGISTRATION,
             "captcha": {
@@ -14,6 +16,7 @@ pub async fn root() -> JsonValue {
                 "key": HCAPTCHA_SITEKEY.to_string()
             },
             "email": *USE_EMAIL,
-        }
+        },
+        "ws": *EXTERNAL_WS_URL,
     })
 }

src/routes/users/add_friend.rs

@@ -1,41 +1,37 @@
-use crate::{
-    database::{
-        entities::{RelationshipStatus, User},
-        get_collection,
-        guards::reference::Ref,
-        permissions::get_relationship,
-    },
-    util::result::Error,
-};
-use crate::{
-    notifications::{events::ClientboundNotification, hive},
-    util::result::Result,
-};
+use crate::database::*;
+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+
 use futures::try_join;
 use mongodb::bson::doc;
-use mongodb::options::{FindOneOptions, Collation};
+use mongodb::options::{Collation, FindOneOptions};
 use rocket_contrib::json::JsonValue;
 
 #[put("/<username>/friend")]
 pub async fn req(user: User, username: String) -> Result<JsonValue> {
     let col = get_collection("users");
-    let doc = col.find_one(
-        doc! {
-            "username": username
-        },
-        FindOneOptions::builder()
-            .collation(Collation::builder().locale("en").strength(2).build())
-            .build(),
-    )
-    .await
-    .map_err(|_| Error::DatabaseError { operation: "find_one", with: "user" })?
-    .ok_or_else(|| Error::UnknownUser)?;
+    let doc = col
+        .find_one(
+            doc! {
+                "username": username
+            },
+            FindOneOptions::builder()
+                .collation(Collation::builder().locale("en").strength(2).build())
+                .build(),
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find_one",
+            with: "user",
+        })?
+        .ok_or_else(|| Error::UnknownUser)?;
 
-    let target_id = doc
-        .get_str("_id")
-        .map_err(|_| Error::DatabaseError { operation: "get_str(_id)", with: "user" })?;
+    let target_id = doc.get_str("_id").map_err(|_| Error::DatabaseError {
+        operation: "get_str(_id)",
+        with: "user",
+    })?;
 
-    match get_relationship(&user, &Ref { id: target_id.to_string() }) {
+    match get_relationship(&user, &target_id) {
         RelationshipStatus::User => return Err(Error::NoEffect),
         RelationshipStatus::Friend => return Err(Error::AlreadyFriends),
         RelationshipStatus::Outgoing => return Err(Error::AlreadySentRequest),
@@ -141,9 +137,6 @@ pub async fn req(user: User, username: String) -> Result<JsonValue> {
                     )
                     .ok();
 
-                    hive::subscribe_if_exists(user.id.clone(), target_id.to_string()).ok();
-                    hive::subscribe_if_exists(target_id.to_string(), user.id.clone()).ok();
-
                     Ok(json!({ "status": "Outgoing" }))
                 }
                 Err(_) => Err(Error::DatabaseError {

src/routes/users/block_user.rs

@@ -1,11 +1,7 @@
-use crate::{
-    database::entities::RelationshipStatus, database::entities::User, database::get_collection,
-    database::guards::reference::Ref, database::permissions::get_relationship, util::result::Error,
-};
-use crate::{
-    notifications::{events::ClientboundNotification, hive},
-    util::result::Result,
-};
+use crate::database::*;
+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+
 use futures::try_join;
 use mongodb::bson::doc;
 use rocket_contrib::json::JsonValue;
@@ -14,7 +10,7 @@ use rocket_contrib::json::JsonValue;
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
     let col = get_collection("users");
 
-    match get_relationship(&user, &target) {
+    match get_relationship(&user, &target.id) {
         RelationshipStatus::User | RelationshipStatus::Blocked => Err(Error::NoEffect),
         RelationshipStatus::BlockedOther => {
             col.update_one(
@@ -94,9 +90,6 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
                     )
                     .ok();
 
-                    hive::subscribe_if_exists(user.id.clone(), target.id.clone()).ok();
-                    hive::subscribe_if_exists(target.id.clone(), user.id.clone()).ok();
-
                     Ok(json!({ "status": "Blocked" }))
                 }
                 Err(_) => Err(Error::DatabaseError {

src/routes/users/fetch_dms.rs

@@ -1,6 +1,6 @@
-use crate::database::entities::User;
-use crate::database::get_collection;
+use crate::database::*;
 use crate::util::result::{Error, Result};
+
 use futures::StreamExt;
 use mongodb::bson::doc;
 use rocket_contrib::json::JsonValue;
@@ -12,11 +12,11 @@ pub async fn req(user: User) -> Result<JsonValue> {
             doc! {
                 "$or": [
                     {
-                        "type": "DirectMessage",
+                        "channel_type": "DirectMessage",
                         "active": true
                     },
                     {
-                        "type": "Group"
+                        "channel_type": "Group"
                     }
                 ],
                 "recipients": user.id

src/routes/users/fetch_relationship.rs

@@ -1,8 +1,9 @@
-use crate::database::{entities::User, guards::reference::Ref, permissions::get_relationship};
+use crate::database::*;
 use crate::util::result::Result;
+
 use rocket_contrib::json::JsonValue;
 
 #[get("/<target>/relationship")]
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
-    Ok(json!({ "status": get_relationship(&user, &target) }))
+    Ok(json!({ "status": get_relationship(&user, &target.id) }))
 }

src/routes/users/fetch_relationships.rs

@@ -1,5 +1,6 @@
-use crate::database::entities::User;
+use crate::database::*;
 use crate::util::result::Result;
+
 use rocket_contrib::json::JsonValue;
 
 #[get("/relationships")]

src/routes/users/fetch_user.rs

@@ -1,22 +1,24 @@
-use crate::database::entities::User;
-use crate::database::guards::reference::Ref;
+use crate::database::*;
 use crate::util::result::{Error, Result};
+
 use rocket_contrib::json::JsonValue;
 
 #[get("/<target>")]
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
-    let mut target = target.fetch_user().await?;
+    let target = target.fetch_user().await?;
 
-    if user.id != target.id {
-        // Check whether we are allowed to fetch this user.
-        let perm = crate::database::permissions::temp_calc_perm(&user, &target).await;
-        if !perm.get_access() {
-            Err(Error::LabelMe)?
-        }
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_user(&target)
+        .for_user_given()
+        .await?;
 
-        // Only return user relationships if the target is the caller.
-        target.relations = None;
+    if !perm.get_access() {
+        Err(Error::LabelMe)?
     }
 
-    Ok(json!(target))
+    Ok(json!(
+        target
+            .from(&user)
+            .with(perm)
+    ))
 }

src/routes/users/get_avatar.rs

@@ -1,7 +1,29 @@
 use rocket::response::NamedFile;
 use std::path::Path;
 
-#[get("/<_target>/avatar")]
-pub async fn req(_target: String) -> Option<NamedFile> {
-    NamedFile::open(Path::new("avatar.png")).await.ok()
+use crate::database::Ref;
+
+#[get("/<target>/avatar")]
+pub async fn req(target: Ref) -> Option<NamedFile> {
+    match target.id.chars().nth(25).unwrap() {
+        '0' | '1' | '2' | '3' | '4' | '5' | '6' | '7' => {
+            NamedFile::open(Path::new("assets/user_red.png")).await.ok()
+        }
+        '8' | '9' | 'A' | 'C' | 'B' | 'D' | 'E' | 'F' => {
+            NamedFile::open(Path::new("assets/user_green.png"))
+                .await
+                .ok()
+        }
+        'G' | 'H' | 'J' | 'K' | 'M' | 'N' | 'P' | 'Q' => {
+            NamedFile::open(Path::new("assets/user_blue.png"))
+                .await
+                .ok()
+        }
+        'R' | 'S' | 'T' | 'V' | 'W' | 'X' | 'Y' | 'Z' => {
+            NamedFile::open(Path::new("assets/user_yellow.png"))
+                .await
+                .ok()
+        }
+        _ => unreachable!(),
+    }
 }

src/routes/users/mod.rs

@@ -1,12 +1,12 @@
 use rocket::Route;
 
 mod add_friend;
-mod get_avatar;
 mod block_user;
 mod fetch_dms;
 mod fetch_relationship;
 mod fetch_relationships;
 mod fetch_user;
+mod get_avatar;
 mod open_dm;
 mod remove_friend;
 mod unblock_user;
@@ -16,11 +16,9 @@ pub fn routes() -> Vec<Route> {
         // User Information
         fetch_user::req,
         get_avatar::req,
-
         // Direct Messaging
         fetch_dms::req,
         open_dm::req,
-
         // Relationships
         fetch_relationships::req,
         fetch_relationship::req,

src/routes/users/open_dm.rs

@@ -1,7 +1,6 @@
-use crate::database::entities::{Channel, User};
-use crate::database::get_collection;
-use crate::database::guards::reference::Ref;
+use crate::database::*;
 use crate::util::result::{Error, Result};
+
 use mongodb::bson::doc;
 use rocket_contrib::json::JsonValue;
 use ulid::Ulid;
@@ -10,12 +9,12 @@ use ulid::Ulid;
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
     let query = if user.id == target.id {
         doc! {
-            "type": "SavedMessages",
+            "channel_type": "SavedMessages",
             "user": &user.id
         }
     } else {
         doc! {
-            "type": "DirectMessage",
+            "channel_type": "DirectMessage",
             "recipients": {
                 "$all": [ &user.id, &target.id ]
             }
@@ -41,10 +40,11 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
                 id,
                 active: false,
                 recipients: vec![user.id, target.id],
+                last_message: None,
             }
         };
 
-        channel.save().await?;
+        channel.clone().publish().await?;
         Ok(json!(channel))
     }
 }

src/routes/users/remove_friend.rs

@@ -1,13 +1,8 @@
-use crate::{
-    database::entities::RelationshipStatus, database::entities::User, database::get_collection,
-    database::guards::reference::Ref, database::permissions::get_relationship, util::result::Error,
-};
-use crate::{
-    notifications::{events::ClientboundNotification, hive},
-    util::result::Result,
-};
+use crate::database::*;
+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+
 use futures::try_join;
-use hive_pubsub::PubSub;
 use mongodb::bson::doc;
 use rocket_contrib::json::JsonValue;
 
@@ -15,11 +10,7 @@ use rocket_contrib::json::JsonValue;
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
     let col = get_collection("users");
 
-    match get_relationship(&user, &target) {
-        RelationshipStatus::Blocked
-        | RelationshipStatus::BlockedOther
-        | RelationshipStatus::User
-        | RelationshipStatus::None => Err(Error::NoEffect),
+    match get_relationship(&user, &target.id) {
         RelationshipStatus::Friend
         | RelationshipStatus::Outgoing
         | RelationshipStatus::Incoming => {
@@ -68,10 +59,6 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
                     )
                     .ok();
 
-                    let hive = hive::get_hive();
-                    hive.unsubscribe(&user.id, &target.id).ok();
-                    hive.unsubscribe(&target.id, &user.id).ok();
-
                     Ok(json!({ "status": "None" }))
                 }
                 Err(_) => Err(Error::DatabaseError {
@@ -80,5 +67,6 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
                 }),
             }
         }
+        _ => Err(Error::NoEffect),
     }
 }

src/routes/users/unblock_user.rs

@@ -1,13 +1,8 @@
-use crate::{
-    database::entities::RelationshipStatus, database::entities::User, database::get_collection,
-    database::guards::reference::Ref, database::permissions::get_relationship, util::result::Error,
-};
-use crate::{
-    notifications::{events::ClientboundNotification, hive},
-    util::result::Result,
-};
+use crate::database::*;
+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+
 use futures::try_join;
-use hive_pubsub::PubSub;
 use mongodb::bson::doc;
 use rocket_contrib::json::JsonValue;
 
@@ -15,15 +10,9 @@ use rocket_contrib::json::JsonValue;
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
     let col = get_collection("users");
 
-    match get_relationship(&user, &target) {
-        RelationshipStatus::None
-        | RelationshipStatus::User
-        | RelationshipStatus::BlockedOther
-        | RelationshipStatus::Incoming
-        | RelationshipStatus::Outgoing
-        | RelationshipStatus::Friend => Err(Error::NoEffect),
+    match get_relationship(&user, &target.id) {
         RelationshipStatus::Blocked => {
-            match get_relationship(&target.fetch_user().await?, &user.as_ref()) {
+            match get_relationship(&target.fetch_user().await?, &user.id) {
                 RelationshipStatus::Blocked => {
                     col.update_one(
                         doc! {
@@ -100,10 +89,6 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
                             )
                             .ok();
 
-                            let hive = hive::get_hive();
-                            hive.unsubscribe(&user.id, &target.id).ok();
-                            hive.unsubscribe(&target.id, &user.id).ok();
-
                             Ok(json!({ "status": "None" }))
                         }
                         Err(_) => Err(Error::DatabaseError {
@@ -115,5 +100,6 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
                 _ => Err(Error::InternalError),
             }
         }
+        _ => Err(Error::NoEffect),
     }
 }

src/util/result.rs

@@ -30,6 +30,22 @@ pub enum Error {
     Blocked,
     #[snafu(display("You have been blocked by this user."))]
     BlockedByOther,
+    #[snafu(display("Not friends with target user."))]
+    NotFriends,
+
+    // ? Channel related errors.
+    #[snafu(display("This channel does not exist!"))]
+    UnknownChannel,
+    #[snafu(display("Cannot edit someone else's message."))]
+    CannotEditMessage,
+    #[snafu(display("Cannot remove yourself from a group, use delete channel instead."))]
+    CannotRemoveYourself,
+    #[snafu(display("Group size is too large."))]
+    GroupTooLarge { max: usize },
+    #[snafu(display("User already part of group."))]
+    AlreadyInGroup,
+    #[snafu(display("User is not part of the group."))]
+    NotInGroup,
 
     // ? General errors.
     #[snafu(display("Failed to validate fields."))]
@@ -41,6 +57,10 @@ pub enum Error {
     },
     #[snafu(display("Internal server error."))]
     InternalError,
+    #[snafu(display("Operation cannot be performed on this object."))]
+    InvalidOperation,
+    #[snafu(display("Already created an object with this nonce."))]
+    DuplicateNonce,
     #[snafu(display("This request had no effect."))]
     NoEffect,
 }
@@ -61,10 +81,20 @@ impl<'r> Responder<'r, 'static> for Error {
             Error::AlreadySentRequest => Status::Conflict,
             Error::Blocked => Status::Conflict,
             Error::BlockedByOther => Status::Forbidden,
+            Error::NotFriends => Status::Forbidden,
+
+            Error::UnknownChannel => Status::NotFound,
+            Error::CannotEditMessage => Status::Forbidden,
+            Error::CannotRemoveYourself => Status::BadRequest,
+            Error::GroupTooLarge { .. } => Status::Forbidden,
+            Error::AlreadyInGroup => Status::Conflict,
+            Error::NotInGroup => Status::NotFound,
 
             Error::FailedValidation { .. } => Status::UnprocessableEntity,
             Error::DatabaseError { .. } => Status::InternalServerError,
             Error::InternalError => Status::InternalServerError,
+            Error::InvalidOperation => Status::BadRequest,
+            Error::DuplicateNonce => Status::Conflict,
             Error::NoEffect => Status::Ok,
         };
 

src/util/variables.rs

@@ -9,6 +9,8 @@ lazy_static! {
         env::var("REVOLT_MONGO_URI").expect("Missing REVOLT_MONGO_URI environment variable.");
     pub static ref PUBLIC_URL: String =
         env::var("REVOLT_PUBLIC_URL").expect("Missing REVOLT_PUBLIC_URL environment variable.");
+    pub static ref EXTERNAL_WS_URL: String =
+        env::var("REVOLT_EXTERNAL_WS_URL").expect("Missing REVOLT_EXTERNAL_WS_URL environment variable.");
     pub static ref HCAPTCHA_KEY: String =
         env::var("REVOLT_HCAPTCHA_KEY").unwrap_or_else(|_| "0x0000000000000000000000000000000000000000".to_string());
     pub static ref HCAPTCHA_SITEKEY: String =
@@ -31,21 +33,21 @@ lazy_static! {
     pub static ref SMTP_HOST: String =
         env::var("REVOLT_SMTP_HOST").unwrap_or_else(|_| "".to_string());
     pub static ref SMTP_USERNAME: String =
-        env::var("SMTP_USERNAME").unwrap_or_else(|_| "".to_string());
+        env::var("REVOLT_SMTP_USERNAME").unwrap_or_else(|_| "".to_string());
     pub static ref SMTP_PASSWORD: String =
-        env::var("SMTP_PASSWORD").unwrap_or_else(|_| "".to_string());
-    pub static ref SMTP_FROM: String = env::var("SMTP_FROM").unwrap_or_else(|_| "".to_string());
+        env::var("REVOLT_SMTP_PASSWORD").unwrap_or_else(|_| "".to_string());
+    pub static ref SMTP_FROM: String = env::var("REVOLT_SMTP_FROM").unwrap_or_else(|_| "".to_string());
+
+    // Application Logic Settings
+    pub static ref MAX_GROUP_SIZE: usize =
+        env::var("REVOLT_MAX_GROUP_SIZE").unwrap_or_else(|_| "50".to_string()).parse().unwrap();
 }
 
 pub fn preflight_checks() {
     if *USE_EMAIL == false {
         #[cfg(not(debug_assertions))]
-        {
-            if !env::var("REVOLT_UNSAFE_NO_EMAIL").map_or(false, |v| v == *"1") {
-                panic!(
-                    "Not letting you run this in production, set REVOLT_UNSAFE_NO_EMAIL=1 to run."
-                );
-            }
+        if !env::var("REVOLT_UNSAFE_NO_EMAIL").map_or(false, |v| v == *"1") {
+            panic!("Running in production without email is not recommended, set REVOLT_UNSAFE_NO_EMAIL=1 to override.");
         }
 
         #[cfg(debug_assertions)]
@@ -54,10 +56,8 @@ pub fn preflight_checks() {
 
     if *USE_HCAPTCHA == false {
         #[cfg(not(debug_assertions))]
-        {
-            if !env::var("REVOLT_UNSAFE_NO_CAPTCHA").map_or(false, |v| v == *"1") {
-                panic!("Not letting you run this in production, set REVOLT_UNSAFE_NO_CAPTCHA=1 to run.");
-            }
+        if !env::var("REVOLT_UNSAFE_NO_CAPTCHA").map_or(false, |v| v == *"1") {
+            panic!("Running in production without CAPTCHA is not recommended, set REVOLT_UNSAFE_NO_CAPTCHA=1 to override.");
         }
 
         #[cfg(debug_assertions)]