feat: initial work on elasticsearch integration

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

Cargo.lock

@@ -146,7 +146,7 @@ dependencies = [
  "bytes 1.10.1",
  "serde",
  "serde_bytes_ng",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -244,6 +244,19 @@ dependencies = [
  "pin-project-lite 0.2.16",
 ]
 
+[[package]]
+name = "async-compression"
+version = "0.4.33"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93c1f86859c1af3d514fa19e8323147ff10ea98684e6c7b307912509f50e67b2"
+dependencies = [
+ "compression-codecs",
+ "compression-core",
+ "futures-core",
+ "pin-project-lite 0.2.16",
+ "tokio 1.50.0",
+]
+
 [[package]]
 name = "async-executor"
 version = "1.13.3"
@@ -272,7 +285,7 @@ dependencies = [
  "futures-lite 2.6.1",
  "once_cell",
  "tokio 0.2.25",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -596,7 +609,7 @@ dependencies = [
  "http 1.3.1",
  "ring",
  "time",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
  "url",
  "zeroize",
@@ -799,7 +812,7 @@ checksum = "1e190749ea56f8c42bf15dd76c65e14f8f765233e6df9b0506d9d934ebef867c"
 dependencies = [
  "futures-util",
  "pin-project-lite 0.2.16",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -878,7 +891,7 @@ dependencies = [
  "rustls 0.23.32",
  "rustls-native-certs 0.8.1",
  "rustls-pki-types",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-rustls 0.26.3",
  "tower",
  "tracing",
@@ -932,7 +945,7 @@ dependencies = [
  "http-body 1.0.1",
  "pin-project-lite 0.2.16",
  "pin-utils",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
 ]
 
@@ -948,7 +961,7 @@ dependencies = [
  "http 0.2.12",
  "http 1.3.1",
  "pin-project-lite 0.2.16",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
  "zeroize",
 ]
@@ -975,7 +988,7 @@ dependencies = [
  "ryu",
  "serde",
  "time",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-util",
 ]
 
@@ -1031,7 +1044,7 @@ dependencies = [
  "serde_path_to_error",
  "serde_urlencoded",
  "sync_wrapper 1.0.2",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower",
  "tower-layer",
  "tower-service",
@@ -1109,7 +1122,7 @@ dependencies = [
  "futures-util",
  "tempfile",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "uuid",
 ]
 
@@ -1418,7 +1431,7 @@ dependencies = [
  "instant",
  "once_cell",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -1564,10 +1577,27 @@ dependencies = [
  "futures-core",
  "memchr",
  "pin-project-lite 0.2.16",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-util",
 ]
 
+[[package]]
+name = "compression-codecs"
+version = "0.4.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "680dc087785c5230f8e8843e2e57ac7c1c90488b6a91b88caa265410568f441b"
+dependencies = [
+ "compression-core",
+ "flate2",
+ "memchr",
+]
+
+[[package]]
+name = "compression-core"
+version = "0.4.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75984efb6ed102a0d42db99afb6c1948f0380d1d91808d5529916e6c08b49d8d"
+
 [[package]]
 name = "concurrent-queue"
 version = "2.5.0"
@@ -2081,7 +2111,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72ff1266be84e4e04a81e2d1cbb998cb271b374fb73ce780245ef96c037c50cd"
 dependencies = [
  "crossbeam-queue",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -2344,6 +2374,41 @@ version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
 
+[[package]]
+name = "elasticsearch"
+version = "9.1.0-alpha.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12bb303aa6e1d28c0c86b6fbfe484fd0fd3f512629aeed1ac4f6b85f81d9834a"
+dependencies = [
+ "base64 0.22.1",
+ "bytes 1.10.1",
+ "dyn-clone",
+ "flate2",
+ "lazy_static",
+ "parking_lot",
+ "percent-encoding",
+ "reqwest 0.12.23",
+ "rustc_version",
+ "serde",
+ "serde_json",
+ "serde_with",
+ "tokio 1.50.0",
+ "url",
+ "void",
+]
+
+[[package]]
+name = "elasticsearch-dsl"
+version = "0.4.26"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c9445aba099fdf52dea22534cf953b918d1e4cbb799c99507c9cc96188d9c230"
+dependencies = [
+ "chrono",
+ "num-traits",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "elliptic-curve"
 version = "0.12.3"
@@ -2779,7 +2844,7 @@ dependencies = [
  "redis-protocol",
  "semver",
  "socket2 0.5.10",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-stream",
  "tokio-util",
  "url",
@@ -2816,9 +2881,9 @@ dependencies = [
 
 [[package]]
 name = "futures"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -2831,9 +2896,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
+checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -2841,15 +2906,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
+checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -2858,9 +2923,9 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
+checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
 
 [[package]]
 name = "futures-lite"
@@ -2898,14 +2963,14 @@ checksum = "45ec6fe3675af967e67c5536c0b9d44e34e6c52f86bedc4ea49c5317b8e94d06"
 dependencies = [
  "futures-channel",
  "futures-task",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
 name = "futures-macro"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
 dependencies = [
  "proc-macro2",
  "quote 1.0.40",
@@ -2914,15 +2979,15 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
 
 [[package]]
 name = "futures-task"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
+checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
 
 [[package]]
 name = "futures-timer"
@@ -2932,9 +2997,9 @@ checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24"
 
 [[package]]
 name = "futures-util"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
+checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -2944,7 +3009,6 @@ dependencies = [
  "futures-task",
  "memchr",
  "pin-project-lite 0.2.16",
- "pin-utils",
  "slab",
 ]
 
@@ -3155,7 +3219,7 @@ dependencies = [
  "http 0.2.12",
  "indexmap 2.13.0",
  "slab",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-util",
  "tracing",
 ]
@@ -3174,7 +3238,7 @@ dependencies = [
  "http 1.3.1",
  "indexmap 2.13.0",
  "slab",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-util",
  "tracing",
 ]
@@ -3326,7 +3390,7 @@ dependencies = [
  "rand 0.8.5",
  "thiserror 1.0.69",
  "tinyvec",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
  "url",
 ]
@@ -3348,7 +3412,7 @@ dependencies = [
  "resolv-conf",
  "smallvec",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
 ]
 
@@ -3514,7 +3578,7 @@ dependencies = [
  "itoa",
  "pin-project-lite 0.2.16",
  "socket2 0.5.10",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower-service",
  "tracing",
  "want",
@@ -3539,7 +3603,7 @@ dependencies = [
  "pin-project-lite 0.2.16",
  "pin-utils",
  "smallvec",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "want",
 ]
 
@@ -3555,7 +3619,7 @@ dependencies = [
  "log",
  "rustls 0.21.12",
  "rustls-native-certs 0.6.3",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-rustls 0.24.1",
 ]
 
@@ -3571,7 +3635,7 @@ dependencies = [
  "hyper-util",
  "rustls 0.22.4",
  "rustls-pki-types",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-rustls 0.25.0",
  "tower-service",
  "webpki-roots 0.26.11",
@@ -3589,7 +3653,7 @@ dependencies = [
  "rustls 0.23.32",
  "rustls-native-certs 0.8.1",
  "rustls-pki-types",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-rustls 0.26.3",
  "tower-service",
 ]
@@ -3603,7 +3667,7 @@ dependencies = [
  "bytes 1.10.1",
  "hyper 0.14.32",
  "native-tls",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-native-tls",
 ]
 
@@ -3618,7 +3682,7 @@ dependencies = [
  "hyper 1.7.0",
  "hyper-util",
  "native-tls",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-native-tls",
  "tower-service",
 ]
@@ -3643,7 +3707,7 @@ dependencies = [
  "pin-project-lite 0.2.16",
  "socket2 0.6.0",
  "system-configuration 0.6.1",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower-service",
  "tracing",
  "windows-registry",
@@ -4543,7 +4607,7 @@ dependencies = [
  "once_cell",
  "quoted_printable",
  "socket2 0.4.10",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -4697,7 +4761,7 @@ dependencies = [
  "prost-types",
  "serde",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -4706,7 +4770,7 @@ version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6ae53eb874eb86e96e8ccffc31b5a00926d46174cbcb1c24ce4e57b1fcc5c8f6"
 dependencies = [
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-stream",
 ]
 
@@ -4716,7 +4780,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "532e84c6cdc5fe774f2b5d9912597b5f3bea561927a48296d03e24549d21c3f6"
 dependencies = [
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-stream",
 ]
 
@@ -4732,9 +4796,9 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.28"
+version = "0.4.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34080505efa8e45a4b816c349525ebe327ceaa8559756f0356cba97ef3bf7432"
+checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
 dependencies = [
  "value-bag",
 ]
@@ -5049,7 +5113,7 @@ dependencies = [
  "log",
  "metrics",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
  "tracing-subscriber",
 ]
@@ -5146,7 +5210,7 @@ dependencies = [
  "strsim 0.11.1",
  "take_mut",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-rustls 0.26.3",
  "tokio-util",
  "typed-builder",
@@ -5190,7 +5254,7 @@ dependencies = [
  "memchr",
  "mime",
  "spin",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-util",
  "version_check",
 ]
@@ -6594,7 +6658,7 @@ dependencies = [
  "ryu",
  "sha1_smol",
  "socket2 0.4.10",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-util",
  "url",
 ]
@@ -6723,7 +6787,7 @@ dependencies = [
  "serde_urlencoded",
  "sync_wrapper 0.1.2",
  "system-configuration 0.5.1",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-native-tls",
  "tower-service",
  "url",
@@ -6739,10 +6803,12 @@ version = "0.12.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d429f34c8092b2d42c7c93cec323bb4adeb7c67698f70839adec842ec10c7ceb"
 dependencies = [
+ "async-compression",
  "base64 0.22.1",
  "bytes 1.10.1",
  "encoding_rs",
  "futures-core",
+ "futures-util",
  "h2 0.4.12",
  "http 1.3.1",
  "http-body 1.0.1",
@@ -6762,8 +6828,9 @@ dependencies = [
  "serde_json",
  "serde_urlencoded",
  "sync_wrapper 1.0.2",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-native-tls",
+ "tokio-util",
  "tower",
  "tower-http 0.6.6",
  "tower-service",
@@ -6825,7 +6892,7 @@ dependencies = [
  "strum_macros",
  "tempfile",
  "thumbhash",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower-http 0.5.2",
  "tracing",
  "tracing-subscriber",
@@ -6872,7 +6939,7 @@ version = "0.11.5"
 dependencies = [
  "indexmap 2.13.0",
  "lru 0.16.3",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -6901,7 +6968,7 @@ dependencies = [
  "revolt-database",
  "revolt-files",
  "revolt-result",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -6989,6 +7056,7 @@ dependencies = [
  "revolt-presence",
  "revolt-ratelimits",
  "revolt-result",
+ "revolt-search",
  "revolt_rocket_okapi",
  "rocket",
  "rocket_authifier",
@@ -7024,7 +7092,7 @@ dependencies = [
  "tempfile",
  "thiserror 2.0.18",
  "tiny-skia",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
  "typenum",
  "usvg",
@@ -7048,7 +7116,7 @@ dependencies = [
  "revolt-result",
  "serde",
  "serde_json",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower-http 0.5.2",
  "tracing",
  "utoipa",
@@ -7076,7 +7144,7 @@ dependencies = [
  "serde",
  "serde_json",
  "tempfile",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tracing",
  "tracing-subscriber",
  "utoipa",
@@ -7162,7 +7230,7 @@ dependencies = [
  "revolt_optional_struct",
  "serde",
  "serde_json",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "ulid 1.2.1",
  "web-push",
 ]
@@ -7200,6 +7268,36 @@ dependencies = [
  "utoipa",
 ]
 
+[[package]]
+name = "revolt-search"
+version = "0.11.5"
+dependencies = [
+ "elasticsearch",
+ "elasticsearch-dsl",
+ "futures",
+ "iso8601-timestamp",
+ "revolt-config",
+ "revolt-database",
+ "revolt-models",
+ "serde",
+ "serde_json",
+ "ulid 1.2.1",
+]
+
+[[package]]
+name = "revolt-searchd"
+version = "0.11.5"
+dependencies = [
+ "amqprs",
+ "async-trait",
+ "log",
+ "revolt-config",
+ "revolt-database",
+ "revolt-search",
+ "serde_json",
+ "tokio 1.50.0",
+]
+
 [[package]]
 name = "revolt-voice-ingress"
 version = "0.11.5"
@@ -7249,7 +7347,7 @@ dependencies = [
  "serde",
  "serde_json",
  "thiserror 1.0.69",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -7405,7 +7503,7 @@ dependencies = [
  "state",
  "tempfile",
  "time",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tokio-stream",
  "tokio-util",
  "ubyte",
@@ -7496,7 +7594,7 @@ dependencies = [
  "stable-pattern",
  "state",
  "time",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "uncased",
 ]
 
@@ -7987,7 +8085,7 @@ dependencies = [
  "sentry-debug-images",
  "sentry-panic",
  "sentry-tracing",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "ureq",
 ]
 
@@ -8962,9 +9060,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.49.0"
+version = "1.50.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72a2903cd7736441aac9df9d7688bd0ce48edccaadf181c3b90be801e81d3d86"
+checksum = "27ad5e34374e03cfffefc301becb44e9dc3c17584f414349ebe29ed26661822d"
 dependencies = [
  "bytes 1.10.1",
  "libc",
@@ -8995,7 +9093,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2"
 dependencies = [
  "native-tls",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -9005,7 +9103,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081"
 dependencies = [
  "rustls 0.21.12",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -9016,7 +9114,7 @@ checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f"
 dependencies = [
  "rustls 0.22.4",
  "rustls-pki-types",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -9026,7 +9124,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "05f63835928ca123f1bef57abbcd23bb2ba0ac9ae1235f1e65bda0d06e7786bd"
 dependencies = [
  "rustls 0.23.32",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -9037,7 +9135,7 @@ checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047"
 dependencies = [
  "futures-core",
  "pin-project-lite 0.2.16",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -9051,7 +9149,7 @@ dependencies = [
  "futures-io",
  "futures-sink",
  "pin-project-lite 0.2.16",
- "tokio 1.49.0",
+ "tokio 1.50.0",
 ]
 
 [[package]]
@@ -9137,7 +9235,7 @@ dependencies = [
  "futures-util",
  "pin-project-lite 0.2.16",
  "sync_wrapper 1.0.2",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -9699,9 +9797,9 @@ checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
 
 [[package]]
 name = "value-bag"
-version = "1.11.1"
+version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "943ce29a8a743eb10d6082545d861b24f9d1b160b7d741e0f2cdf726bec909c5"
+checksum = "7ba6f5989077681266825251a52748b8c1d8a4ad098cc37e440103d0ea717fc0"
 
 [[package]]
 name = "vcpkg"
@@ -9733,6 +9831,12 @@ version = "0.9.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
 
+[[package]]
+name = "void"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d"
+
 [[package]]
 name = "vsimd"
 version = "0.8.0"
@@ -10493,7 +10597,7 @@ dependencies = [
  "serde",
  "serde_json",
  "time",
- "tokio 1.49.0",
+ "tokio 1.50.0",
  "tower-service",
  "url",
 ]

Revolt.toml

@@ -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

@@ -94,3 +94,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/core/config/Revolt.toml

@@ -323,3 +323,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

@@ -413,6 +413,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,
@@ -425,6 +438,7 @@ pub struct Settings {
     pub sentry: Sentry,
     pub production: bool,
     pub disable_events_dont_use: bool,
+    pub elasticsearch: Elasticsearch,
 }
 
 impl Settings {

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

@@ -1,7 +1,7 @@
 use std::collections::HashSet;
 
 use crate::events::rabbit::*;
-use crate::User;
+use crate::{Message, User};
 use amqprs::channel::{BasicPublishArguments, ExchangeDeclareArguments};
 use amqprs::connection::OpenConnectionArguments;
 use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError};
@@ -52,6 +52,15 @@ impl AMQP {
             .await
             .expect("Failed to declare exchange");
 
+        channel
+            .exchange_declare(
+                ExchangeDeclareArguments::new(&config.elasticsearch.exchange, "direct")
+                    .durable(true)
+                    .finish(),
+            )
+            .await
+            .expect("Failed to declare exchange");
+
         AMQP::new(connection, channel)
     }
 
@@ -316,4 +325,104 @@ impl AMQP {
             )
             .await
     }
+
+    pub async fn new_message_search(&self, message: Message) -> Result<(), AMQPError> {
+        let config = revolt_config::config().await;
+
+        let payload = to_string(&message).unwrap();
+
+        debug!(
+            "Sending new message search payload on channel {}: {}",
+            config.elasticsearch.message_queue, payload
+        );
+
+        self.channel
+            .basic_publish(
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.elasticsearch.exchange,
+                    &config.elasticsearch.message_queue,
+                ),
+            )
+            .await
+    }
+
+    pub async fn edit_message_search(&self, message: Message) -> Result<(), AMQPError> {
+        let config = revolt_config::config().await;
+
+        let payload = to_string(&message).unwrap();
+
+        debug!(
+            "Sending edit message search payload on channel {}: {}",
+            config.elasticsearch.message_edit_queue, payload
+        );
+
+        self.channel
+            .basic_publish(
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.elasticsearch.exchange,
+                    &config.elasticsearch.message_edit_queue,
+                ),
+            )
+            .await
+    }
+
+    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.channel
+            .basic_publish(
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.elasticsearch.exchange,
+                    &config.elasticsearch.message_delete_queue,
+                ),
+            )
+            .await
+    }
+
+    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.channel
+            .basic_publish(
+                BasicProperties::default()
+                    .with_content_type("application/json")
+                    .with_persistence(true)
+                    .finish(),
+                payload.into(),
+                BasicPublishArguments::new(
+                    &config.elasticsearch.exchange,
+                    &config.elasticsearch.channel_delete_queue,
+                ),
+            )
+            .await
+    }
 }

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

@@ -78,3 +78,13 @@ pub struct AckPayload {
     pub channel_id: String,
     pub message_id: String,
 }
+
+#[derive(Serialize, Deserialize)]
+pub struct MessageDeletePayload {
+    pub message_id: String
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct ChannelDeletePayload {
+    pub channel_id: String
+}

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

@@ -1,7 +1,7 @@
 #![allow(deprecated)]
 use std::{borrow::Cow, collections::HashMap};
 
-use revolt_config::config;
+use revolt_config::{capture_error, config};
 use revolt_models::v0::{self, MessageAuthor};
 use revolt_permissions::OverrideField;
 use revolt_result::Result;
@@ -339,7 +339,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<()> {
@@ -376,7 +376,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()),
@@ -669,7 +669,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,
@@ -701,7 +701,7 @@ impl Channel {
                         .into_message(id.to_string())
                         .send(
                             db,
-                            Some(amqp),
+                            amqp,
                             MessageAuthor::System {
                                 username: name,
                                 avatar: None,
@@ -714,7 +714,7 @@ impl Channel {
                         .await
                         .ok();
                     } else {
-                        return self.delete(db).await;
+                        return self.delete(db, amqp).await;
                     }
                 }
 
@@ -741,7 +741,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()),
@@ -763,13 +763,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}");
+                capture_error(&e);
+            }
+        }
+
+        Ok(())
     }
 }
 

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::{FeaturesLimits, capture_error, config};
 use revolt_models::v0::{
     self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort,
     MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text,
@@ -671,7 +671,7 @@ 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>,
@@ -726,6 +726,13 @@ impl Message {
             .await;
         }
 
+        if let Some(amqp) = amqp {
+            if let Err(e) = amqp.new_message_search(self.clone()).await {
+                log::error!("Error pushing message to RabbitMQ: {e}");
+                capture_error(&e);
+            }
+        }
+
         Ok(())
     }
 
@@ -994,9 +1001,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();
 
@@ -1006,12 +1014,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(())
     }
 

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

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 use std::time::SystemTime;
 use revolt_result::Result;
 
-use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
+use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage, util::ChunkedDatabaseGenerator};
 
 #[cfg(feature = "mongodb")]
 mod mongodb;
@@ -50,4 +50,5 @@ pub trait AbstractMessages: Sync + Send {
         author: &str,
         since: SystemTime
     ) -> Result<HashMap<String, Vec<String>>>;
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<Message>>;
 }

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};
 use revolt_models::v0::MessageSort;
 use revolt_result::Result;
 use std::collections::{HashMap, HashSet};
@@ -10,7 +10,7 @@ use ulid::Ulid;
 
 use crate::{
     AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath, Message, MessageQuery,
-    MessageTimePeriod, MongoDb, PartialMessage,
+    MessageTimePeriod, MongoDb, PartialMessage, util::ChunkedDatabaseGenerator,
 };
 
 use super::AbstractMessages;
@@ -416,6 +416,31 @@ impl AbstractMessages for MongoDb {
 
         Ok(deleted_messages)
     }
+    
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<Message>> {
+        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)
+            .find(doc! {})
+            .sort(doc ! { "_id": -1i32 })
+            .session(&mut session)
+            .batch_size(1000)
+            .await
+            .inspect_err(|e| log::error!("{e}"))
+            .map_err(|_| create_database_error!("find", COL))?;
+
+        Ok(ChunkedDatabaseGenerator::new_mongo(session, cursor))
+    }
 }
 
 impl IntoDocumentPath for FieldsMessage {

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

@@ -4,7 +4,11 @@ use indexmap::IndexSet;
 use revolt_result::Result;
 use std::time::SystemTime;
 use ulid::Ulid;
-use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage, ReferenceDb};
+
+use crate::{
+    util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
+    PartialMessage, ReferenceDb,
+};
 
 use super::AbstractMessages;
 
@@ -347,4 +351,10 @@ impl AbstractMessages for ReferenceDb {
 
         Ok(deleted_messages)
     }
+    
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<Message>> {
+        Ok(ChunkedDatabaseGenerator::new_reference(
+            self.messages.lock().await.values().cloned().collect(),
+        ))
+    }
 }

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
@@ -210,7 +211,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(),
         }

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

@@ -510,7 +510,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) {
@@ -520,8 +520,10 @@ impl User {
             RelationshipStatus::Blocked => Err(create_error!(Blocked)),
             RelationshipStatus::BlockedOther => Err(create_error!(BlockedByOther)),
             RelationshipStatus::Incoming => {
-                // Accept incoming friend request
+                if let Some(amqp) = amqp {
-                _ = amqp.friend_request_accepted(self, target).await;
+                    // Accept incoming friend request
+                    _ = amqp.friend_request_accepted(self, target).await;
+                };
 
                 self.apply_relationship(
                     db,
@@ -551,7 +553,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(

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

@@ -0,0 +1,54 @@
+#[cfg(feature = "mongodb")]
+use ::mongodb::{ClientSession, SessionCursor};
+use serde::Deserialize;
+
+#[derive(Debug)]
+#[allow(clippy::large_enum_variant)]
+pub enum ChunkedDatabaseGenerator<T> {
+    #[cfg(feature = "mongodb")]
+    MongoDb {
+        session: ClientSession,
+        cursor: SessionCursor<T>,
+    },
+
+    Reference {
+        offset: i32,
+        data: Vec<T>,
+    },
+}
+
+impl<T: for<'d> Deserialize<'d> + Clone> ChunkedDatabaseGenerator<T> {
+    #[cfg(feature = "mongodb")]
+    pub fn new_mongo(session: ClientSession, cursor: SessionCursor<T>) -> Self {
+        Self::MongoDb {
+            session,
+            cursor,
+        }
+    }
+
+    pub fn new_reference(data: Vec<T>) -> Self {
+        Self::Reference {
+            offset: 0,
+            data,
+        }
+    }
+
+    pub async fn next(&mut self) -> Option<T> {
+        match self {
+            #[cfg(feature = "mongodb")]
+            Self::MongoDb { session, cursor } => {
+                let value = cursor.next(session).await;
+                value.map(|val| val.expect("Failed to fetch the next message"))
+            }
+            Self::Reference { offset, data } => {
+                if data.len() as i32 >= *offset {
+                    None
+                } else {
+                    let resp = &data[*offset as usize];
+                    *offset += 1;
+                    Some(resp.clone())
+                }
+            }
+        }
+    }
+}

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

@@ -5,5 +5,7 @@ pub mod idempotency;
 pub mod permissions;
 pub mod reference;
 pub mod test_fixtures;
+pub mod chunked;
 
 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;
 
 pub use bots::*;
 pub use channel_invites::*;
@@ -31,3 +32,4 @@ pub use server_members::*;
 pub use servers::*;
 pub use user_settings::*;
 pub use users::*;
+pub use search::*;

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

@@ -0,0 +1,49 @@
+use iso8601_timestamp::Timestamp;
+
+auto_derived!(
+    pub struct DataChannelMessagesSearch {
+        pub channel: Option<String>,
+        pub server: Option<String>,
+        pub offset: Option<u64>,
+        pub limit: Option<u64>,
+        pub filters: Option<DataChannelMessagesSearchFilters>,
+        pub sort: Option<SortOrder>,
+        pub include_users: Option<bool>,
+    }
+
+    pub struct DataChannelMessagesSearchFilters {
+        pub content: Option<String>,
+        pub author: Option<Vec<String>>,
+        pub mentions: Option<Vec<String>>,
+        pub role_mentions: Option<Vec<String>>,
+        pub before_date: Option<Timestamp>,
+        pub after_date: Option<Timestamp>,
+        pub author_type: Option<Vec<AuthorType>>,
+        pub pinned: Option<bool>,
+        pub components: Option<Vec<MessageComponent>>,
+    }
+
+    #[derive(Copy)]
+    pub enum AuthorType {
+        User,
+        // Bot,
+        Webhook,
+    }
+
+    #[derive(Copy)]
+    pub enum MessageComponent {
+        Image,
+        Video,
+        // Link,
+        File,
+        Embed,
+    }
+
+    #[derive(Copy, Default)]
+    #[cfg_attr(feature = "serde", serde(rename_all = "lowercase"))]
+    pub enum SortOrder {
+        Asc,
+        #[default]
+        Desc,
+    }
+);

crates/core/search/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "revolt-search"
+version = "0.11.5"
+edition = "2024"
+license = "MIT"
+publish = false
+
+[dependencies]
+revolt-config = { version = "0.11.5", path = "../config" }
+revolt-database = { version = "0.11.5", path = "../database" }
+revolt-models = { version = "0.11.5", path = "../models" }
+elasticsearch = "9.1.0-alpha.1"
+serde_json = "1"
+ulid = "1.2.1"
+iso8601-timestamp = { version = "0.2.10", features = ["serde", "bson"] }
+futures = "0.3.32"
+elasticsearch-dsl = "0.4"
+serde = "1"

crates/core/search/src/client.rs

@@ -0,0 +1,315 @@
+use std::fmt::Display;
+
+use elasticsearch::{
+    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 revolt_database::Message;
+use serde_json::{Map, Value, json};
+
+pub use elasticsearch;
+
+use crate::SearchTerms;
+
+#[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:?}"),
+        }
+    }
+}
+
+#[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 }
+    }
+
+    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(())
+        }
+    }
+
+    pub async fn setup_indexes(&self) -> Result<(), Error> {
+        let exception = self
+            .inner
+            .indices()
+            .create(IndicesCreateParts::Index("messages"))
+            .body(json!({
+                // "settings": {
+                //     "index": {
+                //         "sort.field": "_id",
+                //         "sort.order": ["asc", "desc"],
+                //     },
+                // },
+                "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": {  // TODO: images, videos, files
+                            "type": "nested",
+                            "properties": {}
+                        },
+                        // TODO: links
+                    }
+                }
+            }))
+            .send()
+            .await?
+            .exception()
+            .await?;
+
+        if let Some(exception) = exception {
+            Err(exception.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    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")))
+            }
+        }
+
+        // TODO: component filtering
+        // Need to pass FileHash to this to extract metadata
+
+        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())
+        }
+    }
+
+    fn create_message_source(&self, message: Message) -> 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 {
+            map.insert("content".to_string(), Value::String(content));
+        }
+
+        if let Some(attachments) = message.attachments {
+            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));
+        }
+
+        // TODO: bot
+        map.insert(
+            "author_type".to_string(),
+            Value::String(
+                if message.webhook.is_some() {
+                    "webhook"
+                } else {
+                    "user"
+                }
+                .to_string(),
+            ),
+        );
+
+        Value::Object(map)
+    }
+
+    pub async fn index_message(&self, message: Message) -> Result<(), Error> {
+        let id = message.id.clone();
+        let source = self.create_message_source(message);
+
+        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(())
+        }
+    }
+
+    pub async fn edit_message(&self, message: Message) -> Result<(), Error> {
+        let id = message.id.clone();
+        let source = self.create_message_source(message);
+
+        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(())
+        }
+    }
+
+    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(())
+        }
+    }
+
+    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,108 @@
+use iso8601_timestamp::Timestamp;
+use revolt_models::v0;
+use serde::Serialize;
+
+#[derive(Copy, Clone, PartialEq, Eq, Serialize)]
+pub enum AuthorType {
+    User,
+    // Bot,
+    Webhook,
+}
+
+#[derive(Copy, Clone, PartialEq, Eq)]
+pub enum MessageComponent {
+    Image,
+    Video,
+    // Link,
+    File,
+    Embed,
+}
+
+#[derive(Clone, Default, PartialEq)]
+pub struct SearchFilters {
+    pub content: Option<String>,
+    pub author: Option<Vec<String>>,
+    pub mentions: Option<Vec<String>>,
+    pub role_mentions: Option<Vec<String>>,
+    pub before_date: Option<Timestamp>,
+    pub after_date: Option<Timestamp>,
+    pub author_type: Option<Vec<AuthorType>>,
+    pub pinned: Option<bool>,
+    pub components: Option<Vec<MessageComponent>>,
+}
+
+#[derive(Clone, Copy, Default, PartialEq, Eq)] 
+pub enum SortOrder {
+    Asc,
+    #[default]
+    Desc,
+}
+
+#[derive(Clone, PartialEq)]
+pub struct SearchTerms {
+    pub channels: Vec<String>,
+    pub filters: SearchFilters,
+    pub offset: Option<u64>,
+    pub limit: Option<u64>,
+    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/searchd/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "revolt-searchd"
+version = "0.11.5"
+edition = "2024"
+license = "AGPL-3.0-or-later"
+publish = false
+
+[dependencies]
+tokio = { version = "1.50.0", features = ["macros"] }
+revolt-database = { version = "0.11.5", path = "../../core/database" }
+revolt-config = { version = "0.11.5", path = "../../core/config" }
+revolt-search = { version = "0.11.5", path = "../../core/search" }
+amqprs = "1.7.0"
+async-trait = "0.1.89"
+log = "0.4.29"
+serde_json = "1"

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

@@ -0,0 +1,25 @@
+use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use async_trait::async_trait;
+use revolt_database::events::rabbit::{ChannelDeletePayload, MessageDeletePayload};
+use revolt_search::ElasticsearchClient;
+
+pub struct ChannelDeleteConsumer(pub ElasticsearchClient);
+
+#[async_trait]
+impl AsyncConsumer for ChannelDeleteConsumer {
+    async fn consume(
+        &mut self,
+        channel: &Channel,
+        deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) {
+        let payload = serde_json::from_slice::<ChannelDeletePayload>(&content).expect("Failed to decode message");
+
+        if self.0.delete_channel(&payload.channel_id).await.is_ok() {
+            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+        } else {
+            // todo requeue
+        }
+    }
+}

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

@@ -0,0 +1,26 @@
+use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use async_trait::async_trait;
+use revolt_database::Message;
+use revolt_search::ElasticsearchClient;
+
+pub struct MessageConsumer(pub ElasticsearchClient);
+
+#[async_trait]
+impl AsyncConsumer for MessageConsumer {
+    async fn consume(
+        &mut self,
+        channel: &Channel,
+        deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) {
+        let message = serde_json::from_slice::<Message>(&content).expect("Failed to decode message");
+        log::debug!("Received message {message:?}");
+
+        if self.0.index_message(message).await.is_ok() {
+            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+        } else {
+            // todo requeue
+        }
+    }
+}

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

@@ -0,0 +1,25 @@
+use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use async_trait::async_trait;
+use revolt_database::events::rabbit::MessageDeletePayload;
+use revolt_search::ElasticsearchClient;
+
+pub struct MessageDeleteConsumer(pub ElasticsearchClient);
+
+#[async_trait]
+impl AsyncConsumer for MessageDeleteConsumer {
+    async fn consume(
+        &mut self,
+        channel: &Channel,
+        deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) {
+        let payload = serde_json::from_slice::<MessageDeletePayload>(&content).expect("Failed to decode message");
+
+        if self.0.delete_message(&payload.message_id).await.is_ok() {
+            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+        } else {
+            // todo requeue
+        }
+    }
+}

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

@@ -0,0 +1,25 @@
+use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use async_trait::async_trait;
+use revolt_database::Message;
+use revolt_search::ElasticsearchClient;
+
+pub struct MessageEditConsumer(pub ElasticsearchClient);
+
+#[async_trait]
+impl AsyncConsumer for MessageEditConsumer {
+    async fn consume(
+        &mut self,
+        channel: &Channel,
+        deliver: Deliver,
+        _basic_properties: BasicProperties,
+        content: Vec<u8>,
+    ) {
+        let message = serde_json::from_slice::<Message>(&content).expect("Failed to decode message");
+
+        if self.0.edit_message(message).await.is_ok() {
+            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+        } else {
+            // todo requeue
+        }
+    }
+}

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

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

crates/daemons/searchd/src/index.rs

@@ -0,0 +1,14 @@
+use revolt_config::capture_error;
+use revolt_database::{AMQP, Database};
+
+
+pub async fn index_existing_messages(db: Database, amqp: AMQP) {
+    let mut generator = db.fetch_all_messages().await.expect("Database query failed");
+
+    while let Some(message) = generator.next().await {
+        if let Err(e) = amqp.new_message_search(message).await {
+            log::error!("Error pushing message to RabbitMQ: {e}");
+            capture_error(&e);
+        }
+    }
+}

crates/daemons/searchd/src/main.rs

@@ -0,0 +1,151 @@
+use amqprs::{
+    channel::{
+        BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments,
+        QueueDeclareArguments,
+    },
+    connection::{Connection, OpenConnectionArguments},
+    consumer::AsyncConsumer,
+};
+use revolt_config::{Settings, config, configure};
+use revolt_database::{AMQP, DatabaseInfo};
+use revolt_search::ElasticsearchClient;
+use tokio::{signal::ctrl_c, spawn};
+
+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 = Connection::open(&OpenConnectionArguments::new(
+        &config.rabbit.host,
+        config.rabbit.port,
+        &config.rabbit.username,
+        &config.rabbit.password,
+    ))
+    .await
+    .expect("Failed to connect to RabbitMQ");
+
+    let mut channels = Vec::new();
+
+    channels.push(
+        make_queue_and_consume(
+            &config,
+            &connection,
+            &config.elasticsearch.message_queue,
+            consumers::MessageConsumer(client.clone()),
+        )
+        .await,
+    );
+
+    channels.push(
+        make_queue_and_consume(
+            &config,
+            &connection,
+            &config.elasticsearch.message_edit_queue,
+            consumers::MessageEditConsumer(client.clone()),
+        )
+        .await,
+    );
+
+    channels.push(
+        make_queue_and_consume(
+            &config,
+            &connection,
+            &config.elasticsearch.message_delete_queue,
+            consumers::MessageConsumer(client.clone()),
+        )
+        .await,
+    );
+
+    channels.push(
+        make_queue_and_consume(
+            &config,
+            &connection,
+            &config.elasticsearch.channel_delete_queue,
+            consumers::ChannelDeleteConsumer(client.clone()),
+        )
+        .await,
+    );
+
+    let mut task = None;
+
+    if std::env::var("INDEX_ALL_MESSAGES").as_deref().is_ok_and(|v| v == "1") {
+        let amqp = AMQP::new_auto().await;
+        task = Some(spawn(index::index_existing_messages(db, amqp)));
+    }
+
+    ctrl_c().await.expect("Failed to wait for ctrl-c");
+
+    for channel in channels {
+        channel.close().await.unwrap();
+    }
+
+    if let Some(task) = task {
+        task.abort();
+    }
+}
+
+async fn make_queue_and_consume<F: AsyncConsumer + Send + 'static>(
+    config: &Settings,
+    connection: &Connection,
+    queue: &str,
+    consumer: F,
+) -> Channel {
+    let channel = connection.open_channel(None).await.unwrap();
+
+    channel
+        .exchange_declare(
+            ExchangeDeclareArguments::new(&config.elasticsearch.exchange, "direct")
+                .durable(true)
+                .finish(),
+        )
+        .await
+        .expect("Failed to declare pushd exchange");
+
+    _ = channel
+        .queue_declare(QueueDeclareArguments::new(queue).durable(true).finish())
+        .await
+        .unwrap()
+        .unwrap();
+
+    channel
+        .queue_bind(QueueBindArguments::new(
+            queue,
+            &config.elasticsearch.exchange,
+            queue,
+        ))
+        .await
+        .expect("This probably means the revolt.messages exchange does not exist in rabbitmq!");
+
+    let args = BasicConsumeArguments::new(queue, "")
+        .manual_ack(true)
+        .finish();
+
+    let tag = channel.basic_consume(consumer, args).await.unwrap();
+    log::info!("Consuming queue {queue}, tag {tag}");
+
+    channel
+}
+
+#[tokio::main]
+async fn main() {
+    _main().await
+}

crates/delta/Cargo.toml

@@ -84,6 +84,7 @@ revolt-presence = { path = "../core/presence" }
 revolt-result = { path = "../core/result", features = ["rocket", "okapi"] }
 revolt-permissions = { path = "../core/permissions", features = ["schemas"] }
 revolt-ratelimits = { path = "../core/ratelimits", features = ["rocket"] }
+revolt-search = { path = "../core/search" }
 
 # voice
 livekit-api = "0.4.4"

crates/delta/src/main.rs

@@ -12,6 +12,7 @@ use revolt_config::config;
 use revolt_database::events::client::EventV1;
 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;
@@ -24,8 +25,8 @@ use amqprs::{
 };
 use async_std::channel::unbounded;
 use authifier::AuthifierEvent;
-use rocket::data::ToByteUnit;
 use revolt_database::voice::VoiceClient;
+use rocket::data::ToByteUnit;
 
 pub async fn web() -> Rocket<Build> {
     // Get settings
@@ -135,6 +136,15 @@ pub async fn web() -> Rocket<Build> {
         .await
         .expect("Failed to declare exchange");
 
+    channel
+        .exchange_declare(
+            ExchangeDeclareArguments::new(&config.elasticsearch.exchange, "direct")
+                .durable(true)
+                .finish(),
+        )
+        .await
+        .expect("Failed to declare exchange");
+
     let amqp = AMQP::new(connection, channel);
 
     // Launch background task workers
@@ -147,6 +157,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)
@@ -160,6 +177,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/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_delete.rs

@@ -1,6 +1,6 @@
+use revolt_config::capture_error;
 use revolt_database::{
-    util::{permissions::DatabasePermissionQuery, reference::Reference},
+    AMQP, Database, User, util::{permissions::DatabasePermissionQuery, reference::Reference}
-    Database, User,
 };
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::Result;
@@ -14,6 +14,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 +29,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,7 @@
 use iso8601_timestamp::Timestamp;
+use revolt_config::capture_error;
 use revolt_database::{
-    tasks,
+    AMQP, Database, Message, PartialMessage, User, tasks, util::{permissions::DatabasePermissionQuery, reference::Reference}
-    util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Database, Message, PartialMessage, User,
 };
 use revolt_models::v0::{self, Embed};
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
@@ -17,6 +16,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<'_>,
@@ -96,5 +96,10 @@ pub async fn edit(
         }
     }
 
+    if let Err(e) = amqp.edit_message_search(message.clone()).await {
+        log::error!("Error pushing message to RabbitMQ: {e}");
+        capture_error(&e);
+    }
+
     Ok(Json(message.into_model(None, None)))
 }

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,
@@ -46,6 +47,11 @@ pub async fn message_pin(
         )
         .await?;
 
+    if let Err(e) = amqp.edit_message_search(message.clone()).await {
+        log::error!("Error pushing message to RabbitMQ: {e}");
+        capture_error(&e);
+    }
+
     SystemMessage::MessagePinned {
         id: message.id.clone(),
         by: user.id.clone(),

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

@@ -420,7 +420,7 @@ mod test {
         // Delete the message
         message
             .clone()
-            .delete(&harness.db)
+            .delete(&harness.db, None)
             .await
             .expect("Failed to delete message");
 

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,
@@ -39,6 +40,11 @@ pub async fn message_unpin(
         .update(db, PartialMessage::default(), vec![FieldsMessage::Pinned])
         .await?;
 
+    if let Err(e) = amqp.edit_message_search(message.clone()).await {
+        log::error!("Error pushing message to RabbitMQ: {e}");
+        capture_error(&e);
+    }
+
     SystemMessage::MessageUnpinned {
         id: message.id.clone(),
         by: user.id.clone(),

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;
 
 pub fn mount(config: Settings, mut rocket: Rocket<Build>) -> Rocket<Build> {
     let settings = OpenApiSettings::default();
@@ -40,7 +41,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! {
@@ -60,7 +62,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(),
         };
     }
 
@@ -82,7 +85,8 @@ pub fn mount(config: Settings, mut rocket: Rocket<Build>) -> Rocket<Build> {
             "/onboard" => onboard::routes(),
             "/push" => push::routes(),
             "/sync" => sync::routes(),
-            "/webhooks" => webhooks::routes()
+            "/webhooks" => webhooks::routes(),
+            "/search" => search::routes(),
         };
     } else {
         mount_endpoints_and_merged_docs! {
@@ -101,7 +105,8 @@ pub fn mount(config: Settings, mut rocket: Rocket<Build>) -> Rocket<Build> {
             "/auth/mfa" => rocket_authifier::routes::mfa::routes(),
             "/onboard" => onboard::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,144 @@
+use std::collections::HashSet;
+
+use revolt_database::{util::permissions::DatabasePermissionQuery, Database, 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>> {
+    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?
+        .into_iter()
+        .map(|msg| msg.into_model(None, None))
+        .collect::<Vec<_>>();
+
+    // 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),
+    });
+
+    // TODO: abstract the user and member fetching logic out
+    let response = if let Some(true) = data.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, &user, &user_ids).await?;
+
+        v0::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
+            },
+        }
+    } else {
+        v0::BulkMessageResponse::JustMessages(messages)
+    };
+
+    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/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))