refactor(bonfire): remove quark dependency

Cargo.lock

@@ -3588,14 +3588,21 @@ dependencies = [
  "fred",
  "futures",
  "log",
+ "lru 0.7.6",
  "once_cell",
  "querystring",
+ "redis-kiss",
+ "revolt-config",
+ "revolt-database",
+ "revolt-models",
+ "revolt-permissions",
  "revolt-presence",
- "revolt-quark",
+ "revolt-result",
  "rmp-serde",
  "sentry",
  "serde",
  "serde_json",
+ "ulid 0.5.0",
 ]
 
 [[package]]

crates/bonfire/Cargo.toml

@@ -9,14 +9,15 @@ edition = "2021"
 [dependencies]
 # util
 log = "*"
+sentry = "0.31.5"
+lru = "0.7.6"
+ulid = "0.5.0"
 once_cell = "1.9.0"
+redis-kiss = "0.1.4"
 
 # parsing
 querystring = "1.1.0"
 
-# quark
-revolt-quark = { path = "../quark" }
-
 # serde
 bincode = "1.3.3"
 serde_json = "1.0.79"
@@ -33,8 +34,12 @@ async-std = { version = "1.8.0", features = [
 ] }
 
 # core
+revolt-result = { path = "../core/result" }
+revolt-models = { path = "../core/models" }
+revolt-config = { path = "../core/config" }
+revolt-database = { path = "../core/database" }
+revolt-permissions = { version = "0.7.0", path = "../core/permissions" }
 revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
-sentry = "0.31.5"
 
 # redis
 fred = { version = "8.0.1", features = ["subscriber-client"] }

crates/bonfire/src/config.rs

@@ -1,6 +1,6 @@
 use async_tungstenite::tungstenite::{handshake, Message};
 use futures::channel::oneshot::Sender;
-use revolt_quark::{Error, Result};
+use revolt_result::{create_error, Result};
 use serde::{Deserialize, Serialize};
 
 /// Enumeration of supported protocol formats
