src/notifications/events.rs

+use hive_pubsub::PubSub;
+use mongodb::bson::doc;
 use rauth::auth::Session;
+use rocket_contrib::json::JsonValue;
 use serde::{Deserialize, Serialize};
-use snafu::Snafu;
 
-use super::hive::get_hive;
-use crate::database::*;
+use super::hive::{get_hive, subscribe_if_exists};
+use crate::{database::*, util::result::Result};
 
-#[derive(Serialize, Deserialize, Debug, Snafu)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(tag = "error")]
 pub enum WebSocketError {
-    #[snafu(display("This error has not been labelled."))]
     LabelMe,
-    #[snafu(display("Internal server error."))]
-    InternalError,
-    #[snafu(display("Invalid session."))]
+    InternalError { at: String },
     InvalidSession,
-    #[snafu(display("User hasn't completed onboarding."))]
     OnboardingNotFinished,
-    #[snafu(display("Already authenticated with server."))]
     AlreadyAuthenticated,
 }
 
@@ -24,81 +21,243 @@ pub enum WebSocketError {
 #[serde(tag = "type")]
 pub enum ServerboundNotification {
     Authenticate(Session),
+    BeginTyping { channel: String },
+    EndTyping { channel: String },
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum RemoveUserField {
+    ProfileContent,
+    ProfileBackground,
+    StatusText,
+    Avatar,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum RemoveChannelField {
+    Icon,
+    Description
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum RemoveServerField {
+    Icon,
+    Banner,
+    Description,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum RemoveRoleField {
+    Colour,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum RemoveMemberField {
+    Nickname,
+    Avatar,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(tag = "type")]
 pub enum ClientboundNotification {
     Error(WebSocketError),
     Authenticated,
     Ready {
         users: Vec<User>,
+        servers: Vec<Server>,
+        channels: Vec<Channel>,
+        members: Vec<Member>
     },
 
-    /*MessageCreate {
+    Message(Message),
+    MessageUpdate {
         id: String,
-        nonce: Option<String>,
         channel: String,
-        author: String,
-        content: String,
+        data: JsonValue,
     },
-
-    MessageEdit {
+    MessageDelete {
         id: String,
         channel: String,
-        author: String,
-        content: String,
     },
 
-    MessageDelete {
+    ChannelCreate(Channel),
+    ChannelUpdate {
         id: String,
+        data: JsonValue,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        clear: Option<RemoveChannelField>,
     },
-
-    GroupUserJoin {
+    ChannelDelete {
+        id: String,
+    },
+    ChannelGroupJoin {
         id: String,
         user: String,
     },
-
-    GroupUserLeave {
+    ChannelGroupLeave {
         id: String,
         user: String,
     },
-
-    GuildUserJoin {
+    ChannelStartTyping {
         id: String,
         user: String,
     },
-
-    GuildUserLeave {
+    ChannelStopTyping {
         id: String,
         user: String,
-        banned: bool,
     },
-
-    GuildChannelCreate {
+    ChannelAck {
         id: String,
-        channel: String,
-        name: String,
-        description: String,
+        user: String,
+        message_id: String,
     },
 
-    GuildChannelDelete {
+    ServerUpdate {
         id: String,
-        channel: String,
+        data: JsonValue,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        clear: Option<RemoveServerField>,
+    },
+    ServerDelete {
+        id: String,
+    },
+    ServerMemberUpdate {
+        id: MemberCompositeKey,
+        data: JsonValue,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        clear: Option<RemoveMemberField>,
+    },
+    ServerMemberJoin {
+        id: String,
+        user: String,
+    },
+    ServerMemberLeave {
+        id: String,
+        user: String,
+    },
+    ServerRoleUpdate {
+        id: String,
+        role_id: String,
+        data: JsonValue,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        clear: Option<RemoveRoleField>
+    },
+    ServerRoleDelete {
+        id: String,
+        role_id: String
     },
 
-    GuildDelete {
+    UserUpdate {
         id: String,
-    },*/
+        data: JsonValue,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        clear: Option<RemoveUserField>,
+    },
     UserRelationship {
         id: String,
-        user: String,
+        user: User,
         status: RelationshipStatus,
     },
+    UserSettingsUpdate {
+        id: String,
+        update: JsonValue,
+    },
 }
 
 impl ClientboundNotification {
-    pub async fn publish(self, topic: String) -> Result<(), String> {
-        hive_pubsub::backend::mongo::publish(get_hive(), &topic, self).await
+    pub fn publish(self, topic: String) {
+        async_std::task::spawn(async move {
+            prehandle_hook(&self).await.ok(); // ! FIXME: this should be moved to pubsub
+            hive_pubsub::backend::mongo::publish(get_hive(), &topic, self)
+                .await
+                .ok();
+        });
+    }
+
+    pub fn publish_as_user(self, user: String) {
+        self.clone().publish(user.clone());
+
+        async_std::task::spawn(async move {
+            if let Ok(server_ids) = User::fetch_server_ids(&user).await {
+                for server in server_ids {
+                    self.clone().publish(server.clone());
+                }
+            }
+        });
+    }
+}
+
+pub async fn prehandle_hook(notification: &ClientboundNotification) -> Result<()> {
+    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();
+                    }
+                }
+                Channel::TextChannel { server, .. }
+                | Channel::VoiceChannel { server, .. } => {
+                    // ! FIXME: write a better algorithm?
+                    let members = Server::fetch_member_ids(server).await?;
+                    for member in members {
+                        subscribe_if_exists(member.clone(), channel_id.to_string()).ok();
+                    }
+                }
+            }
+        }
+        ClientboundNotification::ServerMemberJoin { id, user } => {
+            let server = Ref::from_unchecked(id.clone()).fetch_server().await?;
+
+            subscribe_if_exists(user.clone(), id.clone()).ok();
+
+            for channel in server.channels {
+                subscribe_if_exists(user.clone(), channel).ok();
+            }
+        }
+        ClientboundNotification::UserRelationship { id, user, status } => {
+            if status != &RelationshipStatus::None {
+                subscribe_if_exists(id.clone(), user.id.clone()).ok();
+            }
+        }
+        _ => {}
+    }
+
+    Ok(())
+}
+
+pub async fn posthandle_hook(notification: &ClientboundNotification) {
+    match &notification {
+        ClientboundNotification::ChannelDelete { id } => {
+            get_hive().hive.drop_topic(&id).ok();
+        }
+        ClientboundNotification::ChannelGroupLeave { id, user } => {
+            get_hive().hive.unsubscribe(user, id).ok();
+        }
+        ClientboundNotification::ServerDelete { id } => {
+            get_hive().hive.drop_topic(&id).ok();
+        }
+        ClientboundNotification::ServerMemberLeave { id, user } => {
+            get_hive().hive.unsubscribe(user, id).ok();
+
+            if let Ok(server) = Ref::from_unchecked(id.clone()).fetch_server().await {
+                for channel in server.channels {
+                    get_hive().hive.unsubscribe(user, &channel).ok();
+                }
+            }
+        }
+        ClientboundNotification::UserRelationship { id, user, status } => {
+            if status == &RelationshipStatus::None {
+                get_hive().hive.unsubscribe(id, &user.id).ok();
+            }
+        }
+        _ => {}
     }
 }

src/notifications/hive.rs

@@ -13,7 +13,12 @@ static HIVE: OnceCell<Hive> = OnceCell::new();
 
 pub async fn init_hive() {
     let hive = MongodbPubSub::new(
-        |ids, notification| {
+        |ids, notification: ClientboundNotification| {
+            let notif = notification.clone();
+            async_std::task::spawn(async move {
+                super::events::posthandle_hook(&notif).await;
+            });
+
             if let Ok(data) = to_string(&notification) {
                 debug!("Pushing out notification. {}", data);
                 websocket::publish(ids, notification);
@@ -36,8 +41,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/payload.rs

-use crate::notifications::events::ClientboundNotification;
+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_bson, Bson},
-    options::FindOptions,
-};
+use mongodb::bson::{doc, from_document};
 
-pub async fn generate_ready(user: User) -> Result<ClientboundNotification> {
-    let mut users = vec![];
+pub async fn generate_ready(mut user: User) -> Result<ClientboundNotification> {
+    let mut user_ids: HashSet<String> = HashSet::new();
 
     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
+        user_ids.extend(
+            relationships
+                .iter()
+                .map(|relationship| relationship.id.clone()),
+        );
+    }
+
+    let members = User::fetch_memberships(&user.id).await?;
+    let server_ids: Vec<String> = members.iter()
+        .map(|x| x.id.server.clone())
+        .collect();
+    
+    let mut cursor = get_collection("servers")
+        .find(
+            doc! {
+                "_id": {
+                    "$in": server_ids
+                }
+            },
+            None,
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find",
+            with: "servers",
+        })?;
+
+    let mut servers = vec![];
+    let mut channel_ids = vec![];
+    while let Some(result) = cursor.next().await {
+        if let Ok(doc) = result {
+            let server: Server = from_document(doc).map_err(|_| Error::DatabaseError {
+                operation: "from_document",
+                with: "server",
+            })?;
+
+            channel_ids.extend(server.channels.iter().cloned());
+            servers.push(server);
+        }
+    }
+
+    let mut cursor = get_collection("channels")
+        .find(
+            doc! {
+                "$or": [
+                    {
+                        "_id": {
+                            "$in": channel_ids
+                        }
+                    },
+                    {
+                        "channel_type": "SavedMessages",
+                        "user": &user.id
+                    },
+                    {
+                        "channel_type": "DirectMessage",
+                        "recipients": &user.id
+                    },
+                    {
+                        "channel_type": "Group",
+                        "recipients": &user.id
                     }
-                },
-                FindOptions::builder()
-                    .projection(doc! { "_id": 1, "username": 1 })
-                    .build(),
-            )
-            .await
-            .map_err(|_| Error::DatabaseError {
-                operation: "find",
-                with: "users",
+                ]
+            },
+            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",
             })?;
 
-        while let Some(result) = cursor.next().await {
-            if let Ok(doc) = result {
-                let mut user: User =
-                    from_bson(Bson::Document(doc)).map_err(|_| Error::DatabaseError {
-                        operation: "from_bson",
-                        with: "user",
-                    })?;
-
-                user.relationship = Some(
-                    relationships
-                        .iter()
-                        .find(|x| user.id == x.id)
-                        .ok_or_else(|| Error::InternalError)?
-                        .status
-                        .clone(),
-                );
-
-                users.push(user);
+            if let Channel::Group { recipients, .. } = &channel {
+                user_ids.extend(recipients.iter().cloned());
+            } else if let Channel::DirectMessage { recipients, .. } = &channel {
+                user_ids.extend(recipients.iter().cloned());
             }
+
+            channels.push(channel);
         }
     }
 
+    user_ids.remove(&user.id);
+    let mut users = if user_ids.len() > 0 {
+        user.fetch_multiple_users(user_ids.into_iter().collect::<Vec<String>>())
+            .await?
+    } else {
+        vec![]
+    };
+
+    user.relationship = Some(RelationshipStatus::User);
+    user.online = Some(true);
     users.push(user);
 
-    Ok(ClientboundNotification::Ready { users })
+    Ok(ClientboundNotification::Ready {
+        users,
+        servers,
+        channels,
+        members
+    })
 }

src/notifications/subscriptions.rs

 use crate::database::*;
 
 use super::hive::get_hive;
+use futures::StreamExt;
 use hive_pubsub::PubSub;
+use mongodb::bson::doc;
+use mongodb::bson::Document;
+use mongodb::options::FindOptions;
 
 pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
     let hive = get_hive();
@@ -13,5 +17,71 @@ pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
         }
     }
 
