chore: ignore WS protocol errors

refactor: redo the bonfire client structure

compose.yml

@@ -34,7 +34,7 @@ services:
       - minio
     entrypoint: >
       /bin/sh -c "while ! /usr/bin/mc ready minio; do
-        /usr/bin/mc alias set minio http://minio:9000 minioautumn minioautumn;
+        /usr/bin/mc config host add minio http://minio:9000 minioautumn minioautumn;
         echo 'Waiting minio...' && sleep 1;
       done; /usr/bin/mc mb minio/revolt-uploads; exit 0;"
 

crates/bonfire/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 [dependencies]
 # util
 log = "*"
+rand = "*"
 sentry = "0.31.5"
 lru = "0.7.6"
 ulid = "0.5.0"
@@ -38,6 +39,7 @@ async-std = { version = "1.8.0", features = [
 # core
 authifier = { version = "1.0.15" }
 revolt-result = { path = "../core/result" }
+revolt-broker = { path = "../core/broker" }
 revolt-models = { path = "../core/models" }
 revolt-config = { path = "../core/config" }
 revolt-database = { path = "../core/database" }
@@ -46,3 +48,6 @@ revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
 
 # redis
 fred = { version = "8.0.1", features = ["subscriber-client"] }
+
+# rabbit
+lapin = { version = "3.0.0" }

crates/bonfire/src/client/core.rs

@@ -0,0 +1,129 @@
+use async_std::{net::TcpStream, sync::Mutex};
+use async_tungstenite::WebSocketStream;
+use futures::{join, SinkExt, StreamExt, TryStreamExt};
+use revolt_config::report_internal_error;
+use revolt_database::{
+    events::{client::EventV1, server::ClientMessage},
+    iso8601_timestamp::Timestamp,
+    Database, User, UserHint,
+};
+use revolt_presence::{create_session, delete_session};
+use revolt_result::create_error;
+
+use crate::{
+    client::{
+        subscriber::client_subscriber,
+        worker::{client_worker, WorkerRef},
+    },
+    config::ProtocolConfiguration,
+    events::state::State,
+};
+
+/// Core event loop of gateway clients
+pub async fn client_core(
+    db: &'static Database,
+    ws: WebSocketStream<TcpStream>,
+    mut config: ProtocolConfiguration,
+) {
+    // Split the socket for simultaneously read and write.
+    let (mut write, mut read) = ws.split();
+
+    // If the user has not provided authentication, request information.
+    if config.get_session_token().is_none() {
+        while let Ok(Some(message)) = read.try_next().await {
+            if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
+                config.set_session_token(token);
+                break;
+            }
+        }
+    }
+
+    // Try to authenticate the user.
+    let Some(token) = config.get_session_token().as_ref() else {
+        write
+            .send(config.encode(&EventV1::Error {
+                data: create_error!(InvalidSession),
+            }))
+            .await
+            .ok();
+        return;
+    };
+
+    let (user, session_id) = match User::from_token(db, token, UserHint::Any).await {
+        Ok(user) => user,
+        Err(err) => {
+            write
+                .send(config.encode(&EventV1::Error { data: err }))
+                .await
+                .ok();
+            return;
+        }
+    };
+
+    info!(
+        "Authenticated user {}#{}",
+        user.username, user.discriminator
+    );
+
+    db.update_session_last_seen(&session_id, Timestamp::now_utc())
+        .await
+        .ok();
+
+    // Create local state.
+    let mut state = State::from(user, session_id);
+    let user_id = state.cache.user_id.clone();
+
+    // Notify socket we have authenticated.
+    if report_internal_error!(write.send(config.encode(&EventV1::Authenticated)).await).is_err() {
+        return;
+    }
+
+    // Download required data to local cache and send Ready payload.
+    let ready_payload = match report_internal_error!(
+        state
+            .generate_ready_payload(db, config.get_ready_payload_fields())
+            .await
+    ) {
+        Ok(ready_payload) => ready_payload,
+        Err(_) => return,
+    };
+
+    if report_internal_error!(write.send(config.encode(&ready_payload)).await).is_err() {
+        return;
+    }
+
+    // Create presence session.
+    let (first_session, session_id) = create_session(&user_id, 0).await;
+
+    // If this was the first session, notify other users that we just went online.
+    if first_session {
+        state.broadcast_presence_change(true).await;
+    }
+
+    {
+        let worker_ref = WorkerRef::from(&state);
+        let write = Mutex::new(write);
+        let (reload, reloaded) = async_channel::bounded(1);
+        let (cancel_1, cancelled_1) = async_channel::bounded(1);
+        let (cancel_2, cancelled_2) = async_channel::bounded(1);
+
+        join!(
+            async {
+                client_subscriber(&write, cancelled_1, reloaded, &config, db, &mut state).await;
+                cancel_2.send(()).await.ok();
+            },
+            async {
+                client_worker(read, &write, cancelled_2, reload, &config, worker_ref).await;
+                cancel_1.send(()).await.ok();
+            }
+        );
+    }
+
+    // Clean up presence session.
+    let last_session = delete_session(&user_id, session_id).await;
+
+    // If this was the last session, notify other users that we just went offline.
+    if last_session {
+        state.broadcast_presence_change(false).await;
+    }
+}

crates/bonfire/src/client/mod.rs

@@ -0,0 +1,3 @@
+pub mod core;
+pub mod subscriber;
+pub mod worker;

crates/bonfire/src/client/subscriber.rs

@@ -0,0 +1,112 @@
+use async_channel::Receiver;
+use async_std::{net::TcpStream, sync::Mutex};
+use async_tungstenite::WebSocketStream;
+use authifier::AuthifierEvent;
+use futures::{pin_mut, select, stream::SplitSink, FutureExt, SinkExt};
+use revolt_broker::event_stream;
+use revolt_database::{events::client::EventV1, Database};
+use sentry::Level;
+
+use crate::{
+    config::ProtocolConfiguration,
+    events::state::{State, SubscriptionStateChange},
+};
+
+/// Event subscriber loop
+pub async fn client_subscriber(
+    write: &Mutex<SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>>,
+    cancelled: Receiver<()>,
+    reloaded: Receiver<()>,
+    protocol_config: &ProtocolConfiguration,
+    db: &'static Database,
+    state: &mut State,
+) {
+    let mut consumer = event_stream::Consumer::new().await;
+    consumer.set_topics(state.subscribed.read().await.clone());
+
+    let mut cancel = false;
+
+    loop {
+        // Reload consumer if subscriptions change
+        if !matches!(state.apply_state().await, SubscriptionStateChange::None) {
+            consumer.set_topics(state.subscribed.read().await.clone());
+        }
+
+        // Read incoming events
+        loop {
+            let reloaded = reloaded.recv().fuse();
+            let cancelled = cancelled.recv().fuse();
+            let delivery = consumer.next().fuse();
+            pin_mut!(delivery, reloaded, cancelled);
+
+            select! {
+                _ = reloaded => {
+                    break;
+                }
+                _ = cancelled => {
+                    cancel = true;
+                    break;
+                }
+                event = delivery => {
+                    if let Some(mut event) = event {
+                        // Handle the event
+                        if let EventV1::Auth(auth) = &event {
+                            if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
+                                if &state.session_id == session_id {
+                                    event = EventV1::Logout;
+                                }
+                            } else if let AuthifierEvent::DeleteAllSessions {
+                                exclude_session_id, ..
+                            } = auth
+                            {
+                                if let Some(excluded) = exclude_session_id {
+                                    if &state.session_id != excluded {
+                                        event = EventV1::Logout;
+                                    }
+                                } else {
+                                    event = EventV1::Logout;
+                                }
+                            }
+                        } else {
+                            let should_send = state.handle_incoming_event_v1(db, &mut event).await;
+                            if !should_send {
+                                continue;
+                            }
+                        }
+
+                        let result = write.lock().await.send(protocol_config.encode(&event)).await;
+                        if let Err(e) = result {
+                            use async_tungstenite::tungstenite::Error;
+                            if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
+                                let err = format!("Error while sending an event: {e:?}");
+                                warn!("{}", err);
+                                sentry::capture_message(&err, Level::Warning);
+                            }
+
+                            cancel = true;
+                            break;
+                        }
+
+                        if let EventV1::Logout = event {
+                            info!("User {} received log out event!", state.user_id);
+                            cancel = true;
+                            break;
+                        }
+
+                        break;
+                    } else {
+                        cancel = true;
+                        break;
+                    }
+                }
+            }
+        }
+
+        // Break out if cancelled
+        if cancel {
+            break;
+        }
+    }
+
+    consumer.dispose_channel().await;
+}

crates/bonfire/src/client/worker.rs

