Cargo fmt.

Code cleanup/formatting

See merge request revolt/delta!2

Cargo.lock

@@ -36,6 +36,16 @@ version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "23b62fc65de8e4e7f52534fb52b0f3ed04746ae267519eef2a83941e8085068b"
 
+[[package]]
+name = "async-attributes"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "efd3d156917d94862e779f356c5acae312b08fd3121e792c857d7928c8088423"
+dependencies = [
+ "quote 1.0.8",
+ "syn 1.0.56",
+]
+
 [[package]]
 name = "async-channel"
 version = "1.5.1"
@@ -110,6 +120,7 @@ version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8f9f84f1280a2b436a2c77c2582602732b6c2f4321d5494d6e799e6c367859a8"
 dependencies = [
+ "async-attributes",
  "async-channel",
  "async-global-executor",
  "async-io",
@@ -2262,7 +2273,7 @@ dependencies = [
 
 [[package]]
 name = "revolt"
-version = "0.3.0-alpha"
+version = "0.3.1"
 dependencies = [
  "async-std",
  "async-tungstenite",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt"
-version = "0.3.0-alpha"
+version = "0.3.1"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"
 
@@ -13,13 +13,12 @@ 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"] }
+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.
 mongodb = { version = "1.1.1", features = ["tokio-runtime"], default-features = false }
 
 once_cell = "1.4.1"

src/database/entities/channel.rs

@@ -1,41 +1,11 @@
-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, to_document};
+use rocket_contrib::json::JsonValue;
 use serde::{Deserialize, Serialize};
 
-/*#[derive(Serialize, Deserialize, Debug, Clone)]
-pub struct LastMessage {
-    id: String,
-    user_id: String,
-    short_content: 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")]
 pub enum Channel {
     SavedMessages {
@@ -52,6 +22,8 @@ pub enum Channel {
     Group {
         #[serde(rename = "_id")]
         id: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        nonce: Option<String>,
         name: String,
         owner: String,
         description: String,
@@ -60,7 +32,15 @@ pub enum Channel {
 }
 
 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 publish(self) -> Result<()> {
         get_collection("channels")
             .insert_one(
                 to_document(&self).map_err(|_| Error::DatabaseError {
@@ -75,6 +55,58 @@ 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, partial: JsonValue) -> Result<()> {
+        let id = self.id().to_string();
+        ClientboundNotification::ChannelUpdate(partial)
+            .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,90 @@
-// 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) -> Result<()> {
+        get_collection("messages")
+            .insert_one(to_bson(&self).unwrap().as_document().unwrap().clone(), None)
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "insert_one",
+                with: "messages",
+            })?;
+
+        let channel = self.channel.clone();
+        ClientboundNotification::Message(self)
+            .publish(channel)
+            .await
+            .ok();
+
+        Ok(())
+    }
+
+    pub async fn publish_update(&self, partial: JsonValue) -> Result<()> {
+        let channel = self.channel.clone();
+        ClientboundNotification::MessageUpdate(partial)
+            .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,6 @@
-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)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub enum RelationshipStatus {
     None,
     User,
@@ -31,42 +25,10 @@ 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,
-            ))
-        }
-    }
-}
-
-impl User {
-    pub fn as_ref(&self) -> Ref {
-        Ref {
-            id: self.id.to_string(),
-        }
-    }
+
+    // ? 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>,
 }

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,34 @@ 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) -> Result<Message> {
+        self.fetch("messages").await
+    }
+}
+
+impl User {
+    pub fn as_ref(&self) -> Ref {
+        Ref {
+            id: self.id.to_string(),
+        }
     }
 }
 

src/database/guards/user.rs

@@ -1 +1,36 @@
+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,
+            ))
+        }
+    }
+}

src/database/migrations/scripts.rs

@@ -2,7 +2,7 @@ use super::super::{get_collection, get_db};
 
 use crate::rocket::futures::StreamExt;
 use log::info;
-use mongodb::bson::{doc, from_bson, Bson};
+use mongodb::bson::{doc, from_document};
 use mongodb::options::FindOptions;
 use serde::{Deserialize, Serialize};
 
@@ -23,7 +23,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;
 

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,60 @@
+use crate::database::*;
+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) -> ChannelPermissions<[u32; 1]> {
+    match target {
+        Channel::SavedMessages { user: owner, .. } => {
+            if &user.id == owner {
+                ChannelPermissions([ChannelPermission::View
+                    + ChannelPermission::SendMessage
+                    + ChannelPermission::ManageMessages])
+            } else {
+                ChannelPermissions([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::calculate(&user, recipient).await;
+
+                    if perms.get_send_message() {
+                        return ChannelPermissions([
+                            ChannelPermission::View + ChannelPermission::SendMessage
+                        ]);
+                    }
+
+                    return ChannelPermissions([ChannelPermission::View as u32]);
+                }
+            }
+
+            ChannelPermissions([0])
+        }
+        Channel::Group { recipients, .. } => {
+            if recipients.iter().find(|x| *x == &user.id).is_some() {
+                ChannelPermissions([ChannelPermission::View + ChannelPermission::SendMessage])
+            } else {
+                ChannelPermissions([0])
+            }
+        }
+    }
+}

src/database/permissions/mod.rs

@@ -1,49 +1,4 @@
-use crate::database::entities::User;
-use num_enum::TryFromPrimitive;
-use std::ops;
+pub mod channel;
+pub mod user;
 
-#[derive(Debug, PartialEq, Eq, TryFromPrimitive, Copy, Clone)]
-#[repr(u32)]
-pub enum UserPermission {
-    Access = 1,
-    SendMessage = 2,
-    Invite = 4,
-}
-
-bitfield! {
-    pub struct UserPermissions(MSB0 [u32]);
-    u32;
-    pub get_access, _: 31;
-    pub get_send_message, _: 30;
-    pub get_invite, _: 29;
-}
-
-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 });
-
-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();
-        }
-    }
-
-    RelationshipStatus::None
-}
+pub use user::get_relationship;

src/database/permissions/user.rs

@@ -0,0 +1,62 @@
+use crate::database::*;
+use num_enum::TryFromPrimitive;
+use std::ops;
+
+#[derive(Debug, PartialEq, Eq, TryFromPrimitive, Copy, Clone)]
+#[repr(u32)]
+pub enum UserPermission {
+    Access = 1,
+    SendMessage = 2,
+    Invite = 4,
+}
+
+bitfield! {
+    pub struct UserPermissions(MSB0 [u32]);
+    u32;
+    pub get_access, _: 31;
+    pub get_send_message, _: 30;
+    pub get_invite, _: 29;
+}
+
+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 });
+
+pub async fn calculate(user: &User, target: &str) -> 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
+
+    let mut permissions: u32 = 0;
+    match get_relationship(&user, &target) {
+        RelationshipStatus::Friend => {
+            return UserPermissions([UserPermission::Access
+                + UserPermission::SendMessage
+                + UserPermission::Invite])
+        }
+        RelationshipStatus::Blocked | RelationshipStatus::BlockedOther => {
+            return UserPermissions([UserPermission::Access as u32])
+        }
+        RelationshipStatus::Incoming | RelationshipStatus::Outgoing => {
+            permissions = UserPermission::Access as u32;
+        }
+        _ => {}
+    }
+
+    UserPermissions([permissions])
+}
+
+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
+}

src/main.rs

@@ -18,17 +18,13 @@ pub mod notifications;
 pub mod routes;
 pub mod util;
 
-use async_std::task;
 use futures::join;
 use log::info;
 use rauth;
 use rocket_cors::AllowedOrigins;
 
-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"));
 

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,94 @@ pub enum ClientboundNotification {
     Error(WebSocketError),
     Authenticated,
     Ready {
-        user: User
-    },
-
-    /*MessageCreate {
-        id: String,
-        nonce: Option<String>,
-        channel: String,
-        author: String,
-        content: String,
-    },
-
-    MessageEdit {
-        id: String,
-        channel: String,
-        author: String,
-        content: String,
+        users: Vec<User>,
+        channels: Vec<Channel>,
     },
 
+    Message(Message),
+    MessageUpdate(JsonValue),
     MessageDelete {
         id: String,
     },
 
-    GroupUserJoin {
+    ChannelCreate(Channel),
+    ChannelUpdate(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::ChannelGroupLeave { id, user } => {
+            get_hive()
+                .hive
+                .unsubscribe(&user.to_string(), &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();
+            }
+        }
+        _ => {}
+    }
+}

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,104 @@
+use crate::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,
+};
+
+use super::websocket::is_online;
+
+pub async fn generate_ready(mut user: User) -> Result<ClientboundNotification> {
+    let mut users = vec![];
+
+    if let Some(relationships) = &user.relations {
+        let user_ids: Vec<String> = relationships
+            .iter()
+            .map(|relationship| relationship.id.clone())
+            .collect();
+
+        let mut cursor = get_collection("users")
+            .find(
+                doc! {
+                    "_id": {
+                        "$in": user_ids
+                    }
+                },
+                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 mut user: User = from_document(doc).map_err(|_| Error::DatabaseError {
+                    operation: "from_document",
+                    with: "user",
+                })?;
+
+                user.relationship = Some(
+                    relationships
+                        .iter()
+                        .find(|x| user.id == x.id)
+                        .ok_or_else(|| Error::InternalError)?
+                        .status
+                        .clone(),
+                );
+
+                user.online = Some(is_online(&user.id));
+
+                users.push(user);
+            }
+        }
+    }
+
+    let mut cursor = get_collection("channels")
+        .find(
+            doc! {
+                "$or": [
+                    {
+                        "type": "SavedMessages",
+                        "user": &user.id
+                    },
+                    {
+                        "type": "DirectMessage",
+                        "recipients": &user.id,
+                        "active": true
+                    },
+                    {
+                        "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 {
+            channels.push(from_document(doc).map_err(|_| Error::DatabaseError {
+                operation: "from_document",
+                with: "channel",
+            })?);
+        }
+    }
+
+    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,35 @@ pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
         }
     }
 
+    let mut cursor = get_collection("channels")
+        .find(
+            doc! {
+                "$or": [
+                    {
+                        "type": "SavedMessages",
+                        "user": &user.id
+                    },
+                    {
+                        "type": "DirectMessage",
+                        "recipients": &user.id,
+                        "active": true
+                    },
+                    {
+                        "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;
@@ -64,69 +63,105 @@ 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"))
+                            .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 +170,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 +209,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,111 @@
+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::channel::calculate(&user, &target).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()
+            .await
+            .ok();
+
+            Ok(())
+        }
+    }
+}

src/routes/channels/fetch_channel.rs

@@ -0,0 +1,16 @@
+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::channel::calculate(&user, &target).await;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    Ok(json!(target))
+}

src/routes/channels/group_add_member.rs

@@ -0,0 +1,70 @@
+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::channel::calculate(&user, &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,
+            format!("<@{}> added <@{}> to the group.", user.id, member.id),
+        )
+        .publish()
+        .await
+        .ok();
+
+        Ok(())
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/group_create.rs

@@ -0,0 +1,78 @@
+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>>(),
+    };
+
+    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,
+            format!("<@{}> removed <@{}> from the group.", user.id, member.id),
+        )
+        .publish()
+        .await
+        .ok();
+
+        Ok(())
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/message_delete.rs

@@ -0,0 +1,24 @@
+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::channel::calculate(&user, &channel).await;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let message = msg.fetch_message().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,56 @@
+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::channel::calculate(&user, &channel).await;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let message = msg.fetch_message().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,17 @@
+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::channel::calculate(&user, &channel).await;
+    if !perm.get_view() {
+        Err(Error::LabelMe)?
+    }
+
+    let message = msg.fetch_message().await?;
+    Ok(json!(message))
+}

src/routes/channels/message_query.rs

@@ -0,0 +1,76 @@
+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::channel::calculate(&user, &target).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,62 @@
+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::channel::calculate(&user, &target).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().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)]

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;

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,6 +1,6 @@
-use crate::database::entities::User;
-use crate::database::guards::reference::Ref;
 use crate::util::result::{Error, Result};
+use crate::{database::*, notifications::websocket::is_online};
+
 use rocket_contrib::json::JsonValue;
 
 #[get("/<target>")]
@@ -9,14 +9,29 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
 
     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;
+        let perm = permissions::user::calculate(&user, &target.id).await;
         if !perm.get_access() {
             Err(Error::LabelMe)?
         }
 
         // Only return user relationships if the target is the caller.
         target.relations = None;
+
+        // Add relevant relationship
+        if let Some(relationships) = &user.relations {
+            target.relationship = relationships
+                .iter()
+                .find(|x| x.id == user.id)
+                .map(|x| x.status.clone())
+                .or_else(|| Some(RelationshipStatus::None));
+        } else {
+            target.relationship = Some(RelationshipStatus::None);
+        }
+    } else {
+        target.relationship = Some(RelationshipStatus::User);
     }
 
+    target.online = Some(is_online(&target.id));
+
     Ok(json!(target))
 }

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;
@@ -44,7 +43,7 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
             }
         };
 
-        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,20 @@ 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("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 +55,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 +79,19 @@ impl<'r> Responder<'r, 'static> for Error {
             Error::AlreadySentRequest => Status::Conflict,
             Error::Blocked => Status::Conflict,
             Error::BlockedByOther => Status::Forbidden,
+            Error::NotFriends => Status::Forbidden,
+
+            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)]