+    let server_ids = User::fetch_server_ids(&user.id)
+        .await
+        .map_err(|_| "Failed to fetch memberships.".to_string())?;
+
+    let channel_ids = get_collection("servers")
+        .find(
+            doc! {
+                "_id": {
+                    "$in": &server_ids
+                }
+            },
+            None,
+        )
+        .await
+        .map_err(|_| "Failed to fetch servers.".to_string())?
+        .filter_map(async move |s| s.ok())
+        .collect::<Vec<Document>>()
+        .await
+        .into_iter()
+        .filter_map(|x| {
+            x.get_array("channels").ok().map(|v| {
+                v.into_iter()
+                    .filter_map(|x| x.as_str().map(|x| x.to_string()))
+                    .collect::<Vec<String>>()
+            })
+        })
+        .flatten()
+        .collect::<Vec<String>>();
+
+    for id in server_ids {
+        hive.subscribe(user.id.clone(), id)?;
+    }
+
+    for id in channel_ids {
+        hive.subscribe(user.id.clone(), id)?;
+    }
+
+    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

@@ -12,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};
@@ -63,81 +66,149 @@ 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();
         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);
-
-                                                match task::block_on(
-                                                    super::payload::generate_ready(user),
-                                                ) {
-                                                    Ok(payload) => {
-                                                        send(payload);
-                                                    }
-                                                    Err(_) => {
-                                                        send(ClientboundNotification::Error(
-                                                            WebSocketError::InternalError,
-                                                        ));
-                                                    }
-                                                }
-                                            } 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::UserUpdate {
+                                                id: id.clone(),
+                                                data: json!({
+                                                    "online": true
+                                                }),
+                                                clear: None
+                                            }
+                                            .publish_as_user(id);
                                         }
                                     }
                                     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,