@@ -37,16 +37,16 @@ impl ProtocolConfiguration {
         match self.format {
             ProtocolFormat::Json => {
                 if let Message::Text(text) = msg {
-                    serde_json::from_str(text).map_err(|_| Error::InternalError)
+                    serde_json::from_str(text).map_err(|_| create_error!(InternalError))
                 } else {
-                    Err(Error::InternalError)
+                    Err(create_error!(InternalError))
                 }
             }
             ProtocolFormat::Msgpack => {
                 if let Message::Binary(buf) = msg {
-                    rmp_serde::from_slice(buf).map_err(|_| Error::InternalError)
+                    rmp_serde::from_slice(buf).map_err(|_| create_error!(InternalError))
                 } else {
-                    Err(Error::InternalError)
+                    Err(create_error!(InternalError))
                 }
             }
         }

crates/bonfire/src/database.rs

@@ -1,5 +1,5 @@
 use once_cell::sync::OnceCell;
-use revolt_quark::{Database, DatabaseInfo};
+use revolt_database::{Database, DatabaseInfo};
 
 static DBCONN: OnceCell<Database> = OnceCell::new();
 
@@ -10,7 +10,9 @@ pub async fn connect() {
         .await
         .expect("Failed to connect to the database.");
 
-    DBCONN.set(database).expect("Setting `Database`");
+    if DBCONN.set(database).is_err() {
+        panic!("couldn't set database")
+    }
 }
 
 /// Get a reference to the current database.

crates/bonfire/src/events/impl.rs

@@ -0,0 +1,552 @@
+use std::collections::HashSet;
+
+use revolt_database::{
+    events::client::EventV1, util::permissions::DatabasePermissionQuery, Channel, Database, Member,
+    MemberCompositeKey, Presence, RelationshipStatus,
+};
+use revolt_models::v0;
+use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
+use revolt_presence::filter_online;
+use revolt_result::Result;
+
+use super::state::{Cache, State};
+
+/// Cache Manager
+impl Cache {
+    /// Check whether the current user can view a channel
+    pub async fn can_view_channel(&self, db: &Database, channel: &Channel) -> bool {
+        match &channel {
+            Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
+                let member = self.members.get(server);
+                let server = self.servers.get(server);
+                let mut query =
+                    DatabasePermissionQuery::new(db, self.users.get(&self.user_id).unwrap())
+                        .channel(&channel);
+                // let mut perms = perms(self.users.get(&self.user_id).unwrap()).channel(channel);
+
+                if let Some(member) = member {
+                    query = query.member(&member);
+                }
+
+                if let Some(server) = server {
+                    query = query.server(server);
+                }
+
+                calculate_channel_permissions(&mut query)
+                    .await
+                    .has_channel_permission(ChannelPermission::ViewChannel)
+            }
+            _ => true,
+        }
+    }
+
+    /// Filter a given vector of channels to only include the ones we can access
+    pub async fn filter_accessible_channels(
+        &self,
+        db: &Database,
+        channels: Vec<Channel>,
+    ) -> Vec<Channel> {
+        let mut viewable_channels = vec![];
+        for channel in channels {
+            if self.can_view_channel(db, &channel).await {
+                viewable_channels.push(channel);
+            }
+        }
+
+        viewable_channels
+    }
+
+    /// Check whether we can subscribe to another user
+    pub fn can_subscribe_to_user(&self, user_id: &str) -> bool {
+        if let Some(user) = self.users.get(&self.user_id) {
+            match user.relationship_with(user_id) {
+                RelationshipStatus::Friend
+                | RelationshipStatus::Incoming
+                | RelationshipStatus::Outgoing
+                | RelationshipStatus::User => true,
+                _ => {
+                    let user_id = &user_id.to_string();
+                    for channel in self.channels.values() {
+                        match channel {
+                            Channel::DirectMessage { recipients, .. }
+                            | Channel::Group { recipients, .. } => {
+                                if recipients.contains(user_id) {
+                                    return true;
+                                }
+                            }
+                            _ => {}
+                        }
+                    }
+
+                    false
+                }
+            }
+        } else {
+            false
+        }
+    }
+}
+
+/// State Manager
+impl State {
+    /// Generate a Ready packet for the current user
+    pub async fn generate_ready_payload(&mut self, db: &Database) -> Result<EventV1> {
+        let user = self.clone_user();
+
+        // Find all relationships to the user.
+        let mut user_ids: HashSet<String> = user
+            .relations
+            .as_ref()
+            .map(|arr| arr.iter().map(|x| x.id.to_string()).collect())
+            .unwrap_or_default();
+
+        // Fetch all memberships with their corresponding servers.
+        let members: Vec<Member> = db.fetch_all_memberships(&user.id).await?;
+        self.cache.members = members
+            .iter()
+            .cloned()
+            .map(|x| (x.id.server.clone(), x))
+            .collect();
+
+        let server_ids: Vec<String> = members.iter().map(|x| x.id.server.clone()).collect();
+        let servers = db.fetch_servers(&server_ids).await?;
+        self.cache.servers = servers.iter().cloned().map(|x| (x.id.clone(), x)).collect();
+
+        // Collect channel ids from servers.
+        let mut channel_ids = vec![];
+        for server in &servers {
+            channel_ids.append(&mut server.channels.clone());
+        }
+
+        // Fetch DMs and server channels.
+        let mut channels = db.find_direct_messages(&user.id).await?;
+        channels.append(&mut db.fetch_channels(&channel_ids).await?);
+
+        // Filter server channels by permission.
+        let channels = self.cache.filter_accessible_channels(db, channels).await;
+
+        // Append known user IDs from DMs.
+        for channel in &channels {
+            match channel {
+                Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
+                    user_ids.extend(&mut recipients.clone().into_iter());
+                }
+                _ => {}
+            }
+        }
+
+        // Fetch presence data for known users.
+        let online_ids = filter_online(&user_ids.iter().cloned().collect::<Vec<String>>()).await;
+
+        // Fetch user data.
+        let users = db
+            .fetch_users(
+                &user_ids
+                    .into_iter()
+                    .filter(|x| x != &user.id)
+                    .collect::<Vec<String>>(),
+            )
+            .await?;
+
+        // Fetch customisations.
+        let emojis = db
+            .fetch_emoji_by_parent_ids(
+                &servers
+                    .iter()
+                    .map(|x| x.id.to_string())
+                    .collect::<Vec<String>>(),
+            )
+            .await?;
+
+        // Copy data into local state cache.
+        self.cache.users = users.iter().cloned().map(|x| (x.id.clone(), x)).collect();
+        self.cache
+            .users
+            .insert(self.cache.user_id.clone(), user.clone());
+        self.cache.channels = channels
+            .iter()
+            .cloned()
+            .map(|x| (x.id().to_string(), x))
+            .collect();
+
+        // Make all users appear from our perspective.
+        let mut users: Vec<v0::User> = users
+            .into_iter()
+            .map(|other_user| {
+                let is_online = online_ids.contains(&other_user.id);
+                other_user.into_known(&user, is_online)
+            })
+            .collect();
+
+        // Make sure we see our own user correctly.
+        users.push(user.into_self().await);
+
+        // Set subscription state internally.
+        self.reset_state();
+        self.insert_subscription(self.private_topic.clone());
+
+        for user in &users {
+            self.insert_subscription(user.id.clone());
+        }
+
+        for server in &servers {
+            self.insert_subscription(server.id.clone());
+        }
+
+        for channel in &channels {
+            self.insert_subscription(channel.id().to_string());
+        }
+
+        Ok(EventV1::Ready {
+            users,
+            servers: servers.into_iter().map(Into::into).collect(),
+            channels: channels.into_iter().map(Into::into).collect(),
+            members: members.into_iter().map(Into::into).collect(),
+            emojis: emojis.into_iter().map(Into::into).collect(),
+        })
+    }
+
+    /// Re-determine the currently accessible server channels
+    pub async fn recalculate_server(&mut self, db: &Database, id: &str, event: &mut EventV1) {
+        if let Some(server) = self.cache.servers.get(id) {
+            let mut channel_ids = HashSet::new();
+            let mut added_channels = vec![];
+            let mut removed_channels = vec![];
+
+            let id = &id.to_string();
+            for (channel_id, channel) in &self.cache.channels {
+                match channel {
+                    Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
+                        if server == id {
+                            channel_ids.insert(channel_id.clone());
+
+                            if self.cache.can_view_channel(db, channel).await {
+                                added_channels.push(channel_id.clone());
+                            } else {
+                                removed_channels.push(channel_id.clone());
+                            }
+                        }
+                    }
+                    _ => {}
+                }
+            }
+
+            let known_ids = server.channels.iter().cloned().collect::<HashSet<String>>();
+
+            let mut bulk_events = vec![];
+
+            for id in added_channels {
+                self.insert_subscription(id);
+            }
+
+            for id in removed_channels {
+                self.remove_subscription(&id);
+                self.cache.channels.remove(&id);
+
+                bulk_events.push(EventV1::ChannelDelete { id });
+            }
+
+            // * NOTE: currently all channels should be cached
+            // * provided that a server was loaded from payload
+            let unknowns = known_ids
+                .difference(&channel_ids)
+                .cloned()
+                .collect::<Vec<String>>();
+
+            if !unknowns.is_empty() {
+                if let Ok(channels) = db.fetch_channels(&unknowns).await {
+                    let viewable_channels =
+                        self.cache.filter_accessible_channels(db, channels).await;
+
+                    for channel in viewable_channels {
+                        self.cache
+                            .channels
+                            .insert(channel.id().to_string(), channel.clone());
+
+                        self.insert_subscription(channel.id().to_string());
+                        bulk_events.push(EventV1::ChannelCreate(channel.into()));
+                    }
+                }
+            }
+
+            if !bulk_events.is_empty() {
+                let mut new_event = EventV1::Bulk { v: bulk_events };
+                std::mem::swap(&mut new_event, event);
+
+                if let EventV1::Bulk { v } = event {
+                    v.push(new_event);
+                }
+            }
+        }
+    }
+
+    /// Push presence change to the user and all associated server topics
+    pub async fn broadcast_presence_change(&self, target: bool) {
+        if if let Some(status) = &self.cache.users.get(&self.cache.user_id).unwrap().status {
+            status.presence != Some(Presence::Invisible)
+        } else {
+            true
+        } {
+            let event = EventV1::UserUpdate {
+                id: self.cache.user_id.clone(),
+                data: v0::PartialUser {
+                    online: Some(target),
+                    ..Default::default()
+                },
+                clear: vec![],
+                event_id: Some(ulid::Ulid::new().to_string()),
+            };
+
+            for server in self.cache.servers.keys() {
+                event.clone().p(server.clone()).await;
+            }
+
+            event.p(self.cache.user_id.clone()).await;
+        }
+    }
+
+    /// Handle an incoming event for protocol version 1
+    pub async fn handle_incoming_event_v1(&mut self, db: &Database, event: &mut EventV1) -> bool {
+        /* Superseded by private topics.
+          if match event {
+            EventV1::UserRelationship { id, .. }
+            | EventV1::UserSettingsUpdate { id, .. }
+            | EventV1::ChannelAck { id, .. } => id != &self.cache.user_id,
+            EventV1::ServerCreate { server, .. } => server.owner != self.cache.user_id,
+            EventV1::ChannelCreate(channel) => match channel {
+                Channel::SavedMessages { user, .. } => user != &self.cache.user_id,
+                Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
+                    !recipients.contains(&self.cache.user_id)
+                }
+                _ => false,
+            },
+            _ => false,
+        } {
+            return false;
+        }*/
+
+        // An event may trigger recalculation of an entire server's permission.
+        // Keep track of whether we need to do anything.
+        let mut queue_server = None;
+
+        // It may also need to sub or unsub a single value.
+        let mut queue_add = None;
+        let mut queue_remove = None;
+
+        match event {
+            EventV1::ChannelCreate(channel) => {
+                let id = channel.id().to_string();
+                self.insert_subscription(id.clone());
+                self.cache.channels.insert(id, channel.clone().into());
+            }
+            EventV1::ChannelUpdate {
+                id, data, clear, ..
+            } => {
+                let could_view: bool = if let Some(channel) = self.cache.channels.get(id) {
+                    self.cache.can_view_channel(db, channel).await
+                } else {
+                    false
+                };
+
+                if let Some(channel) = self.cache.channels.get_mut(id) {
+                    for field in clear {
+                        channel.remove_field(&field.clone().into());
+                    }
+
+                    channel.apply_options(data.clone().into());
+                }
+
+                if !self.cache.channels.contains_key(id) {
+                    if let Ok(channel) = db.fetch_channel(id).await {
+                        self.cache.channels.insert(id.clone(), channel);
+                    }
+                }
+
+                if let Some(channel) = self.cache.channels.get(id) {
+                    let can_view = self.cache.can_view_channel(db, channel).await;
+                    if could_view != can_view {
+                        if can_view {
+                            queue_add = Some(id.clone());
+                            *event = EventV1::ChannelCreate(channel.clone().into());
+                        } else {
+                            queue_remove = Some(id.clone());
+                            *event = EventV1::ChannelDelete { id: id.clone() };
+                        }
+                    }
+                }
+            }
+            EventV1::ChannelDelete { id } => {
+                self.remove_subscription(id);
+                self.cache.channels.remove(id);
+            }
+            EventV1::ChannelGroupJoin { user, .. } => {
+                self.insert_subscription(user.clone());
+            }
+            EventV1::ChannelGroupLeave { id, user, .. } => {
+                if user == &self.cache.user_id {
+                    self.remove_subscription(id);
+                } else if !self.cache.can_subscribe_to_user(user) {
+                    self.remove_subscription(user);
+                }
+            }
+
+            EventV1::ServerCreate {
+                id,
+                server,
+                channels,
+                emojis: _,
+            } => {
+                self.insert_subscription(id.clone());
+                self.cache.servers.insert(id.clone(), server.clone().into());
+                let member = Member {
+                    id: MemberCompositeKey {
+                        server: server.id.clone(),
+                        user: self.cache.user_id.clone(),
+                    },
+                    ..Default::default()
+                };
+                self.cache.members.insert(id.clone(), member);
+
+                for channel in channels {
+                    self.cache
+                        .channels
+                        .insert(channel.id().to_string(), channel.clone().into());
+                }
+
+                queue_server = Some(id.clone());
+            }
+            EventV1::ServerUpdate {
+                id, data, clear, ..
+            } => {
+                if let Some(server) = self.cache.servers.get_mut(id) {
+                    for field in clear {
+                        server.remove_field(&field.clone().into());
+                    }
+
+                    server.apply_options(data.clone().into());
+                }
+
+                if data.default_permissions.is_some() {
+                    queue_server = Some(id.clone());
+                }
+            }
+            EventV1::ServerMemberJoin { .. } => {
+                // We will always receive ServerCreate when joining a new server.
+            }
+            EventV1::ServerMemberLeave { id, user } => {
+                if user == &self.cache.user_id {
+                    self.remove_subscription(id);
+
+                    if let Some(server) = self.cache.servers.remove(id) {
+                        for channel in &server.channels {
+                            self.remove_subscription(channel);
+                            self.cache.channels.remove(channel);
+                        }
+                    }
+                    self.cache.members.remove(id);
+                }
+            }
+            EventV1::ServerDelete { id } => {
+                self.remove_subscription(id);
+
+                if let Some(server) = self.cache.servers.remove(id) {
+                    for channel in &server.channels {
+                        self.remove_subscription(channel);
+                        self.cache.channels.remove(channel);
+                    }
+                }
+                self.cache.members.remove(id);
+            }
+            EventV1::ServerMemberUpdate { id, data, clear } => {
+                if id.user == self.cache.user_id {
+                    if let Some(member) = self.cache.members.get_mut(&id.server) {
+                        for field in &clear.clone() {
+                            member.remove_field(&field.clone().into());
+                        }
+
+                        member.apply_options(data.clone().into());
+                    }
+
+                    if data.roles.is_some() || clear.contains(&v0::FieldsMember::Roles) {
+                        queue_server = Some(id.server.clone());
+                    }
+                }
+            }
+            EventV1::ServerRoleUpdate {
+                id,
+                role_id,
+                data,
+                clear,
+                ..
+            } => {
+                if let Some(server) = self.cache.servers.get_mut(id) {
+                    if let Some(role) = server.roles.get_mut(role_id) {
+                        for field in &clear.clone() {
+                            role.remove_field(&field.clone().into());
+                        }
+
+                        role.apply_options(data.clone().into());
+                    }
+                }
+
+                if data.rank.is_some() || data.permissions.is_some() {
+                    if let Some(member) = self.cache.members.get(id) {
+                        if member.roles.contains(role_id) {
+                            queue_server = Some(id.clone());
+                        }
+                    }
+                }
+            }
+            EventV1::ServerRoleDelete { id, role_id } => {
+                if let Some(server) = self.cache.servers.get_mut(id) {
+                    server.roles.remove(role_id);
+                }
+
+                if let Some(member) = self.cache.members.get(id) {
+                    if member.roles.contains(role_id) {
+                        queue_server = Some(id.clone());
+                    }
+                }
+            }
+
+            EventV1::UserUpdate { event_id, .. } => {
+                if let Some(id) = event_id {
+                    if self.cache.seen_events.contains(id) {
+                        return false;
+                    }
+
+                    self.cache.seen_events.put(id.to_string(), ());
+                }
+
+                *event_id = None;
+            }
+            EventV1::UserRelationship { id, user, .. } => {
+                self.cache.users.insert(id.clone(), user.clone().into());
+
+                if self.cache.can_subscribe_to_user(id) {
+                    self.insert_subscription(id.clone());
+                } else {
+                    self.remove_subscription(id);
+                }
+            }
+
+            _ => {}
+        }
+
+        // Calculate server permissions if requested.
+        if let Some(server_id) = queue_server {
+            self.recalculate_server(db, &server_id, event).await;
+        }
+
+        // Sub / unsub accordingly.
+        if let Some(id) = queue_add {
+            self.insert_subscription(id);
+        }
+
+        if let Some(id) = queue_remove {
+            self.remove_subscription(&id);
+        }
+
+        true
+    }
+}

