diff --git a/Cargo.lock b/Cargo.lock index ac4ed100..461705c8 100644 --- a/Cargo.lock +++ b/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", @@ -6824,7 +6891,7 @@ dependencies = [ "simdutf8", "strum_macros", "tempfile", - "tokio 1.49.0", + "tokio 1.50.0", "tower-http 0.5.2", "tracing", "tracing-subscriber", @@ -6871,7 +6938,7 @@ version = "0.11.5" dependencies = [ "indexmap 2.13.0", "lru 0.16.3", - "tokio 1.49.0", + "tokio 1.50.0", ] [[package]] @@ -6900,7 +6967,7 @@ dependencies = [ "revolt-database", "revolt-files", "revolt-result", - "tokio 1.49.0", + "tokio 1.50.0", ] [[package]] @@ -6988,6 +7055,7 @@ dependencies = [ "revolt-presence", "revolt-ratelimits", "revolt-result", + "revolt-search", "revolt_rocket_okapi", "rocket", "rocket_authifier", @@ -7023,7 +7091,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tiny-skia", - "tokio 1.49.0", + "tokio 1.50.0", "tracing", "typenum", "usvg", @@ -7047,7 +7115,7 @@ dependencies = [ "revolt-result", "serde", "serde_json", - "tokio 1.49.0", + "tokio 1.50.0", "tower-http 0.5.2", "tracing", "utoipa", @@ -7075,7 +7143,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "tokio 1.49.0", + "tokio 1.50.0", "tracing", "tracing-subscriber", "utoipa", @@ -7158,7 +7226,7 @@ dependencies = [ "revolt_optional_struct", "serde", "serde_json", - "tokio 1.49.0", + "tokio 1.50.0", "ulid 1.2.1", "web-push", ] @@ -7196,6 +7264,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" @@ -7245,7 +7343,7 @@ dependencies = [ "serde", "serde_json", "thiserror 1.0.69", - "tokio 1.49.0", + "tokio 1.50.0", ] [[package]] @@ -7401,7 +7499,7 @@ dependencies = [ "state", "tempfile", "time", - "tokio 1.49.0", + "tokio 1.50.0", "tokio-stream", "tokio-util", "ubyte", @@ -7492,7 +7590,7 @@ dependencies = [ "stable-pattern", "state", "time", - "tokio 1.49.0", + "tokio 1.50.0", "uncased", ] @@ -7983,7 +8081,7 @@ dependencies = [ "sentry-debug-images", "sentry-panic", "sentry-tracing", - "tokio 1.49.0", + "tokio 1.50.0", "ureq", ] @@ -8952,9 +9050,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", @@ -8985,7 +9083,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]] @@ -8995,7 +9093,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]] @@ -9006,7 +9104,7 @@ checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f" dependencies = [ "rustls 0.22.4", "rustls-pki-types", - "tokio 1.49.0", + "tokio 1.50.0", ] [[package]] @@ -9016,7 +9114,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]] @@ -9027,7 +9125,7 @@ checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047" dependencies = [ "futures-core", "pin-project-lite 0.2.16", - "tokio 1.49.0", + "tokio 1.50.0", ] [[package]] @@ -9041,7 +9139,7 @@ dependencies = [ "futures-io", "futures-sink", "pin-project-lite 0.2.16", - "tokio 1.49.0", + "tokio 1.50.0", ] [[package]] @@ -9127,7 +9225,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", @@ -9689,9 +9787,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" @@ -9723,6 +9821,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" @@ -10483,7 +10587,7 @@ dependencies = [ "serde", "serde_json", "time", - "tokio 1.49.0", + "tokio 1.50.0", "tower-service", "url", ] diff --git a/Revolt.toml b/Revolt.toml index 0a824e2d..9140e983 100644 --- a/Revolt.toml +++ b/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 = "" \ No newline at end of file diff --git a/compose.yml b/compose.yml index e1e84b6f..eff66834 100644 --- a/compose.yml +++ b/compose.yml @@ -10,6 +10,7 @@ services: image: mongo ports: - "27017:27017" + command: ["--replSet", "rs0"] volumes: - ./.data/db:/data/db ulimits: @@ -74,4 +75,35 @@ services: command: --config /etc/livekit.yml network_mode: "host" volumes: - - ./livekit.yml:/etc/livekit.yml \ No newline at end of file + - ./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"]' \ No newline at end of file diff --git a/crates/core/config/Revolt.toml b/crates/core/config/Revolt.toml index 13028ccc..2a0b5b01 100644 --- a/crates/core/config/Revolt.toml +++ b/crates/core/config/Revolt.toml @@ -313,3 +313,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" \ No newline at end of file diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 8b68ad4f..57c25149 100644 --- a/crates/core/config/src/lib.rs +++ b/crates/core/config/src/lib.rs @@ -408,6 +408,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, @@ -419,6 +432,7 @@ pub struct Settings { pub features: Features, pub sentry: Sentry, pub production: bool, + pub elasticsearch: Elasticsearch, } impl Settings { diff --git a/crates/core/database/src/amqp/amqp.rs b/crates/core/database/src/amqp/amqp.rs index b29105ed..0e4532f2 100644 --- a/crates/core/database/src/amqp/amqp.rs +++ b/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 + } } diff --git a/crates/core/database/src/events/rabbit.rs b/crates/core/database/src/events/rabbit.rs index 6f5c9ab3..cf33c15f 100644 --- a/crates/core/database/src/events/rabbit.rs +++ b/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 +} \ No newline at end of file diff --git a/crates/core/database/src/models/channels/model.rs b/crates/core/database/src/models/channels/model.rs index d8a485b9..ad7ad958 100644 --- a/crates/core/database/src/models/channels/model.rs +++ b/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; @@ -331,7 +331,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<()> { @@ -368,7 +368,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()), @@ -661,7 +661,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, @@ -693,7 +693,7 @@ impl Channel { .into_message(id.to_string()) .send( db, - Some(amqp), + amqp, MessageAuthor::System { username: name, avatar: None, @@ -706,7 +706,7 @@ impl Channel { .await .ok(); } else { - return self.delete(db).await; + return self.delete(db, amqp).await; } } @@ -733,7 +733,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()), @@ -755,13 +755,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(()) } } diff --git a/crates/core/database/src/models/messages/model.rs b/crates/core/database/src/models/messages/model.rs index 11bc2eac..1ecaddfe 100644 --- a/crates/core/database/src/models/messages/model.rs +++ b/crates/core/database/src/models/messages/model.rs @@ -2,7 +2,7 @@ use std::{collections::HashSet, hash::RandomState}; 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, @@ -670,7 +670,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, member: Option, @@ -720,6 +720,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(()) } @@ -988,9 +995,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 = self .attachments + .as_ref() .map(|files| files.iter().map(|file| file.id.to_string()).collect()) .unwrap_or_default(); @@ -1000,12 +1008,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(()) } diff --git a/crates/core/database/src/models/messages/ops.rs b/crates/core/database/src/models/messages/ops.rs index 4d54e83c..59f985a6 100644 --- a/crates/core/database/src/models/messages/ops.rs +++ b/crates/core/database/src/models/messages/ops.rs @@ -1,6 +1,6 @@ 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; @@ -40,4 +40,6 @@ pub trait AbstractMessages: Sync + Send { /// Delete messages from a channel by their ids and corresponding channel id async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()>; + + async fn fetch_all_messages(&self) -> Result>; } diff --git a/crates/core/database/src/models/messages/ops/mongodb.rs b/crates/core/database/src/models/messages/ops/mongodb.rs index 449e71ef..2816c513 100644 --- a/crates/core/database/src/models/messages/ops/mongodb.rs +++ b/crates/core/database/src/models/messages/ops/mongodb.rs @@ -1,12 +1,12 @@ use bson::{to_bson, Document}; use futures::try_join; -use mongodb::options::FindOptions; +use mongodb::options::{FindOptions, ReadConcern}; use revolt_models::v0::MessageSort; use revolt_result::Result; use crate::{ AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath, Message, MessageQuery, - MessageTimePeriod, MongoDb, PartialMessage, + MessageTimePeriod, MongoDb, PartialMessage, util::ChunkedDatabaseGenerator, }; use super::AbstractMessages; @@ -306,6 +306,31 @@ impl AbstractMessages for MongoDb { .map(|_| ()) .map_err(|_| create_database_error!("delete_many", COL)) } + + async fn fetch_all_messages(&self) -> Result> { + 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::(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 { diff --git a/crates/core/database/src/models/messages/ops/reference.rs b/crates/core/database/src/models/messages/ops/reference.rs index ae5b43f0..d0fbe39c 100644 --- a/crates/core/database/src/models/messages/ops/reference.rs +++ b/crates/core/database/src/models/messages/ops/reference.rs @@ -2,7 +2,10 @@ use futures::future::try_join_all; use indexmap::IndexSet; use revolt_result::Result; -use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage, ReferenceDb}; +use crate::{ + util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery, + PartialMessage, ReferenceDb, +}; use super::AbstractMessages; @@ -286,4 +289,10 @@ impl AbstractMessages for ReferenceDb { Ok(()) } + + async fn fetch_all_messages(&self) -> Result> { + Ok(ChunkedDatabaseGenerator::new_reference( + self.messages.lock().await.values().cloned().collect(), + )) + } } diff --git a/crates/core/database/src/models/servers/model.rs b/crates/core/database/src/models/servers/model.rs index 183dd3d2..a01bf2bd 100644 --- a/crates/core/database/src/models/servers/model.rs +++ b/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(), } diff --git a/crates/core/database/src/models/users/model.rs b/crates/core/database/src/models/users/model.rs index 0c2b6cee..a1823f58 100644 --- a/crates/core/database/src/models/users/model.rs +++ b/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 - _ = 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, @@ -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( diff --git a/crates/core/database/src/util/chunked.rs b/crates/core/database/src/util/chunked.rs new file mode 100644 index 00000000..8d14e34e --- /dev/null +++ b/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 { + #[cfg(feature = "mongodb")] + MongoDb { + session: ClientSession, + cursor: SessionCursor, + }, + + Reference { + offset: i32, + data: Vec, + }, +} + +impl Deserialize<'d> + Clone> ChunkedDatabaseGenerator { + #[cfg(feature = "mongodb")] + pub fn new_mongo(session: ClientSession, cursor: SessionCursor) -> Self { + Self::MongoDb { + session, + cursor, + } + } + + pub fn new_reference(data: Vec) -> Self { + Self::Reference { + offset: 0, + data, + } + } + + pub async fn next(&mut self) -> Option { + 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()) + } + } + } + } +} diff --git a/crates/core/database/src/util/mod.rs b/crates/core/database/src/util/mod.rs index 1a03a176..6f27c32d 100644 --- a/crates/core/database/src/util/mod.rs +++ b/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; \ No newline at end of file diff --git a/crates/core/models/src/v0/mod.rs b/crates/core/models/src/v0/mod.rs index 0468493b..fcdd90a3 100644 --- a/crates/core/models/src/v0/mod.rs +++ b/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::*; diff --git a/crates/core/models/src/v0/search.rs b/crates/core/models/src/v0/search.rs new file mode 100644 index 00000000..569c99e6 --- /dev/null +++ b/crates/core/models/src/v0/search.rs @@ -0,0 +1,49 @@ +use iso8601_timestamp::Timestamp; + +auto_derived!( + pub struct DataChannelMessagesSearch { + pub channel: Option, + pub server: Option, + pub offset: Option, + pub limit: Option, + pub filters: Option, + pub sort: Option, + pub include_users: Option, + } + + pub struct DataChannelMessagesSearchFilters { + pub content: Option, + pub author: Option>, + pub mentions: Option>, + pub role_mentions: Option>, + pub before_date: Option, + pub after_date: Option, + pub author_type: Option>, + pub pinned: Option, + pub components: Option>, + } + + #[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, + } +); diff --git a/crates/core/search/Cargo.toml b/crates/core/search/Cargo.toml new file mode 100644 index 00000000..9e51ddda --- /dev/null +++ b/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" diff --git a/crates/core/search/src/client.rs b/crates/core/search/src/client.rs new file mode 100644 index 00000000..900858a7 --- /dev/null +++ b/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 for Error { + fn from(value: elasticsearch::Error) -> Self { + Self::Http(value) + } +} + +impl From 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, 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::().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(()) + } + } +} diff --git a/crates/core/search/src/lib.rs b/crates/core/search/src/lib.rs new file mode 100644 index 00000000..c9e13b8c --- /dev/null +++ b/crates/core/search/src/lib.rs @@ -0,0 +1,5 @@ +mod client; +mod types; + +pub use client::*; +pub use types::*; \ No newline at end of file diff --git a/crates/core/search/src/types.rs b/crates/core/search/src/types.rs new file mode 100644 index 00000000..dc79d63a --- /dev/null +++ b/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, + pub author: Option>, + pub mentions: Option>, + pub role_mentions: Option>, + pub before_date: Option, + pub after_date: Option, + pub author_type: Option>, + pub pinned: Option, + pub components: Option>, +} + +#[derive(Clone, Copy, Default, PartialEq, Eq)] +pub enum SortOrder { + Asc, + #[default] + Desc, +} + +#[derive(Clone, PartialEq)] +pub struct SearchTerms { + pub channels: Vec, + pub filters: SearchFilters, + pub offset: Option, + pub limit: Option, + pub sort: Option, +} + +impl From 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 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 for SortOrder { + fn from(value: v0::SortOrder) -> Self { + match value { + v0::SortOrder::Asc => SortOrder::Asc, + v0::SortOrder::Desc => SortOrder::Desc, + } + } +} + +impl From 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 for elasticsearch_dsl::SortOrder { + fn from(value: SortOrder) -> Self { + match value { + SortOrder::Asc => elasticsearch_dsl::SortOrder::Asc, + SortOrder::Desc => elasticsearch_dsl::SortOrder::Desc, + } + } +} diff --git a/crates/daemons/searchd/Cargo.toml b/crates/daemons/searchd/Cargo.toml new file mode 100644 index 00000000..724f1346 --- /dev/null +++ b/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" diff --git a/crates/daemons/searchd/src/consumers/channel_delete.rs b/crates/daemons/searchd/src/consumers/channel_delete.rs new file mode 100644 index 00000000..dce1b2b8 --- /dev/null +++ b/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, + ) { + let payload = serde_json::from_slice::(&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 + } + } +} \ No newline at end of file diff --git a/crates/daemons/searchd/src/consumers/message.rs b/crates/daemons/searchd/src/consumers/message.rs new file mode 100644 index 00000000..ce288100 --- /dev/null +++ b/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, + ) { + let message = serde_json::from_slice::(&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 + } + } +} \ No newline at end of file diff --git a/crates/daemons/searchd/src/consumers/message_delete.rs b/crates/daemons/searchd/src/consumers/message_delete.rs new file mode 100644 index 00000000..e1a1f17a --- /dev/null +++ b/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, + ) { + let payload = serde_json::from_slice::(&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 + } + } +} \ No newline at end of file diff --git a/crates/daemons/searchd/src/consumers/message_edit.rs b/crates/daemons/searchd/src/consumers/message_edit.rs new file mode 100644 index 00000000..8b8ed42a --- /dev/null +++ b/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, + ) { + let message = serde_json::from_slice::(&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 + } + } +} \ No newline at end of file diff --git a/crates/daemons/searchd/src/consumers/mod.rs b/crates/daemons/searchd/src/consumers/mod.rs new file mode 100644 index 00000000..f51eab9c --- /dev/null +++ b/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::*; \ No newline at end of file diff --git a/crates/daemons/searchd/src/index.rs b/crates/daemons/searchd/src/index.rs new file mode 100644 index 00000000..ab942371 --- /dev/null +++ b/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); + } + } +} \ No newline at end of file diff --git a/crates/daemons/searchd/src/main.rs b/crates/daemons/searchd/src/main.rs new file mode 100644 index 00000000..feca829c --- /dev/null +++ b/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( + 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 +} diff --git a/crates/delta/Cargo.toml b/crates/delta/Cargo.toml index af6de573..59414c21 100644 --- a/crates/delta/Cargo.toml +++ b/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" diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index 882609cc..47e0f9be 100644 --- a/crates/delta/src/main.rs +++ b/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 { // Get settings @@ -135,6 +136,15 @@ pub async fn web() -> Rocket { .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 { // 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 { .manage(cors.clone()) .manage(voice_client) .manage(ratelimits) + .manage(elasticsearch) .attach(ratelimiter::RatelimitFairing) .attach(cors) .configure(rocket::Config { diff --git a/crates/delta/src/routes/bots/invite.rs b/crates/delta/src/routes/bots/invite.rs index ad4f16ad..8f0215c2 100644 --- a/crates/delta/src/routes/bots/invite.rs +++ b/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) } diff --git a/crates/delta/src/routes/channels/channel_delete.rs b/crates/delta/src/routes/channels/channel_delete.rs index f6cc8b75..68ae63cc 100644 --- a/crates/delta/src/routes/channels/channel_delete.rs +++ b/crates/delta/src/routes/channels/channel_delete.rs @@ -47,7 +47,7 @@ pub async fn delete( channel .remove_user_from_group( db, - amqp, + Some(amqp), &user, None, options.leave_silently.unwrap_or_default(), @@ -60,7 +60,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, channel.id(), channel.server()).await?; } diff --git a/crates/delta/src/routes/channels/group_add_member.rs b/crates/delta/src/routes/channels/group_add_member.rs index ac1aa886..5fd1c1c6 100644 --- a/crates/delta/src/routes/channels/group_add_member.rs +++ b/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) } diff --git a/crates/delta/src/routes/channels/group_remove_member.rs b/crates/delta/src/routes/channels/group_remove_member.rs index 995991f9..db09e4c7 100644 --- a/crates/delta/src/routes/channels/group_remove_member.rs +++ b/crates/delta/src/routes/channels/group_remove_member.rs @@ -48,7 +48,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)); diff --git a/crates/delta/src/routes/channels/message_delete.rs b/crates/delta/src/routes/channels/message_delete.rs index 437c80ee..9b1ba4ee 100644 --- a/crates/delta/src/routes/channels/message_delete.rs +++ b/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}, - Database, User, + AMQP, Database, User, util::{permissions::DatabasePermissionQuery, reference::Reference} }; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::Result; @@ -14,6 +14,7 @@ use rocket_empty::EmptyResponse; #[delete("//messages/", rank = 2)] pub async fn delete( db: &State, + amqp: &State, 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) } diff --git a/crates/delta/src/routes/channels/message_edit.rs b/crates/delta/src/routes/channels/message_edit.rs index 53426ec0..ee1ad586 100644 --- a/crates/delta/src/routes/channels/message_edit.rs +++ b/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, - util::{permissions::DatabasePermissionQuery, reference::Reference}, - Database, Message, PartialMessage, User, + AMQP, Database, Message, PartialMessage, User, tasks, util::{permissions::DatabasePermissionQuery, reference::Reference} }; use revolt_models::v0::{self, Embed}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; @@ -17,6 +16,7 @@ use validator::Validate; #[patch("//messages/", data = "")] pub async fn edit( db: &State, + amqp: &State, 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))) } diff --git a/crates/delta/src/routes/channels/message_pin.rs b/crates/delta/src/routes/channels/message_pin.rs index bcc97341..043f0d9b 100644 --- a/crates/delta/src/routes/channels/message_pin.rs +++ b/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(), diff --git a/crates/delta/src/routes/channels/message_search.rs b/crates/delta/src/routes/channels/message_search.rs index 74bc2017..e554c1d9 100644 --- a/crates/delta/src/routes/channels/message_search.rs +++ b/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("//search", data = "")] diff --git a/crates/delta/src/routes/channels/message_send.rs b/crates/delta/src/routes/channels/message_send.rs index d0479fa1..0ed9c09e 100644 --- a/crates/delta/src/routes/channels/message_send.rs +++ b/crates/delta/src/routes/channels/message_send.rs @@ -374,7 +374,7 @@ mod test { // Delete the message message .clone() - .delete(&harness.db) + .delete(&harness.db, None) .await .expect("Failed to delete message"); diff --git a/crates/delta/src/routes/channels/message_unpin.rs b/crates/delta/src/routes/channels/message_unpin.rs index 42184895..b860f657 100644 --- a/crates/delta/src/routes/channels/message_unpin.rs +++ b/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(), diff --git a/crates/delta/src/routes/invites/invite_join.rs b/crates/delta/src/routes/invites/invite_join.rs index a18213f3..a14292e9 100644 --- a/crates/delta/src/routes/invites/invite_join.rs +++ b/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?, diff --git a/crates/delta/src/routes/mod.rs b/crates/delta/src/routes/mod.rs index 9643a365..be67d7a1 100644 --- a/crates/delta/src/routes/mod.rs +++ b/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) -> Rocket { let settings = OpenApiSettings::default(); @@ -40,7 +41,8 @@ pub fn mount(config: Settings, mut rocket: Rocket) -> Rocket { "/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) -> Rocket { "/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) -> Rocket { "/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) -> Rocket { "/auth/mfa" => rocket_authifier::routes::mfa::routes(), "/onboard" => onboard::routes(), "/push" => push::routes(), - "/sync" => sync::routes() + "/sync" => sync::routes(), + "/search" => search::routes(), }; } diff --git a/crates/delta/src/routes/search/mod.rs b/crates/delta/src/routes/search/mod.rs new file mode 100644 index 00000000..91e80e7b --- /dev/null +++ b/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, OpenApi) { + openapi_get_routes_spec![ + search_messages::search_messages + ] +} diff --git a/crates/delta/src/routes/search/search_messages.rs b/crates/delta/src/routes/search/search_messages.rs new file mode 100644 index 00000000..ef662a4f --- /dev/null +++ b/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 = "")] +pub async fn search_messages( + db: &State, + es: &State, + user: User, + data: Json, +) -> Result> { + 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::>(); + + // 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::>() + .into_iter() + .collect::>(); + 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)) +} diff --git a/crates/delta/src/routes/servers/server_delete.rs b/crates/delta/src/routes/servers/server_delete.rs index 475765ca..96507cac 100644 --- a/crates/delta/src/routes/servers/server_delete.rs +++ b/crates/delta/src/routes/servers/server_delete.rs @@ -1,10 +1,7 @@ use revolt_database::{ - util::reference::Reference, - voice::{ - delete_voice_channel, get_user_voice_channel_in_server, remove_user_from_voice_channel, - VoiceClient, - }, - Database, RemovalIntention, User, + AMQP, Database, RemovalIntention, User, util::reference::Reference, voice::{ + VoiceClient, delete_voice_channel, get_user_voice_channel_in_server, remove_user_from_voice_channel + } }; use revolt_models::v0; use revolt_result::Result; @@ -19,6 +16,7 @@ use rocket_empty::EmptyResponse; #[delete("/?")] pub async fn delete( db: &State, + amqp: &State, voice_client: &State, user: User, target: Reference<'_>, @@ -32,7 +30,7 @@ pub async fn delete( delete_voice_channel(voice_client, channel_id, Some(&server.id)).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(db, voice_client, &channel_id, &user.id).await?; diff --git a/crates/delta/src/routes/users/add_friend.rs b/crates/delta/src/routes/users/add_friend.rs index 946c3296..59e1c2fb 100644 --- a/crates/delta/src/routes/users/add_friend.rs +++ b/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)) } diff --git a/crates/delta/src/routes/users/send_friend_request.rs b/crates/delta/src/routes/users/send_friend_request.rs index a87334af..421e4ed8 100644 --- a/crates/delta/src/routes/users/send_friend_request.rs +++ b/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))