+                            ));
+                        }
+                    }
+                    // ! TEMP: verify user part of channel
+                    // ! Could just run permission check here.
+                    ServerboundNotification::BeginTyping { channel } => {
+                        if mutex.lock().unwrap().is_some() {
+                            let user = {
+                                let mutex = mutex.lock().unwrap();
+                                let session = mutex.as_ref().unwrap();
+                                session.user_id.clone()
+                            };
+
+                            ClientboundNotification::ChannelStartTyping {
+                                id: channel.clone(),
+                                user,
+                            }
+                            .publish(channel);
+                        } else {
+                            send(ClientboundNotification::Error(
+                                WebSocketError::AlreadyAuthenticated,
+                            ));
+
+                            return Ok(());
+                        }
+                    }
+                    ServerboundNotification::EndTyping { channel } => {
+                        if mutex.lock().unwrap().is_some() {
+                            let user = {
+                                let mutex = mutex.lock().unwrap();
+                                let session = mutex.as_ref().unwrap();
+                                session.user_id.clone()
+                            };
+
+                            ClientboundNotification::ChannelStopTyping {
+                                id: channel.clone(),
+                                user,
+                            }
+                            .publish(channel);
+                        } else {
+                            send(ClientboundNotification::Error(
+                                WebSocketError::AlreadyAuthenticated,
+                            ));
+
+                            return Ok(());
                         }
                     }
                 }
             }
         }
 