crates/bonfire/src/events/mod.rs

@@ -0,0 +1,2 @@
+pub mod r#impl;
+pub mod state;

crates/bonfire/src/events/state.rs

@@ -0,0 +1,164 @@
+use std::collections::{HashMap, HashSet};
+
+use lru::LruCache;
+use revolt_database::{Channel, Member, Server, User};
+
+/// Enumeration representing some change in subscriptions
+pub enum SubscriptionStateChange {
+    /// No change
+    None,
+    /// Clear all subscriptions
+    Reset,
+    /// Append or remove subscriptions
+    Change {
+        add: Vec<String>,
+        remove: Vec<String>,
+    },
+}
+
+/// Dumb per-state cache implementation
+///
+/// Ideally this would use a global cache that
+/// allows for mutations and could use Rc<> to
+/// track usage. If Rc<> == 1, then it only
+/// remains in global cache, hence should be
+/// dropped.
+///
+/// ------------------------------------------------
+/// We can strip these objects to core information!!
+/// ------------------------------------------------
+#[derive(Debug)]
+pub struct Cache {
+    pub user_id: String,
+
+    pub users: HashMap<String, User>,
+    pub channels: HashMap<String, Channel>,
+    pub members: HashMap<String, Member>,
+    pub servers: HashMap<String, Server>,
+
+    pub seen_events: LruCache<String, ()>,
+}
+
+impl Default for Cache {
+    fn default() -> Self {
+        Cache {
+            user_id: Default::default(),
+
+            users: Default::default(),
+            channels: Default::default(),
+            members: Default::default(),
+            servers: Default::default(),
+
+            seen_events: LruCache::new(20),
+        }
+    }
+}
+
+/// Client state
+pub struct State {
+    pub cache: Cache,
+
+    pub private_topic: String,
+    subscribed: HashSet<String>,
+    state: SubscriptionStateChange,
+}
+
+impl State {
+    /// Create state from User
+    pub fn from(user: User) -> State {
+        let mut subscribed = HashSet::new();
+        let private_topic = format!("{}!", user.id);
+        subscribed.insert(private_topic.clone());
+        subscribed.insert(user.id.clone());
+
+        let mut cache: Cache = Cache {
+            user_id: user.id.clone(),
+            ..Default::default()
+        };
+
+        cache.users.insert(user.id.clone(), user);
+
+        State {
+            cache,
+            subscribed,
+            private_topic,
+            state: SubscriptionStateChange::Reset,
+        }
+    }
+
+    /// Apply currently queued state
+    pub fn apply_state(&mut self) -> SubscriptionStateChange {
+        let state = std::mem::replace(&mut self.state, SubscriptionStateChange::None);
+        if let SubscriptionStateChange::Change { add, remove } = &state {
+            for id in add {
+                self.subscribed.insert(id.clone());
+            }
+
+            for id in remove {
+                self.subscribed.remove(id);
+            }
+        }
+
+        state
+    }
+
+    /// Clone the active user
+    pub fn clone_user(&self) -> User {
+        self.cache.users.get(&self.cache.user_id).unwrap().clone()
+    }
+
+    /// Iterate through all subscriptions
+    pub fn iter_subscriptions(&self) -> std::collections::hash_set::Iter<'_, std::string::String> {
+        self.subscribed.iter()
+    }
+
+    /// Reset the current state
+    pub fn reset_state(&mut self) {
+        self.state = SubscriptionStateChange::Reset;
+        self.subscribed.clear();
+    }
+
+    /// Add a new subscription
+    pub fn insert_subscription(&mut self, subscription: String) {
+        if self.subscribed.contains(&subscription) {
+            return;
+        }
+
+        match &mut self.state {
+            SubscriptionStateChange::None => {
+                self.state = SubscriptionStateChange::Change {
+                    add: vec![subscription.clone()],
+                    remove: vec![],
+                };
+            }
+            SubscriptionStateChange::Change { add, .. } => {
+                add.push(subscription.clone());
+            }
+            SubscriptionStateChange::Reset => {}
+        }
+
+        self.subscribed.insert(subscription);
+    }
+
+    /// Remove existing subscription
+    pub fn remove_subscription(&mut self, subscription: &str) {
+        if !self.subscribed.contains(&subscription.to_string()) {
+            return;
+        }
+
+        match &mut self.state {
+            SubscriptionStateChange::None => {
+                self.state = SubscriptionStateChange::Change {
+                    add: vec![],
+                    remove: vec![subscription.to_string()],
+                };
+            }
+            SubscriptionStateChange::Change { remove, .. } => {
+                remove.push(subscription.to_string());
+            }
+            SubscriptionStateChange::Reset => panic!("Should not remove during a reset!"),
+        }
+
+        self.subscribed.remove(subscription);
+    }
+}