@@ -0,0 +1,124 @@
+use std::{collections::HashSet, sync::Arc};
+
+use async_channel::{Receiver, Sender};
+use async_std::{
+    net::TcpStream,
+    sync::{Mutex, RwLock},
+};
+use async_tungstenite::WebSocketStream;
+use futures::{
+    pin_mut, select,
+    stream::{SplitSink, SplitStream},
+    FutureExt, SinkExt, TryStreamExt,
+};
+use revolt_database::events::{client::EventV1, server::ClientMessage};
+use sentry::Level;
+
+use crate::{config::ProtocolConfiguration, events::state::State};
+
+pub struct WorkerRef {
+    user_id: String,
+    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
+    subscribed: Arc<RwLock<HashSet<String>>>,
+}
+
+impl WorkerRef {
+    pub fn from(state: &State) -> WorkerRef {
+        WorkerRef {
+            user_id: state.user_id.clone(),
+            active_servers: state.active_servers.clone(),
+            subscribed: state.subscribed.clone(),
+        }
+    }
+}
+
+/// Incoming message handling
+pub async fn client_worker(
+    mut read: SplitStream<WebSocketStream<TcpStream>>,
+    write: &Mutex<SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>>,
+    cancelled: Receiver<()>,
+    reload: Sender<()>,
+    config: &ProtocolConfiguration,
+    state: WorkerRef,
+) {
+    loop {
+        let read = read.try_next().fuse();
+        let cancelled = cancelled.recv().fuse();
+        pin_mut!(read, cancelled);
+
+        select! {
+            _ = cancelled => { return; },
+            msg = read => {
+                let msg = match msg {
+                    Ok(Some(msg)) => msg,
+                    Ok(None) => {
+                        warn!("Received a None message!");
+                        return;
+                    }
+                    Err(e) => {
+                        use async_tungstenite::tungstenite::Error;
+                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed | Error::Protocol(_)) {
+                            let err = format!("Error while reading an event: {e:?}");
+                            warn!("{}", err);
+                            sentry::capture_message(&err, Level::Warning);
+                        }
+
+                        return;
+                    }
+                };
+
+                let Ok(payload) = config.decode(&msg) else {
+                    continue;
+                };
+
+                match payload {
+                    ClientMessage::BeginTyping { channel } => {
+                        if !state.subscribed.read().await.contains(&channel) {
+                            continue;
+                        }
+
+                        EventV1::ChannelStartTyping {
+                            id: channel.clone(),
+                            user: state.user_id.clone(),
+                        }
+                        .p(channel.clone())
+                        .await;
+                    }
+                    ClientMessage::EndTyping { channel } => {
+                        if !state.subscribed.read().await.contains(&channel) {
+                            continue;
+                        }
+
+                        EventV1::ChannelStopTyping {
+                            id: channel.clone(),
+                            user: state.user_id.clone(),
+                        }
+                        .p(channel.clone())
+                        .await;
+                    }
+                    ClientMessage::Subscribe { server_id } => {
+                        let mut servers = state.active_servers.lock().await;
+                        let has_item = servers.contains_key(&server_id);
+                        servers.insert(server_id, ());
+
+                        if !has_item {
+                            // Poke the listener to adjust subscriptions
+                            reload.send(()).await.ok();
+                        }
+                    }
+                    ClientMessage::Ping { data, responded } => {
+                        if responded.is_none() {
+                            write
+                                .lock()
+                                .await
+                                .send(config.encode(&EventV1::Pong { data }))
+                                .await
+                                .ok();
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        }
+    }
+}

crates/bonfire/src/events/state.rs

@@ -66,6 +66,7 @@ impl Default for Cache {
 pub struct State {
     pub cache: Cache,
 
+    pub user_id: String,
     pub session_id: String,
     pub private_topic: String,
     pub state: SubscriptionStateChange,
@@ -87,6 +88,7 @@ impl State {
             ..Default::default()
         };
 
+        let user_id = user.id.clone();
         cache.users.insert(user.id.clone(), user);
 
         State {
@@ -96,6 +98,7 @@ impl State {
                 Duration::from_secs(900),
                 5,
             ))),
+            user_id,
             session_id,
             private_topic,
             state: SubscriptionStateChange::Reset,

crates/bonfire/src/main.rs

@@ -6,10 +6,11 @@ use revolt_presence::clear_region;
 #[macro_use]
 extern crate log;
 
-pub mod config;
-pub mod events;
+pub mod client;
 
+mod config;
 mod database;
+mod events;
 mod websocket;
 
 #[async_std::main]

crates/bonfire/src/websocket.rs

@@ -1,40 +1,11 @@
-use std::{collections::HashSet, net::SocketAddr, sync::Arc};
+use std::net::SocketAddr;
 
-use async_tungstenite::WebSocketStream;
-use authifier::AuthifierEvent;
-use fred::{
-    error::RedisErrorKind,
-    interfaces::{ClientLike, EventInterface, PubsubInterface},
-    types::RedisConfig,
-};
-use futures::{
-    channel::oneshot,
-    join, pin_mut, select,
-    stream::{SplitSink, SplitStream},
-    FutureExt, SinkExt, StreamExt, TryStreamExt,
-};
-use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
-use revolt_config::report_internal_error;
-use revolt_database::{
-    events::{client::EventV1, server::ClientMessage},
-    iso8601_timestamp::Timestamp,
-    Database, User, UserHint,
-};
-use revolt_presence::{create_session, delete_session};
+use async_std::net::TcpStream;
+use futures::channel::oneshot;
+use revolt_database::Database;
 
-use async_std::{
-    net::TcpStream,
-    sync::{Mutex, RwLock},
-    task::spawn,
-};
-use revolt_result::create_error;
-use sentry::Level;
-
-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>;
+use crate::client::core::client_core;
+use crate::config::WebsocketHandshakeCallback;
 
 /// Start a new WebSocket client worker given access to the database,
 /// the relevant TCP stream and the remote address of the client.
@@ -54,7 +25,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     };
 
     // Verify we've received a valid config, otherwise we should just drop the connection.
-    let Ok(mut config) = receiver.await else {
+    let Ok(config) = receiver.await else {
         return;
     };
 
@@ -64,446 +35,5 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
         config.get_protocol_format()
     );
 
-    // Split the socket for simultaneously read and write.
-    let (mut write, mut read) = ws.split();
-
-    // If the user has not provided authentication, request information.
-    if config.get_session_token().is_none() {
-        while let Ok(Some(message)) = read.try_next().await {
-            if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
-                config.set_session_token(token);
-                break;
-            }
-        }
-    }
-
-    // Try to authenticate the user.
-    let Some(token) = config.get_session_token().as_ref() else {
-        write
-            .send(config.encode(&EventV1::Error {
-                data: create_error!(InvalidSession),
-            }))
-            .await
-            .ok();
-        return;
-    };
-
-    let (user, session_id) = match User::from_token(db, token, UserHint::Any).await {
-        Ok(user) => user,
-        Err(err) => {
-            write
-                .send(config.encode(&EventV1::Error { data: err }))
-                .await
-                .ok();
-            return;
-        }
-    };
-
-    info!("User {addr:?} authenticated as @{}", user.username);
-
-    db.update_session_last_seen(&session_id, Timestamp::now_utc())
-        .await
-        .ok();
-
-    // Create local state.
-    let mut state = State::from(user, session_id);
-    let user_id = state.cache.user_id.clone();
-
-    // Notify socket we have authenticated.
-    if report_internal_error!(write.send(config.encode(&EventV1::Authenticated)).await).is_err() {
-        return;
-    }
-
-    // Download required data to local cache and send Ready payload.
-    let ready_payload = match report_internal_error!(
-        state
-            .generate_ready_payload(db, config.get_ready_payload_fields())
-            .await
-    ) {
-        Ok(ready_payload) => ready_payload,
-        Err(_) => return,
-    };
-
-    if report_internal_error!(write.send(config.encode(&ready_payload)).await).is_err() {
-        return;
-    }
-
-    // Create presence session.
-    let (first_session, session_id) = create_session(&user_id, 0).await;
-
-    // If this was the first session, notify other users that we just went online.
-    if first_session {
-        state.broadcast_presence_change(true).await;
-    }
-
-    {
-        // Setup channels and mutexes
-        let write = Mutex::new(write);
-        let subscribed = state.subscribed.clone();
-        let active_servers = state.active_servers.clone();
-        let (topic_signal_s, topic_signal_r) = async_channel::unbounded();
-
-        // TODO: this needs to be rewritten
-        // Create channels through which the tasks can signal to each other they need to clean up
-        let (kill_signal_1_s, kill_signal_1_r) = async_channel::bounded(1);
-        let (kill_signal_2_s, kill_signal_2_r) = async_channel::bounded(1);
-
-        // Create a PubSub connection to poll on.
-        let listener = listener_with_kill_signal(
-            db,
-            &mut state,
-            addr,
-            &config,
-            topic_signal_r,
-            kill_signal_1_r,
-            &write,
-            kill_signal_2_s,
-        );
-
-        // Read from WebSocket stream.
-        let worker = worker_with_kill_signal(
-            addr,
-            subscribed,
-            active_servers,
-            user_id.clone(),
-            &config,
-            topic_signal_s,
-            kill_signal_2_r,
-            read,
-            &write,
-            kill_signal_1_s,
-        );
-
-        join!(listener, worker);
-    }
-    // Clean up presence session.
-    let last_session = delete_session(&user_id, session_id).await;
-
-    // If this was the last session, notify other users that we just went offline.
-    if last_session {
-        state.broadcast_presence_change(false).await;
-    }
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn listener_with_kill_signal(
-    db: &'static Database,
-    state: &mut State,
-    addr: SocketAddr,
-    config: &ProtocolConfiguration,
-    topic_signal_r: async_channel::Receiver<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    write: &Mutex<WsWriter>,
-    kill_signal_s: async_channel::Sender<()>,
-) {
-    listener(
-        db,
-        state,
-        addr,
-        config,
-        topic_signal_r,
-        kill_signal_r,
-        write,
-    )
-    .await;
-    kill_signal_s.send(()).await.ok();
-}
-
-async fn listener(
-    db: &'static Database,
-    state: &mut State,
-    addr: SocketAddr,
-    config: &ProtocolConfiguration,
-    topic_signal_r: async_channel::Receiver<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    write: &Mutex<WsWriter>,
-) {
-    let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
-    let subscriber = match report_internal_error!(
-        fred::types::Builder::from_config(redis_config).build_subscriber_client()
-    ) {
-        Ok(subscriber) => subscriber,
-        Err(_) => return,
-    };
-
-    if report_internal_error!(subscriber.init().await).is_err() {
-        return;
-    }
-
-    // Handle Redis connection dropping
-    let (clean_up_s, clean_up_r) = async_channel::bounded(1);
-    let clean_up_s = Arc::new(Mutex::new(clean_up_s));
-    subscriber.on_error(move |err| {
-        if let RedisErrorKind::Canceled = err.kind() {
-            let clean_up_s = clean_up_s.clone();
-            spawn(async move {
-                clean_up_s.lock().await.send(()).await.ok();
-            });
-        }
-
-        Ok(())
-    });
-
-    let mut message_rx = subscriber.message_rx();
-    'out: loop {
-        // Check for state changes for subscriptions.
-        match state.apply_state().await {
-            SubscriptionStateChange::Reset => {
-                if report_internal_error!(subscriber.unsubscribe_all().await).is_err() {
-                    break 'out;
-                }
-
-                let subscribed = state.subscribed.read().await;
-                for id in subscribed.iter() {
-                    if report_internal_error!(subscriber.subscribe(id).await).is_err() {
-                        break 'out;
-                    }
-                }
-
-                #[cfg(debug_assertions)]
-                info!("{addr:?} has reset their subscriptions");
-            }
-            SubscriptionStateChange::Change { add, remove } => {
-                for id in remove {
-                    #[cfg(debug_assertions)]
-                    info!("{addr:?} unsubscribing from {id}");
-
-                    if report_internal_error!(subscriber.unsubscribe(id).await).is_err() {
-                        break 'out;
-                    }
-                }
-
-                for id in add {
-                    #[cfg(debug_assertions)]
-                    info!("{addr:?} subscribing to {id}");
-
-                    if report_internal_error!(subscriber.subscribe(id).await).is_err() {
-                        break 'out;
-                    }
-                }
-            }
-            SubscriptionStateChange::None => {}
-        }
-
-        let t1 = message_rx.recv().fuse();
-        let t2 = topic_signal_r.recv().fuse();
-        let t3 = kill_signal_r.recv().fuse();
-        let t4 = clean_up_r.recv().fuse();
-
-        pin_mut!(t1, t2, t3, t4);
-
-        select! {
-            _ = t4 => {
-                break 'out;
-            },
-            _ = t3 => {
-                break 'out;
-            },
-            _ = t2 => {},
-            message = t1 => {
-                // Handle incoming events.
-                let message = match report_internal_error!(message) {
-                    Ok(message) => message,
-                    Err(_) => break 'out
-                };
-
-                let event = match *REDIS_PAYLOAD_TYPE {
-                    PayloadType::Json => message
-                        .value
-                        .as_str()
-                        .and_then(|s| report_internal_error!(serde_json::from_str::<EventV1>(s.as_ref())).ok()),
-                    PayloadType::Msgpack => message
-                        .value
-                        .as_bytes()
-                        .and_then(|b| report_internal_error!(rmp_serde::from_slice::<EventV1>(b)).ok()),
-                    PayloadType::Bincode => message
-                        .value
-                        .as_bytes()
-                        .and_then(|b| report_internal_error!(bincode::deserialize::<EventV1>(b)).ok()),
-                };
-
-                let Some(mut event) = event else {
-                    let err = format!(
-                        "Failed to deserialise event for {}: `{:?}`",
-                        message.channel,
-                        message
-                            .value
-                    );
-
-                    error!("{}", err);
-                    sentry::capture_message(&err, Level::Error);
-                    break 'out;
-                };
-
-                if let EventV1::Auth(auth) = &event {
-                    if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
-                        if &state.session_id == session_id {
-                            event = EventV1::Logout;
-                        }
-                    } else if let AuthifierEvent::DeleteAllSessions {
-                        exclude_session_id, ..
-                    } = auth
-                    {
-                        if let Some(excluded) = exclude_session_id {
-                            if &state.session_id != excluded {
-                                event = EventV1::Logout;
-                            }
-                        } else {
-                            event = EventV1::Logout;
-                        }
-                    }
-                } else {
-                    let should_send = state.handle_incoming_event_v1(db, &mut event).await;
-                    if !should_send {
-                        continue;
-                    }
-                }
-
-                let result = write.lock().await.send(config.encode(&event)).await;
-                if let Err(e) = result {
-                    use async_tungstenite::tungstenite::Error;
-                    if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
-                        let err = format!("Error while sending an event to {addr:?}: {e:?}");
-                        warn!("{}", err);
-                        sentry::capture_message(&err, Level::Warning);
-                    }
-
-                    break 'out;
-                }
-
-                if let EventV1::Logout = event {
-                    info!("User {addr:?} received log out event!");
-                    break 'out;
-                }
-            }
-        }
-    }
-
-    report_internal_error!(subscriber.quit().await).ok();
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn worker_with_kill_signal(
-    addr: SocketAddr,
-    subscribed: Arc<RwLock<HashSet<String>>>,
-    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
-    user_id: String,
-    config: &ProtocolConfiguration,
-    topic_signal_s: async_channel::Sender<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    read: WsReader,
-    write: &Mutex<WsWriter>,
-    kill_signal_s: async_channel::Sender<()>,
-) {
-    worker(
-        addr,
-        subscribed,
-        active_servers,
-        user_id,
-        config,
-        topic_signal_s,
-        kill_signal_r,
-        read,
-        write,
-    )
-    .await;
-    kill_signal_s.send(()).await.ok();
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn worker(
-    addr: SocketAddr,
-    subscribed: Arc<RwLock<HashSet<String>>>,
-    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
-    user_id: String,
-    config: &ProtocolConfiguration,
-    topic_signal_s: async_channel::Sender<()>,
-    kill_signal_r: async_channel::Receiver<()>,
-    mut read: WsReader,
-    write: &Mutex<WsWriter>,
-) {
-    loop {
-        let t1 = read.try_next().fuse();
-        let t2 = kill_signal_r.recv().fuse();
-
-        pin_mut!(t1, t2);
-
-        select! {
-            _ = t2 => {
-                return;
-            },
-            result = t1 => {
-                let msg = match result {
-                    Ok(Some(msg)) => msg,
-                    Ok(None) => {
-                        warn!("Received a None message!");
-                        sentry::capture_message("Received a None message!", Level::Warning);
-                        return;
-                    }
-                    Err(e) => {
-                        use async_tungstenite::tungstenite::Error;
-                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
-                            let err = format!("Error while reading an event from {addr:?}: {e:?}");
-                            warn!("{}", err);
-                            sentry::capture_message(&err, Level::Warning);
-                        }
-
-                        return;
-                    }
-                };
-
-                let Ok(payload) = config.decode(&msg) else {
-                    continue;
-                };
-
-                match payload {
-                    ClientMessage::BeginTyping { channel } => {
-                        if !subscribed.read().await.contains(&channel) {
-                            continue;
-                        }
-
-                        EventV1::ChannelStartTyping {
-                            id: channel.clone(),
-                            user: user_id.clone(),
-                        }
-                        .p(channel.clone())
-                        .await;
-                    }
-                    ClientMessage::EndTyping { channel } => {
-                        if !subscribed.read().await.contains(&channel) {
-                            continue;
-                        }
-
-                        EventV1::ChannelStopTyping {
-                            id: channel.clone(),
-                            user: user_id.clone(),
-                        }
-                        .p(channel.clone())
-                        .await;
-                    }
-                    ClientMessage::Subscribe { server_id } => {
-                        let mut servers = active_servers.lock().await;
-                        let has_item = servers.contains_key(&server_id);
-                        servers.insert(server_id, ());
-
-                        if !has_item {
-                            // Poke the listener to adjust subscriptions
-                            topic_signal_s.send(()).await.ok();
-                        }
-                    }
-                    ClientMessage::Ping { data, responded } => {
-                        if responded.is_none() {
-                            write
-                                .lock()
-                                .await
-                                .send(config.encode(&EventV1::Pong { data }))
-                                .await
-                                .ok();
-                        }
-                    }
-                    _ => {}
-                }
-            }
-        }
-    }
+    client_core(db, ws, config).await;
 }

crates/core/broker/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "revolt-broker"
+version = "0.8.8"
+edition = "2024"
+license = "AGPL-3.0-or-later"
+authors = ["Paul Makles <me@insrt.uk>"]
+description = "Revolt Backend: Event Broker"
+
+[dependencies]
+# Utility
+log = "0.4"
+rand = "0.9.1"
+
+# RabbitMQ/AMQP client
+lapin = "3.0.0"
+
+# Async runtime
+async-std = { version = "1.8.0" }
+
+# Serialisation
+serde = "1"
+rmp-serde = "1.3.0"
+
+# Core
+revolt-config = { version = "0.8.8", path = "../config" }

crates/core/broker/src/event_stream/consumer.rs