-        future::ok(())
+        Ok(())
     });
 
     pin_mut!(fwd, incoming);
@@ -146,12 +217,28 @@ async fn accept(stream: TcpStream) {
     info!("User {} disconnected.", &addr);
     CONNECTIONS.lock().unwrap().remove(&addr);
 
-    if let Some(session) = session {
-        let mut users = USERS.write().unwrap();
-        users.remove(&session.user_id, &addr);
-        if users.get_left(&session.user_id).is_none() {
-            get_hive().drop_client(&session.user_id).unwrap();
+    let mut offline = None;
+    {
+        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() {
+                get_hive().drop_client(&session.user_id).unwrap();
+                offline = Some(session.user_id.clone());
+            }
+        }
+    }
+
+    if let Some(id) = offline {
+        ClientboundNotification::UserUpdate {
+            id: id.clone(),
+            data: json!({
+                "online": false
+            }),
+            clear: None
         }
+        .publish_as_user(id);
     }
 }
 
@@ -161,10 +248,15 @@ pub fn publish(ids: Vec<String>, notification: ClientboundNotification) {
         let users = USERS.read().unwrap();
         for id in ids {
             // Block certain notifications from reaching users that aren't meant to see them.
-            if let ClientboundNotification::UserRelationship { id: user_id, .. } = &notification {
-                if &id != user_id {
-                    continue;
+            match &notification {
+                ClientboundNotification::UserRelationship { id: user_id, .. }
+                | ClientboundNotification::UserSettingsUpdate { id: user_id, .. }
+                | ClientboundNotification::ChannelAck { user: user_id, .. } => {
+                    if &id != user_id {
+                        continue;
+                    }
                 }
+                _ => {}
             }
 
             if let Some(mut arr) = users.get_left(&id) {
@@ -184,3 +276,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/channel_ack.rs

+use crate::database::*;
+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+
+use mongodb::bson::doc;
+use mongodb::options::UpdateOptions;
+
+#[put("/<target>/ack/<message>")]
+pub async fn req(user: User, target: Ref, message: 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::MissingPermission)?
+    }
+
+    let id = target.id();
+    get_collection("channel_unreads")
+        .update_one(
+            doc! {
+                "_id.channel": id,
+                "_id.user": &user.id
+            },
+            doc! {
+                "$unset": {
+                    "mentions": 1
+                },
+                "$set": {
+                    "last_id": &message.id
+                }
+            },
+            UpdateOptions::builder().upsert(true).build(),
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "update_one",
+            with: "channel_unreads",
+        })?;
+
+    ClientboundNotification::ChannelAck {
+        id: id.to_string(),
+        user: user.id.clone(),
+        message_id: message.id,
+    }
+    .publish(user.id);
+
+    Ok(())
+}

src/routes/channels/channel_delete.rs

+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::MissingPermission)?
+    }
+
+    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());
+
+            Content::SystemMessage(SystemMessage::UserLeft { id: user.id })
+                .send_as_system(&target)
+                .await
+                .ok();
+
+            Ok(())
+        }
+        Channel::TextChannel { .. } |
+        Channel::VoiceChannel { .. } => {
+            if perm.get_manage_channel() {
+                target.delete().await
+            } else {
+                Err(Error::MissingPermission)
+            }
+        }
+    }
+}