crates/bonfire/src/main.rs

@@ -7,6 +7,7 @@ use revolt_presence::clear_region;
 extern crate log;
 
 pub mod config;
+pub mod events;
 
 mod database;
 mod websocket;
@@ -14,7 +15,7 @@ mod websocket;
 #[async_std::main]
 async fn main() {
     // Configure requirements for Bonfire.
-    revolt_quark::configure!();
+    revolt_config::configure!();
     database::connect().await;
 
     // Clean up the current region information.

crates/bonfire/src/websocket.rs

@@ -11,21 +11,17 @@ use futures::{
     stream::{SplitSink, SplitStream},
     FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
-use revolt_presence::{create_session, delete_session};
-use revolt_quark::{
-    events::{
-        client::EventV1,
-        server::ClientMessage,
-        state::{State, SubscriptionStateChange},
-    },
-    models::{user::UserHint, User},
-    redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI},
-    Database,
+use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
+use revolt_database::{
+    events::{client::EventV1, server::ClientMessage},
+    Database, User, UserHint,
 };
+use revolt_presence::{create_session, delete_session};
 
 use async_std::{net::TcpStream, sync::Mutex};
 
 use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
+use crate::events::state::{State, SubscriptionStateChange};
 
 type WsReader = SplitStream<WebSocketStream<TcpStream>>;
 type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;

crates/core/database/src/events/client.rs

@@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize};
 
 use revolt_models::v0::{
     AppendMessage, Channel, Emoji, FieldsChannel, FieldsMember, FieldsRole, FieldsServer,
-    FieldsUser, FieldsWebhook, MemberCompositeKey, Message, PartialChannel, PartialMember,
+    FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialMember,
     PartialMessage, PartialRole, PartialServer, PartialUser, PartialWebhook, Report, Server, User,
     UserSettings, Webhook,
 };
@@ -48,17 +48,17 @@ pub enum EventV1 {
 
     /// Successfully authenticated
     Authenticated,
-    /* /// Basic data to cache
+    /// Basic data to cache
     Ready {
         users: Vec<User>,
         servers: Vec<Server>,
         channels: Vec<Channel>,
         members: Vec<Member>,
-        emojis: Option<Vec<Emoji>>,
+        emojis: Vec<Emoji>,
     },
 
     /// Ping response
-    Pong { data: Ping }, */
+    Pong { data: Ping },
     /// New message
     Message(Message),
 

crates/core/database/src/events/mod.rs

@@ -1 +1,2 @@
 pub mod client;
+pub mod server;

crates/core/database/src/events/server.rs

@@ -0,0 +1,12 @@
+use serde::Deserialize;
+
+use super::client::Ping;
+
+#[derive(Deserialize, Debug)]
+#[serde(tag = "type")]
+pub enum ClientMessage {
+    Authenticate { token: String },
+    BeginTyping { channel: String },
+    EndTyping { channel: String },
+    Ping { data: Ping, responded: Option<()> },
+}

crates/core/database/src/models/users/model.rs

@@ -122,6 +122,16 @@ auto_derived!(
         /// Id of the owner of this bot
         pub owner: String,
     }
+
+    /// Enumeration providing a hint to the type of user we are handling
+    pub enum UserHint {
+        /// Could be either a user or a bot
+        Any,
+        /// Only match bots
+        Bot,
+        /// Only match users
+        User,
+    }
 );
 
 pub static DISCRIMINATOR_SEARCH_SPACE: Lazy<HashSet<String>> = Lazy::new(|| {
@@ -266,6 +276,30 @@ impl User {
         Ok(username)
     }
 
+    /// Find a user from a given token and hint
+    #[async_recursion]
+    pub async fn from_token(db: &Database, token: &str, hint: UserHint) -> Result<User> {
+        match hint {
+            UserHint::Bot => {
+                db.fetch_user(
+                    &db.fetch_bot_by_token(token)
+                        .await
+                        .map_err(|_| create_error!(InternalError))?
+                        .id,
+                )
+                .await
+            }
+            UserHint::User => db.fetch_user_by_token(token).await,
+            UserHint::Any => {
+                if let Ok(user) = User::from_token(db, token, UserHint::User).await {
+                    Ok(user)
+                } else {
+                    User::from_token(db, token, UserHint::Bot).await
+                }
+            }
+        }
+    }
+
     /// Helper function to fetch many users as a mutually connected user
     /// (while optimising the online ID query)
     pub async fn fetch_many_ids_as_mutuals(

crates/core/database/src/util/bridge/v0.rs

@@ -220,6 +220,86 @@ impl From<crate::Channel> for Channel {
     }
 }
 
+impl From<Channel> for crate::Channel {
+    fn from(value: Channel) -> crate::Channel {
+        match value {
+            Channel::SavedMessages { id, user } => crate::Channel::SavedMessages { id, user },
+            Channel::DirectMessage {
+                id,
+                active,
+                recipients,
+                last_message_id,
+            } => crate::Channel::DirectMessage {
+                id,
+                active,
+                recipients,
+                last_message_id,
+            },
+            Channel::Group {
+                id,
+                name,
+                owner,
+                description,
+                recipients,
+                icon,
+                last_message_id,
+                permissions,
+                nsfw,
+            } => crate::Channel::Group {
+                id,
+                name,
+                owner,
+                description,
+                recipients,
+                icon: icon.map(|file| file.into()),
+                last_message_id,
+                permissions,
+                nsfw,
+            },
+            Channel::TextChannel {
+                id,
+                server,
+                name,
+                description,
+                icon,
+                last_message_id,
+                default_permissions,
+                role_permissions,
+                nsfw,
+            } => crate::Channel::TextChannel {
+                id,
+                server,
+                name,
+                description,
+                icon: icon.map(|file| file.into()),
+                last_message_id,
+                default_permissions,
+                role_permissions,
+                nsfw,
+            },
+            Channel::VoiceChannel {
+                id,
+                server,
+                name,
+                description,
+                icon,
+                default_permissions,
+                role_permissions,
+                nsfw,
+            } => crate::Channel::VoiceChannel {
+                id,
+                server,
+                name,
+                description,
+                icon: icon.map(|file| file.into()),
+                default_permissions,
+                role_permissions,
+                nsfw,
+            },
+        }
+    }
+}
+
 impl From<crate::PartialChannel> for PartialChannel {
     fn from(value: crate::PartialChannel) -> Self {
         PartialChannel {
@@ -237,6 +317,23 @@ impl From<crate::PartialChannel> for PartialChannel {
     }
 }
 
+impl From<PartialChannel> for crate::PartialChannel {
+    fn from(value: PartialChannel) -> crate::PartialChannel {
+        crate::PartialChannel {
+            name: value.name,
+            owner: value.owner,
+            description: value.description,
+            icon: value.icon.map(|file| file.into()),
+            nsfw: value.nsfw,
+            active: value.active,
+            permissions: value.permissions,
+            role_permissions: value.role_permissions,
+            default_permissions: value.default_permissions,
+            last_message_id: value.last_message_id,
+        }
+    }
+}
+
 impl From<FieldsChannel> for crate::FieldsChannel {
     fn from(value: FieldsChannel) -> Self {
         match value {
@@ -307,6 +404,25 @@ impl From<crate::File> for File {
     }
 }
 
+impl From<File> for crate::File {
+    fn from(value: File) -> crate::File {
+        crate::File {
+            id: value.id,
+            tag: value.tag,
+            filename: value.filename,
+            metadata: value.metadata.into(),
+            content_type: value.content_type,
+            size: value.size,
+            deleted: value.deleted,
+            reported: value.reported,
+            message_id: value.message_id,
+            user_id: value.user_id,
+            server_id: value.server_id,
+            object_id: value.object_id,
+        }
+    }
+}
+
 impl From<crate::Metadata> for Metadata {
     fn from(value: crate::Metadata) -> Self {
         match value {
@@ -325,6 +441,24 @@ impl From<crate::Metadata> for Metadata {
     }
 }
 
+impl From<Metadata> for crate::Metadata {
+    fn from(value: Metadata) -> crate::Metadata {
+        match value {
+            Metadata::File => crate::Metadata::File,
+            Metadata::Text => crate::Metadata::Text,
+            Metadata::Image { width, height } => crate::Metadata::Image {
+                width: width as isize,
+                height: height as isize,
+            },
+            Metadata::Video { width, height } => crate::Metadata::Video {
+                width: width as isize,
+                height: height as isize,
+            },
+            Metadata::Audio => crate::Metadata::Audio,
+        }
+    }
+}
+
 impl From<crate::Message> for Message {
     fn from(value: crate::Message) -> Self {
         Message {
@@ -480,6 +614,19 @@ impl From<crate::Member> for Member {
     }
 }
 
+impl From<Member> for crate::Member {
+    fn from(value: Member) -> crate::Member {
+        crate::Member {
+            id: value.id.into(),
+            joined_at: value.joined_at,
+            nickname: value.nickname,
+            avatar: value.avatar.map(|f| f.into()),
+            roles: value.roles,
+            timeout: value.timeout,
+        }
+    }
+}
+
 impl From<crate::PartialMember> for PartialMember {
     fn from(value: crate::PartialMember) -> Self {
         PartialMember {
@@ -493,6 +640,19 @@ impl From<crate::PartialMember> for PartialMember {
     }
 }
 
+impl From<PartialMember> for crate::PartialMember {
+    fn from(value: PartialMember) -> crate::PartialMember {
+        crate::PartialMember {
+            id: value.id.map(|id| id.into()),
+            joined_at: value.joined_at,
+            nickname: value.nickname,
+            avatar: value.avatar.map(|f| f.into()),
+            roles: value.roles,
+            timeout: value.timeout,
+        }
+    }
+}
+
 impl From<crate::MemberCompositeKey> for MemberCompositeKey {
     fn from(value: crate::MemberCompositeKey) -> Self {
         MemberCompositeKey {
@@ -502,6 +662,15 @@ impl From<crate::MemberCompositeKey> for MemberCompositeKey {
     }
 }
 
+impl From<MemberCompositeKey> for crate::MemberCompositeKey {
+    fn from(value: MemberCompositeKey) -> crate::MemberCompositeKey {
+        crate::MemberCompositeKey {
+            server: value.server,
+            user: value.user,
+        }
+    }
+}
+
 impl From<crate::FieldsMember> for FieldsMember {
     fn from(value: crate::FieldsMember) -> Self {
         match value {
@@ -562,6 +731,34 @@ impl From<crate::Server> for Server {
     }
 }
 
+impl From<Server> for crate::Server {
+    fn from(value: Server) -> crate::Server {
+        crate::Server {
+            id: value.id,
+            owner: value.owner,
+            name: value.name,
+            description: value.description,
+            channels: value.channels,
+            categories: value
+                .categories
+                .map(|categories| categories.into_iter().map(|v| v.into()).collect()),
+            system_messages: value.system_messages.map(|v| v.into()),
+            roles: value
+                .roles
+                .into_iter()
+                .map(|(k, v)| (k, v.into()))
+                .collect(),
+            default_permissions: value.default_permissions,
+            icon: value.icon.map(|f| f.into()),
+            banner: value.banner.map(|f| f.into()),
+            flags: Some(value.flags as i32),
+            nsfw: value.nsfw,
+            analytics: value.analytics,
+            discoverable: value.discoverable,
+        }
+    }
+}
+
 impl From<crate::PartialServer> for PartialServer {
     fn from(value: crate::PartialServer) -> Self {
         PartialServer {
@@ -588,6 +785,32 @@ impl From<crate::PartialServer> for PartialServer {
     }
 }
 
+impl From<PartialServer> for crate::PartialServer {
+    fn from(value: PartialServer) -> crate::PartialServer {
+        crate::PartialServer {
+            id: value.id,
+            owner: value.owner,
+            name: value.name,
+            description: value.description,
+            channels: value.channels,
+            categories: value
+                .categories
+                .map(|categories| categories.into_iter().map(|v| v.into()).collect()),
+            system_messages: value.system_messages.map(|v| v.into()),
+            roles: value
+                .roles
+                .map(|roles| roles.into_iter().map(|(k, v)| (k, v.into())).collect()),
+            default_permissions: value.default_permissions,
+            icon: value.icon.map(|f| f.into()),
+            banner: value.banner.map(|f| f.into()),
+            flags: value.flags.map(|v| v as i32),
+            nsfw: value.nsfw,
+            analytics: value.analytics,
+            discoverable: value.discoverable,
+        }
+    }
+}
+
 impl From<crate::FieldsServer> for FieldsServer {
     fn from(value: crate::FieldsServer) -> Self {
         match value {
@@ -666,6 +889,18 @@ impl From<crate::Role> for Role {
     }
 }
 
+impl From<Role> for crate::Role {
+    fn from(value: Role) -> crate::Role {
+        crate::Role {
+            name: value.name,
+            permissions: value.permissions,
+            colour: value.colour,
+            hoist: value.hoist,
+            rank: value.rank,
+        }
+    }
+}
+
 impl From<crate::PartialRole> for PartialRole {
     fn from(value: crate::PartialRole) -> Self {
         PartialRole {
@@ -678,6 +913,18 @@ impl From<crate::PartialRole> for PartialRole {
     }
 }
 
+impl From<PartialRole> for crate::PartialRole {
+    fn from(value: PartialRole) -> crate::PartialRole {
+        crate::PartialRole {
+            name: value.name,
+            permissions: value.permissions,
+            colour: value.colour,
+            hoist: value.hoist,
+            rank: value.rank,
+        }
+    }
+}
+
 impl From<crate::FieldsRole> for FieldsRole {
     fn from(value: crate::FieldsRole) -> Self {
         match value {
@@ -871,6 +1118,25 @@ impl crate::User {
     }
 }
 
+impl From<User> for crate::User {
+    fn from(value: User) -> crate::User {
+        crate::User {
+            id: value.id,
+            username: value.username,
+            discriminator: value.discriminator,
+            display_name: value.display_name,
+            avatar: value.avatar.map(Into::into),
+            relations: None,
+            badges: Some(value.badges as i32),
+            status: value.status.map(Into::into),
+            profile: value.profile.map(Into::into),
+            flags: Some(value.flags as i32),
+            privileged: value.privileged,
+            bot: value.bot.map(Into::into),
+        }
+    }
+}
+
 impl From<crate::PartialUser> for PartialUser {
     fn from(value: crate::PartialUser) -> Self {
         PartialUser {
@@ -977,6 +1243,15 @@ impl From<crate::UserStatus> for UserStatus {
     }
 }
 
+impl From<UserStatus> for crate::UserStatus {
+    fn from(value: UserStatus) -> crate::UserStatus {
+        crate::UserStatus {
+            text: value.text,
+            presence: value.presence.map(|presence| presence.into()),
+        }
+    }
+}
+
 impl From<crate::UserProfile> for UserProfile {
     fn from(value: crate::UserProfile) -> Self {
         UserProfile {
@@ -986,6 +1261,15 @@ impl From<crate::UserProfile> for UserProfile {
     }
 }
 
+impl From<UserProfile> for crate::UserProfile {
+    fn from(value: UserProfile) -> crate::UserProfile {
+        crate::UserProfile {
+            content: value.content,
+            background: value.background.map(|file| file.into()),
+        }
+    }
+}
+
 impl From<crate::BotInformation> for BotInformation {
     fn from(value: crate::BotInformation) -> Self {
         BotInformation {
@@ -993,3 +1277,11 @@ impl From<crate::BotInformation> for BotInformation {
         }
     }
 }
+
+impl From<BotInformation> for crate::BotInformation {
+    fn from(value: BotInformation) -> crate::BotInformation {
+        crate::BotInformation {
+            owner: value.owner_id,
+        }
+    }
+}

crates/core/models/src/lib.rs

@@ -30,7 +30,16 @@ macro_rules! auto_derived_partial {
         #[derive(
             OptionalStruct, Debug, Clone, Eq, PartialEq, Serialize, Deserialize, JsonSchema,
         )]
-        #[optional_derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, JsonSchema)]
+        #[optional_derive(
+            Debug,
+            Clone,
+            Eq,
+            PartialEq,
+            Serialize,
+            Deserialize,
+            JsonSchema,
+            Default
+        )]
         #[optional_name = $name]
         #[opt_skip_serializing_none]
         #[opt_some_priority]