chore: update es code to lapin

Signed-off-by: Zomatree <me@zomatree.live>

.mise/config.toml

@@ -2,12 +2,12 @@
 node = "25.4.0"
 pnpm = "10.28.1"
 
-gh = "2.25.0"
+gh = "2.95.0"
 
 rust = "1.92.0"
 "cargo:cargo-nextest" = "0.9.122"
 
-"github:git-town/git-town" = "22.4.0"
+"github:git-town/git-town" = "22.7.1"
 
 [settings]
 experimental = true

Cargo.toml

@@ -23,11 +23,11 @@ async-trait = "0.1.89"
 tokio = "1.49.0"
 async-channel = "2.3.1"
 futures = "0.3.32"
-async-std = "1.8.0"
 async-tungstenite = "0.17.0"
 futures-locks = "0.7.1"
 async-lock = "2.8.0"
 async-recursion = "1.0.4"
+tokio-util = { version = "0.7.18" }
 
 # Error Handling
 anyhow = "1.0.100"
@@ -89,7 +89,7 @@ log = "0.4.29"
 pretty_env_logger = "0.4.0"
 
 # Redis
-redis-kiss = "0.1.4"
+redis-kiss = { version = "0.1.4", default-features = false }
 fred = "8.0.1"
 
 # Serialisation
@@ -160,10 +160,14 @@ opentelemetry-appender-tracing = "0.31.1"
 lapin = "4.7.1"
 
 # Voice
-livekit-api = "0.4.4"
-livekit-protocol = "0.7.4"
+livekit-api = "=0.4.23"
+livekit-protocol = "=0.7.7"
 livekit-runtime = "0.4.0"
 
+# Elasticsearch
+elasticsearch = "9.1.0-alpha.1"
+elasticsearch-dsl = "0.4"
+
 # Other Utilities
 once_cell = "1.9.0"
 config = "0.13.3"
@@ -203,3 +207,4 @@ revolt-permissions = { version = "0.13.7", path = "crates/core/permissions" }
 revolt-presence = { version = "0.13.7", path = "crates/core/presence" }
 revolt-ratelimits = { version = "0.13.7", path = "crates/core/ratelimits" }
 revolt-result = { version = "0.13.7", path = "crates/core/result" }
+revolt-search = { version = "0.13.7", path = "crates/core/search" }

Revolt.toml

@@ -4,7 +4,7 @@
 [database]
 # MongoDB connection URL
 # Defaults to the container name specified in self-hosted
-mongodb = "mongodb://127.0.0.1:27017"
+mongodb = "mongodb://127.0.0.1:27017?directConnection=true&replicaSet=rs0"
 # Redis connection URL
 # Defaults to the container name specified in self-hosted
 redis = "redis://127.0.0.1:6379/"
@@ -68,3 +68,6 @@ access_key_id = "minioautumn"
 secret_access_key = "minioautumn"
 # Bucket to upload to by default
 default_bucket = "revolt-uploads"
+
+[elasticsearch]
+api_key = ""

compose.yml

@@ -86,3 +86,34 @@ services:
     network_mode: "host"
     volumes:
       - ./livekit.yml:/etc/livekit.yml
+
+  elasticsearch:
+    image: elasticsearch:9.3.0
+    volumes:
+      - ./.data/elasticsearch:/usr/share/elasticsearch/data
+    ports:
+      - 9200:9200
+    extra_hosts:
+      - host.docker.internal:host-gateway
+      - model-runner.docker.internal:host-gateway
+    environment:
+      - discovery.type=single-node
+      - xpack.security.enabled=false
+      - xpack.security.http.ssl.enabled=false
+      - indices.id_field_data.enabled=true
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "curl --output /dev/null --silent --head --fail http://elasticsearch:9200",
+        ]
+      interval: 10s
+      timeout: 10s
+      retries: 30
+
+  kibana:
+    image: kibana:9.3.1
+    ports:
+      - 5601:5601
+    environment:
+      ELASTICSEARCH_HOSTS: '["http://elasticsearch:9200"]'

crates/bonfire/Cargo.toml

@@ -14,7 +14,7 @@ sentry = { workspace = true }
 lru = { workspace = true }
 ulid = { workspace = true }
 once_cell = { workspace = true }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 lru_time_cache = { workspace = true }
 async-channel = { workspace = true }
 
@@ -30,8 +30,9 @@ serde = { workspace = true }
 
 # async
 futures = { workspace = true }
-async-tungstenite = { workspace = true, features = ["async-std-runtime"] }
-async-std = { workspace = true }
+async-tungstenite = { workspace = true, features = ["tokio-runtime"] }
+tokio = { workspace = true }
+tokio-util = { workspace = true, features = ["compat"] }
 
 # core
 revolt-result = { workspace = true }

crates/bonfire/src/events/state.rs

@@ -2,7 +2,7 @@ use std::{
     collections::{HashMap, HashSet}, num::NonZeroUsize, sync::Arc, time::Duration
 };
 
-use async_std::sync::{Mutex, RwLock};
+use tokio::sync::{Mutex, RwLock};
 use lru::LruCache;
 use lru_time_cache::{LruCache as LruTimeCache, TimedEntry};
 use revolt_database::{Channel, Member, Server, User};

crates/bonfire/src/main.rs

@@ -1,6 +1,6 @@
 use std::env;
 
-use async_std::net::TcpListener;
+use tokio::net::TcpListener;
 use revolt_presence::clear_region;
 
 #[macro_use]
@@ -12,7 +12,7 @@ pub mod events;
 mod database;
 mod websocket;
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
     // Configure requirements for Bonfire.
     revolt_config::configure!(events);