src/routes/channels/channel_edit.rs

+use crate::notifications::events::ClientboundNotification;
+use crate::util::result::{Error, Result};
+use crate::{database::*, notifications::events::RemoveChannelField};
+
+use mongodb::bson::{doc, to_document};
+use rocket_contrib::json::Json;
+use serde::{Deserialize, Serialize};
+use validator::Validate;
+
+#[derive(Validate, Serialize, Deserialize)]
+pub struct Data {
+    #[validate(length(min = 1, max = 32))]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    name: Option<String>,
+    #[validate(length(min = 0, max = 1024))]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    description: Option<String>,
+    #[validate(length(min = 1, max = 128))]
+    icon: Option<String>,
+    remove: Option<RemoveChannelField>,
+}
+
+#[patch("/<target>", data = "<data>")]
+pub async fn req(user: User, target: Ref, data: Json<Data>) -> Result<()> {
+    let data = data.into_inner();
+    data.validate()
+        .map_err(|error| Error::FailedValidation { error })?;
+
+    if data.name.is_none()
+        && data.description.is_none()
+        && data.icon.is_none()
+        && data.remove.is_none()
+    {
+        return Ok(());
+    }
+
+    let target = target.fetch_channel().await?;
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+
+    if !perm.get_manage_channel() {
+        Err(Error::MissingPermission)?
+    }
+
+    match &target {
+        Channel::Group { id, icon, .. }
+        | Channel::TextChannel { id, icon, .. }
+        | Channel::VoiceChannel { id, icon, .. } => {
+            let mut set = doc! {};
+            let mut unset = doc! {};
+
+            let mut remove_icon = false;
+            if let Some(remove) = &data.remove {
+                match remove {
+                    RemoveChannelField::Icon => {
+                        unset.insert("icon", 1);
+                        remove_icon = true;
+                    }
+                    RemoveChannelField::Description => {
+                        unset.insert("description", 1);
+                    }
+                }
+            }
+
+            if let Some(name) = &data.name {
+                set.insert("name", name);
+            }
+
+            if let Some(description) = &data.description {
+                set.insert("description", description);
+            }
+
+            if let Some(attachment_id) = &data.icon {
+                let attachment =
+                    File::find_and_use(&attachment_id, "icons", "object", target.id()).await?;
+                set.insert(
+                    "icon",
+                    to_document(&attachment).map_err(|_| Error::DatabaseError {
+                        operation: "to_document",
+                        with: "attachment",
+                    })?,
+                );
+
+                remove_icon = true;
+            }
+
+            let mut operations = doc! {};
+            if set.len() > 0 {
+                operations.insert("$set", &set);
+            }
+
+            if unset.len() > 0 {
+                operations.insert("$unset", unset);
+            }
+
+            if operations.len() > 0 {
+                get_collection("channels")
+                    .update_one(doc! { "_id": &id }, operations, None)
+                    .await
+                    .map_err(|_| Error::DatabaseError {
+                        operation: "update_one",
+                        with: "channel",
+                    })?;
+            }
+
+            ClientboundNotification::ChannelUpdate {
+                id: id.clone(),
+                data: json!(set),
+                clear: data.remove,
+            }
+            .publish(id.clone());
+
+            if let Channel::Group { .. } = &target {
+                if let Some(name) = data.name {
+                    Content::SystemMessage(SystemMessage::ChannelRenamed {
+                        name,
+                        by: user.id.clone(),
+                    })
+                    .send_as_system(&target)
+                    .await
+                    .ok();
+                }
+
+                if let Some(_) = data.description {
+                    Content::SystemMessage(SystemMessage::ChannelDescriptionChanged {
+                        by: user.id.clone(),
+                    })
+                    .send_as_system(&target)
+                    .await
+                    .ok();
+                }
+
+                if let Some(_) = data.icon {
+                    Content::SystemMessage(SystemMessage::ChannelIconChanged { by: user.id })
+                        .send_as_system(&target)
+                        .await
+                        .ok();
+                }
+            }
+
+            if remove_icon {
+                if let Some(old_icon) = icon {
+                    old_icon.delete().await?;
+                }
+            }
+
+            Ok(())
+        }
+        _ => Err(Error::InvalidOperation),
+    }
+}

src/routes/channels/channel_fetch.rs

+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::MissingPermission)?
+    }
+
+    Ok(json!(target))
+}