@@ -0,0 +1,180 @@
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
+
+use async_std::stream::StreamExt;
+use lapin::{
+    Channel, Connection,
+    options::BasicAckOptions,
+    types::{AMQPValue, FieldArray, FieldTable, LongLongInt},
+};
+use log::info;
+use rand::Rng;
+use revolt_config::{capture_internal_error, config};
+use serde::de::DeserializeOwned;
+
+use crate::event_stream::{create_channel, get_connection};
+
+pub struct Consumer {
+    #[allow(dead_code)]
+    conn: Arc<Connection>,
+    channel: Channel,
+    tag: String,
+    topics: HashSet<String>,
+    topics_changed: bool,
+    consumer: Option<lapin::Consumer>,
+    offset: Option<LongLongInt>,
+}
+
+impl Consumer {
+    /// Create a new event stream consumer
+    pub async fn new() -> Consumer {
+        let config = config().await;
+        let conn = get_connection().await;
+        let channel = create_channel(&conn, config.rabbit.event_stream).await;
+
+        Consumer {
+            conn,
+            channel,
+            tag: rand::rng()
+                .sample_iter::<char, _>(&rand::distr::StandardUniform)
+                .take(32)
+                .collect(),
+            topics: HashSet::new(),
+            topics_changed: false,
+            consumer: None,
+            offset: None,
+        }
+    }
+
+    /// Update the set of topics
+    pub fn set_topics(&mut self, topics: HashSet<String>) {
+        self.topics = topics;
+        self.topics_changed = true;
+    }
+
+    /// Get the current consumer
+    pub async fn ensure_consumer(&mut self) {
+        if self.topics_changed {
+            info!("Topics changed, disposing the consumer.");
+            self.dispose_consumer().await;
+            self.topics_changed = false;
+        }
+
+        if self.consumer.is_none() {
+            info!("Creating a new consumer, tag={}", self.tag);
+            let config = config().await;
+
+            // Build arguments for consumer
+            let mut args: FieldTable = Default::default();
+
+            // Configure stream filter to select topics we are listening for
+            {
+                let mut filter: FieldArray = Default::default();
+                for topic in &self.topics {
+                    filter.push(AMQPValue::LongString(topic.as_str().into()));
+                }
+
+                args.insert("x-stream-filter".into(), AMQPValue::FieldArray(filter));
+            }
+
+            // Set stream offset if applicable
+            if let Some(offset) = self.offset {
+                args.insert("x-stream-offset".into(), AMQPValue::LongLongInt(offset));
+            }
+
+            // Create the consumer
+            self.consumer = Some(
+                self.channel
+                    .basic_consume(
+                        &config.rabbit.event_stream.queue,
+                        &self.tag,
+                        Default::default(),
+                        args,
+                    )
+                    .await
+                    .unwrap(),
+            );
+        }
+    }
+
+    /// Close the active consumer if one exists
+    pub async fn dispose_consumer(&mut self) {
+        if let Some(consumer) = self.consumer.as_ref() {
+            if consumer.state().is_active() {
+                if let Err(err) = self
+                    .channel
+                    .basic_cancel(&self.tag, Default::default())
+                    .await
+                {
+                    eprintln!("Failed to close consumer! {:?}", err);
+                }
+
+                // is this necessary?
+                // else {
+                // Read the consumer to the end
+                //     while let Some(delivery) = consumer.next().await {
+                //         let delivery = delivery.expect("error in consumer");
+                //         delivery.ack(BasicAckOptions::default()).await.expect("ack");
+                //     }
+                // }
+            }
+
+            self.consumer = None;
+        }
+    }
+
+    /// Close the active channel
+    pub async fn dispose_channel(&mut self) {
+        // Close the channel -- don't do this actually
+        capture_internal_error!(self.channel.close(0, "closing channel").await);
+    }
+
+    /// Get the next item
+    pub async fn next<T: DeserializeOwned>(&mut self) -> Option<T> {
+        self.ensure_consumer().await;
+
+        let consumer = self.consumer.as_mut().unwrap();
+
+        while let Some(Ok(delivery)) = consumer.next().await {
+            // Acknowledgement is required
+            delivery.ack(BasicAckOptions::default()).await.expect("ack");
+
+            // Parse the delivery headers
+            let headers: HashMap<String, AMQPValue> = delivery
+                .properties
+                .headers()
+                .as_ref()
+                .map(|table| {
+                    table
+                        .into_iter()
+                        .map(|(k, v)| (k.to_string(), v.clone()))
+                        .collect()
+                })
+                .unwrap_or_default();
+
+            // Keep track of the current offset
+            let stream_offset = headers
+                .get("x-stream-offset")
+                .expect("`x-stream-offset` not present in message!");
+
+            self.offset = Some(stream_offset.as_long_long_int().unwrap() + 1);
+
+            // Client-side topic filtering (broker uses Bloom filter so may have false-positives)
+            let filter_value = headers
+                .get("x-stream-filter-value")
+                .expect("`x-stream-filter-value` not present in message!")
+                .as_long_string()
+                .expect("`string`")
+                .to_string();
+
+            if self.topics.contains(&filter_value) {
+                // Deserialise the data
+                return Some(rmp_serde::from_slice(&delivery.data).expect("`data`"));
+            }
+        }
+
+        None
+    }
+}

crates/core/broker/src/event_stream/mod.rs

@@ -0,0 +1,7 @@
+mod consumer;
+mod pool;
+mod publish;
+
+pub use consumer::Consumer;
+pub use pool::{create_channel, get_connection};
+pub use publish::publish_event;

crates/core/broker/src/event_stream/pool.rs

@@ -0,0 +1,99 @@
+use async_std::sync::Mutex;
+use lapin::{
+    Connection,
+    options::QueueDeclareOptions,
+    types::{AMQPValue, FieldTable},
+};
+use log::{debug, warn};
+use revolt_config::{RabbitEventStream, config};
+use std::sync::Arc;
+
+use crate::create_client;
+
+/// Get a handle to the event stream
+pub async fn get_connection() -> Arc<Connection> {
+    let config = config().await;
+
+    static CONNECTIONS: Mutex<Vec<Arc<Connection>>> = Mutex::new(Vec::new());
+
+    let mut connections = CONNECTIONS.lock().await;
+    connections.retain(|item| {
+        if item.status().connected() {
+            true
+        } else {
+            warn!(
+                "Dropping connection with status {:?}",
+                item.status().state()
+            );
+
+            false
+        }
+    });
+
+    debug!(
+        "Connections: {}, Clients: {:?}",
+        connections.len(),
+        connections
+            .iter()
+            .map(Arc::strong_count)
+            .collect::<Vec<usize>>()
+    );
+
+    for conn in connections.iter() {
+        if Arc::strong_count(conn) < config.rabbit.event_stream.channels_per_conn {
+            return conn.clone();
+        }
+    }
+
+    let conn = Arc::new(create_client().await);
+
+    connections.push(conn.clone());
+    conn
+}
+
+/// Create a channel
+pub async fn create_channel(
+    conn: &lapin::Connection,
+    event_stream: RabbitEventStream,
+) -> lapin::Channel {
+    let channel = conn.create_channel().await.unwrap();
+
+    let mut args: FieldTable = Default::default();
+
+    args.insert(
+        // set queue type to stream
+        "x-queue-type".into(),
+        AMQPValue::LongString("stream".into()),
+    );
+
+    args.insert(
+        // max. size of the stream
+        "x-max-length-bytes".into(),
+        AMQPValue::LongLongInt(event_stream.stream_max_length_bytes),
+    );
+
+    args.insert(
+        // size of the Bloom filter
+        "x-stream-filter-size-bytes".into(),
+        AMQPValue::LongLongInt(event_stream.filter_size_bytes),
+    );
+
+    channel
+        .queue_declare(
+            &event_stream.queue,
+            QueueDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            args,
+        )
+        .await
+        .unwrap();
+
+    channel
+        .basic_qos(event_stream.qos_prefetch, Default::default())
+        .await
+        .unwrap();
+
+    channel
+}

crates/core/broker/src/event_stream/publish.rs

@@ -0,0 +1,37 @@
+use lapin::{
+    Error,
+    protocol::basic::AMQPProperties,
+    publisher_confirm::PublisherConfirm,
+    types::{AMQPValue, FieldTable},
+};
+use revolt_config::config;
+use serde::Serialize;
+
+use crate::event_stream::{create_channel, get_connection};
+
+/// Publish an event to the message broker
+pub async fn publish_event<T: Serialize>(
+    channel: &str,
+    data: &T,
+) -> Result<PublisherConfirm, Error> {
+    let config = config().await;
+
+    let mut headers: FieldTable = Default::default();
+    headers.insert(
+        "x-stream-filter-value".into(),
+        AMQPValue::LongString(channel.into()),
+    );
+
+    let conn = get_connection().await;
+
+    create_channel(&conn, config.rabbit.event_stream.clone())
+        .await
+        .basic_publish(
+            &config.rabbit.event_stream.exchange,
+            &config.rabbit.event_stream.queue,
+            Default::default(),
+            &rmp_serde::to_vec_named(data).unwrap(),
+            AMQPProperties::default().with_headers(headers),
+        )
+        .await
+}

crates/core/broker/src/lib.rs

@@ -0,0 +1,18 @@
+use revolt_config::config;
+
+pub mod event_stream;
+
+/// Create a lapin client
+pub async fn create_client() -> lapin::Connection {
+    let config = config().await;
+
+    lapin::Connection::connect(
+        &format!(
+            "amqp://{}:{}@{}:{}/%2f",
+            config.rabbit.username, config.rabbit.password, config.rabbit.host, config.rabbit.port
+        ),
+        Default::default(),
+    )
+    .await
+    .unwrap()
+}

crates/core/config/Cargo.toml

@@ -11,9 +11,8 @@ description = "Revolt Backend: Configuration"
 [features]
 anyhow = ["dep:sentry-anyhow"]
 report-macros = ["revolt-result"]
-sentry = ["dep:sentry"]
 test = ["async-std"]
-default = ["test", "sentry"]
+default = ["test", "anyhow"]
 
 [dependencies]
 # Utility
@@ -33,7 +32,7 @@ log = "0.4.14"
 pretty_env_logger = "0.4.0"
 
 # Sentry
-sentry = { version = "0.31.5", optional = true }
+sentry = "0.31.5"
 sentry-anyhow = { version = "0.38.1", optional = true }
 
 # Core

crates/core/config/Revolt.toml

@@ -28,6 +28,20 @@ port = 5672
 username = "rabbituser"
 password = "rabbitpass"
 
+[rabbit.event_stream]
+# Configuration for event brokerage
+# Using default/direct exchange
+exchange = ""
+queue = "revolt.events"
+# Number of channels that can be opened per single TCP connection
+channels_per_conn = 128
+# Maximum size of the stream
+stream_max_length_bytes = 5_000_000_000
+# Size of the Bloom filter
+filter_size_bytes = 26
+# Number of messages to prefetch
+qos_prefetch = 100
+
 [api]
 
 [api.registration]

crates/core/config/src/lib.rs

@@ -6,12 +6,10 @@ use futures_locks::RwLock;
 use once_cell::sync::Lazy;
 use serde::Deserialize;
 
-#[cfg(feature = "sentry")]
 pub use sentry::{capture_error, capture_message, Level};