@@ -33,7 +33,7 @@ async fn main() {
 
     // Start accepting new connections and spawn a client for each connection.
     while let Ok((stream, addr)) = listener.accept().await {
-        async_std::task::spawn(async move {
+        tokio::task::spawn(async move {
             info!("User connected from {addr:?}");
             websocket::client(database::get_db(), stream, addr).await;
             info!("User disconnected from {addr:?}");

crates/bonfire/src/websocket.rs

@@ -21,11 +21,12 @@ use revolt_database::{
 };
 use revolt_presence::{create_session, delete_session};
 
-use async_std::{
+use tokio::{
     net::TcpStream,
     sync::{Mutex, RwLock},
     task::spawn,
 };
+use tokio_util::compat::{TokioAsyncReadCompatExt, Compat};
 use revolt_result::create_error;
 use sentry::Level;
 
@@ -33,8 +34,8 @@ use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
 use revolt_models::v0;
 
-type WsReader = SplitStream<WebSocketStream<TcpStream>>;
-type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
+type WsReader = SplitStream<WebSocketStream<Compat<TcpStream>>>;
+type WsWriter = SplitSink<WebSocketStream<Compat<TcpStream>>, async_tungstenite::tungstenite::Message>;
 
 /// Start a new WebSocket client worker given access to the database,
 /// the relevant TCP stream and the remote address of the client.
@@ -44,7 +45,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     // e.g. wss://example.com?format=json&version=1
     let (sender, receiver) = oneshot::channel();
     let Ok(ws) = async_tungstenite::accept_hdr_async_with_config(
-        stream,
+        stream.compat(),
         WebsocketHandshakeCallback::from(sender),
         None,
     )

crates/core/config/Cargo.toml

@@ -13,7 +13,7 @@ repository = "https://github.com/stoatchat/stoatchat"
 anyhow = ["dep:sentry-anyhow"]
 report-macros = ["revolt-result"]
 sentry = ["dep:sentry"]
-test = ["async-std"]
+test = ["tokio"]
 default = ["sentry"]
 
 [dependencies]
@@ -26,7 +26,7 @@ serde = { workspace = true }
 
 # Async
 futures-locks = { workspace = true }
-async-std = { workspace = true, features = ["attributes"], optional = true }
+tokio = { workspace = true, optional = true }
 
 # Logging
 log = { workspace = true }

crates/core/config/Revolt.test.toml

@@ -1,7 +1,7 @@
 environment = "test"
 
 [database]
-mongodb = "mongodb://localhost"
+mongodb = "mongodb://localhost?directConnection=true&replicaSet=rs0"
 redis = "redis://localhost/"
 
 [rabbit]

crates/core/config/Revolt.toml

@@ -5,7 +5,7 @@ environment = "dev"
 [database]
 # MongoDB connection URL
 # Defaults to the container name specified in self-hosted
-mongodb = "mongodb://database"
+mongodb = "mongodb://database?directConnection=true&replicaSet=rs0"
 # Redis connection URL
 # Defaults to the container name specified in self-hosted
 redis = "redis://redis/"
@@ -346,3 +346,14 @@ proxy = ""
 pushd = ""
 crond = ""
 gifbox = ""
+
+[elasticsearch]
+host = "http://elasticsearch"
+port = 9200
+api_key = ""
+
+exchange = "revolt.messages"
+message_queue = "messages.message"
+message_edit_queue = "messages.message_edit"
+message_delete_queue = "messages.message_delete"
+channel_delete_queue = "messages.channel_delete"

crates/core/config/src/lib.rs

@@ -456,6 +456,19 @@ pub struct Sentry {
     pub gifbox: String,
 }
 
+#[derive(Deserialize, Debug, Clone)]
+pub struct Elasticsearch {
+    pub host: String,
+    pub port: u16,
+    pub api_key: String,
+
+    pub exchange: String,
+    pub message_queue: String,
+    pub message_edit_queue: String,
+    pub message_delete_queue: String,
+    pub channel_delete_queue: String,
+}
+
 #[derive(Deserialize, Debug, Clone)]
 pub struct Settings {
     pub database: Database,
@@ -470,6 +483,7 @@ pub struct Settings {
     pub production: bool,
     pub environment: String,
     pub disable_events_dont_use: bool,
+    pub elasticsearch: Elasticsearch,
 }
 
 impl Settings {
@@ -586,7 +600,7 @@ macro_rules! configure {
 mod tests {
     use crate::init;
 
-    #[async_std::test]
+    #[tokio::test]
     async fn it_works() {
         init().await;
     }

crates/core/database/Cargo.toml

@@ -15,7 +15,7 @@ mongodb = ["dep:mongodb", "bson"]
 
 # ... Other
 tasks = ["isahc", "linkify", "url-escape"]
-async-std-runtime = ["async-std"]
+tokio-runtime = ["tokio"]
 rocket-impl = [
     "rocket",
     "schemars",
@@ -27,11 +27,11 @@ redis-is-patched = ["revolt-presence/redis-is-patched"]
 voice = ["livekit-api", "livekit-protocol", "livekit-runtime"]
 
 # Default Features
-default = ["mongodb", "async-std-runtime", "tasks"]
+default = ["mongodb", "tokio-runtime", "tasks"]
 
 [dependencies]
 # Core
-revolt-config = { workspace = true, features = ["report-macros"] }
+revolt-config = { workspace = true, features = ["report-macros", "anyhow"] }
 revolt-result = { workspace = true }
 revolt-models = { workspace = true, features = ["validator"] }
 revolt-presence = { workspace = true }
@@ -56,6 +56,7 @@ validator = { workspace = true, features = ["derive"] }
 isahc = { workspace = true, features = ["json"], optional = true }
 base32 = { workspace = true }
 sha1 = { workspace = true }
+anyhow = { workspace = true }
 
 # Serialisation
 serde_json = { workspace = true }
@@ -64,7 +65,7 @@ serde = { workspace = true }
 iso8601-timestamp = { workspace = true, features = ["serde", "bson"] }
 
 # Events
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 
 # Database
 bson = { workspace = true, optional = true }
@@ -81,7 +82,7 @@ async-trait = { workspace = true }
 async-recursion = { workspace = true }
 
 # Async
-async-std = { workspace = true, features = ["attributes"], optional = true }
+tokio = { workspace = true, optional = true }
 
 # Axum Impl
 axum = { workspace = true, optional = true }

crates/core/database/src/amqp/amqp.rs

@@ -1,7 +1,7 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 
-use crate::events::rabbit::*;
+use crate::{Message, events::rabbit::*};
 use crate::User;
 use lapin::{
     options::BasicPublishOptions,
@@ -25,6 +25,10 @@ pub struct AMQP {
     ack_notification_message: Arc<Channel>,
     dm_call_updated: Arc<Channel>,
     process_ack: Arc<Channel>,
+    message_search: Arc<Channel>,
+    edit_message_search: Arc<Channel>,
+    delete_message_search: Arc<Channel>,
+    delete_channel_search: Arc<Channel>,
     #[allow(unused)]
     connection: Arc<Connection>,
 }
@@ -40,6 +44,10 @@ impl AMQP {
             ack_notification_message: Self::create_channel(&connection).await,
             dm_call_updated: Self::create_channel(&connection).await,
             process_ack: Self::create_channel(&connection).await,
+            message_search: Self::create_channel(&connection).await,
+            edit_message_search: Self::create_channel(&connection).await,
+            delete_message_search: Self::create_channel(&connection).await,
+            delete_channel_search: Self::create_channel(&connection).await,
             connection,
         }
     }
@@ -379,4 +387,112 @@ impl AMQP {
 
         Ok(())
     }
+
+    pub async fn new_message_search(
+        &self,
+        message: Message,
+        user: Option<User>,
+    ) -> Result<(), AMQPError> {
+        let config = revolt_config::config().await;
+
+        let payload = to_string(&MessageCreatePayload { message, user }).unwrap();
+
+        debug!(
+            "Sending new message search payload on channel {}: {}",
+            config.elasticsearch.message_queue, payload
+        );
+
+        self.message_search
+            .basic_publish(
+                config.elasticsearch.exchange.clone().into(),
+                config.elasticsearch.message_queue.into(),
+                BasicPublishOptions::default(),
+                payload.as_bytes(),
+                AMQPProperties::default()
+                    .with_content_type("application/json".into())
+                    .with_delivery_mode(2),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn edit_message_search(
+        &self,
+        message: Message,
+        user: Option<User>,
+    ) -> Result<(), AMQPError> {
+        let config = revolt_config::config().await;
+
+        let payload = to_string(&MessageEditPayload { message, user }).unwrap();
+
+        debug!(
+            "Sending edit message search payload on channel {}: {}",
+            config.elasticsearch.message_edit_queue, payload
+        );
+
+        self.edit_message_search
+            .basic_publish(
+                config.elasticsearch.exchange.clone().into(),
+                config.elasticsearch.message_edit_queue.into(),
+                BasicPublishOptions::default(),
+                payload.as_bytes(),
+                AMQPProperties::default()
+                    .with_content_type("application/json".into())
+                    .with_delivery_mode(2),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn delete_message_search(&self, message_id: String) -> Result<(), AMQPError> {
+        let config = revolt_config::config().await;
+
+        let payload = to_string(&MessageDeletePayload { message_id }).unwrap();
+
+        debug!(
+            "Sending delete message search payload on channel {}: {}",
+            config.elasticsearch.message_delete_queue, payload
+        );
+
+        self.delete_message_search
+            .basic_publish(
+                config.elasticsearch.exchange.clone().into(),
+                config.elasticsearch.message_delete_queue.into(),
+                BasicPublishOptions::default(),
+                payload.as_bytes(),
+                AMQPProperties::default()
+                    .with_content_type("application/json".into())
+                    .with_delivery_mode(2),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn delete_channel_search(&self, channel_id: String) -> Result<(), AMQPError> {
+        let config = revolt_config::config().await;
+
+        let payload = to_string(&ChannelDeletePayload { channel_id }).unwrap();
+
+        debug!(
+            "Sending delete channel search payload on channel {}: {}",
+            config.elasticsearch.channel_delete_queue, payload
+        );
+
+        self.delete_channel_search
+            .basic_publish(
+                config.elasticsearch.exchange.clone().into(),
+                config.elasticsearch.channel_delete_queue.into(),
+                BasicPublishOptions::default(),
+                payload.as_bytes(),
+                AMQPProperties::default()
+                    .with_content_type("application/json".into())
+                    .with_delivery_mode(2),
+            )
+            .await?;
+
+        Ok(())
+    }
 }

crates/core/database/src/amqp/consumer.rs

@@ -1,7 +1,5 @@
 use std::{
-    future::{ready, Future},
-    pin::Pin,
-    sync::Arc,
+    future::{Future, ready}, marker::PhantomData, pin::Pin, sync::Arc
 };
 
 use anyhow::Result;
@@ -12,14 +10,15 @@ use lapin::{
     BasicProperties, Channel, Connection, ConsumerDelegate, Error as AMQPError,
 };
 use log::debug;
-use revolt_database::Database;
+use crate::Database;
 
 #[async_trait]
-pub trait Consumer: Clone + Send + Sync + 'static {
+pub trait Consumer<T: Clone = ()>: Clone + Send + Sync + 'static {
     async fn create(
         db: Database,
         connection: Arc<Connection>,
         channel: Arc<Channel>,
+        data: T,
     ) -> Self;
     fn channel(&self) -> &Arc<Channel>;
     async fn consume(&self, delivery: Delivery) -> Result<()>;
@@ -65,9 +64,15 @@ pub trait Consumer: Clone + Send + Sync + 'static {
     }
 }
 
-pub struct Delegate<C: Consumer>(pub C);
+pub struct Delegate<C: Consumer<D>, D: Clone>(C, PhantomData<D>);
 
-impl<C: Consumer> ConsumerDelegate for Delegate<C> {
+impl<C: Consumer<D>, D: Clone> Delegate<C, D> {
+    pub fn new(consumer: C) -> Self {
+        Self(consumer, PhantomData)
+    }
+}
+
+impl<C: Consumer<D>, D: Clone + Send + Sync> ConsumerDelegate for Delegate<C, D> {
     fn on_new_delivery(
         &self,
         delivery: DeliveryResult,

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

@@ -1,2 +1,3 @@
 #[allow(clippy::module_inception)]
 pub mod amqp;
+pub mod consumer;

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

@@ -3,7 +3,7 @@ use std::collections::HashMap;
 use revolt_models::v0::PushNotification;
 use serde::{Deserialize, Serialize};
 
-use crate::User;
+use crate::{Message, User};
 
 #[derive(Serialize, Deserialize)]
 pub struct MessageSentPayload {
@@ -79,6 +79,28 @@ pub struct AckPayload {
     pub message_id: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MessageDeletePayload {
+    pub message_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ChannelDeletePayload {
+    pub channel_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MessageCreatePayload {
+    pub message: Message,
+    pub user: Option<User>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MessageEditPayload {
+    pub message: Message,
+    pub user: Option<User>,
+}
+
 /// This is not the same as the AckPayload above, as the state for this event is stored in redis to allow for state updates while the event is queued.
 #[derive(Serialize, Deserialize, Debug)]
 pub struct AckEventPayload {

crates/core/database/src/lib.rs

@@ -25,8 +25,8 @@ pub use mongodb;
 #[macro_use]
 extern crate bson;
 
-#[cfg(not(feature = "async-std-runtime"))]
-compile_error!("async-std-runtime feature must be enabled.");
+#[cfg(not(feature = "tokio-runtime"))]
+compile_error!("tokio-runtime feature must be enabled.");
 
 #[macro_export]
 #[cfg(debug_assertions)]
@@ -109,7 +109,7 @@ pub mod events;
 #[cfg(feature = "tasks")]
 pub mod tasks;
 
-mod amqp;
+pub mod amqp;
 pub use amqp::amqp::AMQP;
 
 #[cfg(feature = "voice")]

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

@@ -11,7 +11,7 @@ auto_derived!(
 
 #[cfg(test)]
 mod tests {
-    #[async_std::test]
+    #[tokio::test]
     async fn migrate() {
         database_test!(|db| async move {
             // Initialise the database

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

@@ -452,7 +452,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
         warn!("This is a destructive operation and will wipe existing permission data (excl. defaults for SendMessage).");
         warn!("Taking a backup is advised.");
         warn!("Continuing in 10 seconds...");
-        async_std::task::sleep(Duration::from_secs(10)).await;
+        tokio::time::sleep(Duration::from_secs(10)).await;
 
         let servers = db.col::<Document>("servers");
         let mut cursor = servers.find(doc! {}).await.unwrap();

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

@@ -164,7 +164,7 @@ impl Bot {
 mod tests {
     use crate::{Bot, FieldsBot, PartialBot, User};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn crud() {
         database_test!(|db| async move {
             let owner = User::create(&db, "Owner".to_string(), None, None)

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

@@ -128,7 +128,7 @@ impl Webhook {
 mod tests {
     use crate::{FieldsWebhook, PartialWebhook, Webhook};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn crud() {
         database_test!(|db| async move {
             let webhook_id = "webhook";

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

@@ -340,7 +340,7 @@ impl Channel {
     pub async fn add_user_to_group(
         &mut self,
         db: &Database,
-        amqp: &AMQP,
+        amqp: Option<&AMQP>,
         user: &User,
         by_id: &str,
     ) -> Result<()> {
@@ -377,7 +377,7 @@ impl Channel {
                 .into_message(id.to_string())
                 .send(
                     db,
-                    Some(amqp),
+                    amqp,
                     MessageAuthor::System {
                         username: &user.username,
                         avatar: user.avatar.as_ref().map(|file| file.id.as_ref()),
@@ -665,7 +665,7 @@ impl Channel {
     pub async fn remove_user_from_group(
         &self,
         db: &Database,
-        amqp: &AMQP,
+        amqp: Option<&AMQP>,
         user: &User,
         by_id: Option<&str>,
         silent: bool,
@@ -697,7 +697,7 @@ impl Channel {
                         .into_message(id.to_string())
                         .send(
                             db,
-                            Some(amqp),
+                            amqp,
                             MessageAuthor::System {
                                 username: name,
                                 avatar: None,
@@ -710,7 +710,7 @@ impl Channel {
                         .await
                         .ok();
                     } else {
-                        return self.delete(db).await;
+                        return self.delete(db, amqp).await;
                     }
                 }
 
@@ -737,7 +737,7 @@ impl Channel {
                     .into_message(id.to_string())
                     .send(
                         db,
-                        Some(amqp),
+                        amqp,
                         MessageAuthor::System {
                             username: &user.username,
                             avatar: user.avatar.as_ref().map(|file| file.id.as_ref()),
@@ -759,13 +759,22 @@ impl Channel {
     }
 
     /// Delete a channel
-    pub async fn delete(&self, db: &Database) -> Result<()> {
+    pub async fn delete(&self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
         let id = self.id().to_string();
         EventV1::ChannelDelete { id: id.clone() }.p(id).await;
         // TODO: missing functionality:
         // - group invites
         // - channels list / categories list on server
-        db.delete_channel(self).await
+        db.delete_channel(self).await?;
+
+        if let Some(amqp) = amqp {
+            if let Err(e) = amqp.delete_channel_search(self.id().to_string()).await {
+                log::error!("Error pushing message to RabbitMQ: {e}");
+                revolt_config::capture_error(&e);
+            }
+        }
+
+        Ok(())
     }
 }
 
@@ -787,7 +796,7 @@ mod tests {
 
     use crate::{fixture, util::permissions::DatabasePermissionQuery};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn permissions_group_channel() {
         database_test!(|db| async move {
             fixture!(db, "group_with_members",
@@ -813,7 +822,7 @@ mod tests {
         });
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn permissions_text_channel() {
         database_test!(|db| async move {
             fixture!(db, "server_with_roles",

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

@@ -1,6 +1,6 @@
 use indexmap::{IndexMap, IndexSet};
 use iso8601_timestamp::Timestamp;
-use revolt_config::{config, FeaturesLimits};
+use revolt_config::{capture_error, config, FeaturesLimits};
 use revolt_models::v0::{
     self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort,
     MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text,
@@ -18,7 +18,7 @@ use crate::{
         bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
         permissions::DatabasePermissionQuery,
     },
-    Channel, Database, Emoji, File, User, AMQP,
+    Channel, Database, Emoji, File, Member, User, AMQP,
 };
 
 #[cfg(feature = "tasks")]
@@ -210,6 +210,13 @@ auto_derived!(
     pub enum FieldsMessage {
         Pinned,
     }
+
+    /// Message along with the user for the author
+    pub struct MessageWithUser {
+        #[serde(flatten)]
+        pub message: Message,
+        pub user: Option<User>,
+    }
 );
 
 pub struct MessageFlagsValue(pub u32);
@@ -272,8 +279,8 @@ impl Message {
         channel: Channel,
         data: DataMessageSend,
         author: MessageAuthor<'_>,
-        user: Option<v0::User>,
-        member: Option<v0::Member>,
+        user: Option<User>,
+        member: Option<Member>,
         limits: FeaturesLimits,
         mut idempotency: IdempotencyKey,
         generate_embeds: bool,
@@ -610,10 +617,10 @@ impl Message {
 
     /// Send a message without any notifications
     pub async fn send_without_notifications(
-        &mut self,
+        &self,
         db: &Database,
-        user: Option<v0::User>,
-        member: Option<v0::Member>,
+        user: Option<User>,
+        member: Option<Member>,
         is_dm: bool,
         generate_embeds: bool,
         // This determines if this function should queue the mentions task or if somewhere else will.
@@ -623,9 +630,18 @@ impl Message {
         db.insert_message(self).await?;
 
         // Fan out events
-        EventV1::Message(self.clone().into_model(user, member))
-            .p(self.channel.to_string())
-            .await;
+        EventV1::Message(self.clone().into_model(
+            match user {
+                Some(user) => {
+                    let is_online = revolt_presence::is_online(&user.id).await;
+                    Some(user.into_known_static(is_online).await)
+                }
+                None => None,
+            },
+            member.map(Into::into),
+        ))
+        .p(self.channel.to_string())
+        .await;
 
         // Update last_message_id
         #[cfg(feature = "tasks")]
@@ -671,10 +687,10 @@ impl Message {
     pub async fn send(
         &mut self,
         db: &Database,
-        _amqp: Option<&AMQP>, // this is optional mostly for tests.
+        amqp: Option<&AMQP>, // this is optional mostly for tests.
         author: MessageAuthor<'_>,
-        user: Option<v0::User>,
-        member: Option<v0::Member>,
+        user: Option<User>,
+        member: Option<Member>,
         channel: &Channel,
         generate_embeds: bool,
     ) -> Result<()> {
@@ -704,7 +720,17 @@ impl Message {
                     messages: vec![(
                         Some(
                             PushNotification::from(
-                                self.clone().into_model(user, member),
+                                self.clone().into_model(
+                                    match user.clone() {
+                                        Some(user) => {
+                                            let is_online =
+                                                revolt_presence::is_online(&user.id).await;
+                                            Some(user.into_known_static(is_online).await)
+                                        }
+                                        None => None,
+                                    },
+                                    member.map(Into::into),
+                                ),
                                 Some(author.clone()),
                                 channel.to_owned().into(),
                             )
@@ -730,6 +756,13 @@ impl Message {
             .await;
         }
 
+        if let Some(amqp) = amqp {
+            if let Err(e) = amqp.new_message_search(self.clone(), user).await {
+                log::error!("Error pushing message to RabbitMQ: {e}");
+                capture_error(&e);
+            }
+        }
+
         Ok(())
     }
 
@@ -782,6 +815,7 @@ impl Message {
     pub async fn update(
         &mut self,
         db: &Database,
+        amqp: Option<&AMQP>,
         partial: PartialMessage,
         remove: Vec<FieldsMessage>,
     ) -> Result<()> {
@@ -803,9 +837,74 @@ impl Message {
         .p(self.channel.clone())
         .await;
 
+        if let Some(amqp) = amqp {
+            if let Err(e) = amqp
+                .edit_message_search(self.clone(), self.fetch_author(db).await)
+                .await
+            {
+                log::error!("Error pushing message to RabbitMQ: {e}");
+                capture_error(&e);
+            }
+        }
+
         Ok(())
     }
 
+    pub async fn fetch_users(
+        db: &Database,
+        messages: &[Message],
+        server_id: Option<&str>,
+    ) -> Result<(Vec<User>, Vec<Member>)> {
+        let user_ids = messages
+            .iter()
+            .flat_map(|m| {
+                let mut users = vec![m.author.clone()];
+                if let Some(system) = &m.system {
+                    match system {
+                        SystemMessage::ChannelDescriptionChanged { by } => users.push(by.clone()),
+                        SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
+                        SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
+                            users.push(from.clone());
+                            users.push(to.clone())
+                        }
+                        SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
+                        SystemMessage::UserAdded { by, id, .. }
+                        | SystemMessage::UserRemove { by, id, .. } => {
+                            users.push(by.clone());
+                            users.push(id.clone());
+                        }
+                        SystemMessage::UserBanned { id, .. }
+                        | SystemMessage::UserKicked { id, .. }
+                        | SystemMessage::UserJoined { id, .. }
+                        | SystemMessage::UserLeft { id, .. } => {
+                            users.push(id.clone());
+                        }
+                        SystemMessage::Text { .. } => {}
+                        SystemMessage::MessagePinned { by, .. } => {
+                            users.push(by.clone());
+                        }
+                        SystemMessage::MessageUnpinned { by, .. } => {
+                            users.push(by.clone());
+                        }
+                        SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
+                    }
+                }
+                users
+            })
+            .collect::<HashSet<String>>()
+            .into_iter()
+            .collect::<Vec<String>>();
+
+        let users = db.fetch_users(&user_ids).await?;
+        let members = if let Some(server_id) = server_id {
+            db.fetch_members(server_id, &user_ids).await?
+        } else {
+            Vec::new()
+        };
+
+        Ok((users, members))
+    }
+
     /// Helper function to fetch many messages with users
     pub async fn fetch_with_users(
         db: &Database,
@@ -814,93 +913,55 @@ impl Message {
         include_users: Option<bool>,
         server_id: Option<&str>,
     ) -> Result<BulkMessageResponse> {
-        let messages: Vec<v0::Message> = db
-            .fetch_messages(query)
-            .await?
-            .into_iter()
-            .map(|msg| msg.into_model(None, None))
-            .collect();
+        let messages = db.fetch_messages(query).await?;
 
         if let Some(true) = include_users {
-            let user_ids = messages
-                .iter()
-                .flat_map(|m| {
-                    let mut users = vec![m.author.clone()];
-                    if let Some(system) = &m.system {
-                        match system {
-                            v0::SystemMessage::ChannelDescriptionChanged { by } => {
-                                users.push(by.clone())
-                            }
-                            v0::SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
-                            v0::SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
-                                users.push(from.clone());
-                                users.push(to.clone())
-                            }
-                            v0::SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
-                            v0::SystemMessage::UserAdded { by, id, .. }
-                            | v0::SystemMessage::UserRemove { by, id, .. } => {
-                                users.push(by.clone());
-                                users.push(id.clone());
-                            }
-                            v0::SystemMessage::UserBanned { id, .. }
-                            | v0::SystemMessage::UserKicked { id, .. }
-                            | v0::SystemMessage::UserJoined { id, .. }
-                            | v0::SystemMessage::UserLeft { id, .. } => {
-                                users.push(id.clone());
-                            }
-                            v0::SystemMessage::Text { .. } => {}
-                            v0::SystemMessage::MessagePinned { by, .. } => {
-                                users.push(by.clone());
-                            }
-                            v0::SystemMessage::MessageUnpinned { by, .. } => {
-                                users.push(by.clone());
-                            }
-                            v0::SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
-                        }
-                    }
-                    users
-                })
-                .collect::<HashSet<String>>()
-                .into_iter()
-                .collect::<Vec<String>>();
-            let users = User::fetch_many_ids_as_mutuals(db, perspective, &user_ids).await?;
+            let (users, members) = Message::fetch_users(db, &messages, server_id).await?;
 
             Ok(BulkMessageResponse::MessagesAndUsers {
-                messages,
-                users,
-                members: if let Some(server_id) = server_id {
-                    Some(
-                        db.fetch_members(server_id, &user_ids)
-                            .await?
-                            .into_iter()
-                            .map(Into::into)
-                            .collect(),
-                    )
-                } else {
-                    None
-                },
+                messages: messages
+                    .into_iter()
+                    .map(|msg| msg.into_model(None, None))
+                    .collect(),
+                users: User::into_mutuals(perspective, users).await,
+                members: Some(members.into_iter().map(Into::into).collect()),
             })
         } else {
-            Ok(BulkMessageResponse::JustMessages(messages))
+            Ok(BulkMessageResponse::JustMessages(
+                messages
+                    .into_iter()
+                    .map(|msg| msg.into_model(None, None))
+                    .collect(),
+            ))
         }
     }
 
     /// Append content to message
     pub async fn append(
         db: &Database,
+        amqp: Option<&AMQP>,
         id: String,
         channel: String,
         append: AppendMessage,
     ) -> Result<()> {
-        db.append_message(&id, &append).await?;
+        if let Some(message) = db.append_message(&id, &append).await? {
+            if let Some(amqp) = amqp {
+                let author = message.fetch_author(db).await;
 
-        EventV1::MessageAppend {
-            id,
-            channel: channel.to_string(),
-            append: append.into(),
+                if let Err(e) = amqp.edit_message_search(message, author).await {
+                    log::error!("Error pushing message to RabbitMQ: {e}");
+                    capture_error(&e);
+                }
+            }
+
+            EventV1::MessageAppend {
+                id,
+                channel: channel.to_string(),
+                append: append.into(),
+            }
+            .p(channel)
+            .await;
         }
-        .p(channel)
-        .await;
 
         Ok(())
     }
@@ -998,9 +1059,10 @@ impl Message {
     }
 
     /// Delete a message
-    pub async fn delete(self, db: &Database) -> Result<()> {
+    pub async fn delete(&self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
         let file_ids: Vec<String> = self
             .attachments
+            .as_ref()
             .map(|files| files.iter().map(|file| file.id.to_string()).collect())
             .unwrap_or_default();
 
@@ -1010,12 +1072,20 @@ impl Message {
 
         db.delete_message(&self.id).await?;
 
+        if let Some(amqp) = amqp {
+            if let Err(e) = amqp.delete_message_search(self.id.clone()).await {
+                log::error!("Error pushing message to RabbitMQ: {e}");
+                capture_error(&e);
+            }
+        }
+
         EventV1::MessageDelete {
-            id: self.id,
+            id: self.id.clone(),
             channel: self.channel.clone(),
         }
-        .p(self.channel)
+        .p(self.channel.clone())
         .await;
+
         Ok(())
     }
 
@@ -1116,6 +1186,14 @@ impl Message {
             FieldsMessage::Pinned => self.pinned = None,
         }
     }
+
+    pub async fn fetch_author(&self, db: &Database) -> Option<User> {
+        if self.webhook.is_some() {
+            None
+        } else {
+            db.fetch_user(&self.author).await.ok()
+        }
+    }
 }
 
 impl SystemMessage {

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

@@ -1,8 +1,11 @@
+use revolt_result::Result;
 use std::collections::HashMap;
 use std::time::SystemTime;
-use revolt_result::Result;
 
-use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
+use crate::{
+    util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
+    MessageWithUser, PartialMessage,
+};
 
 #[cfg(feature = "mongodb")]
 mod mongodb;
@@ -23,10 +26,15 @@ pub trait AbstractMessages: Sync + Send {
     async fn fetch_messages_by_id(&self, ids: &[String]) -> Result<Vec<Message>>;
 
     /// Update a given message with new information
-    async fn update_message(&self, id: &str, message: &PartialMessage, remove: Vec<FieldsMessage>) -> Result<()>;
+    async fn update_message(
+        &self,
+        id: &str,
+        message: &PartialMessage,
+        remove: Vec<FieldsMessage>,
+    ) -> Result<()>;
 
     /// Append information to a given message
-    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()>;
+    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>>;
 
     /// Add a new reaction to a message
     async fn add_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()>;
@@ -48,8 +56,11 @@ pub trait AbstractMessages: Sync + Send {
         &self,
         channels: &[String],
         author: &str,
-        since: SystemTime
+        since: SystemTime,
     ) -> Result<HashMap<String, Vec<String>>>;
 
+    /// Fetches all messages along with their author from every message in decending order
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>>;
+    
     async fn delete_messages_by_user(&self, user_id: &str) -> Result<()>;
 }

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

@@ -1,7 +1,7 @@
 use bson::{to_bson, Document};
 use futures::try_join;
 use futures::StreamExt;
-use mongodb::options::FindOptions;
+use mongodb::options::{FindOptions, ReadConcern, ReturnDocument};
 use revolt_models::v0::MessageSort;
 use revolt_result::Result;
 use std::collections::{HashMap, HashSet};
@@ -9,8 +9,8 @@ use std::time::SystemTime;
 use ulid::Ulid;
 
 use crate::{
-    AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath, Message, MessageQuery,
-    MessageTimePeriod, MongoDb, PartialMessage,
+    util::ChunkedDatabaseGenerator, AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath,
+    Message, MessageQuery, MessageTimePeriod, MessageWithUser, MongoDb, PartialMessage,
 };
 
 use super::AbstractMessages;
@@ -205,7 +205,7 @@ impl AbstractMessages for MongoDb {
     }
 
     /// Append information to a given message
-    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
+    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>> {
         let mut query = doc! {};
 
         if let Some(embeds) = &append.embeds {
@@ -223,18 +223,18 @@ impl AbstractMessages for MongoDb {
         }
 
         if query.is_empty() {
-            return Ok(());
+            return Ok(None);
         }
 
-        self.col::<Document>(COL)
-            .update_one(
+        self.col::<Message>(COL)
+            .find_one_and_update(
                 doc! {
                     "_id": id
                 },
                 query,
             )
+            .return_document(ReturnDocument::After)
             .await
-            .map(|_| ())
             .map_err(|_| create_database_error!("update_one", COL))
     }
 
@@ -417,6 +417,53 @@ impl AbstractMessages for MongoDb {
         Ok(deleted_messages)
     }
 
+    /// Fetches all messages along with their author from every message in decending order
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>> {
+        let mut session = self
+            .start_session()
+            .await
+            .map_err(|_| create_database_error!("start_session", COL))?;
+
+        session
+            .start_transaction()
+            .read_concern(ReadConcern::snapshot())
+            .await
+            .map_err(|_| create_database_error!("start_transaction", COL))?;
+
+        let cursor = self
+            .col::<Message>(COL)
+            .aggregate([
+                doc! {
+                    "$lookup": {
+                        "from": "users",
+                        "localField": "author",
+                        "foreignField": "_id",
+                        "as": "user"
+                    }
+                },
+                doc! {
+                    "$set": {
+                        "user": {
+                            "$first": "$user"
+                        }
+                    }
+                },
+                doc! {
+                    "$sort": {
+                        "_id": -1
+                    }
+                },
+            ])
+            .with_type::<MessageWithUser>()
+            .session(&mut session)
+            .batch_size(1000)
+            .await
+            .inspect_err(|e| log::error!("{e}"))
+            .map_err(|_| create_database_error!("aggregate", COL))?;
+
+        Ok(ChunkedDatabaseGenerator::new_mongo(session, cursor))
+    }
+
     async fn delete_messages_by_user(&self, user_id: &str) -> Result<()> {
         self.delete_bulk_messages(doc! {
             "author": user_id,

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

@@ -1,6 +1,6 @@
 use crate::{
-    AppendMessage, FieldsMessage, Message, MessageQuery,
-    PartialMessage, ReferenceDb,
+    util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
+    MessageWithUser, PartialMessage, ReferenceDb,
 };
 use futures::future::try_join_all;
 use indexmap::IndexSet;
@@ -215,7 +215,7 @@ impl AbstractMessages for ReferenceDb {
     }
 
     /// Append information to a given message
-    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
+    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>> {
         let mut messages = self.messages.lock().await;
         if let Some(message_data) = messages.get_mut(id) {
             if let Some(embeds) = &append.embeds {
@@ -226,9 +226,11 @@ impl AbstractMessages for ReferenceDb {
                         message_data.embeds = Some(embeds.clone());
                     }
                 }
-            }
 
-            Ok(())
+                Ok(Some(message_data.clone()))
+            } else {
+                Ok(None)
+            }
         } else {
             Err(create_error!(NotFound))
         }
@@ -353,6 +355,24 @@ impl AbstractMessages for ReferenceDb {
         Ok(deleted_messages)
     }
 
+    /// Fetches all messages along with their author from every message in decending order
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>> {
+        let users = self.users.lock().await;
+
+        Ok(ChunkedDatabaseGenerator::new_reference(
+            self.messages
+                .lock()
+                .await
+                .values()
+                .cloned()
+                .map(|message| MessageWithUser {
+                    user: users.get(&message.author).cloned(),
+                    message,
+                })
+                .collect(),
+        ))
+    }
+
     async fn delete_messages_by_user(&self, user_id: &str) -> Result<()> {
         let mut messages = self.messages.lock().await;
 

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

@@ -320,7 +320,7 @@ mod tests {
 
     use crate::{Member, PartialMember, RemovalIntention, Server, User};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn muted_member_rejoin() {
         database_test!(|db| async move {
             match db {

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

@@ -1,11 +1,12 @@
 use std::collections::{HashMap, HashSet};
 
+use revolt_config::capture_error;
 use revolt_models::v0::{self, DataCreateServerChannel};
 use revolt_permissions::{OverrideField, DEFAULT_PERMISSION_SERVER};
 use revolt_result::Result;
 use ulid::Ulid;
 
-use crate::{events::client::EventV1, Channel, Database, File, User};
+use crate::{AMQP, Channel, Database, File, User, events::client::EventV1};
 
 auto_derived_partial!(
     /// Server
@@ -214,7 +215,16 @@ impl Server {
     }
 
     /// Delete a server
-    pub async fn delete(self, db: &Database) -> Result<()> {
+    pub async fn delete(self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
+        if let Some(amqp) = amqp {
+            for channel_id in self.channels {
+                if let Err(e) = amqp.delete_channel_search(channel_id).await {
+                    log::error!("Error pushing message to RabbitMQ: {e}");
+                    capture_error(&e);
+                }
+            }
+        }
+
         EventV1::ServerDelete {
             id: self.id.clone(),
         }
@@ -420,7 +430,7 @@ mod tests {
 
     use crate::{fixture, util::permissions::DatabasePermissionQuery};
 
-    #[async_std::test]
+    #[tokio::test]
     async fn permissions() {
         database_test!(|db| async move {
             fixture!(db, "server_with_roles",

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

@@ -381,6 +381,17 @@ impl User {
         )
     }
 
+    pub async fn into_mutuals(perspective: &User, users: Vec<User>) -> Vec<v0::User> {
+        let online_ids =
+            filter_online(&users.iter().map(|user| user.id.clone()).collect::<Vec<_>>()).await;
+
+        join_all(users.into_iter().map(|user| async {
+            let is_online = online_ids.contains(&user.id);
+            user.into_known(perspective, is_online).await
+        }))
+        .await
+    }
+
     /// Find a free discriminator for a given username
     pub async fn find_discriminator(
         db: &Database,
@@ -531,7 +542,7 @@ impl User {
     pub async fn add_friend(
         &mut self,
         db: &Database,
-        amqp: &AMQP,
+        amqp: Option<&AMQP>,
         target: &mut User,
     ) -> Result<()> {
         match self.relationship_with(&target.id) {
@@ -541,8 +552,10 @@ impl User {
             RelationshipStatus::Blocked => Err(create_error!(Blocked)),
             RelationshipStatus::BlockedOther => Err(create_error!(BlockedByOther)),
             RelationshipStatus::Incoming => {
-                // Accept incoming friend request
-                _ = amqp.friend_request_accepted(self, target).await;
+                if let Some(amqp) = amqp {
+                    // Accept incoming friend request
+                    _ = amqp.friend_request_accepted(self, target).await;
+                };
 
                 self.apply_relationship(
                     db,
@@ -572,7 +585,9 @@ impl User {
                     }));
                 }
 
-                _ = amqp.friend_request_received(target, self).await;
+                if let Some(amqp) = amqp {
+                    _ = amqp.friend_request_received(target, self).await;
+                };
 
                 // Send the friend request
                 self.apply_relationship(
@@ -867,13 +882,13 @@ impl User {
     /// - deletes owned bots, servers and messages
     /// - removes user from all groups
     /// - clears relationships
-    pub async fn delete(&mut self, db: &Database) -> Result<()> {
+    pub async fn delete(&mut self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
         for bot in db.fetch_bots_by_user(&self.id).await? {
             bot.delete(db).await?;
         }
 
         for server in db.fetch_owned_servers(&self.id).await? {
-            server.delete(db).await?;
+            server.delete(db, amqp).await?;
         }
 
         self.remove_from_all_groups(db).await?;
@@ -926,7 +941,7 @@ mod tests {
         assert!(User::validate_username(username_https).is_err());
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn username_sanitisation_clean() {
         let username_clean = "Test";
 
@@ -936,7 +951,7 @@ mod tests {
         assert_eq!(username_clean, username_clean_sanitised.unwrap());
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn username_sanitisation_homoglyphs() {
         let username_homoglyphs = "𝔽𝕌Ňℕｙ";
 
@@ -947,7 +962,7 @@ mod tests {
         assert_eq!("funny", username_homoglyphs_sanitised);
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn username_sanitisation_padding() {
         let username_padding = "a";
 
@@ -956,7 +971,7 @@ mod tests {
         assert_eq!("a_", username);
     }
 
-    #[async_std::test]
+    #[tokio::test]
     async fn create_user() {
         use revolt_result::Result;
 

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

@@ -305,6 +305,6 @@ pub async fn worker(db: Database, amqp: AMQP) {
         }
 
         // Sleep for an arbitrary amount of time.
-        async_std::task::sleep(Duration::from_secs(1)).await;
+        tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }

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

@@ -82,6 +82,6 @@ pub async fn worker(db: Database) {
         }
 
         // Sleep for an arbitrary amount of time.
-        async_std::task::sleep(Duration::from_secs(1)).await;
+        tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }

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

@@ -2,7 +2,7 @@
 
 use crate::{Database, AMQP};
 
-use async_std::task;
+use tokio::task;
 use std::time::Instant;
 
 const WORKER_COUNT: usize = 5;
@@ -16,7 +16,7 @@ pub fn start_workers(db: Database, amqp: AMQP) {
     for _ in 0..WORKER_COUNT {
         task::spawn(ack::worker(db.clone(), amqp.clone()));
         task::spawn(last_message_id::worker(db.clone()));
-        task::spawn(process_embeds::worker(db.clone()));
+        task::spawn(process_embeds::worker(db.clone(), amqp.clone()));
     }
 }
 

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

@@ -1,4 +1,4 @@
-use crate::{models::Message, AppendMessage, Database};
+use crate::{AMQP, AppendMessage, Database, models::Message};
 
 use futures::future::join_all;
 use linkify::{LinkFinder, LinkKind};
@@ -7,11 +7,11 @@ use revolt_config::config;
 use revolt_result::Result;
 
 use async_lock::Semaphore;
-use async_std::task::spawn;
 use deadqueue::limited::Queue;
 use once_cell::sync::Lazy;
 use revolt_models::v0::Embed;
 use std::{collections::HashSet, sync::Arc};
+use tokio::task::spawn;
 
 use isahc::prelude::*;
 
@@ -41,7 +41,7 @@ pub async fn queue(channel: String, id: String, content: String) {
 }
 
 /// Start a new worker
-pub async fn worker(db: Database) {
+pub async fn worker(db: Database, amqp: AMQP) {
     let semaphore = Arc::new(Semaphore::new(
         config().await.api.workers.max_concurrent_connections,
     ));
@@ -49,6 +49,7 @@ pub async fn worker(db: Database) {
     loop {
         let task = Q.pop().await;
         let db = db.clone();
+        let amqp = amqp.clone();
         let semaphore = semaphore.clone();
 
         spawn(async move {
@@ -64,6 +65,7 @@ pub async fn worker(db: Database) {
             if let Ok(embeds) = embeds {
                 if let Err(err) = Message::append(
                     &db,
+                    Some(&amqp),
                     task.id,
                     task.channel,
                     AppendMessage {
@@ -159,6 +161,7 @@ pub async fn generate(
         .await
         .into_iter()
         .flatten()
+        .flatten()
         .collect::<Vec<Embed>>();
 
     // Prevent database update when no embeds are found.

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

@@ -5,7 +5,7 @@ use revolt_result::{create_error, Result};
 #[cfg(feature = "rocket-impl")]
 use revolt_result::Error;
 
-use async_std::sync::Mutex;
+use tokio::sync::Mutex;
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 

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

@@ -13,5 +13,5 @@ pub mod reference;
 pub mod shield;
 pub mod test_fixtures;
 
+pub use chunked::ChunkedDatabaseGenerator;
 pub use funcs::*;
-pub use chunked::ChunkedDatabaseGenerator;

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

@@ -14,6 +14,7 @@ mod server_members;
 mod servers;
 mod user_settings;
 mod users;
+mod search;
 mod accounts;
 mod mfa_tickets;
 mod sessions;
@@ -34,6 +35,7 @@ pub use server_members::*;
 pub use servers::*;
 pub use user_settings::*;
 pub use users::*;
+pub use search::*;
 pub use accounts::*;
 pub use mfa_tickets::*;
-pub use sessions::*;
+pub use sessions::*;

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

@@ -0,0 +1,74 @@
+use iso8601_timestamp::Timestamp;
+use std::collections::HashSet;
+
+auto_derived!(
+    /// Options for searching messages in a server or channel
+    pub struct DataChannelMessagesSearch {
+        /// Channel to search in
+        pub channel: Option<String>,
+        /// Server to search in
+        pub server: Option<String>,
+
+        /// Filter options
+        pub filters: Option<DataChannelMessagesSearchFilters>,
+
+        /// What index to start the search at
+        pub offset: Option<u64>,
+        /// Max amount of messages to return
+        pub limit: Option<u64>,
+        /// Sort order
+        pub sort: Option<SortOrder>,
+    }
+
+    /// Message search filters
+    pub struct DataChannelMessagesSearchFilters {
+        /// Message content
+        pub content: Option<String>,
+        /// Specific user
+        pub author: Option<HashSet<String>>,
+
+        /// Mentions a user
+        pub mentions: Option<HashSet<String>>,
+        /// Mentions a role
+        pub role_mentions: Option<HashSet<String>>,
+
+        /// Send before a specific date
+        pub before_date: Option<Timestamp>,
+        /// Sent after a specific date
+        pub after_date: Option<Timestamp>,
+
+        /// What type of user sent the message
+        pub author_type: Option<HashSet<AuthorType>>,
+        /// Whether the message is pinned or not
+        pub pinned: Option<bool>,
+        /// Require message to have a specific component type
+        pub components: Option<HashSet<MessageComponent>>,
+    }
+
+    /// Message author type
+    #[derive(Copy, Hash)]
+    pub enum AuthorType {
+        User,
+        Bot,
+        Webhook,
+    }
+
+    /// Message component
+    #[derive(Copy, Hash)]
+    pub enum MessageComponent {
+        Image,
+        Video,
+        Link,
+        File,
+        Embed,
+    }
+
+    /// Message sort order
+    #[derive(Copy, Default)]
+    #[cfg_attr(feature = "serde", serde(rename_all = "lowercase"))]
+    pub enum SortOrder {
+        Asc,
+        #[default]
+        Desc,
+    }
+);

crates/core/permissions/Cargo.toml

@@ -18,7 +18,7 @@ try-from-primitive = ["dep:num_enum"]
 
 [dev-dependencies]
 # Async
-async-std = { workspace = true, features = ["attributes"] }
+tokio = { workspace = true }
 
 [dependencies]
 # Core

crates/core/permissions/src/test.rs

@@ -4,7 +4,7 @@ use crate::{
     DEFAULT_PERMISSION_SERVER, DEFAULT_PERMISSION_VIEW_ONLY,
 };
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_user_permissions() {
     /// Scenario in which we are friends with a user
     /// and we have a DM channel open with them
@@ -102,7 +102,7 @@ async fn validate_user_permissions() {
     }
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_group_permissions() {
     /// Scenario in which we are in a group channel with only talking permission
     struct Scenario {}
@@ -202,7 +202,7 @@ async fn validate_group_permissions() {
     }
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_server_permissions() {
     /// Scenario in which we are in a server channel where:
     /// - the server grants reading history and sending messages by default
@@ -314,7 +314,7 @@ async fn validate_server_permissions() {
     }
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn validate_timed_out_member() {
     /// Scenario in which we are in a server that we have been timed out from
     struct Scenario {}

crates/core/presence/Cargo.toml

@@ -14,7 +14,7 @@ redis-is-patched = []
 
 [dev-dependencies]
 # Async
-async-std = { workspace = true, features = ["attributes"] }
+tokio = { workspace = true }
 
 # Config for loading Redis URI
 revolt-config = { workspace = true }
@@ -26,4 +26,4 @@ rand = { workspace = true }
 once_cell = { workspace = true }
 
 # Redis
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }

crates/core/presence/src/lib.rs

@@ -195,7 +195,7 @@ mod tests {
     use crate::{clear_region, create_session, delete_session, filter_online, is_online};
     use rand::Rng;
 
-    #[async_std::test]
+    #[tokio::test]
     async fn it_works() {
         revolt_config::config().await;
 

crates/core/search/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "revolt-search"
+version = "0.13.7"
+edition = "2024"
+license = "MIT"
+publish = false
+
+[dependencies]
+revolt-config = { workspace = true }
+revolt-database = { workspace = true }
+revolt-models = { workspace = true }
+elasticsearch = { workspace = true }
+serde_json = { workspace = true }
+ulid = { workspace = true }
+iso8601-timestamp = { workspace = true, features = ["serde", "bson"] }
+futures = { workspace = true }
+elasticsearch-dsl = { workspace = true }
+serde = { workspace = true }
+linkify = { workspace = true }

crates/core/search/src/client.rs

@@ -0,0 +1,413 @@
+use std::fmt::Display;
+
+use elasticsearch::{
+    BulkOperation, BulkParts, CreateParts, DeleteByQueryParts, DeleteParts, Elasticsearch,
+    IndexParts, SearchParts,
+    auth::Credentials,
+    http::{
+        response::Exception,
+        transport::{SingleNodeConnectionPool, TransportBuilder},
+    },
+    indices::{IndicesCreateParts, IndicesDeleteParts},
+};
+use elasticsearch_dsl::{FieldSort, Query, Search, SearchResponse, Sort};
+use linkify::{LinkFinder, LinkKind};
+use revolt_database::{Database, Message, MessageWithUser, User};
+use serde_json::{Map, Value, json, to_value};
+
+pub use elasticsearch;
+
+use crate::{AuthorType, MessageComponent, SearchTerms};
+
+/// Elasticsearch errors
+#[derive(Debug)]
+pub enum Error {
+    Http(elasticsearch::Error),
+    Exception(Exception),
+}
+
+impl From<elasticsearch::Error> for Error {
+    fn from(value: elasticsearch::Error) -> Self {
+        Self::Http(value)
+    }
+}
+
+impl From<Exception> for Error {
+    fn from(value: Exception) -> Self {
+        Self::Exception(value)
+    }
+}
+
+impl std::error::Error for Error {}
+impl Display for Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Error::Http(error) => write!(f, "Http error: {error}"),
+            Error::Exception(exception) => write!(f, "Elasticsearch error: {exception:?}"),
+        }
+    }
+}
+
+/// Higher level elasticsearch API more fit for our specific usecase
+#[derive(Debug, Clone)]
+pub struct ElasticsearchClient {
+    pub inner: Elasticsearch,
+}
+
+impl ElasticsearchClient {
+    pub fn new(host: &str, port: u16, key: String) -> Self {
+        let pool =
+            SingleNodeConnectionPool::new(format!("{host}:{port}").as_str().try_into().unwrap());
+        let transport = TransportBuilder::new(pool)
+            .auth(Credentials::EncodedApiKey(key))
+            .build()
+            .unwrap();
+
+        let inner = Elasticsearch::new(transport);
+
+        Self { inner }
+    }
+
+    /// Delete messages index along with all documents
+    pub async fn delete_indexes(&self) -> Result<(), Error> {
+        let exception = self
+            .inner
+            .indices()
+            .delete(IndicesDeleteParts::Index(&["messages"]))
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Create the messages index
+    pub async fn setup_indexes(&self) -> Result<(), Error> {
+        let exception = self
+            .inner
+            .indices()
+            .create(IndicesCreateParts::Index("messages"))
+            .body(json!({
+                "mappings": {
+                    "properties": {
+                        "content": {"type": "text"},
+                        "author": {"type": "keyword"},
+                        "author_type": {"type": "keyword"},
+                        "channel": {"type": "keyword"},
+                        "mentions": {"type": "keyword"},
+                        "role_mentions": {"type": "keyword"},
+                        "pinned": {"type": "boolean"},
+                        "embeds": {
+                            "properties": {}
+                        },
+                        "attachments": {
+                            "type": "nested",
+                            "dynamic": false,
+                            "properties": {
+                                "metadata.type": {
+                                    "type": "keyword"
+                                }
+                            }
+                        },
+                        "has_link": { "type": "boolean" },
+                    }
+                }
+            }))
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Performs a search for messages, returns a vec of message ids
+    pub async fn search(&self, terms: SearchTerms) -> Result<Vec<String>, Error> {
+        let mut query = Query::bool().filter(Query::terms("channel", terms.channels));
+
+        if let Some(content) = terms.filters.content {
+            query = query.filter(Query::r#match("content", content))
+        }
+
+        if let Some(author) = terms.filters.author {
+            query = query.filter(Query::terms("author", author))
+        }
+
+        if let Some(mentions) = terms.filters.mentions {
+            query = query.filter(Query::terms("mentions", mentions))
+        }
+
+        if let Some(author_type) = terms.filters.author_type {
+            query = query.filter(Query::terms("author_type", author_type))
+        }
+
+        if let Some(pinned) = terms.filters.pinned {
+            if pinned {
+                query = query.filter(Query::exists("pinned"))
+            } else {
+                query = query.filter(Query::bool().must_not(Query::exists("pinned")))
+            }
+        }
+
+        if let Some(components) = terms.filters.components {
+            let mut components_query = Query::bool();
+            let mut attachments_query = Query::bool();
+
+            for component in components {
+                match component {
+                    MessageComponent::Image => {
+                        attachments_query = attachments_query
+                            .should(Query::term("attachments.metadata.type", "Image"))
+                    }
+                    MessageComponent::Video => {
+                        attachments_query = attachments_query
+                            .should(Query::term("attachments.metadata.type", "Video"))
+                    }
+                    MessageComponent::Link => {
+                        components_query = components_query.should(Query::exists("has_link"))
+                    }
+                    MessageComponent::File => {
+                        attachments_query = attachments_query.should(Query::exists("attachments"))
+                    }
+                    MessageComponent::Embed => {
+                        components_query = components_query.should(Query::exists("embeds"))
+                    }
+                };
+            }
+
+            query = query
+                .filter(components_query.should(Query::nested("attachments", attachments_query)));
+        }
+
+        let search = Search::new()
+            .query(query)
+            .stats(false)
+            .sort(Sort::FieldSort(
+                FieldSort::new("_id".to_string()).order(terms.sort.unwrap_or_default().into()),
+            ));
+
+        let response = self
+            .inner
+            .search(SearchParts::Index(&["messages"]))
+            .stored_fields(&[])
+            .body(search)
+            .size(terms.limit.unwrap_or(100) as i64)
+            .from(terms.offset.unwrap_or(0) as i64)
+            .send()
+            .await?;
+
+        if response.status_code().is_success() {
+            let messages = response.json::<SearchResponse>().await?;
+            Ok(messages.hits.hits.into_iter().map(|hit| hit.id).collect())
+        } else {
+            Err(response
+                .exception()
+                .await?
+                .expect("No exception with error response.")
+                .into())
+        }
+    }
+
+    /// Creates a source for a message which can be stored and indexed into elasticsearch
+    fn create_message_source(
+        &self,
+        _db: &Database,
+        message: Message,
+        author: Option<User>,
+    ) -> Value {
+        let mut map = Map::new();
+
+        map.insert("channel".to_string(), Value::String(message.channel));
+
+        map.insert("author".to_string(), Value::String(message.author));
+
+        if let Some(content) = message.content {
+            // Is there a better way to handle this? can elasticsearch index links itself?
+            // Maybe in the future store the domains and be able to filter by that as well
+            let mut finder = LinkFinder::new();
+            finder.kinds(&[LinkKind::Url]);
+
+            if finder.links(&content).next().is_some() {
+                map.insert("has_link".to_string(), Value::Bool(true));
+            }
+
+            map.insert("content".to_string(), Value::String(content));
+        }
+
+        if let Some(attachments) = message.attachments {
+            // TODO: fetch the file metadata from FileHash because of File.metadata deprecation
+            // let metadata = attachment.as_hash(db).await.expect("Failed to fetch FileHash").metadata;
+
+            map.insert(
+                "attachments".to_string(),
+                serde_json::to_value(attachments).unwrap(),
+            );
+        }
+
+        if let Some(embeds) = message.embeds {
+            map.insert("embeds".to_string(), serde_json::to_value(embeds).unwrap());
+        }
+
+        if let Some(mentions) = message.mentions {
+            map.insert(
+                "mentions".to_string(),
+                serde_json::to_value(mentions).unwrap(),
+            );
+        }
+
+        if let Some(role_mentions) = message.role_mentions {
+            map.insert(
+                "role_mentions".to_string(),
+                serde_json::to_value(role_mentions).unwrap(),
+            );
+        }
+
+        if let Some(pinned) = message.pinned {
+            map.insert("pinned".to_string(), Value::Bool(pinned));
+        }
+
+        // This will turn bot author type to user author type if this is ran on a deleted message,
+        // due to the author not existing anymore so fetching will fail, this is probably niche enough
+        // to not really matter, might try fix in the future.
+        map.insert(
+            "author_type".to_string(),
+            to_value(if message.webhook.is_some() {
+                AuthorType::Webhook
+            } else if author.is_some_and(|user| user.bot.is_some()) {
+                AuthorType::Bot
+            } else {
+                AuthorType::User
+            })
+            .unwrap(),
+        );
+
+        Value::Object(map)
+    }
+
+    /// Bulk uploads and indexes messages to elasticsearch
+    pub async fn bulk_index_messages(
+        &self,
+        db: &Database,
+        messages: Vec<MessageWithUser>,
+    ) -> Result<(), Error> {
+        let mut ops = Vec::<BulkOperation<Value>>::new();
+
+        for message in messages {
+            let id = message.message.id.clone();
+            let source = self.create_message_source(db, message.message, message.user);
+
+            ops.push(BulkOperation::create(source).id(id).into());
+        }
+
+        let exception = self
+            .inner
+            .bulk(BulkParts::Index("messages"))
+            .body(ops)
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Uploads and indexes a single message to elasticsearch
+    pub async fn index_message(
+        &self,
+        db: &Database,
+        message: Message,
+        author: Option<User>,
+    ) -> Result<(), Error> {
+        let id = message.id.clone();
+        let source = self.create_message_source(db, message, author);
+
+        let exception = self
+            .inner
+            .create(CreateParts::IndexId("messages", &id))
+            .body(source)
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Updates or upserts an existing message to elasticsearch
+    pub async fn edit_message(
+        &self,
+        db: &Database,
+        message: Message,
+        author: Option<User>,
+    ) -> Result<(), Error> {
+        let id = message.id.clone();
+        let source = self.create_message_source(db, message, author);
+
+        let exception = self
+            .inner
+            .index(IndexParts::IndexId("messages", &id))
+            .body(source)
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Deletes a message from elasticsearch
+    pub async fn delete_message(&self, message_id: &str) -> Result<(), Error> {
+        let exception = self
+            .inner
+            .delete(DeleteParts::IndexId("messages", message_id))
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Deletes all messages in a channel from elasticsearch
+    pub async fn delete_channel(&self, channel_id: &str) -> Result<(), Error> {
+        let exception = self
+            .inner
+            .delete_by_query(DeleteByQueryParts::Index(&["messages"]))
+            .body(Search::new().query(Query::term("channel", channel_id)))
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+}

crates/core/search/src/lib.rs

@@ -0,0 +1,5 @@
+mod client;
+mod types;
+
+pub use client::*;
+pub use types::*;

crates/core/search/src/types.rs

@@ -0,0 +1,134 @@
+use std::collections::HashSet;
+
+use iso8601_timestamp::Timestamp;
+use revolt_models::v0;
+use serde::Serialize;
+
+/// Message author type
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Hash)]
+pub enum AuthorType {
+    User,
+    Bot,
+    Webhook,
+}
+
+/// Message component
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
+pub enum MessageComponent {
+    Image,
+    Video,
+    Link,
+    File,
+    Embed,
+}
+
+/// Message search filters
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct SearchFilters {
+    /// Message content
+    pub content: Option<String>,
+    /// Specific user
+    pub author: Option<HashSet<String>>,
+
+    /// Mentions a user
+    pub mentions: Option<HashSet<String>>,
+    /// Mentions a role
+    pub role_mentions: Option<HashSet<String>>,
+
+    /// Send before a specific date
+    pub before_date: Option<Timestamp>,
+    /// Sent after a specific date
+    pub after_date: Option<Timestamp>,
+
+    /// What type of user sent the message
+    pub author_type: Option<HashSet<AuthorType>>,
+    /// Whether the message is pinned or not
+    pub pinned: Option<bool>,
+    /// Require message to have a specific component type
+    pub components: Option<HashSet<MessageComponent>>,
+}
+
+/// Message sort order
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum SortOrder {
+    Asc,
+    #[default]
+    Desc,
+}
+
+/// Options for searching messages in a server or channel
+#[derive(Debug, Clone, PartialEq)]
+pub struct SearchTerms {
+    /// Channels to search in
+    pub channels: Vec<String>,
+
+    /// Filter options
+    pub filters: SearchFilters,
+
+    /// What index to start the search at
+    pub offset: Option<u64>,
+    /// Max amount of messages to return
+    pub limit: Option<u64>,
+    /// Sort order
+    pub sort: Option<SortOrder>,
+}
+
+impl From<v0::AuthorType> for AuthorType {
+    fn from(value: v0::AuthorType) -> Self {
+        match value {
+            v0::AuthorType::User => AuthorType::User,
+            v0::AuthorType::Bot => AuthorType::Bot,
+            v0::AuthorType::Webhook => AuthorType::Webhook,
+        }
+    }
+}
+
+impl From<v0::MessageComponent> for MessageComponent {
+    fn from(value: v0::MessageComponent) -> Self {
+        match value {
+            v0::MessageComponent::Image => MessageComponent::Image,
+            v0::MessageComponent::Video => MessageComponent::Video,
+            v0::MessageComponent::Link => MessageComponent::Link,
+            v0::MessageComponent::File => MessageComponent::File,
+            v0::MessageComponent::Embed => MessageComponent::Embed,
+        }
+    }
+}
+
+impl From<v0::SortOrder> for SortOrder {
+    fn from(value: v0::SortOrder) -> Self {
+        match value {
+            v0::SortOrder::Asc => SortOrder::Asc,
+            v0::SortOrder::Desc => SortOrder::Desc,
+        }
+    }
+}
+
+impl From<v0::DataChannelMessagesSearchFilters> for SearchFilters {
+    fn from(value: v0::DataChannelMessagesSearchFilters) -> Self {
+        Self {
+            content: value.content,
+            author: value.author,
+            mentions: value.mentions,
+            role_mentions: value.role_mentions,
+            before_date: value.before_date,
+            after_date: value.after_date,
+            author_type: value
+                .author_type
+                .map(|types| types.into_iter().map(Into::into).collect()),
+            pinned: value.pinned,
+            components: value
+                .components
+                .map(|types| types.into_iter().map(Into::into).collect()),
+        }
+    }
+}
+
+impl From<SortOrder> for elasticsearch_dsl::SortOrder {
+    fn from(value: SortOrder) -> Self {
+        match value {
+            SortOrder::Asc => elasticsearch_dsl::SortOrder::Asc,
+            SortOrder::Desc => elasticsearch_dsl::SortOrder::Desc,
+        }
+    }
+}

crates/daemons/crond/Cargo.toml

@@ -18,7 +18,7 @@ tokio = { workspace = true }
 futures = { workspace = true }
 
 # Redis
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 
 # RabbitMQ
 lapin = { workspace = true }

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

@@ -1,10 +1,10 @@
 use std::time::Duration;
 
-use revolt_database::Database;
+use revolt_database::{Database, AMQP};
 use revolt_result::Result;
 use tokio::time::sleep;
 
-pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
+pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
     loop {
         let accounts = db.fetch_accounts_due_for_deletion().await?;
         let count = accounts.len();
@@ -12,7 +12,7 @@ pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
         for mut account in accounts {
             let mut user = db.fetch_user(&account.id).await?;
 
-            user.delete(&db).await?;
+            user.delete(&db, Some(&amqp)).await?;
             account.mark_deleted(&db).await?;
         }
 

crates/daemons/pushd/Cargo.toml

@@ -20,7 +20,7 @@ fcm_v1 = { workspace = true }
 web-push = { workspace = true }
 isahc = { workspace = true, features = ["json"], optional = true }
 revolt_a2 = { workspace = true, features = ["ring"] }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 tokio = { workspace = true }
 async-trait = { workspace = true }
 ulid = { workspace = true }

crates/daemons/pushd/src/consumers/inbound/ack.rs

@@ -1,10 +1,9 @@
 use std::sync::Arc;
 
-use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 
 #[derive(Clone)]
 #[allow(unused)]
@@ -16,11 +15,7 @@ pub struct AckConsumer {
 
 #[async_trait]
 impl Consumer for AckConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/inbound/dm_call.rs

@@ -1,11 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 
 #[derive(Clone)]
 #[allow(unused)]
@@ -17,11 +16,7 @@ pub struct DmCallConsumer {
 
 #[async_trait]
 impl Consumer for DmCallConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs

@@ -1,11 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 
 #[derive(Clone)]
 #[allow(unused)]
@@ -17,11 +16,7 @@ pub struct FRAcceptedConsumer {
 
 #[async_trait]
 impl Consumer for FRAcceptedConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/inbound/fr_received.rs

@@ -1,11 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 
 #[derive(Clone)]
 #[allow(unused)]
@@ -17,11 +16,7 @@ pub struct FRReceivedConsumer {
 
 #[async_trait]
 impl Consumer for FRReceivedConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/inbound/generic.rs

@@ -1,11 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 
 #[derive(Clone)]
 #[allow(unused)]
@@ -17,11 +16,7 @@ pub struct GenericConsumer {
 
 #[async_trait]
 impl Consumer for GenericConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/inbound/mass_mention.rs

@@ -4,13 +4,13 @@ use std::{
     sync::Arc,
 };
 
-use crate::utils::{render_notification_content, Consumer};
+use crate::utils::render_notification_content;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
 use revolt_database::{
-    events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
-    MessageFlagsValue,
+    amqp::consumer::Consumer, events::rabbit::*,
+    util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member, MessageFlagsValue,
 };
 use revolt_models::v0::{MessageFlags, PushNotification};
 use revolt_result::ToRevoltError;
@@ -29,11 +29,7 @@ impl MassMessageConsumer {
         push: &PushNotification,
         users: &[String],
     ) -> Result<()> {
-        if let Ok(sessions) = self
-            .db
-            .fetch_sessions_with_subscription(users)
-            .await
-        {
+        if let Ok(sessions) = self.db.fetch_sessions_with_subscription(users).await {
             let config = revolt_config::config().await;
             for session in sessions {
                 if let Some(sub) = session.subscription {
@@ -72,11 +68,7 @@ impl MassMessageConsumer {
 
 #[async_trait]
 impl Consumer for MassMessageConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/inbound/message.rs

@@ -1,11 +1,11 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::utils::{render_notification_content, Consumer};
+use crate::utils::render_notification_content;
 use anyhow::Result;
 use async_trait::async_trait;
 use lapin::{message::Delivery, Channel, Connection};
 use log::debug;
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 
 #[derive(Clone)]
 #[allow(unused)]
@@ -17,11 +17,7 @@ pub struct MessageConsumer {
 
 #[async_trait]
 impl Consumer for MessageConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<Channel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
         Self {
             db,
             connection,

crates/daemons/pushd/src/consumers/outbound/apn.rs

@@ -1,6 +1,5 @@
 use std::{borrow::Cow, collections::BTreeMap, io::Cursor, sync::Arc};
 
-use crate::utils::Consumer;
 use anyhow::Result;
 use async_trait::async_trait;
 use base64::{
@@ -15,7 +14,7 @@ use revolt_a2::{
     },
     Client, ClientConfig, Endpoint, Error, ErrorBody, ErrorReason, Priority, PushType, Response,
 };
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 use revolt_models::v0::{Channel, Message, PushNotification};
 use serde::Serialize;
 
@@ -123,11 +122,7 @@ impl ApnsOutboundConsumer {
 
 #[async_trait]
 impl Consumer for ApnsOutboundConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<AMQPChannel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<AMQPChannel>, _: ()) -> Self {
         let config = revolt_config::config().await;
 
         if config.pushd.apn.pkcs8.is_empty()

crates/daemons/pushd/src/consumers/outbound/fcm.rs

@@ -1,6 +1,5 @@
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
-use crate::utils::Consumer;
 use anyhow::{bail, Result};
 use async_trait::async_trait;
 use fcm_v1::{
@@ -10,7 +9,7 @@ use fcm_v1::{
 };
 use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
 use revolt_config::config;
-use revolt_database::{events::rabbit::*, Database};
+use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
 use serde_json::Value;
 
 /// Custom notification data
@@ -126,11 +125,7 @@ pub struct FcmOutboundConsumer {
 
 #[async_trait]
 impl Consumer for FcmOutboundConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<AMQPChannel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<AMQPChannel>, _: ()) -> Self {
         let config = revolt_config::config().await;
 
         Self {

crates/daemons/pushd/src/consumers/outbound/vapid.rs

@@ -1,7 +1,5 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::utils::Consumer;
-
 use anyhow::{anyhow, bail, Result};
 use async_trait::async_trait;
 use base64::{
@@ -9,7 +7,9 @@ use base64::{
     Engine as _,
 };
 use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
-use revolt_database::{events::rabbit::*, util::format_display_name, Database};
+use revolt_database::{
+    amqp::consumer::Consumer, events::rabbit::*, util::format_display_name, Database,
+};
 use web_push::{
     ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
     WebPushClient, WebPushError, WebPushMessageBuilder,
@@ -27,11 +27,7 @@ pub struct VapidOutboundConsumer {
 
 #[async_trait]
 impl Consumer for VapidOutboundConsumer {
-    async fn create(
-        db: Database,
-        connection: Arc<Connection>,
-        channel: Arc<AMQPChannel>,
-    ) -> Self {
+    async fn create(db: Database, connection: Arc<Connection>, channel: Arc<AMQPChannel>, _: ()) -> Self {
         let config = revolt_config::config().await;
 
         if config.pushd.vapid.private_key.is_empty() || config.pushd.vapid.public_key.is_empty() {

crates/daemons/pushd/src/main.rs

@@ -9,7 +9,7 @@ use lapin::{
     Channel, Connection, ConnectionProperties,
 };
 use revolt_config::{config, Settings};
-use revolt_database::Database;
+use revolt_database::{Database, amqp::consumer::{Consumer, Delegate}};
 use tokio::signal::ctrl_c;
 
 mod consumers;
@@ -23,8 +23,6 @@ use consumers::{
     outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer},
 };
 
-use crate::utils::{Consumer, Delegate};
-
 #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
 async fn main() {
     // Configure logging and environment
@@ -275,11 +273,12 @@ where
         consumer.tag()
     );
 
-    let delegate = Delegate(
+    let delegate = Delegate::new(
         F::create(
             db.clone(),
             connection.clone(),
             channel.clone(),
+            ()
         )
         .await,
     );

crates/daemons/pushd/src/utils/mod.rs

@@ -1,5 +1,3 @@
 mod renderer;
-mod consumer;
 
-pub use renderer::render_notification_content;
-pub use consumer::{Consumer, Delegate};
+pub use renderer::render_notification_content;

crates/daemons/searchd/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "revolt-searchd"
+version = "0.13.7"
+edition = "2024"
+license = "AGPL-3.0-or-later"
+publish = false
+
+[dependencies]
+tokio = { workspace = true, features = ["macros"] }
+revolt-database = { workspace = true }
+revolt-config = { workspace = true }
+revolt-search = { workspace = true }
+lapin = { workspace = true }
+async-trait = { workspace = true }
+log = { workspace = true }
+serde_json = { workspace = true }
+anyhow = { workspace = true }

crates/daemons/searchd/src/consumers/channel_delete.rs

@@ -0,0 +1,59 @@
+use std::sync::Arc;
+
+use lapin::{Channel, Connection, message::Delivery, options::{BasicAckOptions, BasicRejectOptions}};
+use async_trait::async_trait;
+use revolt_database::{Database, events::rabbit::ChannelDeletePayload, amqp::consumer::Consumer};
+use revolt_search::ElasticsearchClient;
+use anyhow::Result;
+
+#[allow(unused)]
+#[derive(Clone)]
+pub struct ChannelDeleteConsumer {
+    client: ElasticsearchClient,
+    database: Database,
+    connection: Arc<Connection>,
+    channel: Arc<Channel>,
+}
+
+#[async_trait]
+impl Consumer<ElasticsearchClient> for ChannelDeleteConsumer {
+    async fn create(database: Database, connection: Arc<Connection>, channel: Arc<Channel>, client: ElasticsearchClient) -> Self {
+        Self {
+            client,
+            database,
+            connection,
+            channel,
+        }
+    }
+
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload = serde_json::from_slice::<ChannelDeletePayload>(&delivery.data)
+            .expect("Failed to decode message");
+        log::debug!("Received channel delete {payload:?}");
+
+        if self
+            .client
+            .delete_channel(&payload.channel_id)
+            .await
+            .is_ok()
+        {
+            self.channel
+                .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
+                .await
+                .expect("Failed to ack");
+        } else {
+            self.channel
+                .basic_reject(delivery.delivery_tag, BasicRejectOptions {
+                    requeue: true,
+                })
+                .await
+                .expect("Failed to reject");
+        };
+
+        Ok(())
+    }
+}

crates/daemons/searchd/src/consumers/message.rs

@@ -0,0 +1,60 @@
+use std::sync::Arc;
+
+use lapin::{Channel, Connection, message::Delivery, options::{BasicAckOptions, BasicRejectOptions}};
+
+use async_trait::async_trait;
+use revolt_database::{Database, events::rabbit::MessageCreatePayload, amqp::consumer::Consumer};
+use revolt_search::ElasticsearchClient;
+use anyhow::Result;
+
+#[allow(unused)]
+#[derive(Clone)]
+pub struct MessageConsumer {
+    client: ElasticsearchClient,
+    database: Database,
+    connection: Arc<Connection>,
+    channel: Arc<Channel>,
+}
+
+#[async_trait]
+impl Consumer<ElasticsearchClient> for MessageConsumer {
+    async fn create(database: Database, connection: Arc<Connection>, channel: Arc<Channel>, client: ElasticsearchClient) -> Self {
+        Self {
+            client,
+            database,
+            connection,
+            channel,
+        }
+    }
+
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload = serde_json::from_slice::<MessageCreatePayload>(&delivery.data)
+            .expect("Failed to decode message");
+        log::debug!("Received message {payload:?}");
+
+        if self
+            .client
+            .index_message(&self.database, payload.message, payload.user)
+            .await
+            .is_ok()
+        {
+            self.channel
+                .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
+                .await
+                .expect("Failed to ack");
+        } else {
+            self.channel
+                .basic_reject(delivery.delivery_tag, BasicRejectOptions {
+                    requeue: true,
+                })
+                .await
+                .expect("Failed to reject");
+        };
+
+        Ok(())
+    }
+}

crates/daemons/searchd/src/consumers/message_delete.rs

@@ -0,0 +1,61 @@
+use std::sync::Arc;
+
+use lapin::{Channel, Connection, message::Delivery, options::{BasicAckOptions, BasicRejectOptions}};
+
+use async_trait::async_trait;
+use revolt_database::{Database, events::rabbit::MessageDeletePayload, amqp::consumer::Consumer};
+use revolt_search::ElasticsearchClient;
+use anyhow::Result;
+
+#[allow(unused)]
+#[derive(Clone)]
+pub struct MessageDeleteConsumer {
+    client: ElasticsearchClient,
+    database: Database,
+    connection: Arc<Connection>,
+    channel: Arc<Channel>,
+}
+
+
+#[async_trait]
+impl Consumer<ElasticsearchClient> for MessageDeleteConsumer {
+    async fn create(database: Database, connection: Arc<Connection>, channel: Arc<Channel>, client: ElasticsearchClient) -> Self {
+        Self {
+            client,
+            database,
+            connection,
+            channel,
+        }
+    }
+
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload = serde_json::from_slice::<MessageDeletePayload>(&delivery.data)
+            .expect("Failed to decode message");
+        log::debug!("Received message delete {payload:?}");
+
+        if self
+            .client
+            .delete_message(&payload.message_id)
+            .await
+            .is_ok()
+        {
+            self.channel
+                .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
+                .await
+                .expect("Failed to ack");
+        } else {
+            self.channel
+                .basic_reject(delivery.delivery_tag, BasicRejectOptions {
+                    requeue: true,
+                })
+                .await
+                .expect("Failed to reject");
+        };
+
+        Ok(())
+    }
+}

crates/daemons/searchd/src/consumers/message_edit.rs

@@ -0,0 +1,60 @@
+use std::sync::Arc;
+
+
+use lapin::{Channel, Connection, message::Delivery, options::{BasicAckOptions, BasicRejectOptions}};
+
+use async_trait::async_trait;
+use revolt_database::{Database, events::rabbit::MessageEditPayload, amqp::consumer::Consumer};
+use revolt_search::ElasticsearchClient;
+use anyhow::Result;
+
+#[derive(Clone)]
+pub struct MessageEditConsumer {
+    client: ElasticsearchClient,
+    database: Database,
+    connection: Arc<Connection>,
+    channel: Arc<Channel>,
+}
+
+#[async_trait]
+impl Consumer<ElasticsearchClient> for MessageEditConsumer {
+    async fn create(database: Database, connection: Arc<Connection>, channel: Arc<Channel>, client: ElasticsearchClient) -> Self {
+        Self {
+            client,
+            database,
+            connection,
+            channel,
+        }
+    }
+
+    fn channel(&self) -> &Arc<Channel> {
+        &self.channel
+    }
+
+    async fn consume(&self, delivery: Delivery) -> Result<()> {
+        let payload = serde_json::from_slice::<MessageEditPayload>(&delivery.data)
+            .expect("Failed to decode message");
+        log::debug!("Received edit message {payload:?}");
+
+        if self
+            .client
+            .edit_message(&self.database, payload.message, payload.user)
+            .await
+            .is_ok()
+        {
+            self.channel
+                .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
+                .await
+                .expect("Failed to ack");
+        } else {
+            self.channel
+                .basic_reject(delivery.delivery_tag, BasicRejectOptions {
+                    requeue: true,
+                })
+                .await
+                .expect("Failed to reject");
+        };
+
+        Ok(())
+    }
+}

crates/daemons/searchd/src/consumers/mod.rs

@@ -0,0 +1,9 @@
+mod channel_delete;
+mod message;
+mod message_delete;
+mod message_edit;
+
+pub use channel_delete::*;
+pub use message::*;
+pub use message_delete::*;
+pub use message_edit::*;

crates/daemons/searchd/src/index.rs

@@ -0,0 +1,36 @@
+use revolt_config::capture_error;
+use revolt_database::Database;
+use revolt_search::ElasticsearchClient;
+
+pub async fn index_existing_messages(db: Database, client: ElasticsearchClient) {
+    log::info!("Starting bulk indexing.");
+
+    let mut generator = db
+        .fetch_all_messages()
+        .await
+        .expect("Database query failed");
+
+    let mut chunk = Vec::new();
+
+    while let Some(message) = generator.next().await.expect("Failed to fetch batch") {
+        chunk.push(message);
+
+        if chunk.len() >= 1000
+            && let Err(e) = client
+                .bulk_index_messages(&db, std::mem::take(&mut chunk))
+                .await
+        {
+            log::error!("Error bulk indexing messages: {e}");
+            capture_error(&e);
+        }
+    }
+
+    if !chunk.is_empty()
+        && let Err(e) = client.bulk_index_messages(&db, chunk).await
+    {
+        log::error!("Error bulk indexing messages: {e}");
+        capture_error(&e);
+    }
+
+    log::info!("Finished bulk indexing.")
+}

crates/daemons/searchd/src/main.rs

@@ -0,0 +1,206 @@
+use lapin::{
+    options::{BasicConsumeOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions},
+    types::{AMQPValue, FieldTable},
+    Channel, Connection, ConnectionProperties,
+};
+use log::info;
+use revolt_config::{Settings, config, configure};
+use revolt_database::{Database, DatabaseInfo, amqp::consumer::{Consumer, Delegate}};
+use revolt_search::ElasticsearchClient;
+use tokio::{signal::ctrl_c, spawn};
+use std::{marker::PhantomData, sync::Arc};
+
+mod consumers;
+mod index;
+
+async fn _main() {
+    configure!(api);
+    let config = config().await;
+
+    let db = DatabaseInfo::Auto.connect().await.unwrap();
+
+    let client = ElasticsearchClient::new(
+        &config.elasticsearch.host,
+        config.elasticsearch.port,
+        config.elasticsearch.api_key.clone(),
+    );
+
+    if std::env::var("REMAKE_MESSAGES_INDEX")
+        .as_deref()
+        .is_ok_and(|v| v == "1")
+    {
+        if client.delete_indexes().await.is_err() {
+            log::info!("Index does not existing, skipping.")
+        };
+
+        client.setup_indexes().await.unwrap();
+    };
+
+    let connection = Arc::new(
+        Connection::connect(
+            &format!(
+                "amqp://{}:{}@{}:{}",
+                &config.rabbit.username,
+                &config.rabbit.password,
+                &config.rabbit.host,
+                &config.rabbit.port,
+            ),
+            ConnectionProperties::default(),
+        )
+        .await
+        .expect("Failed to connect to RabbitMQ"),
+    );
+
+    let mut channels = Vec::new();
+
+    channels.push(
+        make_queue_and_consume::<consumers::MessageConsumer>(
+            &client,
+            &db,
+            &connection,
+            &config,
+            &config.elasticsearch.message_queue,
+        )
+        .await,
+    );
+
+    channels.push(
+        make_queue_and_consume::<consumers::MessageEditConsumer>(
+            &client,
+            &db,
+            &connection,
+            &config,
+            &config.elasticsearch.message_edit_queue,
+        )
+        .await,
+    );
+
+    channels.push(
+        make_queue_and_consume::<consumers::MessageDeleteConsumer>(
+            &client,
+            &db,
+            &connection,
+            &config,
+            &config.elasticsearch.message_delete_queue,
+        )
+        .await,
+    );
+
+    channels.push(
+        make_queue_and_consume::<consumers::ChannelDeleteConsumer>(
+            &client,
+            &db,
+            &connection,
+            &config,
+            &config.elasticsearch.channel_delete_queue,
+        )
+        .await,
+    );
+
+    let mut task = None;
+
+    if std::env::var("INDEX_ALL_MESSAGES")
+        .as_deref()
+        .is_ok_and(|v| v == "1")
+    {
+        task = Some(spawn(index::index_existing_messages(db, client.clone())));
+    }
+
+    ctrl_c().await.unwrap();
+
+    for channel in channels {
+        let _ = channel.close(0, "close".into()).await;
+    }
+
+    if let Some(task) = task {
+        task.abort();
+    }
+}
+
+async fn make_queue_and_consume<F: Consumer<ElasticsearchClient>>(
+    client: &ElasticsearchClient,
+    db: &Database,
+    connection: &Arc<Connection>,
+    config: &Settings,
+    queue_name: &str,
+) -> Arc<Channel> {
+    let channel = Arc::new(connection.create_channel().await.unwrap());
+
+    channel
+        .exchange_declare(
+            config.elasticsearch.exchange.clone().into(),
+            lapin::ExchangeKind::Direct,
+            ExchangeDeclareOptions {
+                durable: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .expect("Failed to declare exchange");
+
+    let mut table = FieldTable::default();
+    table.insert("x-queue-type".try_into().unwrap(), AMQPValue::LongString("quorum".into()));
+
+    let args = QueueDeclareOptions {
+        durable: true,
+        ..Default::default()
+    };
+
+    channel
+        .queue_declare(queue_name.into(), args, table)
+        .await
+        .unwrap();
+
+    channel
+        .queue_bind(
+            queue_name.into(),
+            config.elasticsearch.exchange.clone().into(),
+            queue_name.into(),
+            QueueBindOptions::default(),
+            FieldTable::default(),
+        )
+        .await
+        .expect(
+            "This probably means the revolt.messages exchange does not exist in rabbitmq!",
+        );
+
+
+    let consumer = channel
+        .basic_consume(
+            queue_name.into(),
+            "".into(),
+            BasicConsumeOptions {
+                no_ack: true,
+                ..Default::default()
+            },
+            FieldTable::default(),
+        )
+        .await
+        .unwrap();
+    info!(
+        "Consuming routing key {} as queue {}, tag {}",
+        queue_name,
+        queue_name,
+        consumer.tag()
+    );
+
+    let delegate = Delegate::new(
+        F::create(
+            db.clone(),
+            connection.clone(),
+            channel.clone(),
+            client.clone(),
+        )
+        .await,
+    );
+
+    consumer.set_delegate(delegate);
+
+    channel
+}
+
+#[tokio::main]
+async fn main() {
+    _main().await
+}

crates/daemons/voice-ingress/Cargo.toml

@@ -13,7 +13,7 @@ log = { workspace = true }
 sentry = { workspace = true }
 lru = { workspace = true }
 ulid = { workspace = true }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 chrono = { workspace = true }
 
 # Serde
@@ -27,11 +27,6 @@ rocket_empty = { workspace = true }
 
 # Async
 futures = { workspace = true }
-async-std = { workspace = true, features = [
-    "tokio1",
-    "tokio02",
-    "attributes",
-] }
 
 # Core
 revolt-result = { workspace = true, features = ["rocket"] }

crates/delta/Cargo.toml

@@ -11,7 +11,7 @@ publish = false
 [dependencies]
 # Test
 rand = { workspace = true }
-redis-kiss = { workspace = true }
+redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
 
 # Utility
 lru = { workspace = true }
@@ -42,11 +42,7 @@ futures = { workspace = true }
 chrono = { workspace = true }
 async-channel = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
-async-std = { workspace = true, features = [
-    "tokio1",
-    "tokio02",
-    "attributes",
-] }
+tokio = { workspace = true }
 
 # internal util
 lettre = { workspace = true }
@@ -81,6 +77,7 @@ revolt-presence = { workspace = true }
 revolt-result = { workspace = true, features = ["rocket", "okapi"] }
 revolt-permissions = { workspace = true, features = ["schemas"] }
 revolt-ratelimits = { workspace = true, features = ["rocket"] }
+revolt-search = { workspace = true }
 
 # voice
 livekit-api = { workspace = true }

crates/delta/src/main.rs

@@ -11,6 +11,7 @@ pub mod util;
 use revolt_config::config;
 use revolt_database::AMQP;
 use revolt_ratelimits::rocket as ratelimiter;
+use revolt_search::ElasticsearchClient;
 use rocket::{Build, Rocket};
 use rocket_cors::{AllowedOrigins, CorsOptions};
 use rocket_prometheus::PrometheusMetrics;
@@ -79,6 +80,13 @@ pub async fn web() -> Rocket<Build> {
     // Ratelimits
     let ratelimits = ratelimiter::RatelimitStorage::new(util::ratelimits::DeltaRatelimits);
 
+    // Search
+    let elasticsearch = ElasticsearchClient::new(
+        &config.elasticsearch.host,
+        config.elasticsearch.port,
+        config.elasticsearch.api_key.clone(),
+    );
+
     routes::mount(config, rocket)
         .attach(prometheus.clone())
         .mount("/metrics", prometheus)
@@ -90,6 +98,7 @@ pub async fn web() -> Rocket<Build> {
         .manage(cors.clone())
         .manage(voice_client)
         .manage(ratelimits)
+        .manage(elasticsearch)
         .attach(ratelimiter::RatelimitFairing)
         .attach(cors)
         .configure(rocket::Config {

crates/delta/src/routes/account/create_account.rs

@@ -2,7 +2,7 @@
 //! POST /account/create
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use revolt_config::config;
 use revolt_database::{
     util::{

crates/delta/src/routes/account/resend_verification.rs

@@ -2,7 +2,7 @@
 //! POST /account/reverify
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use rocket::{serde::json::Json, State};
 use rocket_empty::EmptyResponse;
 use revolt_result::Result;

crates/delta/src/routes/account/send_password_reset.rs

@@ -2,7 +2,7 @@
 //! POST /account/reset_password
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use rocket::serde::json::Json;
 use rocket::State;
 use rocket_empty::EmptyResponse;

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

@@ -56,7 +56,7 @@ pub async fn invite_bot(
                 .throw_if_lacking_channel_permission(ChannelPermission::InviteOthers)?;
 
             channel
-                .add_user_to_group(db, amqp, &bot_user, &user.id)
+                .add_user_to_group(db, Some(amqp), &bot_user, &user.id)
                 .await
                 .map(|_| EmptyResponse)
         }

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

@@ -50,7 +50,7 @@ pub async fn delete(
             channel
                 .remove_user_from_group(
                     db,
-                    amqp,
+                    Some(amqp),
                     &user,
                     None,
                     options.leave_silently.unwrap_or_default(),
@@ -65,7 +65,7 @@ pub async fn delete(
         }
         Channel::TextChannel { .. } => {
             permissions.throw_if_lacking_channel_permission(ChannelPermission::ManageChannel)?;
-            channel.delete(db).await?;
+            channel.delete(db, Some(amqp)).await?;
 
             delete_voice_channel(voice_client, &UserVoiceChannel::from_channel(&channel)).await?;
         }

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

@@ -39,7 +39,7 @@ pub async fn add_member(
             }
 
             channel
-                .add_user_to_group(db, amqp, &member, &user.id)
+                .add_user_to_group(db, Some(amqp), &member, &user.id)
                 .await
                 .map(|_| EmptyResponse)
         }

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

@@ -46,7 +46,7 @@ pub async fn remove_member(
         }
 
         channel
-            .remove_user_from_group(db, amqp, &member, Some(&user.id), false)
+            .remove_user_from_group(db, Some(amqp), &member, Some(&user.id), false)
             .await?;
     } else {
         return Err(create_error!(InvalidOperation));

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

@@ -1,6 +1,6 @@
 use revolt_database::{
     util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Database, PartialMessage, User,
+    Database, PartialMessage, User, AMQP,
 };
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::Result;
@@ -16,6 +16,7 @@ use rocket_empty::EmptyResponse;
 #[delete("/<target>/messages/<msg>/reactions")]
 pub async fn clear_reactions(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     user: User,
     target: Reference<'_>,
     msg: Reference<'_>,
@@ -33,11 +34,12 @@ pub async fn clear_reactions(
     message
         .update(
             db,
+            Some(amqp),
             PartialMessage {
                 reactions: Some(Default::default()),
                 ..Default::default()
             },
-            vec![]
+            vec![],
         )
         .await
         .map(|_| EmptyResponse)

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

@@ -1,6 +1,5 @@
 use revolt_database::{
-    util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Database, User,
+    AMQP, Database, User, util::{permissions::DatabasePermissionQuery, reference::Reference}
 };
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::Result;
@@ -14,6 +13,7 @@ use rocket_empty::EmptyResponse;
 #[delete("/<target>/messages/<msg>", rank = 2)]
 pub async fn delete(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     user: User,
     target: Reference<'_>,
     msg: Reference<'_>,
@@ -28,5 +28,5 @@ pub async fn delete(
             .throw_if_lacking_channel_permission(ChannelPermission::ManageMessages)?;
     }
 
-    message.delete(db).await.map(|_| EmptyResponse)
+    message.delete(db, Some(amqp)).await.map(|_| EmptyResponse)
 }

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

@@ -1,8 +1,9 @@
 use iso8601_timestamp::Timestamp;
+use revolt_config::capture_error;
 use revolt_database::{
     tasks,
     util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Database, Message, PartialMessage, User,
+    Database, Message, PartialMessage, User, AMQP,
 };
 use revolt_models::v0::{self, Embed};
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
@@ -17,6 +18,7 @@ use validator::Validate;
 #[patch("/<target>/messages/<msg>", data = "<edit>")]
 pub async fn edit(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     user: User,
     target: Reference<'_>,
     msg: Reference<'_>,
@@ -82,7 +84,7 @@ pub async fn edit(
 
     partial.embeds = Some(new_embeds);
 
-    message.update(db, partial, vec![]).await?;
+    message.update(db, Some(amqp), partial, vec![]).await?;
 
     // Queue up a task for processing embeds if the we have sufficient permissions
     if permissions.has_channel_permission(ChannelPermission::SendEmbeds) {

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

@@ -1,3 +1,4 @@
+use revolt_config::capture_error;
 use revolt_database::{
     util::{permissions::DatabasePermissionQuery, reference::Reference},
     Channel, Database, PartialMessage, SystemMessage, User, AMQP,
@@ -38,6 +39,7 @@ pub async fn message_pin(
     message
         .update(
             db,
+            Some(amqp),
             PartialMessage {
                 pinned: Some(true),
                 ..Default::default()
@@ -46,6 +48,14 @@ pub async fn message_pin(
         )
         .await?;
 
+    if let Err(e) = amqp
+        .edit_message_search(message.clone(), message.fetch_author(db).await)
+        .await
+    {
+        log::error!("Error pushing message to RabbitMQ: {e}");
+        capture_error(&e);
+    }
+
     SystemMessage::MessagePinned {
         id: message.id.clone(),
         by: user.id.clone(),
@@ -116,8 +126,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.into()),
+            Some(user.clone()),
+            Some(member),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,

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

@@ -10,6 +10,7 @@ use validator::Validate;
 
 /// # Search for Messages
 ///
+/// DEPRECATED: use /search instead.
 /// This route searches for messages within the given parameters.
 #[openapi(tag = "Messaging")]
 #[post("/<target>/search", data = "<options>")]

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

@@ -162,16 +162,10 @@ pub async fn message_send(
     // Make sure we have server member (edge case if server owner)
     query.are_we_a_member().await;
 
-    // Create model user / members
-    let model_user = user
-        .clone()
-        .into_known_static(revolt_presence::is_online(&user.id).await)
-        .await;
-
-    let model_member: Option<v0::Member> = query
+    let member = query
         .member_ref()
         .as_ref()
-        .map(|member| member.clone().into_owned().into());
+        .map(|member| member.clone().into_owned());
 
     Ok(Json(
         Message::create_from_api(
@@ -180,15 +174,22 @@ pub async fn message_send(
             channel,
             data,
             v0::MessageAuthor::User(&author),
-            Some(model_user.clone()),
-            model_member.clone(),
+            Some(user.clone()),
+            member.clone(),
             user.limits().await,
             idempotency,
             permissions.has_channel_permission(ChannelPermission::SendEmbeds),
             allow_mentions,
         )
         .await?
-        .into_model(Some(model_user), model_member),
+        .into_model(
+            Some(
+                user.clone()
+                    .into_known_static(revolt_presence::is_online(&user.id).await)
+                    .await,
+            ),
+            member.map(Into::into),
+        ),
     ))
 }
 
@@ -298,8 +299,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.clone().into()),
+            Some(user.clone()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,
@@ -338,8 +339,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.clone().into()),
+            Some(user.clone()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -387,8 +388,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.clone().into()),
+            Some(user.clone()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("2".to_string()),
             false,
@@ -433,8 +434,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.clone().into()),
+            Some(user.clone()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -457,7 +458,7 @@ mod test {
         // Delete the message
         message
             .clone()
-            .delete(&harness.db)
+            .delete(&harness.db, None)
             .await
             .expect("Failed to delete message");
 
@@ -484,8 +485,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.clone().into()),
+            Some(user.clone()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("3".to_string()),
             false,
@@ -521,7 +522,7 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
             Some(member.clone().into()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("4".to_string()),
@@ -552,7 +553,7 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
             Some(member.clone().into()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("4".to_string()),
@@ -607,13 +608,8 @@ mod test {
                     .into(&harness.db, Some(&other_user))
                     .await,
             ),
-            Some(
-                other_user
-                    .clone()
-                    .into(&harness.db, Some(&other_user))
-                    .await,
-            ),
-            Some(other_member.clone().into()),
+            Some(other_user.clone()),
+            Some(other_member.clone()),
             other_user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -652,13 +648,8 @@ mod test {
                     .into(&harness.db, Some(&other_user))
                     .await,
             ),
-            Some(
-                other_user
-                    .clone()
-                    .into(&harness.db, Some(&other_user))
-                    .await,
-            ),
-            Some(other_member.clone().into()),
+            Some(other_user.clone()),
+            Some(other_member.clone()),
             other_user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -722,13 +713,8 @@ mod test {
                     .into(&harness.db, Some(&other_user))
                     .await,
             ),
-            Some(
-                other_user
-                    .clone()
-                    .into(&harness.db, Some(&other_user))
-                    .await,
-            ),
-            Some(other_member.clone().into()),
+            Some(other_user.clone()),
+            Some(other_member.clone()),
             other_user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,

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

@@ -1,3 +1,4 @@
+use revolt_config::capture_error;
 use revolt_database::{
     util::{permissions::DatabasePermissionQuery, reference::Reference},
     Channel, Database, FieldsMessage, PartialMessage, SystemMessage, User, AMQP,
@@ -36,9 +37,22 @@ pub async fn message_unpin(
     }
 
     message
-        .update(db, PartialMessage::default(), vec![FieldsMessage::Pinned])
+        .update(
+            db,
+            Some(amqp),
+            PartialMessage::default(),
+            vec![FieldsMessage::Pinned],
+        )
         .await?;
 
+    if let Err(e) = amqp
+        .edit_message_search(message.clone(), message.fetch_author(db).await)
+        .await
+    {
+        log::error!("Error pushing message to RabbitMQ: {e}");
+        capture_error(&e);
+    }
+
     SystemMessage::MessageUnpinned {
         id: message.id.clone(),
         by: user.id.clone(),
@@ -114,8 +128,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
-            Some(member.into()),
+            Some(user.clone()),
+            Some(member),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,

crates/delta/src/routes/invites/invite_join.rs

@@ -35,7 +35,7 @@ pub async fn join(
             channel, creator, ..
         } => {
             let mut channel = db.fetch_channel(channel).await?;
-            channel.add_user_to_group(db, amqp, &user, creator).await?;
+            channel.add_user_to_group(db, Some(amqp), &user, creator).await?;
             if let Channel::Group { recipients, .. } = &channel {
                 Ok(Json(InviteJoinResponse::Group {
                     users: User::fetch_many_ids_as_mutuals(db, &user, recipients).await?,

crates/delta/src/routes/mod.rs

@@ -17,6 +17,7 @@ mod servers;
 mod sync;
 mod users;
 mod webhooks;
+mod search;
 mod account;
 mod session;
 mod mfa;
@@ -43,7 +44,8 @@ pub fn mount(config: Settings, mut rocket: Rocket<Build>) -> Rocket<Build> {
             "/policy" => policy::routes(),
             "/push" => push::routes(),
             "/sync" => sync::routes(),
-            "/webhooks" => webhooks::routes()
+            "/webhooks" => webhooks::routes(),
+            "/search" => search::routes(),
         };
     } else {
         mount_endpoints_and_merged_docs! {
@@ -63,7 +65,8 @@ pub fn mount(config: Settings, mut rocket: Rocket<Build>) -> Rocket<Build> {
             "/onboard" => onboard::routes(),
             "/policy" => policy::routes(),
             "/push" => push::routes(),
-            "/sync" => sync::routes()
+            "/sync" => sync::routes(),
+            "/search" => search::routes(),
         };
     }
 

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

@@ -0,0 +1,10 @@
+use revolt_rocket_okapi::revolt_okapi::openapi3::OpenApi;
+use rocket::Route;
+
+mod search_messages;
+
+pub fn routes() -> (Vec<Route>, OpenApi) {
+    openapi_get_routes_spec![
+        search_messages::search_messages
+    ]
+}

crates/delta/src/routes/search/search_messages.rs

@@ -0,0 +1,88 @@
+use revolt_database::{util::permissions::DatabasePermissionQuery, Database, Message, User};
+use revolt_models::v0;
+use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
+use revolt_result::{create_error, Result, ToRevoltError};
+use revolt_search::{ElasticsearchClient, SearchTerms};
+use rocket::{serde::json::Json, State};
+
+#[openapi(tag = "Search")]
+#[post("/", data = "<data>")]
+pub async fn search_messages(
+    db: &State<Database>,
+    es: &State<ElasticsearchClient>,
+    user: User,
+    data: Json<v0::DataChannelMessagesSearch>,
+) -> Result<Json<v0::BulkMessageResponse>> {
+    if user.bot.is_some() {
+        return Err(create_error!(IsBot));
+    }
+
+    let data = data.into_inner();
+
+    if (data.server.is_some() && data.channel.is_some())
+        || (data.server.is_none() && data.channel.is_none())
+    {
+        return Err(create_error!(InvalidOperation));
+    }
+
+    let (server_id, channels) = if let Some(server_id) = data.server {
+        let server = db.fetch_server(&server_id).await?;
+        let server_channels = db.fetch_channels(&server.channels).await?;
+        let mut channels = Vec::new();
+
+        let query = DatabasePermissionQuery::new(db, &user).server(&server);
+
+        for channel in server_channels {
+            let perms = calculate_channel_permissions(&mut query.clone().channel(&channel)).await;
+
+            if perms.has(ChannelPermission::ReadMessageHistory as u64) {
+                channels.push(channel.id().to_string())
+            }
+        }
+
+        (Some(server_id.clone()), channels)
+    } else {
+        let channel_id = data.channel.unwrap();
+        let channel = db.fetch_channel(&channel_id).await?;
+        let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
+        let perms = calculate_channel_permissions(&mut query).await;
+
+        if !perms.has(ChannelPermission::ReadMessageHistory as u64) {
+            return Err(create_error!(InvalidOperation));
+        }
+
+        (channel.server().map(ToString::to_string), vec![channel_id])
+    };
+
+    let terms = SearchTerms {
+        channels,
+        filters: data.filters.map(Into::into).unwrap_or_default(),
+        offset: data.offset,
+        limit: data.limit.map(|limit| limit.min(100)),
+        sort: data.sort.map(Into::into),
+    };
+
+    let message_ids = es.search(terms).await.to_internal_error()?;
+
+    let mut messages = db.fetch_messages_by_id(&message_ids).await?;
+
+    // Fetching the messages looses the order so resort
+    let sort_order = data.sort.unwrap_or_default();
+    messages.sort_by(|a, b| match sort_order {
+        v0::SortOrder::Asc => a.id.cmp(&b.id),
+        v0::SortOrder::Desc => b.id.cmp(&a.id),
+    });
+
+    let (users, members) = Message::fetch_users(db, &messages, server_id.as_deref()).await?;
+
+    let response = v0::BulkMessageResponse::MessagesAndUsers {
+        messages: messages
+            .into_iter()
+            .map(|msg| msg.into_model(None, None))
+            .collect(),
+        users: User::into_mutuals(&user, users).await,
+        members: Some(members.into_iter().map(Into::into).collect()),
+    };
+
+    Ok(Json(response))
+}

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

@@ -4,7 +4,7 @@ use revolt_database::{
         delete_voice_channel, get_user_voice_channel_in_server, remove_user_from_voice_channel,
         UserVoiceChannel, VoiceClient,
     },
-    Database, RemovalIntention, User,
+    AMQP, Database, RemovalIntention, User,
 };
 use revolt_models::v0;
 use revolt_result::Result;
@@ -19,6 +19,7 @@ use rocket_empty::EmptyResponse;
 #[delete("/<target>?<options..>")]
 pub async fn delete(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     voice_client: &State<VoiceClient>,
     user: User,
     target: Reference<'_>,
@@ -39,7 +40,7 @@ pub async fn delete(
             .await?;
         }
 
-        server.delete(db).await
+        server.delete(db, Some(amqp)).await
     } else {
         if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? {
             remove_user_from_voice_channel(

crates/delta/src/routes/session/login.rs

@@ -3,7 +3,7 @@
 use std::ops::Add;
 use std::time::Duration;
 
-use async_std::task::sleep;
+use tokio::time::sleep;
 use iso8601_timestamp::Timestamp;
 use revolt_database::{
     util::{email::normalise_email, password::assert_safe},

crates/delta/src/routes/users/add_friend.rs

@@ -22,6 +22,6 @@ pub async fn add(
         return Err(create_error!(IsBot));
     }
 
-    user.add_friend(db, amqp, &mut target).await?;
+    user.add_friend(db, Some(amqp), &mut target).await?;
     Ok(Json(target.into(db, &user).await))
 }

crates/delta/src/routes/users/send_friend_request.rs

@@ -23,7 +23,7 @@ pub async fn send_friend_request(
             return Err(create_error!(IsBot));
         }
 
-        user.add_friend(db, amqp, &mut target).await?;
+        user.add_friend(db, Some(amqp), &mut target).await?;
         Ok(Json(target.into(db, &user).await))
     } else {
         Err(create_error!(InvalidProperty))

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

@@ -1,4 +1,4 @@
-use revolt_database::{util::reference::Reference, Database};
+use revolt_database::{util::reference::Reference, Database, AMQP};
 use revolt_result::{create_error, Result};
 use rocket::State;
 use rocket_empty::EmptyResponse;
@@ -10,6 +10,7 @@ use rocket_empty::EmptyResponse;
 #[delete("/<webhook_id>/<token>/<message_id>")]
 pub async fn webhook_delete_message(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     webhook_id: Reference<'_>,
     token: String,
     message_id: Reference<'_>,
@@ -23,5 +24,5 @@ pub async fn webhook_delete_message(
         return Err(create_error!(CannotDeleteMessage));
     }
 
-    message.delete(db).await.map(|_| EmptyResponse)
+    message.delete(db, Some(amqp)).await.map(|_| EmptyResponse)
 }

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

@@ -2,6 +2,7 @@ use iso8601_timestamp::Timestamp;
 use revolt_config::config;
 use revolt_database::{
     tasks::process_embeds::queue, util::reference::Reference, Database, Message, PartialMessage,
+    AMQP,
 };
 use revolt_models::v0::{self, DataEditMessage, Embed};
 use revolt_models::validator::Validate;
@@ -15,6 +16,7 @@ use rocket::{serde::json::Json, State};
 #[patch("/<webhook_id>/<token>/<message_id>", data = "<data>")]
 pub async fn webhook_edit_message(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     webhook_id: Reference<'_>,
     token: String,
     message_id: Reference<'_>,
@@ -73,7 +75,7 @@ pub async fn webhook_edit_message(
 
     partial.embeds = Some(new_embeds);
 
-    message.update(db, partial, vec![]).await?;
+    message.update(db, Some(amqp), partial, vec![]).await?;
 
     // Queue up a task for processing embeds
     if let Some(content) = edit.content {

crates/delta/src/util/test.rs

@@ -181,8 +181,8 @@ impl TestHarness {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&self.db, Some(user)).await),
-            Some(user.clone().into(&self.db, Some(user)).await),
-            Some(member.clone().into()),
+            Some(user.clone()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,

crates/services/autumn/Dockerfile

@@ -5,8 +5,8 @@ FROM debian:12 AS debian
 # Bundle Stage
 FROM gcr.io/distroless/cc-debian12:nonroot
 COPY --from=builder /home/rust/src/target/release/revolt-autumn ./
-COPY --from=mwader/static-ffmpeg:7.0.2 /ffmpeg /usr/local/bin/
-COPY --from=mwader/static-ffmpeg:7.0.2 /ffprobe /usr/local/bin/
+COPY --from=mwader/static-ffmpeg:7.1.1 /ffmpeg /usr/local/bin/
+COPY --from=mwader/static-ffmpeg:7.1.1 /ffprobe /usr/local/bin/
 COPY --from=debian /usr/bin/uname /usr/bin/uname
 
 EXPOSE 14704

crates/services/january/Dockerfile

@@ -5,8 +5,8 @@ FROM debian:12 AS debian
 # Bundle Stage
 FROM gcr.io/distroless/cc-debian12:nonroot
 COPY --from=builder /home/rust/src/target/release/revolt-january ./
-COPY --from=mwader/static-ffmpeg:7.0.2 /ffmpeg /usr/local/bin/
-COPY --from=mwader/static-ffmpeg:7.0.2 /ffprobe /usr/local/bin/
+COPY --from=mwader/static-ffmpeg:7.1.1 /ffmpeg /usr/local/bin/
+COPY --from=mwader/static-ffmpeg:7.1.1 /ffprobe /usr/local/bin/
 COPY --from=debian /usr/bin/uname /usr/bin/uname
 
 EXPOSE 14705