src/routes/channels/group_add_member.rs

+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_invite_others() {
+        Err(Error::MissingPermission)?
+    }
+
+    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());
+
+        Content::SystemMessage(SystemMessage::UserAdded {
+            id: member.id,
+            by: user.id,
+        })
+        .send_as_system(&channel)
+        .await
+        .ok();
+
+        Ok(())
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/group_create.rs

+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> {
+    let info = info.into_inner();
+    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 {
+        match get_relationship(&user, target) {
+            RelationshipStatus::Friend | RelationshipStatus::User => {}
+            _ => {
+                return 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),
+        name: info.name,
+        description: info.description,
+        owner: user.id,
+        recipients: set.into_iter().collect::<Vec<String>>(),
+        icon: None,
+        last_message: None,
+        permissions: None
+    };
+
+    channel.clone().publish().await?;
+
+    Ok(json!(channel))
+}

src/routes/channels/group_remove_member.rs

+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::MissingPermission)?
+        }
+
+        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());
+
+        Content::SystemMessage(SystemMessage::UserRemove {
+            id: member.id,
+            by: user.id,
+        })
+        .send_as_system(&channel)
+        .await
+        .ok();
+
+        Ok(())
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/invite_create.rs

+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use mongodb::bson::doc;
+use nanoid::nanoid;
+use rocket_contrib::json::JsonValue;
+
+lazy_static! {
+    static ref ALPHABET: [char; 54] = [
+        '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H',
+        'J', 'K', 'M', 'N', 'P', 'Q', 'R', 'S', 'T', 'V', 'W', 'X', 'Y', 'Z', 'a', 'b', 'c', 'd',
+        'e', 'f', 'g', 'h', 'j', 'k', 'm', 'n', 'p', 'q', 'r', 's', 't', 'v', 'w', 'x', 'y', 'z'
+    ];
+}
+
+#[post("/<target>/invites")]
+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_invite_others() {
+        return Err(Error::MissingPermission);
+    }
+
+    let code = nanoid!(8, &*ALPHABET);
+    match &target {
+        Channel::Group { .. } => {
+            unimplemented!()
+        }
+        Channel::TextChannel { id, server, .. }
+        | Channel::VoiceChannel { id, server, .. } => {
+            Invite::Server {
+                code: code.clone(),
+                creator: user.id,
+                server: server.clone(),
+                channel: id.clone(),
+            }
+            .save()
+            .await?;
+
+            Ok(json!({ "code": code }))
+        }
+        _ => Err(Error::InvalidOperation),
+    }
+}

src/routes/channels/members_fetch.rs

+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use rocket_contrib::json::JsonValue;
+
+#[get("/<target>/members")]
+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::MissingPermission)?
+    }
+
+    if let Channel::Group { recipients, .. } = target {
+        Ok(json!(user.fetch_multiple_users(recipients).await?))
+    } else {
+        Err(Error::InvalidOperation)
+    }
+}

src/routes/channels/message_delete.rs

+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?;
+    channel.has_messaging()?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::MissingPermission)?
+    }
+
+    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

+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use chrono::Utc;
+use mongodb::bson::{doc, Bson, DateTime, Document};
+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?;
+    channel.has_messaging()?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::MissingPermission)?
+    }
+
+    let mut message = msg.fetch_message(&channel).await?;
+    if message.author != user.id {
+        Err(Error::CannotEditMessage)?
+    }
+
+    let edited = Utc::now();
+    let mut set = doc! {
+        "content": &edit.content,
+        "edited": Bson::DateTime(edited)
+    };
+
+    message.content = Content::Text(edit.content.clone());
+    let mut update = json!({ "content": edit.content, "edited": DateTime(edited) });
+
+    if let Some(embeds) = &message.embeds {
+        let new_embeds: Vec<Document> = vec![];
+
+        for embed in embeds {
+            match embed {
+                Embed::Website(_) | Embed::Image(_) | Embed::None => {} // Otherwise push to new_embeds.
+            }
+        }
+
+        let obj = update.as_object_mut().unwrap();
+        obj.insert("embeds".to_string(), json!(new_embeds).0);
+        set.insert("embeds", new_embeds);
+    }
+
+    get_collection("messages")
+        .update_one(
+            doc! {
+                "_id": &message.id
+            },
+            doc! {
+                "$set": set
+            },
+            None,
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "update_one",
+            with: "message",
+        })?;
+
+    message.publish_update(update).await
+}

src/routes/channels/message_fetch.rs

