chore: ignore WS protocol errors

refactor: redo the bonfire client structure

Cargo.toml

@@ -11,6 +11,8 @@ members = [
 
 [patch.crates-io]
 redis23 = { package = "redis", version = "0.23.3", git = "https://github.com/revoltchat/redis-rs", rev = "523b2937367e17bd0073722bf6e23d06042cb4e4" }
+#authifier = { package = "authifier", version = "1.0.10", path = "../authifier/crates/authifier" }
+#rocket_authifier = { package = "rocket_authifier", version = "1.0.10", path = "../authifier/crates/rocket_authifier" }
 
 # I'm 99% sure this is overloading the GitHub worker
 # hence builds have been failing since, let's just

README.md

@@ -114,7 +114,8 @@ If you'd like to change anything, create a `Revolt.overrides.toml` file and spec
 > And corresponding Revolt configuration:
 >
 > ```toml
-> # Revolt.overrides.toml
+> #     Revolt.overrides.toml
+> # and Revolt.test-overrides.toml
 > [database]
 > mongodb = "mongodb://127.0.0.1:14017"
 > redis = "redis://127.0.0.1:14079/"

crates/bonfire/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-bonfire"
-version = "0.8.7"
+version = "0.8.8"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 
@@ -9,6 +9,7 @@ edition = "2021"
 [dependencies]
 # util
 log = "*"
+rand = "*"
 sentry = "0.31.5"
 lru = "0.7.6"
 ulid = "0.5.0"
@@ -38,11 +39,15 @@ 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" }
-revolt-permissions = { version = "0.8.7", path = "../core/permissions" }
+revolt-permissions = { version = "0.8.8", path = "../core/permissions" }
 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,39 +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},
-    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.
@@ -53,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;
     };
 
@@ -63,442 +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);
-
-    // 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

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-config"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -36,4 +36,4 @@ sentry = "0.31.5"
 sentry-anyhow = { version = "0.38.1", optional = true }
 
 # Core
-revolt-result = { version = "0.8.7", path = "../result", optional = true }
+revolt-result = { version = "0.8.8", path = "../result", optional = true }

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

@@ -60,6 +60,9 @@ static CONFIG_SEARCH_PATHS: [&str; 3] = [
     "/Revolt.toml",
 ];
 
+/// Path to search for test overrides
+static TEST_OVERRIDE_PATH: &str = "Revolt.test-overrides.toml";
+
 /// Configuration builder
 static CONFIG_BUILDER: Lazy<RwLock<Config>> = Lazy::new(|| {
     RwLock::new({
@@ -73,6 +76,20 @@ static CONFIG_BUILDER: Lazy<RwLock<Config>> = Lazy::new(|| {
                 include_str!("../Revolt.test.toml"),
                 FileFormat::Toml,
             ));
+
+            // recursively search upwards for an overrides file (if there is one)
+            if let Ok(cwd) = std::env::current_dir() {
+                let mut path = Some(cwd.as_path());
+                while let Some(current_path) = path {
+                    let target_path = current_path.join(TEST_OVERRIDE_PATH);
+                    if target_path.exists() {
+                        builder = builder
+                            .add_source(File::new(target_path.to_str().unwrap(), FileFormat::Toml));
+                    }
+
+                    path = current_path.parent();
+                }
+            }
         }
 
         for path in CONFIG_SEARCH_PATHS {
@@ -91,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)]
@@ -388,6 +417,11 @@ pub async fn read() -> Config {
 pub async fn config() -> Settings {
     let mut config = read().await.try_deserialize::<Settings>().unwrap();
 
+    // inject REDIS_URI for redis-kiss library
+    if std::env::var("REDIS_URL").is_err() {
+        std::env::set_var("REDIS_URI", config.database.redis.clone());
+    }
+
     // auto-detect production nodes
     if config.hosts.api.contains("https") && config.hosts.api.contains("revolt.chat") {
         config.production = true;
@@ -406,12 +440,6 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry:
         std::env::set_var("ROCKET_ADDRESS", "0.0.0.0");
     }
 
-    if std::env::var("REDIS_URL").is_err() {
-        // Configure redis-kiss library
-        let config = config().await;
-        std::env::set_var("REDIS_URI", config.database.redis);
-    }
-
     pretty_env_logger::init();
     log::info!("Starting {release}");
 

crates/core/database/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -24,19 +24,20 @@ default = ["mongodb", "async-std-runtime", "tasks"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.8.7", path = "../config", features = [
+revolt-config = { version = "0.8.8", path = "../config", features = [
     "report-macros",
 ] }
-revolt-result = { version = "0.8.7", path = "../result" }
-revolt-models = { version = "0.8.7", path = "../models", features = [
+revolt-result = { version = "0.8.8", path = "../result" }
+revolt-models = { version = "0.8.8", path = "../models", features = [
     "validator",
 ] }
-revolt-presence = { version = "0.8.7", path = "../presence" }
-revolt-permissions = { version = "0.8.7", path = "../permissions", features = [
+revolt-presence = { version = "0.8.8", path = "../presence" }
+revolt-permissions = { version = "0.8.8", path = "../permissions", features = [
     "serde",
     "bson",
 ] }
-revolt-parser = { version = "0.8.7", path = "../parser" }
+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"] }
@@ -101,3 +103,4 @@ authifier = { version = "1.0.15", features = ["rocket_impl"] }
 
 # RabbitMQ
 amqprs = { version = "1.7.0" }
+lapin = { version = "3.0.0" }

crates/core/database/fixtures/group_with_members.json

@@ -3,18 +3,21 @@
     "_object_type": "User",
     "_id": "__ID:0__",
     "username": "Owner",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
     "discriminator": "0001"
   },
   {
     "_object_type": "User",
     "_id": "__ID:1__",
     "username": "Member",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
     "discriminator": "0001"
   },
   {
     "_object_type": "User",
     "_id": "__ID:2__",
     "username": "Member",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
     "discriminator": "0002"
   },
   {
@@ -23,6 +26,9 @@
     "channel_type": "Group",
     "name": "My Group",
     "owner": "__ID:0__",
-    "recipients": ["__ID:0__", "__ID:1__"]
+    "recipients": [
+      "__ID:0__",
+      "__ID:1__"
+    ]
   }
-]
+]

crates/core/database/fixtures/server_with_roles.json

@@ -3,18 +3,21 @@
     "_object_type": "User",
     "_id": "__ID:0__",
     "username": "Owner",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
     "discriminator": "0001"
   },
   {
     "_object_type": "User",
     "_id": "__ID:1__",
     "username": "Moderator",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
     "discriminator": "0001"
   },
   {
     "_object_type": "User",
     "_id": "__ID:2__",
     "username": "User",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
     "discriminator": "0001"
   },
   {
@@ -39,7 +42,9 @@
     "_id": "__ID:4__",
     "owner": "__ID:0__",
     "name": "Server",
-    "channels": ["__ID:3__"],
+    "channels": [
+      "__ID:3__"
+    ],
     "roles": {
       "__ID:5__": {
         "name": "Moderator",
@@ -47,7 +52,7 @@
           "a": 545270208,
           "d": 0
         },
-        "rank": 3
+        "rank": 1
       },
       "__ID:6__": {
         "name": "Owner",
@@ -66,7 +71,9 @@
       "user": "__ID:0__",
       "server": "__ID:4__"
     },
-    "roles": ["__ID:6__"],
+    "roles": [
+      "__ID:6__"
+    ],
     "joined_at": 1698318340195
   },
   {
@@ -75,7 +82,9 @@
       "user": "__ID:1__",
       "server": "__ID:4__"
     },
-    "roles": ["__ID:5__"],
+    "roles": [
+      "__ID:5__"
+    ],
     "joined_at": 1698318340195
   },
   {
@@ -86,4 +95,4 @@
     },
     "joined_at": 1698318340195
   }
-]
+]

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};
 
@@ -165,6 +166,9 @@ pub enum EventV1 {
     /// Server role deleted
     ServerRoleDelete { id: String, role_id: String },
 
+    /// Server roles ranks updated
+    ServerRoleRanksUpdate { id: String, ranks: Vec<String> },
+
     /// Update existing user
     UserUpdate {
         id: String,
@@ -250,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/models/admin_migrations/ops/mongodb/scripts.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashSet, ops::BitXor, time::Duration};
+use std::{
+    collections::{HashMap, HashSet},
+    ops::BitXor,
+    time::Duration,
+};
 
 use crate::{
     mongodb::{
@@ -22,7 +26,7 @@ struct MigrationInfo {
     revision: i32,
 }
 
-pub const LATEST_REVISION: i32 = 41; // MUST BE +1 to last migration
+pub const LATEST_REVISION: i32 = 42; // MUST BE +1 to last migration
 
 pub async fn migrate_database(db: &MongoDb) {
     let migrations = db.col::<Document>("migrations");
@@ -1165,6 +1169,63 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
             .expect("failed to update users");
     }
 
+    if revision <= 41 {
+        info!(
+            "Running migration [revision 41 / 05-06-2025]: convert role ranks to uniform numbers."
+        );
+
+        #[derive(Serialize, Deserialize, Clone)]
+        struct Role {
+            pub rank: i64,
+        }
+
+        #[derive(Serialize, Deserialize, Clone)]
+        struct Server {
+            #[serde(rename = "_id")]
+            pub id: String,
+            #[serde(default = "HashMap::<String, Role>::new")]
+            pub roles: HashMap<String, Role>,
+        }
+
+        let mut servers = db
+            .db()
+            .collection::<Server>("servers")
+            .find(doc! {
+                "roles": {
+                    "$exists": true,
+                    "$ne": []
+                }
+            })
+            .await
+            .unwrap()
+            .filter_map(|s| async { s.ok() })
+            .boxed();
+
+        while let Some(server) = servers.next().await {
+            let mut ordered_roles = server.roles.clone().into_iter().collect::<Vec<_>>();
+            ordered_roles.sort_by(|(_, role_a), (_, role_b)| role_a.rank.cmp(&role_b.rank));
+            let ordered_roles = ordered_roles
+                .into_iter()
+                .map(|(id, _)| id)
+                .collect::<Vec<_>>();
+
+            let mut doc = doc! {};
+
+            for id in server.roles.keys() {
+                doc.insert(
+                    format!("roles.{id}.rank"),
+                    ordered_roles.iter().position(|x| id == x).unwrap() as i64,
+                );
+            }
+
+            db.db()
+                .collection::<Server>("servers")
+                .update_one(doc! { "_id": &server.id }, doc! { "$set": doc })
+                .await
+                .unwrap();
+        }
+    }
+
     // Reminder to update LATEST_REVISION when adding new migrations.
     LATEST_REVISION.max(revision)
 }

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

@@ -181,7 +181,7 @@ impl Server {
     }
 
     /// Update server data
-    pub async fn update<'a>(
+    pub async fn update(
         &mut self,
         db: &Database,
         partial: PartialServer,
@@ -228,6 +228,13 @@ impl Server {
         }
     }
 
+    /// Ordered roles list
+    pub fn ordered_roles(&self) -> Vec<(String, Role)> {
+        let mut ordered_roles = self.roles.clone().into_iter().collect::<Vec<_>>();
+        ordered_roles.sort_by(|(_, role_a), (_, role_b)| role_a.rank.cmp(&role_b.rank));
+        ordered_roles
+    }
+
     /// Set role permission on a server
     pub async fn set_role_permission(
         &mut self,
@@ -253,6 +260,37 @@ impl Server {
             Err(create_error!(NotFound))
         }
     }
+
+    /// Reorders the server's roles rankings
+    pub async fn set_role_ordering(&mut self, db: &Database, new_order: Vec<String>) -> Result<()> {
+        // New order must always contain every role
+        debug_assert_eq!(self.roles.len(), new_order.len());
+
+        // Set the role's ranks to the positions in the vec
+        for (rank, id) in new_order.iter().enumerate() {
+            self.roles.get_mut(id).unwrap().rank = rank as i64;
+        }
+
+        db.update_server(
+            &self.id,
+            &PartialServer {
+                roles: Some(self.roles.clone()),
+                ..Default::default()
+            },
+            Vec::new(),
+        )
+        .await?;
+
+        // Publish bulk update event
+        EventV1::ServerRoleRanksUpdate {
+            id: self.id.clone(),
+            ranks: new_order,
+        }
+        .p(self.id.clone())
+        .await;
+
+        Ok(())
+    }
 }
 
 impl Role {

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

@@ -7,11 +7,5 @@ mod rocket;
 #[cfg(feature = "rocket-impl")]
 mod schema;
 
-#[cfg(feature = "axum-impl")]
-pub use self::axum::*;
-#[cfg(feature = "rocket-impl")]
-pub use self::rocket::*;
-#[cfg(feature = "rocket-impl")]
-pub use self::schema::*;
 pub use model::*;
 pub use ops::*;

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

@@ -1,4 +1,5 @@
 use authifier::models::Session;
+use iso8601_timestamp::Timestamp;
 use revolt_result::Result;
 
 use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
@@ -61,4 +62,6 @@ pub trait AbstractUsers: Sync + Send {
 
     /// Remove push subscription for a session by session id (TODO: remove)
     async fn remove_push_subscription_by_session_id(&self, session_id: &str) -> Result<()>;
+
+    async fn update_session_last_seen(&self, session_id: &str, when: Timestamp) -> Result<()>;
 }

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

@@ -1,6 +1,7 @@
 use ::mongodb::options::{Collation, CollationStrength, FindOneOptions, FindOptions};
 use authifier::models::Session;
 use futures::StreamExt;
+use iso8601_timestamp::Timestamp;
 use revolt_result::Result;
 
 use crate::DocumentId;
@@ -317,7 +318,26 @@ impl AbstractUsers for MongoDb {
             )
             .await
             .map(|_| ())
-            .map_err(|_| create_database_error!("update_one", COL))
+            .map_err(|_| create_database_error!("update_one", "sessions"))
+    }
+
+    async fn update_session_last_seen(&self, session_id: &str, when: Timestamp) -> Result<()> {
+        let formatted: &str = &when.format();
+
+        self.col::<Session>("sessions")
+            .update_one(
+                doc! {
+                    "_id": session_id
+                },
+                doc! {
+                    "$set": {
+                        "last_seen": formatted
+                    }
+                },
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_one", "sessions"))
     }
 }
 

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

@@ -1,4 +1,5 @@
 use authifier::models::Session;
+use iso8601_timestamp::Timestamp;
 use revolt_result::Result;
 
 use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
@@ -168,4 +169,8 @@ impl AbstractUsers for ReferenceDb {
     async fn remove_push_subscription_by_session_id(&self, _session_id: &str) -> Result<()> {
         todo!()
     }
+
+    async fn update_session_last_seen(&self, _session_id: &str, _when: Timestamp) -> Result<()> {
+        todo!()
+    }
 }

crates/core/database/src/tasks/authifier_relay.rs

@@ -4,7 +4,7 @@ use once_cell::sync::Lazy;
 
 use crate::events::client::EventV1;
 
-static Q: Lazy<(Sender<AuthifierEvent>, Receiver<AuthifierEvent>)> = Lazy::new(|| unbounded());
+static Q: Lazy<(Sender<AuthifierEvent>, Receiver<AuthifierEvent>)> = Lazy::new(unbounded);
 
 /// Get sender
 pub fn sender() -> Sender<AuthifierEvent> {

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

@@ -15,8 +15,7 @@ impl crate::Bot {
             avatar: user.avatar.map(|x| x.id).unwrap_or_default(),
             description: user
                 .profile
-                .map(|profile| profile.content)
-                .flatten()
+                .and_then(|profile| profile.content)
                 .unwrap_or_default(),
         }
     }
@@ -1114,7 +1113,7 @@ impl crate::User {
     }
 
     /// Convert user object into user model without presence information
-    pub async fn into_known_static<'a>(self, is_online: bool) -> User {
+    pub async fn into_known_static(self, is_online: bool) -> User {
         let badges = self.get_badges().await;
 
         User {

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

@@ -54,7 +54,7 @@ use revolt_rocket_okapi::{
 use schemars::schema::{InstanceType, SchemaObject, SingleOrVec};
 
 #[cfg(feature = "rocket-impl")]
-impl<'r> OpenApiFromRequest<'r> for IdempotencyKey {
+impl OpenApiFromRequest<'_> for IdempotencyKey {
     fn from_request_input(
         _gen: &mut OpenApiGenerator,
         _name: String,

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

@@ -104,7 +104,7 @@ impl PermissionQuery for DatabasePermissionQuery<'_> {
                 .unwrap_or_default();
 
             self.cached_mutual_connection = Some(value);
-            matches!(value, true)
+            value
         } else {
             false
         }

crates/core/files/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-files"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -20,10 +20,10 @@ typenum = "1.17.0"
 aws-config = "1.5.5"
 aws-sdk-s3 = { version = "1.46.0", features = ["behavior-version-latest"] }
 
-revolt-config = { version = "0.8.7", path = "../config", features = [
+revolt-config = { version = "0.8.8", path = "../config", features = [
     "report-macros",
 ] }
-revolt-result = { version = "0.8.7", path = "../result" }
+revolt-result = { version = "0.8.8", path = "../result" }
 
 # image processing
 jxl-oxide = "0.8.1"

crates/core/models/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-models"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -20,8 +20,8 @@ default = ["serde", "partials", "rocket"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.8.7", path = "../config" }
-revolt-permissions = { version = "0.8.7", path = "../permissions" }
+revolt-config = { version = "0.8.8", path = "../config" }
+revolt-permissions = { version = "0.8.8", path = "../permissions" }
 
 # Utility
 regex = "1.11"

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

@@ -310,14 +310,14 @@ impl Channel {
     /// This returns a Result because the recipient name can't be determined here without a db call,
     /// which can't be done since this is models, which can't reference the database crate.
     ///
-    /// If it returns Err, you need to fetch the name from the db.
-    pub fn name(&self) -> Result<&str, ()> {
+    /// If it returns None, you need to fetch the name from the db.
+    pub fn name(&self) -> Option<&str> {
         match self {
-            Channel::DirectMessage { .. } => Err(()),
-            Channel::SavedMessages { .. } => Ok("Saved Messages"),
+            Channel::DirectMessage { .. } => None,
+            Channel::SavedMessages { .. } => Some("Saved Messages"),
             Channel::TextChannel { name, .. }
             | Channel::Group { name, .. }
-            | Channel::VoiceChannel { name, .. } => Ok(name),
+            | Channel::VoiceChannel { name, .. } => Some(name),
         }
     }
 }

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

@@ -267,7 +267,7 @@ auto_derived!(
         pub hoist: Option<bool>,
         /// Ranking position
         ///
-        /// Smaller values take priority.
+        /// **Removed** - no effect, use the edit server role positions route
         pub rank: Option<i64>,
         /// Fields to remove from role object
         #[cfg_attr(feature = "validator", validate(length(min = 1)))]
@@ -286,4 +286,9 @@ auto_derived!(
         /// Whether to not send a leave message
         pub leave_silently: Option<bool>,
     }
+
+    /// New role positions
+    pub struct DataEditRoleRanks {
+        pub ranks: Vec<String>,
+    }
 );

crates/core/parser/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-parser"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 description = "Revolt Backend: Message Parser"

crates/core/permissions/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-permissions"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -21,7 +21,7 @@ async-std = { version = "1.8.0", features = ["attributes"] }
 
 [dependencies]
 # Core
-revolt-result = { version = "0.8.7", path = "../result" }
+revolt-result = { version = "0.8.8", path = "../result" }
 
 # Utility
 auto_ops = "0.3.0"

crates/core/presence/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-presence"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -15,6 +15,9 @@ redis-is-patched = []
 # Async
 async-std = { version = "1.8.0", features = ["attributes"] }
 
+# Config for loading Redis URI
+revolt-config = { version = "0.8.8", path = "../config" }
+
 [dependencies]
 # Utility
 log = "0.4.17"

crates/core/presence/src/lib.rs

@@ -197,6 +197,8 @@ mod tests {
 
     #[async_std::test]
     async fn it_works() {
+        revolt_config::config().await;
+
         // Clear the region before we start the tests:
         clear_region(None).await;
 

crates/core/result/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-result"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/daemons/crond/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-crond"
-version = "0.8.7"
+version = "0.8.8"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
 edition = "2021"
@@ -16,7 +16,7 @@ log = "0.4"
 tokio = { version = "1" }
 
 # Core
-revolt-database = { version = "0.8.7", path = "../../core/database" }
-revolt-result = { version = "0.8.7", path = "../../core/result" }
-revolt-config = { version = "0.8.7", path = "../../core/config" }
-revolt-files = { version = "0.8.7", path = "../../core/files" }
+revolt-database = { version = "0.8.8", path = "../../core/database" }
+revolt-result = { version = "0.8.8", path = "../../core/result" }
+revolt-config = { version = "0.8.8", path = "../../core/config" }
+revolt-files = { version = "0.8.8", path = "../../core/files" }

crates/daemons/pushd/Cargo.toml

@@ -1,19 +1,19 @@
 [package]
 name = "revolt-pushd"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 
 [dependencies]
-revolt-result = { version = "0.8.7", path = "../../core/result" }
-revolt-config = { version = "0.8.7", path = "../../core/config", features = [
+revolt-result = { version = "0.8.8", path = "../../core/result" }
+revolt-config = { version = "0.8.8", path = "../../core/config", features = [
     "report-macros",
 ] }
-revolt-database = { version = "0.8.7", path = "../../core/database" }
-revolt-models = { version = "0.8.7", path = "../../core/models", features = [
+revolt-database = { version = "0.8.8", path = "../../core/database" }
+revolt-models = { version = "0.8.8", path = "../../core/models", features = [
     "validator",
 ] }
-revolt-presence = { version = "0.8.7", path = "../../core/presence", features = [
+revolt-presence = { version = "0.8.8", path = "../../core/presence", features = [
     "redis-is-patched",
 ] }
 

crates/delta/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-delta"
-version = "0.8.7"
+version = "0.8.8"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"
@@ -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",

crates/delta/fixtures/server_with_many_roles.json

@@ -0,0 +1,114 @@
+[
+  {
+    "_object_type": "User",
+    "_id": "__ID:0__",
+    "username": "Owner",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
+    "discriminator": "0001"
+  },
+  {
+    "_object_type": "User",
+    "_id": "__ID:1__",
+    "username": "Moderator",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
+    "discriminator": "0001"
+  },
+  {
+    "_object_type": "User",
+    "_id": "__ID:2__",
+    "username": "User",
+    "last_acknowledged_policy_change": "2025-06-07T04:04:48+0000",
+    "discriminator": "0001"
+  },
+  {
+    "_object_type": "Channel",
+    "_id": "__ID:3__",
+    "channel_type": "TextChannel",
+    "name": "General",
+    "server": "__ID:4__",
+    "default_permissions": {
+      "a": 0,
+      "d": 1048576
+    },
+    "role_permissions": {
+      "__ID:5__": {
+        "a": 1048576,
+        "d": 0
+      }
+    }
+  },
+  {
+    "_object_type": "Server",
+    "_id": "__ID:4__",
+    "owner": "__ID:0__",
+    "name": "Server",
+    "channels": [
+      "__ID:3__"
+    ],
+    "roles": {
+      "__ID:5__": {
+        "name": "Moderator",
+        "permissions": {
+          "a": 545270216,
+          "d": 0
+        },
+        "rank": 1
+      },
+      "__ID:6__": {
+        "name": "Owner",
+        "permissions": {
+          "a": 0,
+          "d": 0
+        },
+        "rank": 0
+      },
+      "__ID:7__": {
+        "name": "Lower Rank 1",
+        "permissions": {
+          "a": 0,
+          "d": 0
+        },
+        "rank": 2
+      },
+      "__ID:8__": {
+        "name": "Lower Rank 2",
+        "permissions": {
+          "a": 0,
+          "d": 0
+        },
+        "rank": 2
+      }
+    },
+    "default_permissions": 4000322560
+  },
+  {
+    "_object_type": "ServerMember",
+    "_id": {
+      "user": "__ID:0__",
+      "server": "__ID:4__"
+    },
+    "roles": [
+      "__ID:6__"
+    ],
+    "joined_at": 1698318340195
+  },
+  {
+    "_object_type": "ServerMember",
+    "_id": {
+      "user": "__ID:1__",
+      "server": "__ID:4__"
+    },
+    "roles": [
+      "__ID:5__"
+    ],
+    "joined_at": 1698318340195
+  },
+  {
+    "_object_type": "ServerMember",
+    "_id": {
+      "user": "__ID:2__",
+      "server": "__ID:4__"
+    },
+    "joined_at": 1698318340195
+  }
+]

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

@@ -16,7 +16,7 @@ pub async fn fetch_public_bot(
     target: Reference,
 ) -> Result<Json<PublicBot>> {
     let bot = db.fetch_bot(&target.id).await?;
-    if !bot.public && user.map_or(true, |x| x.id != bot.owner) {
+    if !bot.public && user.is_none_or(|x| x.id != bot.owner) {
         return Err(create_error!(NotFound));
     }
 

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

@@ -84,7 +84,8 @@ pub async fn message_send(
     // Create model user / members
     let model_user = user
         .clone()
-        .into_known_static(revolt_presence::is_online(&user.id).await).await;
+        .into_known_static(revolt_presence::is_online(&user.id).await)
+        .await;
 
     let model_member: Option<v0::Member> = query
         .member_ref()
@@ -491,7 +492,7 @@ mod test {
         let (_, _, other_user) = harness.new_user().await;
         let (server, _) = harness.new_server(&user).await;
         let channel = harness.new_channel(&server).await;
-        let (role_id, mut role) = harness
+        let (role_id, _role) = harness
             .new_role(
                 &server,
                 1,

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

@@ -41,7 +41,7 @@ pub async fn call(
     // - If not, create it.
     let client = reqwest::Client::new();
     let result = client
-        .get(&format!(
+        .get(format!(
             "{}/room/{}",
             config.hosts.voso_legacy,
             channel.id()
@@ -59,7 +59,7 @@ pub async fn call(
             reqwest::StatusCode::OK => (),
             reqwest::StatusCode::NOT_FOUND => {
                 if (client
-                    .post(&format!(
+                    .post(format!(
                         "{}/room/{}",
                         config.hosts.voso_legacy,
                         channel.id()
@@ -81,7 +81,7 @@ pub async fn call(
 
     // Then create a user for the room.
     if let Ok(response) = client
-        .post(&format!(
+        .post(format!(
             "{}/room/{}/user/{}",
             config.hosts.voso_legacy,
             channel.id(),

crates/delta/src/routes/policy/acknowledge_policy_changes.rs

@@ -1,11 +1,7 @@
-use revolt_database::{events::client::EventV1, Database, Report, Snapshot, SnapshotContent, User};
-use revolt_models::v0::{ReportStatus, ReportedContent};
-use revolt_result::{create_error, Result};
-use serde::Deserialize;
-use ulid::Ulid;
-use validator::Validate;
+use revolt_database::{Database, User};
+use revolt_result::Result;
 
-use rocket::{serde::json::Json, State};
+use rocket::State;
 
 /// # Acknowledge Policy Changes
 ///

crates/delta/src/routes/root.rs

@@ -136,7 +136,6 @@ pub async fn root() -> Result<Json<RevoltConfig>> {
 }
 
 #[cfg(test)]
-#[cfg(feature = "FIXME: THIS TEST CAUSES cargo test TO SEG FAULT, I HAVE NO CLUE HOW")]
 mod test {
     use crate::rocket;
     use rocket::http::Status;

crates/delta/src/routes/servers/mod.rs

@@ -18,6 +18,7 @@ mod roles_create;
 mod roles_delete;
 mod roles_edit;
 mod roles_fetch;
+mod roles_edit_positions;
 mod server_ack;
 mod server_create;
 mod server_delete;
@@ -47,6 +48,7 @@ pub fn routes() -> (Vec<Route>, OpenApi) {
         roles_delete::delete,
         permissions_set::set_role_permission,
         permissions_set_default::set_default_permissions,
-        emoji_list::list_emoji
+        emoji_list::list_emoji,
+        roles_edit_positions::edit_role_ranks
     ]
 }

crates/delta/src/routes/servers/roles_edit.rs

@@ -12,7 +12,7 @@ use validator::Validate;
 ///
 /// Edit a role by its id.
 #[openapi(tag = "Server Permissions")]
-#[patch("/<target>/roles/<role_id>", data = "<data>")]
+#[patch("/<target>/roles/<role_id>", data = "<data>", rank = 1)]
 pub async fn edit(
     db: &State<Database>,
     user: User,
@@ -45,22 +45,14 @@ pub async fn edit(
             name,
             colour,
             hoist,
-            rank,
             remove,
+            ..
         } = data;
 
-        // Prevent us from moving a role above other roles
-        if let Some(rank) = &rank {
-            if rank <= &member_rank {
-                return Err(create_error!(NotElevated));
-            }
-        }
-
         let partial = PartialRole {
             name,
             colour,
             hoist,
-            rank,
             ..Default::default()
         };
 

crates/delta/src/routes/servers/roles_edit_positions.rs

@@ -0,0 +1,177 @@
+use revolt_database::{
+    util::{permissions::DatabasePermissionQuery, reference::Reference},
+    Database, User,
+};
+use revolt_models::v0;
+use revolt_permissions::{calculate_server_permissions, ChannelPermission};
+use revolt_result::{create_error, Result};
+use rocket::{serde::json::Json, State};
+
+/// # Edits server roles ranks
+///
+/// Edit's server role's ranks.
+#[openapi(tag = "Server Permissions")]
+#[patch("/<target>/roles/ranks", data = "<data>")]
+pub async fn edit_role_ranks(
+    db: &State<Database>,
+    user: User,
+    target: Reference,
+    data: Json<v0::DataEditRoleRanks>,
+) -> Result<Json<v0::Server>> {
+    let data = data.into_inner();
+
+    let mut server = target.as_server(db).await?;
+    let mut query = DatabasePermissionQuery::new(db, &user).server(&server);
+    calculate_server_permissions(&mut query)
+        .await
+        .throw_if_lacking_channel_permission(ChannelPermission::ManageRole)?;
+
+    let existing_order = server
+        .ordered_roles()
+        .into_iter()
+        .map(|(id, _)| id)
+        .collect::<Vec<_>>();
+
+    let new_order = data.ranks.clone().into_iter().collect::<Vec<_>>();
+
+    // Verify all roles are in the new ordering
+    if data.ranks.len() != server.roles.len()
+        || !server.roles.iter().all(|(id, _)| data.ranks.contains(id))
+    {
+        return Err(create_error!(InvalidOperation));
+    }
+
+    // Don't have to check what the user can't modify if they are the server owner
+    if server.owner != user.id {
+        let member_top_rank = query.get_member_rank();
+
+        if server
+            .roles
+            .iter()
+            // Find all roles above the member which we should not be able to reorder
+            .filter(|(_, role)| {
+                if let Some(top_rank) = member_top_rank {
+                    role.rank <= top_rank
+                } else {
+                    true
+                }
+            })
+            // Check if user is trying to reorder roles they can't reorder (as found previously)
+            .any(|(id, _)| {
+                existing_order
+                    .iter()
+                    .position(|existing_id| id == existing_id)
+                    != new_order.iter().position(|new_id| id == new_id)
+            })
+        {
+            return Err(create_error!(NotElevated));
+        }
+    }
+
+    server.set_role_ordering(db, new_order).await?;
+
+    Ok(Json(server.into()))
+}
+
+#[cfg(test)]
+mod test {
+    use revolt_database::fixture;
+    use revolt_models::v0;
+    use rocket::http::{ContentType, Header, Status};
+
+    use crate::util::test::TestHarness;
+
+    #[rocket::async_test]
+    async fn edit_role_rankings() {
+        let harness = TestHarness::new().await;
+
+        fixture!(harness.db, "server_with_many_roles",
+            owner user 0
+            moderator user 1
+            server server 4);
+
+        // Moderator can re-order the roles below them
+        let (_, moderator_session) = harness.account_from_user(moderator.id).await;
+        let mut target_order: Vec<String> = server
+            .ordered_roles()
+            .into_iter()
+            .map(|(id, _)| id)
+            .collect();
+
+        // Swap the two lower ranked roles
+        target_order.swap(2, 3);
+
+        let response = harness
+            .client
+            .patch(format!("/servers/{}/roles/ranks", server.id))
+            .header(ContentType::JSON)
+            .body(
+                json!(v0::DataEditRoleRanks {
+                    ranks: target_order.clone()
+                })
+                .to_string(),
+            )
+            .header(Header::new(
+                "x-session-token",
+                moderator_session.token.to_string(),
+            ))
+            .dispatch()
+            .await;
+
+        assert_eq!(response.status(), Status::Ok);
+        drop(response);
+
+        // ... but not above them
+        let mut target_order: Vec<String> = server
+            .ordered_roles()
+            .into_iter()
+            .map(|(id, _)| id)
+            .collect();
+
+        // Swap the two lower ranked roles
+        target_order.swap(0, 1);
+
+        let response = harness
+            .client
+            .patch(format!("/servers/{}/roles/ranks", server.id))
+            .header(ContentType::JSON)
+            .body(
+                json!(v0::DataEditRoleRanks {
+                    ranks: target_order.clone()
+                })
+                .to_string(),
+            )
+            .header(Header::new(
+                "x-session-token",
+                moderator_session.token.to_string(),
+            ))
+            .dispatch()
+            .await;
+
+        assert_eq!(response.status(), Status::Forbidden);
+        drop(response);
+
+        // The owner can set any order they want
+        let (_, owner_session) = harness.account_from_user(owner.id).await;
+
+        let response = harness
+            .client
+            .patch(format!("/servers/{}/roles/ranks", server.id))
+            .header(ContentType::JSON)
+            .body(
+                json!(v0::DataEditRoleRanks {
+                    ranks: target_order.clone()
+                })
+                .to_string(),
+            )
+            .header(Header::new(
+                "x-session-token",
+                owner_session.token.to_string(),
+            ))
+            .dispatch()
+            .await;
+
+        assert_eq!(response.status(), Status::Ok);
+        drop(response);
+    }
+}

crates/delta/src/routes/webhooks/webhook_execute.rs

@@ -36,11 +36,11 @@ pub async fn webhook_execute(
     let permissions: PermissionValue = webhook.permissions.into();
     permissions.throw_if_lacking_channel_permission(ChannelPermission::SendMessage)?;
 
-    if data.attachments.as_ref().map_or(false, |v| !v.is_empty()) {
+    if data.attachments.as_ref().is_some_and(|v| !v.is_empty()) {
         permissions.throw_if_lacking_channel_permission(ChannelPermission::UploadFiles)?;
     }
 
-    if data.embeds.as_ref().map_or(false, |v| !v.is_empty()) {
+    if data.embeds.as_ref().is_some_and(|v| !v.is_empty()) {
         permissions.throw_if_lacking_channel_permission(ChannelPermission::SendEmbeds)?;
     }
 

crates/delta/src/routes/webhooks/webhook_execute_github.rs

@@ -621,7 +621,7 @@ pub struct Event {
 #[derive(Debug, JsonSchema)]
 pub struct EventHeader<'r>(pub &'r str);
 
-impl<'r> std::ops::Deref for EventHeader<'r> {
+impl std::ops::Deref for EventHeader<'_> {
     type Target = str;
 
     fn deref(&self) -> &Self::Target {

crates/delta/src/util/ratelimiter.rs

@@ -109,8 +109,6 @@ fn resolve_bucket<'r>(request: &'r rocket::Request<'_>) -> (&'r str, Option<&'r
     };
 
     if let Some(segment) = segment {
-        let resource = resource;
-
         let method = request.method();
         match (segment, resource, method) {
             ("users", target, Method::Patch) => ("user_edit", target),
@@ -254,7 +252,7 @@ impl<'r> FromRequest<'r> for Ratelimiter {
     }
 }
 
-impl<'r> OpenApiFromRequest<'r> for Ratelimiter {
+impl OpenApiFromRequest<'_> for Ratelimiter {
     fn from_request_input(
         _gen: &mut OpenApiGenerator,
         _name: String,

crates/delta/src/util/test.rs

@@ -1,10 +1,13 @@
+use std::collections::HashMap;
+
 use authifier::{
-    models::{Account, Session},
+    models::{Account, EmailVerification, Session},
     Authifier,
 };
 use futures::StreamExt;
+use lapin::types::AMQPValue;
 use rand::Rng;
-use redis_kiss::redis::aio::PubSub;
+use revolt_broker::event_stream;
 use revolt_database::{
     events::client::EventV1, Channel, Database, Member, Message, Server, User, AMQP,
 };
@@ -19,7 +22,7 @@ pub struct TestHarness {
     authifier: Authifier,
     pub db: Database,
     pub amqp: AMQP,
-    sub: PubSub,
+    consumer: lapin::Consumer,
     event_buffer: Vec<(String, EventV1)>,
 }
 
@@ -31,11 +34,19 @@ impl TestHarness {
             .await
             .expect("valid rocket instance");
 
-        let mut sub = redis_kiss::open_pubsub_connection()
-            .await
-            .expect("`PubSub`");
+        let tag: String = rand::thread_rng()
+            .sample_iter::<char, _>(&rand::distributions::Standard)
+            .take(32)
+            .collect();
 
-        sub.psubscribe("*").await.unwrap();
+        static QUEUE_NAME: &str = "revolt.events";
+        let conn = event_stream::get_connection().await;
+
+        let consumer = event_stream::create_channel(&conn, config.rabbit.event_stream)
+            .await
+            .basic_consume(QUEUE_NAME, &tag, Default::default(), Default::default())
+            .await
+            .unwrap();
 
         let db = client
             .rocket()
@@ -68,7 +79,7 @@ impl TestHarness {
             authifier,
             db,
             amqp,
-            sub,
+            consumer,
             event_buffer: vec![],
         }
     }
@@ -83,30 +94,41 @@ impl TestHarness {
     }
 
     pub async fn new_user(&self) -> (Account, Session, User) {
-        let account = Account::new(
-            &self.authifier,
-            format!("{}@revolt.chat", TestHarness::rand_string()),
-            "password".to_string(),
-            false,
-        )
-        .await
-        .expect("`Account`");
+        let user = User::create(&self.db, TestHarness::rand_string(), None, None)
+            .await
+            .expect("`User`");
+
+        let (account, session) = self.account_from_user(user.id.clone()).await;
+
+        (account, session, user)
+    }
+
+    pub async fn account_from_user(&self, id: String) -> (Account, Session) {
+        let account = Account {
+            id,
+            email: format!("{}@revolt.chat", TestHarness::rand_string()),
+            password: Default::default(),
+            email_normalised: Default::default(),
+            deletion: None,
+            disabled: false,
+            lockout: None,
+            mfa: Default::default(),
+            password_reset: None,
+            verification: EmailVerification::Verified,
+        };
+
+        self.authifier
+            .database
+            .save_account(&account)
+            .await
+            .expect("`Account`");
 
         let session = account
             .create_session(&self.authifier, String::new())
             .await
             .expect("`Session`");
 
-        let user = User::create(
-            &self.db,
-            TestHarness::rand_string(),
-            account.id.to_string(),
-            None,
-        )
-        .await
-        .expect("`User`");
-
-        (account, session, user)
+        (account, session)
     }
 
     pub async fn new_server(&self, user: &User) -> (Server, Vec<Channel>) {
@@ -198,10 +220,7 @@ impl TestHarness {
         (channel.clone(), member, message)
     }
 
-    pub async fn with_session<'c>(
-        session: Session,
-        request: LocalRequest<'c>,
-    ) -> LocalResponse<'c> {
+    pub async fn with_session(session: Session, request: LocalRequest<'_>) -> LocalResponse<'_> {
         request
             .header(Header::new("x-session-token", session.token.to_string()))
             .dispatch()
@@ -219,16 +238,33 @@ impl TestHarness {
             }
         }
 
-        let mut stream = self.sub.on_message();
-        while let Some(item) = stream.next().await {
-            let msg_topic = item.get_channel_name();
-            let payload: EventV1 = redis_kiss::decode_payload(&item).unwrap();
+        while let Some(Ok(delivery)) = self.consumer.next().await {
+            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();
 
-            if topic == msg_topic && predicate(&payload) {
+            let filter_value = headers
+                .get("x-stream-filter-value")
+                .expect("`x-stream-filter-value` not present in message!")
+                .as_long_string()
+                .unwrap()
+                .to_string();
+
+            let payload: EventV1 = rmp_serde::from_slice(&delivery.data).unwrap();
+
+            if topic == filter_value && predicate(&payload) {
                 return payload;
             }
 
-            self.event_buffer.push((msg_topic.to_string(), payload));
+            self.event_buffer.push((filter_value, payload));
         }
 
         // WARNING: if predicate is never satisfied, this will never return

crates/services/autumn/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-autumn"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 
@@ -43,12 +43,12 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 # Core crates
-revolt-files = { version = "0.8.7", path = "../../core/files" }
-revolt-config = { version = "0.8.7", path = "../../core/config" }
-revolt-database = { version = "0.8.7", path = "../../core/database", features = [
+revolt-files = { version = "0.8.8", path = "../../core/files" }
+revolt-config = { version = "0.8.8", path = "../../core/config" }
+revolt-database = { version = "0.8.8", path = "../../core/database", features = [
     "axum-impl",
 ] }
-revolt-result = { version = "0.8.7", path = "../../core/result", features = [
+revolt-result = { version = "0.8.8", path = "../../core/result", features = [
     "utoipa",
     "axum",
 ] }

crates/services/january/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-january"
-version = "0.8.7"
+version = "0.8.8"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 
@@ -32,13 +32,13 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 # Core crates
-revolt-config = { version = "0.8.7", path = "../../core/config" }
-revolt-models = { version = "0.8.7", path = "../../core/models" }
-revolt-result = { version = "0.8.7", path = "../../core/result", features = [
+revolt-config = { version = "0.8.8", path = "../../core/config" }
+revolt-models = { version = "0.8.8", path = "../../core/models" }
+revolt-result = { version = "0.8.8", path = "../../core/result", features = [
     "utoipa",
     "axum",
 ] }
-revolt-files = { version = "0.8.7", path = "../../core/files" }
+revolt-files = { version = "0.8.8", path = "../../core/files" }
 
 # Axum / web server
 axum = { version = "0.7.5" }

crates/services/january/src/website_embed.rs

@@ -327,7 +327,7 @@ pub async fn populate_special(original_url: String, metadata: &mut WebsiteMetada
             Special::Twitch { .. } => metadata.colour = Some("#7B68EE".to_string()),
             Special::Lightspeed { .. } => metadata.colour = Some("#7445D9".to_string()),
             Special::Spotify { .. } => metadata.colour = Some("#1ABC9C".to_string()),
-            Special::Soundcloud { .. } => metadata.colour = Some("#FF7F50".to_string()),
+            Special::Soundcloud => metadata.colour = Some("#FF7F50".to_string()),
             Special::AppleMusic { .. } => metadata.colour = Some("#FA233B".to_string()),
             _ => {}
         }

justfile

@@ -1,7 +1,7 @@
 publish:
     cargo publish --package revolt-parser
-    cargo publish --package revolt-config
     cargo publish --package revolt-result
+    cargo publish --package revolt-config
     cargo publish --package revolt-files
     cargo publish --package revolt-permissions
     cargo publish --package revolt-models