-#[cfg(feature = "anyhow")]
 pub use sentry_anyhow::capture_anyhow;
 
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! report_error {
     ( $expr: expr, $error: ident $( $tt:tt )? ) => {
@@ -26,7 +24,7 @@ macro_rules! report_error {
     };
 }
 
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! capture_internal_error {
     ( $expr: expr ) => {
@@ -37,7 +35,7 @@ macro_rules! capture_internal_error {
     };
 }
 
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! report_internal_error {
     ( $expr: expr ) => {
@@ -110,12 +108,24 @@ pub struct Database {
     pub redis: String,
 }
 
+#[derive(Deserialize, Debug, Clone)]
+pub struct RabbitEventStream {
+    pub exchange: String,
+    pub queue: String,
+    pub channels_per_conn: usize,
+    pub stream_max_length_bytes: i64,
+    pub filter_size_bytes: i64,
+    pub qos_prefetch: u16,
+}
+
 #[derive(Deserialize, Debug, Clone)]
 pub struct Rabbit {
     pub host: String,
     pub port: u16,
     pub username: String,
     pub password: String,
+
+    pub event_stream: RabbitEventStream,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -421,7 +431,6 @@ pub async fn config() -> Settings {
 }
 
 /// Configure logging and common Rust variables
-#[cfg(feature = "sentry")]
 pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry::ClientInitGuard> {
     if std::env::var("RUST_LOG").is_err() {
         std::env::set_var("RUST_LOG", "info");
@@ -447,7 +456,6 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry:
     }
 }
 
-#[cfg(feature = "sentry")]
 #[macro_export]
 macro_rules! configure {
     ($application: ident) => {

crates/core/database/Cargo.toml

@@ -10,12 +10,12 @@ description = "Revolt Backend: Database Implementation"
 
 [features]
 # Databases
-mongodb = ["dep:mongodb", "bson", "authifier/database-mongodb"]
+mongodb = ["dep:mongodb", "bson"]
 
 # ... Other
 tasks = ["isahc", "linkify", "url-escape"]
-async-std-runtime = ["async-std", "authifier/async-std-runtime"]
-rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi", "authifier/rocket_impl"]
+async-std-runtime = ["async-std"]
+rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi"]
 axum-impl = ["axum"]
 redis-is-patched = ["revolt-presence/redis-is-patched"]
 
@@ -37,6 +37,7 @@ revolt-permissions = { version = "0.8.8", path = "../permissions", features = [
     "bson",
 ] }
 revolt-parser = { version = "0.8.8", path = "../parser" }
+revolt-broker = { version = "0.8.8", path = "../broker" }
 
 # Utility
 log = "0.4"
@@ -56,6 +57,7 @@ isahc = { optional = true, version = "1.7", features = ["json"] }
 
 # Serialisation
 serde_json = "1"
+rmp-serde = "1.0.0"
 revolt_optional_struct = "0.2.0"
 serde = { version = "1", features = ["derive"] }
 iso8601-timestamp = { version = "0.2.10", features = ["serde", "bson"] }
@@ -91,8 +93,14 @@ rocket = { version = "0.5.1", default-features = false, features = [
 revolt_okapi = { version = "0.9.1", optional = true }
 revolt_rocket_okapi = { version = "0.10.0", optional = true }
 
+# Notifications
+fcm_v1 = "0.3.0"
+web-push = "0.10.0"
+revolt_a2 = { version = "0.10", default-features = false, features = ["ring"] }
+
 # Authifier
-authifier = { version = "1.0.15" }
+authifier = { version = "1.0.15", features = ["rocket_impl"] }
 
 # RabbitMQ
 amqprs = { version = "1.7.0" }
+lapin = { version = "3.0.0" }

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

@@ -1,4 +1,3 @@
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 
@@ -14,7 +13,6 @@ use authifier::Authifier;
 use rand::Rng;
 use revolt_config::config;
 
-#[cfg(feature = "mongodb")]
 pub use self::mongodb::*;
 pub use self::reference::*;
 
@@ -27,10 +25,8 @@ pub enum DatabaseInfo {
     /// Use the mock database
     Reference,
     /// Connect to MongoDB
-    #[cfg(feature = "mongodb")]
     MongoDb { uri: String, database_name: String },
     /// Use existing MongoDB connection
-    #[cfg(feature = "mongodb")]
     MongoDbFromClient(::mongodb::Client, String),
 }
 
@@ -40,7 +36,6 @@ pub enum Database {
     /// Mock database
     Reference(ReferenceDb),
     /// MongoDB database
-    #[cfg(feature = "mongodb")]
     MongoDb(MongoDb),
 }
 
@@ -50,7 +45,7 @@ impl DatabaseInfo {
     pub async fn connect(self) -> Result<Database, String> {
         let config = config().await;
 
-        match self {
+        Ok(match self {
             DatabaseInfo::Auto => {
                 if std::env::var("TEST_DB").is_ok() {
                     DatabaseInfo::Test(format!(
@@ -58,20 +53,16 @@ impl DatabaseInfo {
                         rand::thread_rng().gen_range(1_000_000..10_000_000)
                     ))
                     .connect()
-                    .await
+                    .await?
                 } else if !config.database.mongodb.is_empty() {
-                    #[cfg(feature = "mongodb")]
-                    return DatabaseInfo::MongoDb {
+                    DatabaseInfo::MongoDb {
                         uri: config.database.mongodb,
                         database_name: "revolt".to_string(),
                     }
                     .connect()
-                    .await;
-
-                    #[cfg(not(feature = "mongodb"))]
-                    return Err("MongoDB not enabled.".to_string())
+                    .await?
                 } else {
-                    DatabaseInfo::Reference.connect().await
+                    DatabaseInfo::Reference.connect().await?
                 }
             }
             DatabaseInfo::Test(database_name) => {
@@ -79,36 +70,30 @@ impl DatabaseInfo {
                     .expect("`TEST_DB` environment variable should be set to REFERENCE or MONGODB")
                     .as_str()
                 {
-                    "REFERENCE" => DatabaseInfo::Reference.connect().await,
+                    "REFERENCE" => DatabaseInfo::Reference.connect().await?,
                     "MONGODB" => {
-                        #[cfg(feature = "mongodb")]
-                        return DatabaseInfo::MongoDb {
+                        DatabaseInfo::MongoDb {
                             uri: config.database.mongodb,
                             database_name,
                         }
                         .connect()
-                        .await;
-
-                        #[cfg(not(feature = "mongodb"))]
-                        return Err("MongoDB not enabled.".to_string())
+                        .await?
                     }
                     _ => unreachable!("must specify REFERENCE or MONGODB"),
                 }
             }
-            DatabaseInfo::Reference => Ok(Database::Reference(Default::default())),
-            #[cfg(feature = "mongodb")]
+            DatabaseInfo::Reference => Database::Reference(Default::default()),
             DatabaseInfo::MongoDb { uri, database_name } => {
                 let client = ::mongodb::Client::with_uri_str(uri)
                     .await
                     .map_err(|_| "Failed to init db connection.".to_string())?;
 
-                Ok(Database::MongoDb(MongoDb(client, database_name)))
+                Database::MongoDb(MongoDb(client, database_name))
             }
-            #[cfg(feature = "mongodb")]
             DatabaseInfo::MongoDbFromClient(client, database_name) => {
-                Ok(Database::MongoDb(MongoDb(client, database_name)))
+                Database::MongoDb(MongoDb(client, database_name))
             }
-        }
+        })
     }
 }
 
@@ -234,16 +219,12 @@ impl Database {
         Authifier {
             database: match self {
                 Database::Reference(_) => Default::default(),
-                #[cfg(feature = "mongodb")]
                 Database::MongoDb(MongoDb(client, _)) => authifier::Database::MongoDb(
                     authifier::database::MongoDb(client.database("revolt")),
                 ),
             },
             config: auth_config,
-            #[cfg(feature = "tasks")]
             event_channel: Some(crate::tasks::authifier_relay::sender()),
-            #[cfg(not(feature = "tasks"))]
-            event_channel: None,
         }
     }
 }

crates/core/database/src/drivers/mongodb.rs

@@ -10,6 +10,7 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 
 database_derived!(
+    #[cfg(feature = "mongodb")]
     /// MongoDB implementation
     pub struct MongoDb(pub ::mongodb::Client, pub String);
 );

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

@@ -1,4 +1,5 @@
 use authifier::AuthifierEvent;
+use revolt_broker::event_stream;
 use revolt_result::Error;
 use serde::{Deserialize, Serialize};
 
@@ -144,13 +145,7 @@ pub enum EventV1 {
     },
 
     /// User joins server
-    ServerMemberJoin {
-        id: String,
-        // Deprecated: use member.id.user
-        #[deprecated = "Use member.id.user instead"]
-        user: String,
-        member: Member,
-    },
+    ServerMemberJoin { id: String, user: String },
 
     /// User left server
     ServerMemberLeave {
@@ -259,14 +254,16 @@ pub enum EventV1 {
 impl EventV1 {
     /// Publish helper wrapper
     pub async fn p(self, channel: String) {
-        #[cfg(not(debug_assertions))]
-        redis_kiss::p(channel, self).await;
-
         #[cfg(debug_assertions)]
         info!("Publishing event to {channel}: {self:?}");
 
+        let result = event_stream::publish_event(&channel, &self).await;
+
+        #[cfg(not(debug_assertions))]
+        result.ok();
+
         #[cfg(debug_assertions)]
-        redis_kiss::publish(channel, self).await.unwrap();
+        result.unwrap();
     }
 
     /// Publish user event

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

@@ -1,8 +1,8 @@
-use serde::{Serialize, Deserialize};
+use serde::Deserialize;
 
 use super::client::Ping;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 #[serde(tag = "type")]
 pub enum ClientMessage {
     Authenticate { token: String },

crates/core/database/src/lib.rs

@@ -25,9 +25,6 @@ pub use mongodb;
 #[macro_use]
 extern crate bson;
 
-#[cfg(not(feature = "async-std-runtime"))]
-compile_error!("async-std-runtime feature must be enabled.");
-
 #[macro_export]
 #[cfg(debug_assertions)]
 macro_rules! query {
@@ -106,7 +103,6 @@ pub mod util;
 pub use models::*;
 
 pub mod events;
-#[cfg(feature = "tasks")]
 pub mod tasks;
 
 mod amqp;

crates/core/database/src/models/admin_migrations/ops.rs

@@ -1,4 +1,3 @@
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/bots/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{Bot, FieldsBot, PartialBot};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/channel_invites/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Invite;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/channel_unreads/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::ChannelUnread;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/channel_webhooks/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{FieldsWebhook, PartialWebhook, Webhook};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -8,13 +8,10 @@ use serde::{Deserialize, Serialize};
 use ulid::Ulid;
 
 use crate::{
-    events::client::EventV1, Database, File, PartialServer,
+    events::client::EventV1, tasks::ack::AckEvent, Database, File, IntoDocumentPath, PartialServer,
     Server, SystemMessage, User, AMQP,
 };
 
-#[cfg(feature = "mongodb")]
-use crate::IntoDocumentPath;
-
 auto_derived!(
     #[serde(tag = "channel_type")]
     pub enum Channel {
@@ -649,11 +646,10 @@ impl Channel {
         .private(user.to_string())
         .await;
 
-        #[cfg(feature = "tasks")]
         crate::tasks::ack::queue_ack(
             self.id().to_string(),
             user.to_string(),
-            crate::tasks::ack::AckEvent::AckMessage {
+            AckEvent::AckMessage {
                 id: message.to_string(),
             },
         )
@@ -770,7 +766,6 @@ impl Channel {
     }
 }
 
-#[cfg(feature = "mongodb")]
 impl IntoDocumentPath for FieldsChannel {
     fn as_path(&self) -> Option<&'static str> {
         Some(match self {

crates/core/database/src/models/channels/ops.rs

@@ -1,7 +1,5 @@
 use crate::{revolt_result::Result, Channel, FieldsChannel, PartialChannel};
 use revolt_permissions::OverrideField;
-
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/emojis/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Emoji;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/file_hashes/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::FileHash;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/files/ops.rs

@@ -4,7 +4,6 @@ use crate::File;
 
 use super::FileUsedFor;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -14,6 +14,7 @@ use validator::Validate;
 
 use crate::{
     events::client::EventV1,
+    tasks::{self, ack::AckEvent},
     util::{
         bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
         permissions::DatabasePermissionQuery,
@@ -21,9 +22,6 @@ use crate::{
     Channel, Database, Emoji, File, User, AMQP,
 };
 
-#[cfg(feature = "tasks")]
-use crate::tasks::{self, ack::AckEvent};
-
 auto_derived_partial!(
     /// Message
     pub struct Message {
@@ -618,11 +616,9 @@ impl Message {
             .await;
 
         // Update last_message_id
-        #[cfg(feature = "tasks")]
         tasks::last_message_id::queue(self.channel.to_string(), self.id.to_string(), is_dm).await;
 
         // Add mentions for affected users
-        #[cfg(feature = "tasks")]
         if !mentions_elsewhere {
             if let Some(mentions) = &self.mentions {
                 tasks::ack::queue_message(
@@ -641,7 +637,6 @@ impl Message {
         }
 
         // Generate embeds
-        #[cfg(feature = "tasks")]
         if generate_embeds {
             if let Some(content) = &self.content {
                 tasks::process_embeds::queue(
@@ -678,12 +673,10 @@ impl Message {
         )
         .await?;
 
-
         if !self.has_suppressed_notifications()
             && (self.mentions.is_some() || self.contains_mass_push_mention())
         {
             // send Push notifications
-            #[cfg(feature = "tasks")]
             tasks::ack::queue_message(
                 self.channel.to_string(),
                 AckEvent::ProcessMessage {

crates/core/database/src/models/messages/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -38,10 +38,7 @@ pub use servers::*;
 pub use user_settings::*;
 pub use users::*;
 
-use crate::{Database, ReferenceDb};
-
-#[cfg(feature = "mongodb")]
-use crate::MongoDb;
+use crate::{Database, MongoDb, ReferenceDb};
 
 pub trait AbstractDatabase:
     Sync
@@ -69,8 +66,6 @@ pub trait AbstractDatabase:
 }
 
 impl AbstractDatabase for ReferenceDb {}
-
-#[cfg(feature = "mongodb")]
 impl AbstractDatabase for MongoDb {}
 
 impl std::ops::Deref for Database {
@@ -79,7 +74,6 @@ impl std::ops::Deref for Database {
     fn deref(&self) -> &Self::Target {
         match &self {
             Database::Reference(dummy) => dummy,
-            #[cfg(feature = "mongodb")]
             Database::MongoDb(mongo) => mongo,
         }
     }

crates/core/database/src/models/policy_changes/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::PolicyChange;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/ratelimit_events/ops.rs

@@ -1,8 +1,6 @@
 use std::time::Duration;
 
 use crate::{revolt_result::Result, RatelimitEvent, RatelimitEventType};
-
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/safety_reports/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Report;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/safety_snapshots/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::Snapshot;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/server_bans/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{MemberCompositeKey, ServerBan};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -30,9 +30,6 @@ auto_derived_partial!(
         /// Timestamp this member is timed out until
         #[serde(skip_serializing_if = "Option::is_none")]
         pub timeout: Option<Timestamp>,
-        // This value only exists in the database, not the models.
-        // If it is not-None, the database layer should return None to member fetching queries.
-        // pub pending_deletion_at: Option<Timestamp>
     },
     "PartialMember"
 );
@@ -53,7 +50,6 @@ auto_derived!(
         Avatar,
         Roles,
         Timeout,
-        JoinedAt,
     }
 
     /// Member removal intention
@@ -94,7 +90,7 @@ impl Member {
             return Err(create_error!(AlreadyInServer));
         }
 
-        let mut member = Member {
+        let member = Member {
             id: MemberCompositeKey {
                 server: server.id.to_string(),
                 user: user.id.to_string(),
@@ -102,9 +98,7 @@ impl Member {
             ..Default::default()
         };
 
-        if let Some(updated) = db.insert_or_merge_member(&member).await? {
-            member = updated;
-        }
+        db.insert_member(&member).await?;
 
         let should_fetch = channels.is_none();
         let mut channels = channels.unwrap_or_default();
@@ -130,7 +124,6 @@ impl Member {
         EventV1::ServerMemberJoin {
             id: server.id.clone(),
             user: user.id.clone(),
-            member: member.clone().into(),
         }
         .p(server.id.clone())
         .await;
@@ -193,7 +186,6 @@ impl Member {
 
     pub fn remove_field(&mut self, field: &FieldsMember) {
         match field {
-            FieldsMember::JoinedAt => (),
             FieldsMember::Avatar => self.avatar = None,
             FieldsMember::Nickname => self.nickname = None,
             FieldsMember::Roles => self.roles.clear(),
@@ -232,7 +224,7 @@ impl Member {
         intention: RemovalIntention,
         silent: bool,
     ) -> Result<()> {
-        db.soft_delete_member(&self.id).await?;
+        db.delete_member(&self.id).await?;
 
         EventV1::ServerMemberLeave {
             id: self.id.server.to_string(),
@@ -268,74 +260,3 @@ impl Member {
         Ok(())
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use iso8601_timestamp::{Duration, Timestamp};
-    use revolt_models::v0::DataCreateServer;
-
-    use crate::{Member, PartialMember, RemovalIntention, Server, User};
-
-    #[async_std::test]
-    async fn muted_member_rejoin() {
-        database_test!(|db| async move {
-            match db {
-                crate::Database::Reference(_) => return,
-                crate::Database::MongoDb(_) => (),
-            }
-            let owner = User::create(&db, "Server Owner".to_string(), None, None)
-                .await
-                .unwrap();
-
-            let kickable_user = User::create(&db, "Member".to_string(), None, None)
-                .await
-                .unwrap();
-
-            let server = Server::create(
-                &db,
-                DataCreateServer {
-                    name: "Server".to_string(),
-                    description: None,
-                    nsfw: None,
-                },
-                &owner,
-                false,
-            )
-            .await
-            .unwrap()
-            .0;
-
-            Member::create(&db, &server, &owner, None).await.unwrap();
-            let mut kickable_member = Member::create(&db, &server, &kickable_user, None)
-                .await
-                .unwrap()
-                .0;
-
-            kickable_member
-                .update(
-                    &db,
-                    PartialMember {
-                        timeout: Some(Timestamp::now_utc() + Duration::minutes(5)),
-                        ..Default::default()
-                    },
-                    vec![],
-                )
-                .await
-                .unwrap();
-
-            assert!(kickable_member.in_timeout());
-
-            kickable_member
-                .remove(&db, &server, RemovalIntention::Kick, false)
-                .await
-                .unwrap();
-
-            let kickable_member = Member::create(&db, &server, &kickable_user, None)
-                .await
-                .unwrap()
-                .0;
-
-            assert!(kickable_member.in_timeout())
-        });
-    }
-}

crates/core/database/src/models/server_members/ops.rs

@@ -1,20 +1,16 @@
-#[cfg(feature = "mongodb")]
-use ::mongodb::{ClientSession, SessionCursor};
-
+use ::mongodb::SessionCursor;
 use revolt_result::Result;
 
 use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 
 #[derive(Debug)]
-#[allow(clippy::large_enum_variant)]
 pub enum ChunkedServerMembersGenerator {
     #[cfg(feature = "mongodb")]
     MongoDb {
-        session: ClientSession,
+        session: ::mongodb::ClientSession,
         cursor: Option<SessionCursor<Member>>,
     },
 
@@ -26,7 +22,7 @@ pub enum ChunkedServerMembersGenerator {
 
 impl ChunkedServerMembersGenerator {
     #[cfg(feature = "mongodb")]
-    pub fn new_mongo(session: ClientSession, cursor: SessionCursor<Member>) -> Self {
+    pub fn new_mongo(session: ::mongodb::ClientSession, cursor: SessionCursor<Member>) -> Self {
         ChunkedServerMembersGenerator::MongoDb {
             session,
             cursor: Some(cursor),
@@ -73,13 +69,13 @@ impl ChunkedServerMembersGenerator {
 #[async_trait]
 pub trait AbstractServerMembers: Sync + Send {
     /// Insert a new server member into the database
-    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>>;
+    async fn insert_member(&self, member: &Member) -> Result<()>;
 
     /// Fetch a server member by their id
     async fn fetch_member(&self, server_id: &str, user_id: &str) -> Result<Member>;
 
     /// Fetch all members in a server
-    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>>;
+    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>>;
 
     /// Fetch all members in a server as an iterator
     async fn fetch_all_members_chunked(
@@ -100,10 +96,10 @@ pub trait AbstractServerMembers: Sync + Send {
     ) -> Result<ChunkedServerMembersGenerator>;
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>>;
+    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>>;
 
     /// Fetch multiple members by their ids
-    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>>;
+    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>>;
 
     /// Fetch member count of a server
     async fn fetch_member_count(&self, server_id: &str) -> Result<usize>;
@@ -119,14 +115,6 @@ pub trait AbstractServerMembers: Sync + Send {
         remove: Vec<FieldsMember>,
     ) -> Result<()>;
 
-    /// Marks a user as no longer a member of a server, while retaining the database value.
-    /// This is used to keep information such as timeouts in place, but will remove information such as join date and applied roles.
-    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
-
-    /// Forcibly delete a server member by their id.
-    /// This will cancel any pending timeouts or other longer term actions, and they will not be reapplied on rejoin.
-    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
-
-    /// Fetch all members who have been marked for deletion.
-    async fn remove_dangling_members(&self) -> Result<()>;
+    /// Delete a server member by their id
+    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
 }

crates/core/database/src/models/server_members/ops/mongodb.rs

@@ -1,6 +1,4 @@
-use bson::Document;
 use futures::StreamExt;
-use iso8601_timestamp::Timestamp;
 use mongodb::options::ReadConcern;
 use revolt_result::Result;
 
@@ -13,42 +11,9 @@ static COL: &str = "server_members";
 
 #[async_trait]
 impl AbstractServerMembers for MongoDb {
-    /// Insert a new server member (or use the existing member if one is found)
-    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
-        let existing: Result<Option<Document>> = query!(
-            self,
-            find_one,
-            COL,
-            doc! {
-                "_id.server": &member.id.server,
-                "_id.user": &member.id.user,
-                "pending_deletion_at": {"$exists": true}
-            }
-        );
-        // Update the existing record if it exist, otherwise make a new record
-        if existing.is_ok_and(|x| x.is_some()) {
-            self.col::<Member>(COL)
-                .find_one_and_update(
-                    doc! {
-                        "_id.server": &member.id.server,
-                        "_id.user": &member.id.user,
-                    },
-                    doc! {
-                        "$set": {
-                            "joined_at": member.joined_at.duration_since(Timestamp::UNIX_EPOCH).whole_seconds(),
-                        },
-                        "$unset": {
-                            "pending_deletion_at": ""
-                        }
-                    },
-                )
-                .return_document(mongodb::options::ReturnDocument::After)
-                .await
-                .map_err(|_| create_database_error!("update_one", COL))
-        } else {
-            query!(self, insert_one, COL, &member).map(|_| ())?;
-            Ok(None)
-        }
+    /// Insert a new server member into the database
+    async fn insert_member(&self, member: &Member) -> Result<()> {
+        query!(self, insert_one, COL, &member).map(|_| ())
     }
 
     /// Fetch a server member by their id
@@ -59,20 +24,18 @@ impl AbstractServerMembers for MongoDb {
             COL,
             doc! {
                 "_id.server": server_id,
-                "_id.user": user_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.user": user_id
             }
         )?
         .ok_or_else(|| create_error!(NotFound))
     }
 
     /// Fetch all members in a server
-    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
-                "_id.server": server_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.server": server_id
             })
             .await
             .map_err(|_| create_database_error!("find", COL))?
@@ -176,12 +139,11 @@ impl AbstractServerMembers for MongoDb {
     }
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
-                "_id.user": user_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.user": user_id
             })
             .await
             .map_err(|_| create_database_error!("find", COL))?
@@ -197,12 +159,11 @@ impl AbstractServerMembers for MongoDb {
     }
 
     /// Fetch multiple members by their ids
-    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
+    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
                 "_id.server": server_id,
-                "pending_deletion_at": {"$exists": false},
                 "_id.user": {
                     "$in": ids
                 }
@@ -224,8 +185,7 @@ impl AbstractServerMembers for MongoDb {
     async fn fetch_member_count(&self, server_id: &str) -> Result<usize> {
         self.col::<Member>(COL)
             .count_documents(doc! {
-                "_id.server": server_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.server": server_id
             })
             .await
             .map(|c| c as usize)
@@ -236,8 +196,7 @@ impl AbstractServerMembers for MongoDb {
     async fn fetch_server_count(&self, user_id: &str) -> Result<usize> {
         self.col::<Member>(COL)
             .count_documents(doc! {
-                "_id.user": user_id,
-                "pending_deletion_at": {"$exists": false}
+                "_id.user": user_id
             })
             .await
             .map(|c| c as usize)
@@ -266,42 +225,8 @@ impl AbstractServerMembers for MongoDb {
         .map(|_| ())
     }
 
-    /// Marks a member for deletion.
-    /// This will remove the record if the user has no pending actions (eg. timeout),
-    /// otherwise will slate the record for deletion by revolt_crond once the actions expire.
-    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
-        let member = self.fetch_member(&id.server, &id.user).await;
-        if let Ok(member) = member {
-            if member.in_timeout() {
-                self.col::<Document>(COL)
-                    .update_many(
-                        doc! {
-                            "_id.server": &id.server,
-                            "_id.user": &id.user,
-                        },
-                        doc! {
-                            "$set": {"pending_deletion_at": format!("{}", member.timeout.unwrap().format())},
-                            "$unset": {
-                                "joined_at": "",
-                                "avatar": "",
-                                "nickname": "",
-                                "roles": ""
-                            }
-                        },
-                    )
-                    .await
-                    .map(|_| ())
-                    .map_err(|_| create_database_error!("update_many", COL))
-            } else {
-                self.force_delete_member(id).await
-            }
-        } else {
-            Err(create_database_error!("fetch_member", COL))
-        }
-    }
-
     /// Delete a server member by their id
-    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
         query!(
             self,
             delete_one,
@@ -313,25 +238,11 @@ impl AbstractServerMembers for MongoDb {
         )
         .map(|_| ())
     }
-
-    async fn remove_dangling_members(&self) -> Result<()> {
-        let now = Timestamp::now_utc();
-        let date = bson::to_bson(&now).expect("Failed to serialize timestamp");
-
-        self.col::<Document>(COL)
-            .delete_many(doc! {
-                "pending_deletion_at": {"$lt": date}
-            })
-            .await
-            .map(|_| ())
-            .map_err(|_| create_database_error!("count_documents", COL))
-    }
 }
 
 impl IntoDocumentPath for FieldsMember {
     fn as_path(&self) -> Option<&'static str> {
         Some(match self {
-            FieldsMember::JoinedAt => "joined_at",
             FieldsMember::Avatar => "avatar",
             FieldsMember::Nickname => "nickname",
             FieldsMember::Roles => "roles",

crates/core/database/src/models/server_members/ops/reference.rs

@@ -8,13 +8,13 @@ use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
 #[async_trait]
 impl AbstractServerMembers for ReferenceDb {
     /// Insert a new server member into the database
-    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
+    async fn insert_member(&self, member: &Member) -> Result<()> {
         let mut server_members = self.server_members.lock().await;
         if server_members.contains_key(&member.id) {
             Err(create_database_error!("insert", "member"))
         } else {
             server_members.insert(member.id.clone(), member.clone());
-            Ok(None)
+            Ok(())
         }
     }
 
@@ -31,7 +31,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch all members in a server
-    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(server_members
             .values()
@@ -105,7 +105,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(server_members
             .values()
@@ -115,7 +115,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch multiple members by their ids
-    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
+    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(ids
             .iter()
@@ -169,26 +169,8 @@ impl AbstractServerMembers for ReferenceDb {
         }
     }
 
-    /// Soft delete a member
-    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
-        let mut server_members = self.server_members.lock().await;
-
-        let member = server_members.get_mut(id);
-        if let Some(member) = member {
-            if member.in_timeout() {
-                panic!("Soft deletion is not implemented.")
-            } else if server_members.remove(id).is_some() {
-                Ok(())
-            } else {
-                Err(create_error!(NotFound))
-            }
-        } else {
-            Err(create_error!(NotFound))
-        }
-    }
-
     /// Delete a server member by their id
-    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
         let mut server_members = self.server_members.lock().await;
         if server_members.remove(id).is_some() {
             Ok(())
@@ -196,8 +178,4 @@ impl AbstractServerMembers for ReferenceDb {
             Err(create_error!(NotFound))
         }
     }
-
-    async fn remove_dangling_members(&self) -> Result<()> {
-        todo!()
-    }
 }

crates/core/database/src/models/servers/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::{FieldsRole, FieldsServer, PartialRole, PartialServer, Role, Server};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/user_settings/ops.rs

@@ -2,7 +2,6 @@ use revolt_result::Result;
 
 use crate::UserSettings;
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

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

@@ -1,21 +1,14 @@
-use axum::{
-    extract::{FromRef, FromRequestParts},
-    http::request::Parts,
-};
+use axum::{extract::FromRequestParts, http::request::Parts};
 
 use revolt_result::{create_error, Error, Result};
 
 use crate::{Database, User};
 
 #[async_trait::async_trait]
-impl<S: Send + Sync> FromRequestParts<S> for User
-where
-    Database: FromRef<S>,
-{
+impl FromRequestParts<Database> for User {
     type Rejection = Error;
 
-    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<User> {
-        let db = Database::from_ref(state);
+    async fn from_request_parts(parts: &mut Parts, db: &Database) -> Result<User> {
         if let Some(Ok(bot_token)) = parts.headers.get("x-bot-token").map(|v| v.to_str()) {
             let bot = db.fetch_bot_by_token(bot_token).await?;
             db.fetch_user(&bot.id).await

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

@@ -4,7 +4,6 @@ use revolt_result::Result;
 
 use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
 
-#[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 

crates/core/database/src/models/users/ops/mongodb.rs

@@ -212,34 +212,16 @@ impl AbstractUsers for MongoDb {
         partial: &PartialUser,
         remove: Vec<FieldsUser>,
     ) -> Result<()> {
-        if remove.contains(&FieldsUser::StatusText) && partial.status.is_some() {
-            // stupid-ass workaround to fix mongo conflicting the same item
-            let _: Result<()> = query!(
-                self,
-                update_one_by_id,
-                COL,
-                id,
-                PartialUser {
-                    ..Default::default()
-                },
-                remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
-                None
-            )
-            .map(|_| ());
-
-            query!(self, update_one_by_id, COL, id, partial, vec![], None).map(|_| ())
-        } else {
-            query!(
-                self,
-                update_one_by_id,
-                COL,
-                id,
-                partial,
-                remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
-                None
-            )
-            .map(|_| ())
-        }
+        query!(
+            self,
+            update_one_by_id,
+            COL,
+            id,
+            partial,
+            remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
+            None
+        )
+        .map(|_| ())
     }
 
     /// Set relationship with another user

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

@@ -707,7 +707,6 @@ impl From<crate::FieldsMember> for FieldsMember {
             crate::FieldsMember::Nickname => FieldsMember::Nickname,
             crate::FieldsMember::Roles => FieldsMember::Roles,
             crate::FieldsMember::Timeout => FieldsMember::Timeout,
-            crate::FieldsMember::JoinedAt => FieldsMember::JoinedAt,
         }
     }
 }
@@ -719,7 +718,6 @@ impl From<FieldsMember> for crate::FieldsMember {
             FieldsMember::Nickname => crate::FieldsMember::Nickname,
             FieldsMember::Roles => crate::FieldsMember::Roles,
             FieldsMember::Timeout => crate::FieldsMember::Timeout,
-            FieldsMember::JoinedAt => crate::FieldsMember::JoinedAt,
         }
     }
 }

crates/core/database/src/util/reference.rs

@@ -14,40 +14,41 @@ use crate::{
 };
 
 /// Reference to some object in the database
-pub struct Reference<'a> {
+#[derive(Serialize, Deserialize)]
+pub struct Reference {
     /// Id of object
-    pub id: &'a str,
+    pub id: String,
 }
 
-impl<'a> Reference<'a> {
+impl Reference {
     /// Create a Ref from an unchecked string
-    pub fn from_unchecked(id: &'a str) -> Reference<'a> {
+    pub fn from_unchecked(id: String) -> Reference {
         Reference { id }
     }
 
     /// Fetch ban from Ref
     pub async fn as_ban(&self, db: &Database, server: &str) -> Result<ServerBan> {
-        db.fetch_ban(server, self.id).await
+        db.fetch_ban(server, &self.id).await
     }
 
     /// Fetch bot from Ref
     pub async fn as_bot(&self, db: &Database) -> Result<Bot> {
-        db.fetch_bot(self.id).await
+        db.fetch_bot(&self.id).await
     }
 
     /// Fetch emoji from Ref
     pub async fn as_emoji(&self, db: &Database) -> Result<Emoji> {
-        db.fetch_emoji(self.id).await
+        db.fetch_emoji(&self.id).await
     }
 
     /// Fetch channel from Ref
     pub async fn as_channel(&self, db: &Database) -> Result<Channel> {
-        db.fetch_channel(self.id).await
+        db.fetch_channel(&self.id).await
     }
 
     /// Fetch invite from Ref or create invite to server if discoverable
     pub async fn as_invite(&self, db: &Database) -> Result<Invite> {
-        if ulid::Ulid::from_str(self.id).is_ok() {
+        if ulid::Ulid::from_str(&self.id).is_ok() {
             let server = self.as_server(db).await?;
             if !server.discoverable {
                 return Err(create_error!(NotFound));
@@ -64,18 +65,18 @@ impl<'a> Reference<'a> {
                     .ok_or(create_error!(NotFound))?,
             })
         } else {
-            db.fetch_invite(self.id).await
+            db.fetch_invite(&self.id).await
         }
     }
 
     /// Fetch message from Ref
     pub async fn as_message(&self, db: &Database) -> Result<Message> {
-        db.fetch_message(self.id).await
+        db.fetch_message(&self.id).await
     }
 
     /// Fetch message from Ref and validate channel
     pub async fn as_message_in_channel(&self, db: &Database, channel: &str) -> Result<Message> {
-        let msg = db.fetch_message(self.id).await?;
+        let msg = db.fetch_message(&self.id).await?;
         if msg.channel != channel {
             return Err(create_error!(NotFound));
         }
@@ -85,36 +86,36 @@ impl<'a> Reference<'a> {
 
     /// Fetch member from Ref
     pub async fn as_member(&self, db: &Database, server: &str) -> Result<Member> {
-        db.fetch_member(server, self.id).await
+        db.fetch_member(server, &self.id).await
     }
 
     /// Fetch server from Ref
     pub async fn as_server(&self, db: &Database) -> Result<Server> {
-        db.fetch_server(self.id).await
+        db.fetch_server(&self.id).await
     }
 
     /// Fetch user from Ref
     pub async fn as_user(&self, db: &Database) -> Result<User> {
-        db.fetch_user(self.id).await
+        db.fetch_user(&self.id).await
     }
 
     /// Fetch webhook from Ref
     pub async fn as_webhook(&self, db: &Database) -> Result<Webhook> {
-        db.fetch_webhook(self.id).await
+        db.fetch_webhook(&self.id).await
     }
 }
 
 #[cfg(feature = "rocket-impl")]
-impl<'r> FromParam<'r> for Reference<'r> {
+impl<'r> FromParam<'r> for Reference {
     type Error = &'r str;
 
     fn from_param(param: &'r str) -> Result<Self, Self::Error> {
-        Ok(Reference::from_unchecked(param))
+        Ok(Reference::from_unchecked(param.into()))
     }
 }
 
 #[cfg(feature = "rocket-impl")]
-impl<'a> JsonSchema for Reference<'a> {
+impl JsonSchema for Reference {
     fn schema_name() -> String {
         "Id".to_string()
     }

crates/core/database/src/util/test_fixtures.rs

@@ -39,9 +39,7 @@ pub async fn load_fixture(db: &Database, input: &str) -> HashMap<String, String>
             LoadedFixture::User(user) => db.insert_user(&user).await.unwrap(),
             LoadedFixture::Channel(channel) => db.insert_channel(&channel).await.unwrap(),
             LoadedFixture::Server(server) => db.insert_server(&server).await.unwrap(),
-            LoadedFixture::ServerMember(member) => {
-                db.insert_or_merge_member(&member).await.unwrap();
-            }
+            LoadedFixture::ServerMember(member) => db.insert_member(&member).await.unwrap(),
         }
     }
 

crates/core/files/src/lib.rs

@@ -80,9 +80,6 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
         .decrypt_in_place(nonce, b"", &mut buf)
         .map_err(|_| create_error!(InternalError))?;
 
-    // Remove the authentication tag bytes that were added during encryption
-    buf.truncate(buf.len() - AUTHENTICATION_TAG_SIZE_BYTES);
-
     Ok(buf)
 }
 

crates/core/models/src/v0/bots.rs

@@ -132,8 +132,8 @@ auto_derived!(
         #[cfg_attr(feature = "validator", validate(length(min = 1, max = 2048)))]
         pub interactions_url: Option<String>,
         /// Fields to remove from bot object
-        #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsBot>,
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
+        pub remove: Option<Vec<FieldsBot>>,
     }
 
     /// Where we are inviting a bot to

crates/core/models/src/v0/channels.rs

@@ -207,7 +207,7 @@ auto_derived!(
 
         /// Fields to remove from channel
         #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsChannel>,
+        pub remove: Option<Vec<FieldsChannel>>,
     }
 
     /// Create new group

crates/core/models/src/v0/messages.rs

@@ -215,7 +215,7 @@ auto_derived!(
     #[derive(Default)]
     #[cfg_attr(feature = "validator", derive(Validate))]
     pub struct SendableEmbed {
-        #[cfg_attr(feature = "validator", validate(length(min = 1, max = 256)))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1, max = 128)))]
         pub icon_url: Option<String>,
         #[cfg_attr(feature = "validator", validate(length(min = 1, max = 256)))]
         pub url: Option<String>,

crates/core/models/src/v0/server_members.rs

@@ -77,7 +77,6 @@ auto_derived!(
         Avatar,
         Roles,
         Timeout,
-        JoinedAt,
     }
 
     /// Member removal intention
@@ -125,7 +124,7 @@ auto_derived!(
         /// Timestamp this member is timed out until
         pub timeout: Option<Timestamp>,
         /// Fields to remove from channel object
-        #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsMember>,
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
+        pub remove: Option<Vec<FieldsMember>>,
     }
 );

crates/core/models/src/v0/servers.rs

@@ -175,8 +175,6 @@ auto_derived!(
         /// Ranking position
         ///
         /// Smaller values take priority.
-        ///
-        /// **Removed** - no effect, use the edit server role positions route
         pub rank: Option<i64>,
     }
 
@@ -249,8 +247,8 @@ auto_derived!(
         pub analytics: Option<bool>,
 
         /// Fields to remove from server object
-        #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsServer>,
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
+        pub remove: Option<Vec<FieldsServer>>,
     }
 
     /// New role information
@@ -272,8 +270,8 @@ auto_derived!(
         /// **Removed** - no effect, use the edit server role positions route
         pub rank: Option<i64>,
         /// Fields to remove from role object
-        #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsRole>,
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
+        pub remove: Option<Vec<FieldsRole>>,
     }
 
     /// New role permissions

crates/core/models/src/v0/users.rs

@@ -245,8 +245,8 @@ auto_derived!(
         pub flags: Option<i32>,
 
         /// Fields to remove from user object
-        #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsUser>,
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
+        pub remove: Option<Vec<FieldsUser>>,
     }
 
     /// User flag reponse
@@ -255,14 +255,12 @@ auto_derived!(
         pub flags: i32,
     }
 
-    /// Mutual friends, servers, groups and DMs response
+    /// Mutual friends and servers response
     pub struct MutualResponse {
         /// Array of mutual user IDs that both users are friends with
         pub users: Vec<String>,
         /// Array of mutual server IDs that both users are in
         pub servers: Vec<String>,
-        /// Array of mutual group and dm IDs that both users are in
-        pub channels: Vec<String>,
     }
 
     /// Bot information for if the user is a bot

crates/core/parser/Cargo.toml

@@ -2,8 +2,7 @@
 name = "revolt-parser"
 version = "0.8.8"
 edition = "2021"
-license = "MIT"
-authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]
+license = "AGPL-3.0-or-later"
 description = "Revolt Backend: Message Parser"
 
 [dependencies]

crates/core/parser/LICENSE

@@ -1,9 +0,0 @@
-MIT License
-
-Copyright (c) 2024 Pawel Makles
-
-Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

crates/core/parser/src/lib.rs

@@ -4,18 +4,15 @@ use logos::Logos;
 
 #[derive(Debug, Clone, Logos, PartialEq)]
 #[logos(skip "\n")]
-#[logos(subpattern id="[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}")]
-pub enum MessageToken<'a> {
+pub enum MessageToken {
     #[token("\\")]
     Escape,
-    #[regex("```[^`\n]*", |_| 3)]
-    #[regex("``", |_| 2)]
-    #[regex("`", |_| 1)]
+    #[regex("(```[^`\n]*)|(``)|`", |lex| lex.slice().to_owned().chars().filter(|&c| c == '`').count())]
     CodeblockMarker(usize),
-    #[regex("<@(?&id)>", |lex| &lex.slice()[2..lex.slice().len() - 1])]
-    UserMention(&'a str),
-    #[regex("<%(?&id)>", |lex| &lex.slice()[2..lex.slice().len() - 1],)]
-    RoleMention(&'a str),
+    #[regex("<@[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}>", |lex| lex.slice()[2..lex.slice().len() - 1].to_owned())]
+    UserMention(String),
+    #[regex("<%[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}>", |lex| lex.slice()[2..lex.slice().len() - 1].to_owned())]
+    RoleMention(String),
     #[token("@everyone")]
     MentionEveryone,
     #[token("@online")]
@@ -30,13 +27,13 @@ pub struct MessageResults {
     pub mentions_online: bool
 }
 
-struct MessageParserIterator<'a, I> {
+struct MessageParserIterator<I> {
     inner: I,
-    temp: VecDeque<MessageToken<'a>>
+    temp: VecDeque<MessageToken>
 }
 
-impl<'a, I: Iterator<Item = MessageToken<'a>>> Iterator for MessageParserIterator<'a, I> {
-    type Item = MessageToken<'a>;
+impl<I: Iterator<Item = MessageToken>> Iterator for MessageParserIterator<I> {
+    type Item = MessageToken;
 
     fn next(&mut self) -> Option<Self::Item> {
         if !self.temp.is_empty() {
@@ -83,8 +80,8 @@ pub fn parse_message(text: &str) -> MessageResults {
         match token {
             MessageToken::Escape => {}
             MessageToken::CodeblockMarker(_) => {},
-            MessageToken::UserMention(id) => { results.user_mentions.insert(id.to_string()); },
-            MessageToken::RoleMention(id) => { results.role_mentions.insert(id.to_string()); },
+            MessageToken::UserMention(id) => { results.user_mentions.insert(id); },
+            MessageToken::RoleMention(id) => { results.role_mentions.insert(id); },
             MessageToken::MentionEveryone => results.mentions_everyone = true,
             MessageToken::MentionOnline => results.mentions_online = true,
         };
@@ -109,7 +106,7 @@ mod tests {
         let output = parse_message_iter("Hello <@01FD58YK5W7QRV5H3D64KTQYX3>.").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 1);
-        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
     }
 
     #[test]
@@ -117,7 +114,7 @@ mod tests {
         let output = parse_message_iter("Hello <%01FD58YK5W7QRV5H3D64KTQYX3>.").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 1);
-        assert_eq!(output[0], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[0], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
     }
 
     #[test]
@@ -141,8 +138,8 @@ mod tests {
         let output = parse_message_iter("Hello <@01FD58YK5W7QRV5H3D64KTQYX3>, <%01FD58YK5W7QRV5H3D64KTQYX3>, @everyone and @online.").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 4);
-        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
-        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
         assert_eq!(output[2], MessageToken::MentionEveryone);
         assert_eq!(output[3], MessageToken::MentionOnline);
     }
@@ -152,8 +149,8 @@ mod tests {
         let output = parse_message_iter("<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 4);
-        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
-        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
         assert_eq!(output[2], MessageToken::MentionEveryone);
         assert_eq!(output[3], MessageToken::MentionOnline);
     }
@@ -173,14 +170,15 @@ mod tests {
 
         assert_eq!(output.len(), 5);
         assert_eq!(output[0], MessageToken::CodeblockMarker(3));
-        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
-        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
         assert_eq!(output[3], MessageToken::MentionEveryone);
         assert_eq!(output[4], MessageToken::MentionOnline);
     }
 
     #[test]
     fn test_inline_codeblock_no_mentions() {
+
         let output = parse_message_iter("`<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online`").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 2);
@@ -194,21 +192,12 @@ mod tests {
 
         assert_eq!(output.len(), 5);
         assert_eq!(output[0], MessageToken::CodeblockMarker(1));
-        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
-        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
         assert_eq!(output[3], MessageToken::MentionEveryone);
         assert_eq!(output[4], MessageToken::MentionOnline);
     }
 
-    #[test]
-    fn test_codeblock_with_language_no_mentions() {
-        let output = parse_message_iter("```rust\n<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online```").collect::<Vec<_>>();
-
-        assert_eq!(output.len(), 2);
-        assert_eq!(output[0], MessageToken::CodeblockMarker(3));
-        assert_eq!(output[1], MessageToken::CodeblockMarker(3));
-    }
-
     #[test]
     fn test_double_inline_codeblock() {
         let output = parse_message_iter("``this should not ping @everyone``").collect::<Vec<_>>();

crates/core/permissions/src/models/mod.rs

@@ -8,18 +8,10 @@ pub use server::*;
 pub use user::*;
 
 /// Holds a permission value to manipulate.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[derive(Clone, Debug)]
 pub struct PermissionValue(u64);
 
 impl PermissionValue {
-    pub fn from_raw(value: u64) -> Self {
-        Self(value)
-    }
-
-    pub fn into_raw(self) -> u64 {
-        self.0
-    }
-
     /// Apply a given override to this value
     pub fn apply(&mut self, v: Override) {
         self.allow(v.allow);

crates/core/ratelimits/Cargo.toml

@@ -1,26 +0,0 @@
-[package]
-name = "revolt-ratelimits"
-version = "0.8.8"
-edition = "2024"
-
-[features]
-rocket = ["dep:rocket", "dep:revolt_rocket_okapi", "revolt-database/rocket-impl"]
-axum = ["dep:axum", "revolt-database/axum-impl"]
-
-default = ["rocket", "axum"]
-
-[dependencies]
-revolt-database = { version = "0.8.8", path = "../database"}
-revolt-result = { version = "0.8.8", path = "../result" }
-revolt-config = { version = "0.8.8", path = "../config" }
-
-rocket = { version = "0.5.1", optional = true }
-revolt_rocket_okapi = { version = "0.10.0", optional = true }
-
-axum = { version = "0.7.5", optional = true, features = ["macros"] }
-
-serde = { version = "1", features = ["derive"] }
-authifier = { version = "1.0.15" }
-dashmap = "5.2.0"
-async-trait = "0.1.81"
-log = "0.4"

crates/core/ratelimits/src/axum.rs

@@ -1,194 +0,0 @@
-use std::net::SocketAddr;
-
-use async_trait::async_trait;
-use axum::{
-    Json, RequestPartsExt, Router,
-    body::Body,
-    extract::{ConnectInfo, FromRef, FromRequestParts, State},
-    http::{HeaderValue, Request, StatusCode, request::Parts},
-    middleware::Next,
-    response::{IntoResponse, Response},
-    routing::get,
-};
-use revolt_database::{Database, User};
-use revolt_config::config;
-
-use crate::ratelimiter::{RatelimitInformation, Ratelimiter, RequestKind};
-
-#[derive(Clone, Copy)]
-pub struct AxumRequestKind;
-
-impl RequestKind for AxumRequestKind {
-    type R<'a> = Parts;
-}
-
-pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<AxumRequestKind>;
-
-fn to_ip(parts: &Parts) -> String {
-    parts
-        .extensions
-        .get::<ConnectInfo<SocketAddr>>()
-        .map(|info| info.ip().to_string())
-        .unwrap_or_default()
-}
-
-async fn to_real_ip(parts: &Parts) -> String {
-    if config().await.api.security.trust_cloudflare {
-        parts
-            .headers
-            .get("CF-Connecting-IP")
-            .map(|x| x.to_str().unwrap().to_string())
-            .unwrap_or_else(|| to_ip(parts))
-    } else {
-        to_ip(parts)
-    }
-}
-
-#[async_trait]
-impl<S: Send + Sync> FromRequestParts<S> for Ratelimiter
-where
-    Database: FromRef<S>,
-    RatelimitStorage: FromRef<S>,
-{
-    type Rejection = Json<Ratelimiter>;
-
-    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
-        if parts
-            .extensions
-            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
-            .is_none()
-        {
-            let storage = RatelimitStorage::from_ref(state);
-
-            let identifier = if let Ok(user) = parts.extract_with_state::<User, _>(state).await {
-                user.id
-            } else {
-                to_real_ip(parts).await
-            };
-
-            let (bucket, resource) = storage.resolver.resolve_bucket(parts);
-            let limit = storage.resolver.resolve_bucket_limit(bucket);
-
-            let ratelimiter =
-                Ratelimiter::from(&storage.map, &identifier, limit, (bucket, resource));
-
-            parts.extensions.insert(ratelimiter.map_err(Json));
-        };
-
-        *parts
-            .extensions
-            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
-            .unwrap()
-    }
-}
-
-#[async_trait]
-impl<S: Send + Sync> FromRequestParts<S> for RatelimitInformation
-where
-    Database: FromRef<S>,
-    RatelimitStorage: FromRef<S>,
-{
-    type Rejection = Json<RatelimitInformation>;
-
-    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
-        if parts
-            .extensions
-            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
-            .is_none()
-        {
-            let ratelimiter = parts.extract_with_state::<Ratelimiter, S>(state).await;
-
-            parts.extensions.insert(ratelimiter);
-        };
-
-        let ratelimiter = *parts
-            .extensions
-            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
-            .unwrap();
-
-        match ratelimiter {
-            Ok(ratelimter) => Ok(RatelimitInformation::Success(ratelimter)),
-            Err(ratelimiter) => Err(Json(RatelimitInformation::Failure {
-                retry_after: ratelimiter.reset,
-            })),
-        }
-    }
-}
-
-pub async fn ratelimit_middleware(
-    State(database): State<Database>,
-    State(ratelimit_storage): State<RatelimitStorage>,
-    request: Request<Body>,
-    next: Next,
-) -> Response {
-    #[derive(axum::extract::FromRef)]
-    struct TempState {
-        database: Database,
-        ratelimit_storage: RatelimitStorage,
-    }
-
-    let state = TempState {
-        database,
-        ratelimit_storage,
-    };
-
-    let (mut parts, body) = request.into_parts();
-
-    let res = Ratelimiter::from_request_parts(&mut parts, &state).await;
-
-    let (Ok(ratelimiter) | Err(Json(ratelimiter))) = &res;
-
-    let mut response = if res.is_ok() {
-        let request = Request::from_parts(parts, body);
-
-        next.run(request).await
-    } else {
-        let ratelimit_info = RatelimitInformation::from_request_parts(&mut parts, &state).await;
-
-        ratelimit_info.map(Json).into_response()
-    };
-
-    let Ratelimiter {
-        key,
-        limit,
-        remaining,
-        reset,
-    } = ratelimiter;
-
-    let headers = response.headers_mut();
-
-    headers.insert(
-        "X-RateLimit-Limit",
-        HeaderValue::from_str(&limit.to_string()).unwrap(),
-    );
-    headers.insert(
-        "X-RateLimit-Bucket",
-        HeaderValue::from_str(&key.to_string()).unwrap(),
-    );
-    headers.insert(
-        "X-RateLimit-Remaining",
-        HeaderValue::from_str(&remaining.to_string()).unwrap(),
-    );
-    headers.insert(
-        "X-RateLimit-Reset-After",
-        HeaderValue::from_str(&reset.to_string()).unwrap(),
-    );
-
-    if res.is_err() {
-        *response.status_mut() = StatusCode::TOO_MANY_REQUESTS;
-    };
-
-    response
-}
-
-async fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
-    Json(info)
-}
-
-pub fn routes<S: Clone + Send + Sync + 'static>() -> Router<S>
-where
-    Database: FromRef<S>,
-    RatelimitStorage: FromRef<S>,
-{
-    Router::new().route("/ratelimit", get(ratelimit_info))
-}

crates/core/ratelimits/src/lib.rs

@@ -1,7 +0,0 @@
-pub mod ratelimiter;
-
-#[cfg(feature = "rocket")]
-pub mod rocket;
-
-#[cfg(feature = "axum")]
-pub mod axum;

crates/core/ratelimits/src/ratelimiter.rs

@@ -1,145 +0,0 @@
-use std::collections::hash_map::DefaultHasher;
-use std::hash::Hasher;
-use std::ops::Add;
-use std::sync::Arc;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
-
-use serde::Serialize;
-
-use dashmap::DashMap;
-
-pub trait RequestKind {
-    type R<'a>;
-}
-
-pub trait RatelimitResolver<R>: Send + Sync {
-    fn resolve_bucket<'a>(&self, request: &'a R) -> (&'a str, Option<&'a str>);
-    fn resolve_bucket_limit(&self, bucket: &str) -> u32;
-}
-
-#[derive(Clone)]
-pub struct RatelimitStorage<K: RequestKind> {
-    pub resolver: Arc<dyn for<'a> RatelimitResolver<K::R<'a>>>,
-    pub map: Arc<DashMap<u64, Entry>>,
-}
-
-impl<K: RequestKind> RatelimitStorage<K> {
-    pub fn new<R: for<'a> RatelimitResolver<K::R<'a>> + 'static>(resolver: R) -> Self {
-        Self {
-            resolver: Arc::new(resolver),
-            map: Arc::new(DashMap::new()),
-        }
-    }
-}
-
-/// Ratelimit Bucket
-#[derive(Clone, Copy, Debug)]
-pub struct Entry {
-    used: u32,
-    reset: u128,
-}
-
-/// Get the current time from Unix Epoch as a Duration
-fn now() -> Duration {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .expect("Time went backwards...")
-}
-
-impl Entry {
-    /// Find bucket by its key
-    pub fn from(map: &DashMap<u64, Entry>, key: u64) -> Entry {
-        map.get(&key).map(|x| *x).unwrap_or_else(|| Entry {
-            used: 0,
-            reset: now().add(Duration::from_secs(10)).as_millis(),
-        })
-    }
-
-    /// Deduct one unit from the bucket and save
-    pub fn deduct(&mut self) {
-        let current_time = now().as_millis();
-        if current_time > self.reset {
-            self.used = 1;
-            self.reset = now().add(Duration::from_secs(10)).as_millis();
-        } else {
-            self.used += 1;
-        }
-    }
-
-    /// Save information
-    pub fn save(self, map: &DashMap<u64, Entry>, key: u64) {
-        map.insert(key, self);
-    }
-
-    /// Get remaining units in the bucket
-    pub fn get_remaining(&self, limit: u32) -> u32 {
-        if now().as_millis() > self.reset {
-            limit
-        } else {
-            limit - self.used
-        }
-    }
-
-    /// Get how long bucket has until reset
-    pub fn left_until_reset(&self) -> u128 {
-        let current_time = now().as_millis();
-        self.reset.saturating_sub(current_time)
-    }
-}
-
-/// Ratelimit Guard
-#[derive(Serialize, Clone, Copy, Debug)]
-#[allow(dead_code)]
-pub struct Ratelimiter {
-    pub key: u64,
-    pub limit: u32,
-    pub remaining: u32,
-    pub reset: u128,
-}
-
-impl Ratelimiter {
-    /// Generate guard from identifier and target bucket
-    pub fn from(
-        map: &DashMap<u64, Entry>,
-        identifier: &str,
-        limit: u32,
-        (bucket, resource): (&str, Option<&str>),
-    ) -> Result<Ratelimiter, Ratelimiter> {
-        let mut key = DefaultHasher::new();
-        key.write(identifier.as_bytes());
-        key.write(bucket.as_bytes());
-
-        if let Some(id) = resource {
-            key.write(id.as_bytes());
-        }
-
-        let key = key.finish();
-        let mut entry = Entry::from(map, key);
-
-        let remaining = entry.get_remaining(limit);
-        let reset = entry.left_until_reset();
-        let mut ratelimiter = Ratelimiter {
-            key,
-            limit,
-            remaining,
-            reset,
-        };
-        if remaining == 0 {
-            return Err(ratelimiter);
-        }
-
-        entry.deduct();
-        entry.save(map, key);
-        ratelimiter.remaining -= 1;
-        ratelimiter.reset = entry.left_until_reset();
-
-        Ok(ratelimiter)
-    }
-}
-
-#[derive(Serialize)]
-#[serde(untagged)]
-pub enum RatelimitInformation {
-    Success(Ratelimiter),
-    Failure { retry_after: u128 },
-}

crates/core/ratelimits/src/rocket.rs

@@ -1,163 +0,0 @@
-use async_trait::async_trait;
-use log::info;
-use rocket::fairing::{Fairing, Info, Kind};
-use rocket::http::uri::Origin;
-use rocket::http::{Method, Status};
-use rocket::request::{FromRequest, Outcome};
-use rocket::serde::json::Json;
-use rocket::{Data, Request, Response, State};
-use revolt_config::config;
-
-use revolt_rocket_okapi::r#gen::OpenApiGenerator;
-use revolt_rocket_okapi::request::{OpenApiFromRequest, RequestHeaderInput};
-
-use authifier::models::Session;
-
-use crate::ratelimiter::RequestKind;
-use crate::ratelimiter::{RatelimitInformation, Ratelimiter};
-
-#[derive(Clone, Copy)]
-pub struct RocketRequestKind;
-
-impl RequestKind for RocketRequestKind {
-    type R<'a> = Request<'a>;
-}
-
-pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<RocketRequestKind>;
-
-/// Find the remote IP of the client
-fn to_ip(request: &'_ rocket::Request<'_>) -> String {
-    request
-        .remote()
-        .map(|x| x.ip().to_string())
-        .unwrap_or_default()
-}
-
-/// Find the actual IP of the client
-async fn to_real_ip(request: &'_ rocket::Request<'_>) -> String {
-    if config().await.api.security.trust_cloudflare {
-        request
-            .headers()
-            .get_one("CF-Connecting-IP")
-            .map(|x| x.to_string())
-            .unwrap_or_else(|| to_ip(request))
-    } else {
-        to_ip(request)
-    }
-}
-
-#[async_trait]
-impl<'r> FromRequest<'r> for Ratelimiter {
-    type Error = Ratelimiter;
-
-    async fn from_request<'a>(request: &'r rocket::Request<'a>) -> Outcome<Self, Self::Error> {
-        let ratelimiter = request
-            .local_cache_async(async {
-                use rocket::outcome::Outcome;
-
-                let storage = request.guard::<&State<RatelimitStorage>>().await.unwrap();
-
-                let identifier = if let Outcome::Success(session) = request.guard::<Session>().await
-                {
-                    session.id
-                } else {
-                    to_real_ip(request).await
-                };
-
-                let (bucket, resource) = storage.resolver.resolve_bucket(request);
-                let limit = storage.resolver.resolve_bucket_limit(bucket);
-
-                Ratelimiter::from(&storage.map, &identifier, limit, (bucket, resource))
-            })
-            .await;
-
-        match ratelimiter {
-            Ok(ratelimiter) => Outcome::Success(*ratelimiter),
-            Err(ratelimiter) => Outcome::Error((Status::TooManyRequests, *ratelimiter)),
-        }
-    }
-}
-
-impl OpenApiFromRequest<'_> for Ratelimiter {
-    fn from_request_input(
-        _gen: &mut OpenApiGenerator,
-        _name: String,
-        _required: bool,
-    ) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
-        Ok(RequestHeaderInput::None)
-    }
-}
-
-/// Attach ratelimiter to the Rocket application
-pub struct RatelimitFairing;
-
-#[async_trait]
-impl Fairing for RatelimitFairing {
-    fn info(&self) -> Info {
-        Info {
-            name: "Ratelimiter",
-            kind: Kind::Request | Kind::Response,
-        }
-    }
-
-    async fn on_request(&self, request: &mut Request<'_>, _: &mut Data<'_>) {
-        use rocket::outcome::Outcome;
-        if let Outcome::Error(_) = request.guard::<Ratelimiter>().await {
-            info!(
-                "User rate-limited on route {}! (IP = {:?})",
-                request.uri(),
-                to_real_ip(request).await
-            );
-
-            request.set_method(Method::Get);
-            request.set_uri(Origin::parse("/ratelimit").unwrap())
-        }
-    }
-
-    async fn on_response<'r>(&self, request: &'r Request<'_>, response: &mut Response<'r>) {
-        let guard = request.guard::<Ratelimiter>().await;
-        let (Outcome::Success(ratelimiter) | Outcome::Error((_, ratelimiter))) = guard else {
-            unreachable!()
-        };
-        let Ratelimiter {
-            key,
-            limit,
-            remaining,
-            reset,
-        } = ratelimiter;
-
-        response.set_raw_header("X-RateLimit-Limit", limit.to_string());
-        response.set_raw_header("X-RateLimit-Bucket", key.to_string());
-        response.set_raw_header("X-RateLimit-Remaining", remaining.to_string());
-        response.set_raw_header("X-RateLimit-Reset-After", reset.to_string());
-
-        if guard.is_error() {
-            response.set_status(Status::TooManyRequests);
-        }
-    }
-}
-
-#[async_trait]
-impl<'r> FromRequest<'r> for RatelimitInformation {
-    type Error = u128;
-
-    async fn from_request(request: &'r rocket::Request<'_>) -> Outcome<Self, Self::Error> {
-        let info = match request.guard::<Ratelimiter>().await {
-            Outcome::Success(ratelimiter) => RatelimitInformation::Success(ratelimiter),
-            Outcome::Error((_, ratelimiter)) => RatelimitInformation::Failure {
-                retry_after: ratelimiter.reset,
-            },
-            _ => unreachable!(),
-        };
-        Outcome::Success(info)
-    }
-}
-
-#[rocket::get("/ratelimit")]
-fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
-    Json(info)
-}
-
-pub fn routes() -> Vec<rocket::Route> {
-    rocket::routes![ratelimit_info]
-}

crates/daemons/crond/src/main.rs

@@ -1,7 +1,7 @@
 use revolt_config::configure;
 use revolt_database::DatabaseInfo;
 use revolt_result::Result;
-use tasks::{file_deletion, prune_dangling_files, prune_members};
+use tasks::{file_deletion, prune_dangling_files};
 use tokio::try_join;
 
 pub mod tasks;
@@ -13,8 +13,7 @@ async fn main() -> Result<()> {
     let db = DatabaseInfo::Auto.connect().await.expect("database");
     try_join!(
         file_deletion::task(db.clone()),
-        prune_dangling_files::task(db.clone()),
-        prune_members::task(db.clone())
+        prune_dangling_files::task(db)
     )
     .map(|_| ())
 }

crates/daemons/crond/src/tasks/mod.rs

@@ -1,3 +1,2 @@
 pub mod file_deletion;
 pub mod prune_dangling_files;
-pub mod prune_members;

crates/daemons/crond/src/tasks/prune_members.rs

@@ -1,18 +0,0 @@
-use std::time::Duration;
-
-use log::warn;
-use revolt_database::Database;
-use revolt_result::Result;
-use tokio::time::sleep;
-
-pub async fn task(db: Database) -> Result<()> {
-    loop {
-        let success = db.remove_dangling_members().await;
-        if let Err(s) = success {
-            revolt_config::capture_error(&s);
-            warn!("Failed to prune dangling members: {:?}", &s);
-        }
-
-        sleep(Duration::from_secs(90)).await;
-    }
-}

crates/daemons/pushd/Cargo.toml

@@ -8,7 +8,6 @@ license = "AGPL-3.0-or-later"
 revolt-result = { version = "0.8.8", path = "../../core/result" }
 revolt-config = { version = "0.8.8", path = "../../core/config", features = [
     "report-macros",
-    "anyhow"
 ] }
 revolt-database = { version = "0.8.8", path = "../../core/database" }
 revolt-models = { version = "0.8.8", path = "../../core/models", features = [

crates/delta/Cargo.toml

@@ -32,6 +32,7 @@ ulid = "0.4.1"
 nanoid = "0.4.0"
 
 # serde
+rmp-serde = "1.0.0"
 serde_json = "1.0.57"
 serde = { version = "1.0.115", features = ["derive"] }
 validator = { version = "0.16", features = ["derive"] }
@@ -65,10 +66,12 @@ revolt_rocket_okapi = { version = "0.10.0", features = ["swagger"] }
 
 # rabbit
 amqprs = { version = "1.7.0" }
+lapin = { version = "3.0.0" }
 
 # core
 authifier = "1.0.15"
 revolt-config = { path = "../core/config" }
+revolt-broker = { path = "../core/broker" }
 revolt-database = { path = "../core/database", features = [
     "rocket-impl",
     "redis-is-patched",
@@ -81,7 +84,6 @@ revolt-models = { path = "../core/models", features = [
 revolt-presence = { path = "../core/presence" }
 revolt-result = { path = "../core/result", features = ["rocket", "okapi"] }
 revolt-permissions = { path = "../core/permissions", features = ["schemas"] }
-revolt-ratelimits = { path = "../core/ratelimits", features = ["rocket"] }
 
 [build-dependencies]
 vergen = "7.5.0"

crates/delta/src/main.rs

@@ -11,7 +11,6 @@ pub mod util;
 use revolt_config::config;
 use revolt_database::events::client::EventV1;
 use revolt_database::AMQP;
-use revolt_ratelimits::rocket as ratelimiter;
 use rocket::{Build, Rocket};
 use rocket_cors::{AllowedOrigins, CorsOptions};
 use rocket_prometheus::PrometheusMetrics;
@@ -123,22 +122,18 @@ pub async fn web() -> Rocket<Build> {
     let rocket = rocket::build();
     let prometheus = PrometheusMetrics::new();
 
-    // Ratelimits
-    let ratelimits = ratelimiter::RatelimitStorage::new(util::ratelimits::DeltaRatelimits);
-
     routes::mount(config, rocket)
         .attach(prometheus.clone())
         .mount("/metrics", prometheus)
         .mount("/", rocket_cors::catch_all_options_routes())
-        .mount("/", ratelimiter::routes())
+        .mount("/", util::ratelimiter::routes())
         .mount("/swagger/", swagger)
         .mount("/0.8/swagger/", swagger_0_8)
         .manage(authifier)
         .manage(db)
         .manage(amqp)
         .manage(cors.clone())
-        .manage(ratelimits)
-        .attach(ratelimiter::RatelimitFairing)
+        .attach(util::ratelimiter::RatelimitFairing)
         .attach(cors)
         .configure(rocket::Config {
             limits: rocket::data::Limits::default().limit("string", 5.megabytes()),

crates/delta/src/routes/bots/delete.rs

@@ -11,7 +11,7 @@ use rocket_empty::EmptyResponse;
 pub async fn delete_bot(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<EmptyResponse> {
     let bot = target.as_bot(db).await?;
     if bot.owner != user.id {

crates/delta/src/routes/bots/edit.rs

@@ -14,7 +14,7 @@ use validator::Validate;
 pub async fn edit_bot(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     data: Json<DataEditBot>,
 ) -> Result<Json<v0::BotWithUserResponse>> {
     let data = data.into_inner();
@@ -37,7 +37,7 @@ pub async fn edit_bot(
     if data.public.is_none()
         && data.analytics.is_none()
         && data.interactions_url.is_none()
-        && data.remove.is_empty()
+        && data.remove.is_none()
     {
         return Ok(Json(v0::BotWithUserResponse {
             bot: bot.into(),
@@ -64,6 +64,7 @@ pub async fn edit_bot(
         db,
         partial,
         remove
+            .unwrap_or_default()
             .into_iter()
             .map(|v| v.into())
             .collect(),
@@ -99,7 +100,7 @@ mod test {
             .body(
                 json!(v0::DataEditBot {
                     public: Some(true),
-                    remove: vec![FieldsBot::Token],
+                    remove: Some(vec![FieldsBot::Token]),
                     ..Default::default()
                 })
                 .to_string(),

crates/delta/src/routes/bots/fetch.rs

@@ -11,7 +11,7 @@ use rocket::{serde::json::Json, State};
 pub async fn fetch_bot(
     db: &State<Database>,
     user: User,
-    bot: Reference<'_>,
+    bot: Reference,
 ) -> Result<Json<FetchBotResponse>> {
     if user.bot.is_some() {
         return Err(create_error!(IsBot));

crates/delta/src/routes/bots/fetch_public.rs

@@ -13,9 +13,9 @@ use rocket::State;
 pub async fn fetch_public_bot(
     db: &State<Database>,
     user: Option<User>,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<PublicBot>> {
-    let bot = db.fetch_bot(target.id).await?;
+    let bot = db.fetch_bot(&target.id).await?;
     if !bot.public && user.is_none_or(|x| x.id != bot.owner) {
         return Err(create_error!(NotFound));
     }

crates/delta/src/routes/bots/invite.rs

@@ -20,7 +20,7 @@ pub async fn invite_bot(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     dest: Json<v0::InviteBotDestination>,
 ) -> Result<EmptyResponse> {
     if user.bot.is_some() {
@@ -168,8 +168,8 @@ mod test {
             .await;
 
         match event {
-            EventV1::ServerMemberJoin { member, .. } => {
-                assert_eq!(bot.id, member.id.user);
+            EventV1::ServerMemberJoin { user, .. } => {
+                assert_eq!(bot.id, user);
             }
             _ => unreachable!(),
         }

crates/delta/src/routes/channels/channel_ack.rs

@@ -15,8 +15,8 @@ use rocket_empty::EmptyResponse;
 pub async fn ack(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
-    message: Reference<'_>,
+    target: Reference,
+    message: Reference,
 ) -> Result<EmptyResponse> {
     if user.bot.is_some() {
         return Err(create_error!(IsBot));
@@ -29,7 +29,7 @@ pub async fn ack(
         .throw_if_lacking_channel_permission(ChannelPermission::ViewChannel)?;
 
     channel
-        .ack(&user.id, message.id)
+        .ack(&user.id, &message.id)
         .await
         .map(|_| EmptyResponse)
 }

crates/delta/src/routes/channels/channel_delete.rs

@@ -17,7 +17,7 @@ pub async fn delete(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     options: v0::OptionsChannelDelete,
 ) -> Result<EmptyResponse> {
     let mut channel = target.as_channel(db).await?;

crates/delta/src/routes/channels/channel_edit.rs

@@ -17,7 +17,7 @@ pub async fn edit(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     data: Json<v0::DataEditChannel>,
 ) -> Result<Json<v0::Channel>> {
     let data = data.into_inner();
@@ -38,7 +38,7 @@ pub async fn edit(
         && data.icon.is_none()
         && data.nsfw.is_none()
         && data.owner.is_none()
-        && data.remove.is_empty()
+        && data.remove.is_none()
     {
         return Ok(Json(channel.into()));
     }
@@ -112,21 +112,23 @@ pub async fn edit(
             nsfw,
             ..
         } => {
-            if data.remove.contains(&v0::FieldsChannel::Icon) {
-                if let Some(icon) = &icon {
-                    db.mark_attachment_as_deleted(&icon.id).await?;
+            if let Some(fields) = &data.remove {
+                if fields.contains(&v0::FieldsChannel::Icon) {
+                    if let Some(icon) = &icon {
+                        db.mark_attachment_as_deleted(&icon.id).await?;
+                    }
                 }
-            }
 
-            for field in &data.remove {
-                match field {
-                    v0::FieldsChannel::Description => {
-                        description.take();
+                for field in fields {
+                    match field {
+                        v0::FieldsChannel::Description => {
+                            description.take();
+                        }
+                        v0::FieldsChannel::Icon => {
+                            icon.take();
+                        }
+                        _ => {}
                     }
-                    v0::FieldsChannel::Icon => {
-                        icon.take();
-                    }
-                    _ => {}
                 }
             }
 
@@ -212,7 +214,11 @@ pub async fn edit(
                 .update(
                     db,
                     partial,
-                    data.remove.into_iter().map(|f| f.into()).collect(),
+                    data.remove
+                        .unwrap_or_default()
+                        .into_iter()
+                        .map(|f| f.into())
+                        .collect(),
                 )
                 .await?;
         }

crates/delta/src/routes/channels/channel_fetch.rs

@@ -16,7 +16,7 @@ use rocket::{serde::json::Json, State};
 pub async fn fetch(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<v0::Channel>> {
     let channel = target.as_channel(db).await?;
 

crates/delta/src/routes/channels/group_add_member.rs

@@ -17,8 +17,8 @@ pub async fn add_member(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    group_id: Reference<'_>,
-    member_id: Reference<'_>,
+    group_id: Reference,
+    member_id: Reference,
 ) -> Result<EmptyResponse> {
     if user.bot.is_some() {
         return Err(create_error!(IsBot));

crates/delta/src/routes/channels/group_remove_member.rs

@@ -14,8 +14,8 @@ pub async fn remove_member(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    target: Reference<'_>,
-    member: Reference<'_>,
+    target: Reference,
+    member: Reference,
 ) -> Result<EmptyResponse> {
     if user.bot.is_some() {
         return Err(create_error!(IsBot));

crates/delta/src/routes/channels/invite_create.rs

@@ -18,7 +18,7 @@ use rocket::{serde::json::Json, State};
 pub async fn create_invite(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<v0::Invite>> {
     if user.bot.is_some() {
         return Err(create_error!(IsBot));

crates/delta/src/routes/channels/members_fetch.rs

@@ -17,7 +17,7 @@ use rocket::{serde::json::Json, State};
 pub async fn fetch_members(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<Vec<v0::User>>> {
     let channel = target.as_channel(db).await?;
     let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);

crates/delta/src/routes/channels/message_bulk_delete.rs

@@ -22,7 +22,7 @@ use validator::Validate;
 pub async fn bulk_delete_messages(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     options: Json<v0::OptionsBulkDelete>,
 ) -> Result<EmptyResponse> {
     let options = options.into_inner();
@@ -51,7 +51,7 @@ pub async fn bulk_delete_messages(
         .await
         .throw_if_lacking_channel_permission(ChannelPermission::ManageMessages)?;
 
-    Message::bulk_delete(db, target.id, options.ids)
+    Message::bulk_delete(db, &target.id, options.ids)
         .await
         .map(|_| EmptyResponse)
 }

crates/delta/src/routes/channels/message_clear_reactions.rs

@@ -17,8 +17,8 @@ use rocket_empty::EmptyResponse;
 pub async fn clear_reactions(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
-    msg: Reference<'_>,
+    target: Reference,
+    msg: Reference,
 ) -> Result<EmptyResponse> {
     let channel = target.as_channel(db).await?;
     let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);

crates/delta/src/routes/channels/message_delete.rs

@@ -15,10 +15,10 @@ use rocket_empty::EmptyResponse;
 pub async fn delete(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
-    msg: Reference<'_>,
+    target: Reference,
+    msg: Reference,
 ) -> Result<EmptyResponse> {
-    let message = msg.as_message_in_channel(db, target.id).await?;
+    let message = msg.as_message_in_channel(db, &target.id).await?;
 
     if message.author != user.id {
         let channel = target.as_channel(db).await?;

crates/delta/src/routes/channels/message_edit.rs

@@ -18,8 +18,8 @@ use validator::Validate;
 pub async fn edit(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
-    msg: Reference<'_>,
+    target: Reference,
+    msg: Reference,
     edit: Json<v0::DataEditMessage>,
 ) -> Result<Json<v0::Message>> {
     let edit = edit.into_inner();

crates/delta/src/routes/channels/message_fetch.rs

@@ -15,8 +15,8 @@ use rocket::{serde::json::Json, State};
 pub async fn fetch(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
-    msg: Reference<'_>,
+    target: Reference,
+    msg: Reference,
 ) -> Result<Json<v0::Message>> {
     let channel = target.as_channel(db).await?;
     let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);

crates/delta/src/routes/channels/message_pin.rs

@@ -1,7 +1,4 @@
-use revolt_database::{
-    util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Channel, Database, PartialMessage, SystemMessage, User, AMQP,
-};
+use revolt_database::{util::{permissions::DatabasePermissionQuery, reference::Reference}, Channel, Database, PartialMessage, SystemMessage, User, AMQP};
 use revolt_models::v0::MessageAuthor;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
@@ -17,8 +14,8 @@ pub async fn message_pin(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    target: Reference<'_>,
-    msg: Reference<'_>,
+    target: Reference,
+    msg: Reference,
 ) -> Result<EmptyResponse> {
     let channel = target.as_channel(db).await?;
 
@@ -172,7 +169,7 @@ mod test {
             })
             .await;
 
-        let updated_message = Reference::from_unchecked(&message.id)
+        let updated_message = Reference::from_unchecked(message.id)
             .as_message(&harness.db)
             .await
             .expect("Failed to find updated message");

crates/delta/src/routes/channels/message_query.rs

@@ -16,7 +16,7 @@ use validator::Validate;
 pub async fn query(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     options: v0::OptionsQueryMessages,
 ) -> Result<Json<v0::BulkMessageResponse>> {
     options.validate().map_err(|error| {

crates/delta/src/routes/channels/message_react.rs

@@ -15,9 +15,9 @@ use rocket_empty::EmptyResponse;
 pub async fn react_message(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
-    msg: Reference<'_>,
-    emoji: Reference<'_>,
+    target: Reference,
+    msg: Reference,
+    emoji: Reference,
 ) -> Result<EmptyResponse> {
     let channel = target.as_channel(db).await?;
     let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -30,7 +30,7 @@ pub async fn react_message(
 
     // Add the reaction
     message
-        .add_reaction(db, &user, emoji.id)
+        .add_reaction(db, &user, &emoji.id)
         .await
         .map(|_| EmptyResponse)
 }

crates/delta/src/routes/channels/message_search.rs

@@ -16,7 +16,7 @@ use validator::Validate;
 pub async fn search(
     db: &State<Database>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     options: Json<v0::DataMessageSearch>,
 ) -> Result<Json<v0::BulkMessageResponse>> {
     if user.bot.is_some() {

crates/delta/src/routes/channels/message_send.rs

@@ -21,7 +21,7 @@ pub async fn message_send(
     db: &State<Database>,
     amqp: &State<AMQP>,
     user: User,
-    target: Reference<'_>,
+    target: Reference,
     data: Json<v0::DataMessageSend>,
     idempotency: IdempotencyKey,
 ) -> Result<Json<v0::Message>> {
@@ -202,7 +202,7 @@ mod test {
         Member::create(&harness.db, &server, &user, Some(channels.clone()))
             .await
             .expect("Failed to create member");
-        let member = Reference::from_unchecked(&user.id)
+        let member = Reference::from_unchecked(user.id.clone())
             .as_member(&harness.db, &server.id)
             .await
             .expect("Failed to get member");
@@ -242,7 +242,7 @@ mod test {
         Member::create(&harness.db, &server, &second_user, Some(channels.clone()))
             .await
             .expect("Failed to create second member");
-        let mut second_member = Reference::from_unchecked(&second_user.id)
+        let mut second_member = Reference::from_unchecked(second_user.id.clone())
             .as_member(&harness.db, &server.id)
             .await
             .expect("Failed to get second member");