+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?;
+    channel.has_messaging()?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&channel)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::MissingPermission)?
+    }
+
+    let message = msg.fetch_message(&channel).await?;
+    Ok(json!(message))
+}

src/routes/channels/message_query.rs

+use std::collections::HashSet;
+
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use futures::{StreamExt, try_join};
+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(Serialize, Deserialize, FromFormValue)]
+pub enum Sort {
+    Latest,
+    Oldest,
+}
+
+#[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>,
+    sort: Option<Sort>,
+    // Specifying 'nearby' ignores 'before', 'after' and 'sort'.
+    // It will also take half of limit rounded as the limits to each side.
+    // It also fetches the message ID specified.
+    #[validate(length(min = 26, max = 26))]
+    nearby: Option<String>,
+    include_users: Option<bool>,
+}
+
+#[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?;
+    target.has_messaging()?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::MissingPermission)?
+    }
+
+    let mut messages = vec![];
+
+    let collection = get_collection("messages");
+    let limit = options.limit.unwrap_or(50);
+    let channel = target.id();
+    if let Some(nearby) = &options.nearby {
+        let mut cursors = try_join!(
+            collection.find(
+                doc! {
+                    "channel": channel,
+                    "_id": {
+                        "$gte": &nearby
+                    }
+                },
+                FindOptions::builder()
+                    .limit(limit / 2 + 1)
+                    .sort(doc! {
+                        "_id": 1
+                    })
+                    .build(),
+            ),
+            collection.find(
+                doc! {
+                    "channel": channel,
+                    "_id": {
+                        "$lt": &nearby
+                    }
+                },
+                FindOptions::builder()
+                    .limit(limit / 2)
+                    .sort(doc! {
+                        "_id": -1
+                    })
+                    .build(),
+            )
+        )
+        .map_err(|_| Error::DatabaseError {
+            operation: "find",
+            with: "messages",
+        })?;
+
+        while let Some(result) = cursors.0.next().await {
+            if let Ok(doc) = result {
+                messages.push(
+                    from_document::<Message>(doc).map_err(|_| Error::DatabaseError {
+                        operation: "from_document",
+                        with: "message",
+                    })?,
+                );
+            }
+        }
+
+        while let Some(result) = cursors.1.next().await {
+            if let Ok(doc) = result {
+                messages.push(
+                    from_document::<Message>(doc).map_err(|_| Error::DatabaseError {
+                        operation: "from_document",
+                        with: "message",
+                    })?,
+                );
+            }
+        }
+    } else {
+        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 sort: i32 = if let Sort::Latest = options.sort.as_ref().unwrap_or_else(|| &Sort::Latest) {
+            -1
+        } else {
+            1
+        };
+
+        let mut cursor = collection
+            .find(
+                query,
+                FindOptions::builder()
+                    .limit(limit)
+                    .sort(doc! {
+                        "_id": sort
+                    })
+                    .build(),
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "find",
+                with: "messages",
+            })?;
+
+        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",
+                    })?,
+                );
+            }
+        }
+    }
+
+    if options.include_users.unwrap_or_else(|| false) {
+        let mut ids = HashSet::new();
+        for message in &messages {
+            ids.insert(message.author.clone());
+        }
+
+        ids.remove(&user.id);
+        let user_ids = ids.into_iter().collect();
+        let users = user.fetch_multiple_users(user_ids).await?;
+
+        if let Channel::TextChannel { server, .. } = target {
+            Ok(json!({
+                "messages": messages,
+                "users": users,
+                "members": Server::fetch_members(&server).await?
+            }))
+        } else {
+            Ok(json!({
+                "messages": messages,
+                "users": users,
+            }))
+        }
+    } else {
+        Ok(json!(messages))
+    }
+}

src/routes/channels/message_query_stale.rs

+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use futures::StreamExt;
+use mongodb::bson::{doc, from_document};
+use rocket_contrib::json::{Json, JsonValue};
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct Options {
+    ids: Vec<String>,
+}
+
+#[post("/<target>/messages/stale", data = "<data>")]
+pub async fn req(user: User, target: Ref, data: Json<Options>) -> Result<JsonValue> {
+    if data.ids.len() > 150 {
+        return Err(Error::TooManyIds);
+    }
+
+    let target = target.fetch_channel().await?;
+    target.has_messaging()?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::MissingPermission)?
+    }
+
+    let mut cursor = get_collection("messages")
+        .find(
+            doc! {
+                "_id": {
+                    "$in": &data.ids
+                },
+                "channel": target.id()
+            },
+            None,
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find",
+            with: "messages",
+        })?;
+
+    let mut updated = vec![];
+    let mut found_ids = vec![];
+    while let Some(result) = cursor.next().await {
+        if let Ok(doc) = result {
+            let msg = from_document::<Message>(doc).map_err(|_| Error::DatabaseError {
+                operation: "from_document",
+                with: "message",
+            })?;
+
+            found_ids.push(msg.id.clone());
+            if msg.edited.is_some() {
+                updated.push(msg);
+            }
+        }
+    }
+
+    let mut deleted = vec![];
+    for id in &data.ids {
+        if found_ids.iter().find(|x| *x == id).is_none() {
+            deleted.push(id);
+        }
+    }
+
+    Ok(json!({
+        "updated": updated,
+        "deleted": deleted
+    }))
+}

src/routes/channels/message_search.rs

+use std::collections::HashSet;
+
+use crate::database::*;
+use crate::util::result::{Error, Result};
+
+use futures::StreamExt;
+use mongodb::{
+    bson::{doc, from_document},
+    options::FindOptions,
+};
+use rocket_contrib::json::{Json, JsonValue};
+use serde::{Deserialize, Serialize};
+use validator::Validate;
+
+#[derive(Serialize, Deserialize, FromFormValue)]
+pub enum Sort {
+    Relevance,
+    Latest,
+    Oldest,
+}
+
+impl Default for Sort {
+    fn default() -> Sort {
+        Sort::Relevance
+    }
+}
+
+#[derive(Validate, Serialize, Deserialize, FromForm)]
+pub struct Options {
+    #[validate(length(min = 1, max = 64))]
+    query: String,
+
+    #[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>,
+    #[serde(default = "Sort::default")]
+    sort: Sort,
+    include_users: Option<bool>,
+}
+
+#[post("/<target>/search", data = "<options>")]
+pub async fn req(user: User, target: Ref, options: Json<Options>) -> Result<JsonValue> {
+    options
+        .validate()
+        .map_err(|error| Error::FailedValidation { error })?;
+
+    let target = target.fetch_channel().await?;
+    target.has_messaging()?;
+
+    let perm = permissions::PermissionCalculator::new(&user)
+        .with_channel(&target)
+        .for_channel()
+        .await?;
+    if !perm.get_view() {
+        Err(Error::MissingPermission)?
+    }
+
+    let mut messages = vec![];
+    let limit = options.limit.unwrap_or(50);
+
+    let mut filter = doc! {
+        "channel": target.id(),
+        "$text": {
+            "$search": &options.query
+        }
+    };
+
+    if let Some(doc) = match (&options.before, &options.after) {
+        (Some(before), Some(after)) => Some(doc! {
+            "lt": before,
+            "gt": after
+        }),
+        (Some(before), _) => Some(doc! {
+            "lt": before
+        }),
+        (_, Some(after)) => Some(doc! {
+            "gt": after
+        }),
+        _ => None
+    } {
+        filter.insert("_id", doc);
+    }
+
+    let mut cursor = get_collection("messages")
+        .find(
+            filter,
+            FindOptions::builder()
+                .projection(
+                    if let Sort::Relevance = &options.sort {
+                        doc! {
+                            "score": {
+                                "$meta": "textScore"
+                            }
+                        }
+                    } else {
+                        doc! {}
+                    }
+                )
+                .limit(limit)
+                .sort(
+                    match &options.sort {
+                        Sort::Relevance => doc! {
+                            "score": {
+                                "$meta": "textScore"
+                            }
+                        },
+                        Sort::Latest => doc! {
+                            "_id": -1
+                        },
+                        Sort::Oldest => doc! {
+                            "_id": 1
+                        }
+                    }
+                )
+                .build(),
+        )
+        .await
+        .map_err(|_| Error::DatabaseError {
+            operation: "find",
+            with: "messages",
+        })?;
+    
+    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",
+                })?,
+            );
+        }
+    }
+
+    if options.include_users.unwrap_or_else(|| false) {
+        let mut ids = HashSet::new();
+        for message in &messages {
+            ids.insert(message.author.clone());
+        }
+
+        ids.remove(&user.id);
+        let user_ids = ids.into_iter().collect();
+        let users = user.fetch_multiple_users(user_ids).await?;
+
+        if let Channel::TextChannel { server, .. } = target {
+            Ok(json!({
+                "messages": messages,
+                "users": users,
+                "members": Server::fetch_members(&server).await?
+            }))
+        } else {
+            Ok(json!({
+                "messages": messages,
+                "users": users,
+            }))
+        }
+    } else {
+        Ok(json!(messages))
+    }
+}