Compare commits

..
Author SHA1 Message Date
IAmTomahawkx 7cebf6e35b chore: add some routes and descriptions
Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>
2026-06-28 19:38:16 -07:00
IAmTomahawkx db6cd1a014 Merge branch 'feat/admin-api' of github.com:stoatchat/stoatchat into feat/admin-api 2026-06-28 18:08:18 -07:00
İspik 35a11f7916 feat: add admin account enable route
Signed-off-by: İspik <ispik@ispik.dev>
2026-06-28 18:16:35 +03:00
İspik feb936a278 feat: add admin account email verify route
Signed-off-by: İspik <ispik@ispik.dev>
2026-06-28 18:02:59 +03:00
İspik f10aaa5bbe feat: add admin account email change route
Signed-off-by: İspik <ispik@ispik.dev>
2026-06-28 17:49:45 +03:00
İspik 0e8f220dfb fix: add admin routes back
Signed-off-by: İspik <ispik@ispik.dev>
2026-06-28 17:49:12 +03:00
İspik cf4b9ba2ca fix: fix the issues caused by authifier migration
Signed-off-by: İspik <ispik@ispik.dev>
2026-06-28 16:47:16 +03:00
İspik 08fb35c3f9 Merge branch 'main' into feat/admin-api
# Conflicts:
#	crates/core/database/src/drivers/mod.rs
#	crates/core/database/src/drivers/reference.rs
#	crates/core/database/src/models/admin_migrations/ops/mongodb/init.rs
#	crates/core/database/src/models/channels/ops.rs
#	crates/core/database/src/models/server_members/ops.rs
#	crates/core/database/src/models/server_members/ops/mongodb.rs
#	crates/core/database/src/models/server_members/ops/reference.rs
#	crates/core/database/src/util/bridge/v0.rs
#	crates/core/result/src/axum.rs
#	crates/core/result/src/lib.rs
#	crates/core/result/src/rocket.rs
#	crates/delta/src/routes/mod.rs
#	crates/delta/src/routes/servers/server_edit.rs
2026-06-28 16:08:48 +03:00
Zomatree a22378c35c fix(docs): update react version (#842)
Signed-off-by: Zomatree <me@zomatree.live>
2026-06-27 04:24:00 +01:00
Zomatree 164be6a2da Revert "chore(deps): update dependency gh to v2.95.0" (#840)
Signed-off-by: Zomatree <me@zomatree.live>
2026-06-25 18:57:12 +01:00
Zomatree 9217c6e49b ci: fix concurrency group (#837)
Signed-off-by: Zomatree <me@zomatree.live>
2026-06-25 13:38:39 +01:00
İspik 9fb344a433 fix: remove unnecessary mut 2026-06-13 12:00:38 +03:00
İspik 948fb97e89 fix: channel edit endpoint would fail due to expecting user 2026-06-13 11:58:03 +03:00
İspik 05bceb6d4a feat: add admin channel wipe endpoint 2026-06-13 11:33:23 +03:00
İspik e48d92c635 feat: add admin channel edit endpoint 2026-06-12 23:48:26 +03:00
İspik 3f4df765eb fix: add missing parameters 2026-06-12 23:28:18 +03:00
İspik a3d390c085 Merge branch 'main' into feat/admin-api
# Conflicts:
#	crates/delta/src/routes/mod.rs
2026-06-12 23:00:48 +03:00
IAmTomahawkx db4b26251f Merge branch 'main' into feat/admin-api 2026-06-03 11:10:48 -07:00
ispik 93a00ffbf8 Merge branch 'main' into feat/admin-api
# Conflicts:
#	crates/core/database/src/util/mod.rs
#	crates/core/result/src/axum.rs
2026-05-23 13:26:21 +03:00
ispik 2d8b7481bc fix: use the correct permissions
Signed-off-by: ispik <ispik@ispik.dev>
2026-04-27 10:40:07 +03:00
ispik 8bd5542e71 Merge branch 'main' into feat/admin-api
# Conflicts:
#	crates/core/database/src/models/users/model.rs
2026-04-27 10:29:19 +03:00
ispik 533cc5835a feat: add admin channel deletion endpoint 2026-04-18 19:31:38 +03:00
ispik 80b749d6fc Merge branch 'main' into feat/admin-api 2026-04-18 18:54:02 +03:00
ispik 74ce10162c Merge branch 'main' into feat/admin-api 2026-04-12 14:50:04 +03:00
ispik 9a67cb139c feat: implement admin account deletion endpoint 2026-04-12 14:49:41 +03:00
ispik 5e98a01624 refactor: disable account before creating audit action 2026-04-12 14:41:02 +03:00
ispik 9fe7de709a fix: refactor account disabling logic to use disable method 2026-04-12 14:38:29 +03:00
ispik 5b9a909923 fix: update audit action for accurate account actions 2026-04-12 14:18:22 +03:00
ispik 913946cfa1 feat: expose account modules in mod.rs and add it to routes 2026-04-12 14:15:40 +03:00
ispik 01e0c30596 fix: update permissions for accounts routes to ManageAccounts 2026-04-12 14:12:15 +03:00
ispik 1bb12b416a fix: update account disable endpoint to return EmptyResponse 2026-04-12 13:53:30 +03:00
ispik 5e577358ac feat: add optional case parameter to admin account disable endpoint 2026-04-12 13:49:34 +03:00
ispik 6e78f0b9c0 feat: implement admin account disable functionality 2026-04-11 14:00:50 +03:00
ispik 26faa18d6e chore: add files to mod.rs 2026-04-11 12:19:22 +03:00
ispik a152d825a2 fix: update Reference type to be lifetime parameterized in admin API functions 2026-04-11 01:01:57 +03:00
ispik 85ad261cc3 Merge branch 'main' into feat/admin-api
# Conflicts:
#	crates/core/database/src/util/bridge/v0.rs
#	crates/core/result/src/axum.rs
#	crates/core/result/src/lib.rs
#	crates/core/result/src/rocket.rs
2026-04-11 00:49:36 +03:00
IAmTomahawkx f857ca6926 Merge branch 'feat/admin-api' of github.com:stoatchat/stoatchat into feat/admin-api 2026-02-17 16:58:53 -08:00
IAmTomahawkx 73ce052e1a feat: Working server/comment routes 2026-02-17 16:43:05 -08:00
IAmTomahawkx 4fd71c98d6 feat: Implement models and meta routes. Add empty files for routes still to be implemented. 2026-02-17 16:37:18 -08:00
IAmTomahawkx 96f046bc76 create models 2026-02-17 16:37:18 -08:00
IAmTomahawkx 6ecfd14b07 Addd admin models to database 2026-02-17 16:36:57 -08:00
IAmTomahawkx bf4530563a feat: Working server/comment routes 2025-07-22 03:27:05 -07:00
IAmTomahawkx ff36d71ad5 feat: Implement models and meta routes. Add empty files for routes still to be implemented. 2025-07-02 18:43:23 -07:00
IAmTomahawkx f8eb324543 create models 2025-06-20 16:40:01 -07:00
IAmTomahawkx 03db5b1ddf Addd admin models to database 2025-06-08 01:18:07 -07:00
241 changed files with 4877 additions and 2193 deletions
+1 -1
View File
@@ -12,7 +12,7 @@ permissions:
packages: write
concurrency:
group: ${{ github.head_ref || github.ref }}
group: docker-test-${{ github.head_ref || github.ref }}
cancel-in-progress: true
jobs:
+1 -1
View File
@@ -6,7 +6,7 @@ on:
pull_request:
concurrency:
group: ${{ github.head_ref || github.ref }}
group: rust-build-test-${{ github.head_ref || github.ref }}
cancel-in-progress: true
env:
+1 -1
View File
@@ -2,7 +2,7 @@
node = "25.4.0"
pnpm = "10.28.1"
gh = "2.95.0"
gh = "2.25.0"
rust = "1.92.0"
"cargo:cargo-nextest" = "0.9.122"
Generated
+28 -199
View File
@@ -300,18 +300,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "async-compression"
version = "0.4.42"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac"
dependencies = [
"compression-codecs",
"compression-core",
"pin-project-lite",
"tokio",
]
[[package]]
name = "async-executor"
version = "1.14.0"
@@ -1337,15 +1325,6 @@ dependencies = [
"alloc-stdlib",
]
[[package]]
name = "bs58"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf88ba1141d185c399bee5288d850d63b8369520c1eafc32a0430b5b6c287bf4"
dependencies = [
"tinyvec",
]
[[package]]
name = "bson"
version = "2.15.0"
@@ -1512,9 +1491,9 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "chacha20"
version = "0.10.1"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601"
dependencies = [
"cfg-if",
"cpufeatures 0.3.0",
@@ -1530,7 +1509,6 @@ dependencies = [
"iana-time-zone",
"js-sys",
"num-traits",
"serde",
"wasm-bindgen",
"windows-link",
]
@@ -1603,23 +1581,6 @@ dependencies = [
"tokio-util",
]
[[package]]
name = "compression-codecs"
version = "0.4.38"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce2548391e9c1929c21bf6aa2680af86fe4c1b33e6cea9ac1cfeec0bd11218cf"
dependencies = [
"compression-core",
"flate2",
"memchr",
]
[[package]]
name = "compression-core"
version = "0.4.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc14f565cf027a105f7a44ccf9e5b424348421a1d8952a8fc9d499d313107789"
[[package]]
name = "concurrent-queue"
version = "2.5.0"
@@ -2540,41 +2501,6 @@ version = "1.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e"
[[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",
"dyn-clone",
"flate2",
"lazy_static",
"parking_lot",
"percent-encoding",
"reqwest 0.12.28",
"rustc_version",
"serde",
"serde_json",
"serde_with",
"tokio",
"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.13.8"
@@ -3833,22 +3759,6 @@ dependencies = [
"tokio-native-tls",
]
[[package]]
name = "hyper-tls"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0"
dependencies = [
"bytes",
"http-body-util",
"hyper 1.10.1",
"hyper-util",
"native-tls",
"tokio",
"tokio-native-tls",
"tower-service",
]
[[package]]
name = "hyper-util"
version = "0.1.20"
@@ -4231,7 +4141,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24d4e5d712dd664b11e778d1cfc06c79ba2700d6bc1771e44fb7b6a4656b487d"
dependencies = [
"generic-array 1.4.3",
"schemars 0.8.22",
"schemars",
"serde",
"time",
]
@@ -4355,9 +4265,9 @@ dependencies = [
[[package]]
name = "js-sys"
version = "0.3.103"
version = "0.3.102"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
checksum = "03d04c30968dffe80775bd4d7fb676131cd04a1fb46d2686dbffbaec2d9dfd31"
dependencies = [
"cfg-if",
"futures-util",
@@ -7125,7 +7035,7 @@ dependencies = [
"http 0.2.12",
"http-body 0.4.6",
"hyper 0.14.32",
"hyper-tls 0.5.0",
"hyper-tls",
"ipnet",
"js-sys",
"log",
@@ -7164,11 +7074,9 @@ dependencies = [
"http-body-util",
"hyper 1.10.1",
"hyper-rustls 0.27.9",
"hyper-tls 0.6.0",
"hyper-util",
"js-sys",
"log",
"native-tls",
"percent-encoding",
"pin-project-lite",
"quinn",
@@ -7180,7 +7088,6 @@ dependencies = [
"serde_urlencoded",
"sync_wrapper 1.0.2",
"tokio",
"tokio-native-tls",
"tokio-rustls 0.26.4",
"tower",
"tower-http 0.6.11",
@@ -7377,7 +7284,6 @@ dependencies = [
name = "revolt-database"
version = "0.13.7"
dependencies = [
"anyhow",
"async-lock 2.8.0",
"async-recursion",
"async-trait",
@@ -7419,11 +7325,12 @@ dependencies = [
"revolt_rocket_okapi",
"rocket",
"rust-argon2",
"schemars 0.8.22",
"schemars",
"serde",
"serde_json",
"sha1 0.10.6",
"tokio",
"tokio-stream",
"totp-lite",
"ulid",
"unicode-segmentation",
@@ -7464,13 +7371,12 @@ dependencies = [
"revolt-presence",
"revolt-ratelimits",
"revolt-result",
"revolt-search",
"revolt_rocket_okapi",
"rocket",
"rocket_cors",
"rocket_empty",
"rocket_prometheus",
"schemars 0.8.22",
"schemars",
"serde",
"serde_json",
"tokio",
@@ -7574,7 +7480,7 @@ dependencies = [
"revolt-permissions",
"revolt_optional_struct",
"rocket",
"schemars 0.8.22",
"schemars",
"serde",
"serde_json",
"utoipa",
@@ -7598,7 +7504,7 @@ dependencies = [
"num_enum",
"once_cell",
"revolt-result",
"schemars 0.8.22",
"schemars",
"serde",
"tokio",
]
@@ -7670,45 +7576,13 @@ dependencies = [
"revolt_okapi",
"revolt_rocket_okapi",
"rocket",
"schemars 0.8.22",
"schemars",
"sentry",
"serde",
"serde_json",
"utoipa",
]
[[package]]
name = "revolt-search"
version = "0.13.7"
dependencies = [
"elasticsearch",
"elasticsearch-dsl",
"futures",
"iso8601-timestamp",
"linkify",
"revolt-config",
"revolt-database",
"revolt-models",
"serde",
"serde_json",
"ulid",
]
[[package]]
name = "revolt-searchd"
version = "0.13.7"
dependencies = [
"anyhow",
"async-trait",
"lapin",
"log",
"revolt-config",
"revolt-database",
"revolt-search",
"serde_json",
"tokio",
]
[[package]]
name = "revolt-voice-ingress"
version = "0.13.7"
@@ -7773,7 +7647,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23bfdf7ae769c3042fe727f6e5c17363b02a64b4b33ad60c3e5f73b26df7835b"
dependencies = [
"log",
"schemars 0.8.22",
"schemars",
"serde",
"serde_json",
]
@@ -7799,7 +7673,7 @@ dependencies = [
"revolt_okapi",
"revolt_rocket_okapi_codegen",
"rocket",
"schemars 0.8.22",
"schemars",
"serde",
"serde_json",
]
@@ -8374,30 +8248,6 @@ dependencies = [
"serde_json",
]
[[package]]
name = "schemars"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cd191f9397d57d581cddd31014772520aa448f65ef991055d7f61582c65165f"
dependencies = [
"dyn-clone",
"ref-cast",
"serde",
"serde_json",
]
[[package]]
name = "schemars"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2b42f36aa1cd011945615b92222f6bf73c599a102a300334cd7f8dbeec726cc"
dependencies = [
"dyn-clone",
"ref-cast",
"serde",
"serde_json",
]
[[package]]
name = "schemars_derive"
version = "0.8.22"
@@ -8812,18 +8662,8 @@ version = "3.21.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c"
dependencies = [
"base64 0.22.1",
"bs58",
"chrono",
"hex",
"indexmap 1.9.3",
"indexmap 2.14.0",
"schemars 0.9.0",
"schemars 1.2.1",
"serde_core",
"serde_json",
"serde_with_macros",
"time",
]
[[package]]
@@ -9795,17 +9635,12 @@ version = "0.6.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
dependencies = [
"async-compression",
"bitflags 2.13.0",
"bytes",
"futures-core",
"futures-util",
"http 1.4.2",
"http-body 1.0.1",
"http-body-util",
"pin-project-lite",
"tokio",
"tokio-util",
"tower",
"tower-layer",
"tower-service",
@@ -10246,9 +10081,9 @@ dependencies = [
[[package]]
name = "uuid"
version = "1.23.4"
version = "1.23.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53"
checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7"
dependencies = [
"getrandom 0.4.3",
"js-sys",
@@ -10345,12 +10180,6 @@ 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"
@@ -10408,9 +10237,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen"
version = "0.2.126"
version = "0.2.125"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
checksum = "8ddb3f79143bced6de84270411622a2699cee572fc0875aeaf1e7867cf9fca1a"
dependencies = [
"cfg-if",
"once_cell",
@@ -10421,9 +10250,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-futures"
version = "0.4.76"
version = "0.4.75"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d"
checksum = "503b14d284f2c8dac03b819967e155ea753f573586193b2b2c95990cb5d69280"
dependencies = [
"js-sys",
"wasm-bindgen",
@@ -10431,9 +10260,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro"
version = "0.2.126"
version = "0.2.125"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
checksum = "4e21a184b13fb19e157296e2c46056aec9092264fab83e4ba59e68c61b323c3d"
dependencies = [
"quote 1.0.46",
"wasm-bindgen-macro-support",
@@ -10441,9 +10270,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro-support"
version = "0.2.126"
version = "0.2.125"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
checksum = "fecefd9c35bd935a20fc3fc344b5f29138961e4f47fb03297d88f2587afb5ebd"
dependencies = [
"bumpalo",
"proc-macro2",
@@ -10454,9 +10283,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-shared"
version = "0.2.126"
version = "0.2.125"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
checksum = "23939e44bb9a5d7576fa2b563dc2e136628f1224e88a8deed09e04858b77871f"
dependencies = [
"unicode-ident",
]
@@ -10485,9 +10314,9 @@ dependencies = [
[[package]]
name = "web-sys"
version = "0.3.103"
version = "0.3.102"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141"
checksum = "a6430a72df5eb332242960fe84b3002a241163998241eb596d4f739b9757061d"
dependencies = [
"js-sys",
"wasm-bindgen",
+1 -5
View File
@@ -28,6 +28,7 @@ futures-locks = "0.7.1"
async-lock = "2.8.0"
async-recursion = "1.0.4"
tokio-util = { version = "0.7.18" }
tokio-stream = "0.1.18"
# Error Handling
anyhow = "1.0.100"
@@ -164,10 +165,6 @@ livekit-api = "=0.4.23"
livekit-protocol = "=0.7.7"
livekit-runtime = "0.4.0"
# Elasticsearch
elasticsearch = "9.1.0-alpha.1"
elasticsearch-dsl = "0.4"
# Other Utilities
once_cell = "1.9.0"
config = "0.13.3"
@@ -207,4 +204,3 @@ revolt-permissions = { version = "0.13.7", path = "crates/core/permissions" }
revolt-presence = { version = "0.13.7", path = "crates/core/presence" }
revolt-ratelimits = { version = "0.13.7", path = "crates/core/ratelimits" }
revolt-result = { version = "0.13.7", path = "crates/core/result" }
revolt-search = { version = "0.13.7", path = "crates/core/search" }
+1 -4
View File
@@ -4,7 +4,7 @@
[database]
# MongoDB connection URL
# Defaults to the container name specified in self-hosted
mongodb = "mongodb://127.0.0.1:27017?directConnection=true&replicaSet=rs0"
mongodb = "mongodb://127.0.0.1:27017"
# Redis connection URL
# Defaults to the container name specified in self-hosted
redis = "redis://127.0.0.1:6379/"
@@ -68,6 +68,3 @@ access_key_id = "minioautumn"
secret_access_key = "minioautumn"
# Bucket to upload to by default
default_bucket = "revolt-uploads"
[elasticsearch]
api_key = ""
-31
View File
@@ -86,34 +86,3 @@ 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"]'
+1 -1
View File
@@ -1,7 +1,7 @@
environment = "test"
[database]
mongodb = "mongodb://localhost?directConnection=true&replicaSet=rs0"
mongodb = "mongodb://localhost"
redis = "redis://localhost/"
[rabbit]
+7 -12
View File
@@ -5,7 +5,7 @@ environment = "dev"
[database]
# MongoDB connection URL
# Defaults to the container name specified in self-hosted
mongodb = "mongodb://database?directConnection=true&replicaSet=rs0"
mongodb = "mongodb://database"
# Redis connection URL
# Defaults to the container name specified in self-hosted
redis = "redis://redis/"
@@ -70,6 +70,9 @@ trust_cloudflare = false
easypwned = ""
# Tenor API Key
tenor_key = ""
# admin api machine keys
# admin_keys = ["key_1", "key_2"]
admin_keys = []
[api.security.captcha]
# hCaptcha configuration
@@ -219,11 +222,14 @@ default_bucket = "revolt-uploads"
[features]
# Feature gate options
webhooks_enabled = false
admin_api_enabled = false
# Enable push notifications for mass pings (everyone, online, roles)
# When false this will still ping in-client but will not send notifications from pushd
mass_mentions_send_notifications = true
# Can role/everyone pings be used at all
mass_mentions_enabled = true
# Show the admin api in the OpenAPI spec
admin_api_show_spec = false
[features.limits]
@@ -346,14 +352,3 @@ 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"
+3 -14
View File
@@ -228,6 +228,7 @@ pub struct ApiSecurity {
pub trust_cloudflare: bool,
pub easypwned: String,
pub tenor_key: String,
pub admin_keys: Vec<String>,
}
#[derive(Deserialize, Debug, Clone)]
@@ -439,6 +440,8 @@ pub struct Features {
pub webhooks_enabled: bool,
pub mass_mentions_send_notifications: bool,
pub mass_mentions_enabled: bool,
pub admin_api_enabled: bool,
pub admin_api_show_spec: bool,
#[serde(default)]
pub advanced: FeaturesAdvanced,
@@ -456,19 +459,6 @@ 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,
@@ -483,7 +473,6 @@ pub struct Settings {
pub production: bool,
pub environment: String,
pub disable_events_dont_use: bool,
pub elasticsearch: Elasticsearch,
}
impl Settings {
+2 -2
View File
@@ -31,7 +31,7 @@ default = ["mongodb", "tokio-runtime", "tasks"]
[dependencies]
# Core
revolt-config = { workspace = true, features = ["report-macros", "anyhow"] }
revolt-config = { workspace = true, features = ["report-macros"] }
revolt-result = { workspace = true }
revolt-models = { workspace = true, features = ["validator"] }
revolt-presence = { workspace = true }
@@ -56,7 +56,6 @@ validator = { workspace = true, features = ["derive"] }
isahc = { workspace = true, features = ["json"], optional = true }
base32 = { workspace = true }
sha1 = { workspace = true }
anyhow = { workspace = true }
# Serialisation
serde_json = { workspace = true }
@@ -83,6 +82,7 @@ async-recursion = { workspace = true }
# Async
tokio = { workspace = true, optional = true }
tokio-stream = { workspace = true }
# Axum Impl
axum = { workspace = true, optional = true }
+1 -117
View File
@@ -1,7 +1,7 @@
use std::collections::HashSet;
use std::sync::Arc;
use crate::{Message, events::rabbit::*};
use crate::events::rabbit::*;
use crate::User;
use lapin::{
options::BasicPublishOptions,
@@ -25,10 +25,6 @@ pub struct AMQP {
ack_notification_message: Arc<Channel>,
dm_call_updated: Arc<Channel>,
process_ack: Arc<Channel>,
message_search: Arc<Channel>,
edit_message_search: Arc<Channel>,
delete_message_search: Arc<Channel>,
delete_channel_search: Arc<Channel>,
#[allow(unused)]
connection: Arc<Connection>,
}
@@ -44,10 +40,6 @@ impl AMQP {
ack_notification_message: Self::create_channel(&connection).await,
dm_call_updated: Self::create_channel(&connection).await,
process_ack: Self::create_channel(&connection).await,
message_search: Self::create_channel(&connection).await,
edit_message_search: Self::create_channel(&connection).await,
delete_message_search: Self::create_channel(&connection).await,
delete_channel_search: Self::create_channel(&connection).await,
connection,
}
}
@@ -387,112 +379,4 @@ impl AMQP {
Ok(())
}
pub async fn new_message_search(
&self,
message: Message,
user: Option<User>,
) -> Result<(), AMQPError> {
let config = revolt_config::config().await;
let payload = to_string(&MessageCreatePayload { message, user }).unwrap();
debug!(
"Sending new message search payload on channel {}: {}",
config.elasticsearch.message_queue, payload
);
self.message_search
.basic_publish(
config.elasticsearch.exchange.clone().into(),
config.elasticsearch.message_queue.into(),
BasicPublishOptions::default(),
payload.as_bytes(),
AMQPProperties::default()
.with_content_type("application/json".into())
.with_delivery_mode(2),
)
.await?;
Ok(())
}
pub async fn edit_message_search(
&self,
message: Message,
user: Option<User>,
) -> Result<(), AMQPError> {
let config = revolt_config::config().await;
let payload = to_string(&MessageEditPayload { message, user }).unwrap();
debug!(
"Sending edit message search payload on channel {}: {}",
config.elasticsearch.message_edit_queue, payload
);
self.edit_message_search
.basic_publish(
config.elasticsearch.exchange.clone().into(),
config.elasticsearch.message_edit_queue.into(),
BasicPublishOptions::default(),
payload.as_bytes(),
AMQPProperties::default()
.with_content_type("application/json".into())
.with_delivery_mode(2),
)
.await?;
Ok(())
}
pub async fn delete_message_search(&self, message_id: String) -> Result<(), AMQPError> {
let config = revolt_config::config().await;
let payload = to_string(&MessageDeletePayload { message_id }).unwrap();
debug!(
"Sending delete message search payload on channel {}: {}",
config.elasticsearch.message_delete_queue, payload
);
self.delete_message_search
.basic_publish(
config.elasticsearch.exchange.clone().into(),
config.elasticsearch.message_delete_queue.into(),
BasicPublishOptions::default(),
payload.as_bytes(),
AMQPProperties::default()
.with_content_type("application/json".into())
.with_delivery_mode(2),
)
.await?;
Ok(())
}
pub async fn delete_channel_search(&self, channel_id: String) -> Result<(), AMQPError> {
let config = revolt_config::config().await;
let payload = to_string(&ChannelDeletePayload { channel_id }).unwrap();
debug!(
"Sending delete channel search payload on channel {}: {}",
config.elasticsearch.channel_delete_queue, payload
);
self.delete_channel_search
.basic_publish(
config.elasticsearch.exchange.clone().into(),
config.elasticsearch.channel_delete_queue.into(),
BasicPublishOptions::default(),
payload.as_bytes(),
AMQPProperties::default()
.with_content_type("application/json".into())
.with_delivery_mode(2),
)
.await?;
Ok(())
}
}
-1
View File
@@ -1,3 +1,2 @@
#[allow(clippy::module_inception)]
pub mod amqp;
pub mod consumer;
+1 -1
View File
@@ -2,9 +2,9 @@
mod mongodb;
mod reference;
use rand::Rng;
use revolt_config::config;
use revolt_result::Result;
#[cfg(feature = "mongodb")]
pub use self::mongodb::*;
+14 -4
View File
@@ -1,17 +1,27 @@
use std::{collections::HashMap, sync::Arc};
use std::{
collections::{BTreeMap, HashMap},
sync::Arc,
};
use futures::lock::Mutex;
use crate::{
Bot, Channel, ChannelCompositeKey, ChannelUnread, Emoji, File, FileHash, Invite, Member,
MemberCompositeKey, Message, PolicyChange, RatelimitEvent, Report, Server, ServerBan, Snapshot,
User, UserSettings, Webhook, Account, AccountInvite, Session, MFATicket
AdminAuditItem, AdminCase, AdminComment, AdminStrike, AdminToken, AdminUser, Bot, Channel,
ChannelCompositeKey, ChannelUnread, Emoji, File, FileHash, Invite, Member, MemberCompositeKey,
Message, PolicyChange, RatelimitEvent, Report, Server, ServerBan, Snapshot, User, UserSettings,
Webhook, Account, AccountInvite, Session, MFATicket
};
database_derived!(
/// Reference implementation
#[derive(Default, Debug)]
pub struct ReferenceDb {
pub admin_audits: Arc<Mutex<BTreeMap<String, AdminAuditItem>>>,
pub admin_cases: Arc<Mutex<HashMap<String, AdminCase>>>,
pub admin_comments: Arc<Mutex<HashMap<String, AdminComment>>>,
pub admin_strikes: Arc<Mutex<HashMap<String, AdminStrike>>>,
pub admin_tokens: Arc<Mutex<HashMap<String, AdminToken>>>,
pub admin_users: Arc<Mutex<HashMap<String, AdminUser>>>,
pub bots: Arc<Mutex<HashMap<String, Bot>>>,
pub channels: Arc<Mutex<HashMap<String, Channel>>>,
pub channel_invites: Arc<Mutex<HashMap<String, Invite>>>,
+10
View File
@@ -281,6 +281,16 @@ pub enum EventV1 {
id: String,
},
/// Channel wiped
///
/// Clients should remove all associated data:
/// - Messages
/// - Unreads
/// - Voice States
ChannelWipe {
id: String,
},
/// User joins a group
ChannelGroupJoin {
id: String,
+1 -23
View File
@@ -3,7 +3,7 @@ use std::collections::HashMap;
use revolt_models::v0::PushNotification;
use serde::{Deserialize, Serialize};
use crate::{Message, User};
use crate::User;
#[derive(Serialize, Deserialize)]
pub struct MessageSentPayload {
@@ -79,28 +79,6 @@ pub struct AckPayload {
pub message_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageDeletePayload {
pub message_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChannelDeletePayload {
pub channel_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageCreatePayload {
pub message: Message,
pub user: Option<User>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageEditPayload {
pub message: Message,
pub user: Option<User>,
}
/// This is not the same as the AckPayload above, as the state for this event is stored in redis to allow for state updates while the event is queued.
#[derive(Serialize, Deserialize, Debug)]
pub struct AckEventPayload {
+1 -1
View File
@@ -109,7 +109,7 @@ pub mod events;
#[cfg(feature = "tasks")]
pub mod tasks;
pub mod amqp;
mod amqp;
pub use amqp::amqp::AMQP;
#[cfg(feature = "voice")]
@@ -0,0 +1,5 @@
mod models;
mod ops;
pub use models::*;
pub use ops::*;
@@ -0,0 +1,45 @@
use iso8601_timestamp::Timestamp;
use revolt_models::v0::AdminAuditItemActions;
use crate::util::basic::transform_optional_string;
auto_derived! {
pub struct AdminAuditItem {
/// The audit item ID
#[serde(rename = "_id")]
pub id: String,
/// The moderator who performed the action
pub mod_id: String,
/// The action performed (previously 'permission'). Should be one of v0::AdminAuditActionAction
pub action: String,
/// The relevant case ID, if applicable
pub case_id: Option<String>,
/// The object the action was taken against, if applicable
pub target_id: Option<String>,
/// The context of the action, if applicable (eg. search phrases)
pub context: Option<String>,
/// When the action occurred, in iso8601.
pub timestamp: String
}
}
impl AdminAuditItem {
pub fn new(
mod_id: &str,
action: AdminAuditItemActions,
case_id: Option<&str>,
target_id: Option<&str>,
context: Option<&str>,
) -> AdminAuditItem {
let id = ulid::Ulid::new().to_string();
AdminAuditItem {
id,
mod_id: mod_id.to_string(),
action: action.to_string(),
case_id: transform_optional_string(case_id),
target_id: transform_optional_string(target_id),
context: transform_optional_string(context),
timestamp: Timestamp::now_utc().format_short().to_string(),
}
}
}
@@ -0,0 +1,17 @@
mod mongodb;
mod reference;
use revolt_result::Result;
use crate::models::admin_audits::AdminAuditItem;
#[async_trait]
pub trait AbstractAdminAudits: Sync + Send {
async fn admin_audit_insert(&self, audit: AdminAuditItem) -> Result<()>;
async fn admin_audit_fetch(
&self,
before_id: Option<&str>,
limit: i64,
) -> Result<Vec<AdminAuditItem>>;
}
@@ -0,0 +1,48 @@
use tokio_stream::StreamExt;
use revolt_result::Result;
use crate::AdminAuditItem;
use crate::MongoDb;
use super::AbstractAdminAudits;
static COL: &str = "admin_audits";
#[async_trait]
impl AbstractAdminAudits for MongoDb {
async fn admin_audit_insert(&self, audit: AdminAuditItem) -> Result<()> {
query!(self, insert_one, COL, audit).map(|_| ())
}
async fn admin_audit_fetch(
&self,
before_id: Option<&str>,
limit: i64,
) -> Result<Vec<AdminAuditItem>> {
if let Some(before) = before_id {
Ok(self
.col::<AdminAuditItem>(COL)
.find(doc! {
"_id": { "$lt": before}
})
.sort(doc! {"_id": -1})
.limit(limit)
.await
.map_err(|_| create_database_error!("find", COL))?
.filter_map(|f| f.ok())
.collect()
.await)
} else {
Ok(self
.col::<AdminAuditItem>(COL)
.find(doc! {})
.sort(doc! {"_id": -1})
.limit(limit)
.await
.map_err(|_| create_database_error!("find", COL))?
.filter_map(|f| f.ok())
.collect()
.await)
}
}
}
@@ -0,0 +1,45 @@
use revolt_result::Result;
use crate::AdminAuditItem;
use crate::ReferenceDb;
use super::AbstractAdminAudits;
#[async_trait]
impl AbstractAdminAudits for ReferenceDb {
async fn admin_audit_insert(&self, audit: AdminAuditItem) -> Result<()> {
let mut admin_audits = self.admin_audits.lock().await;
if admin_audits.contains_key(&audit.id) {
Err(create_database_error!("insert", "admin_audits"))
} else {
admin_audits.insert(audit.id.to_string(), audit.clone());
Ok(())
}
}
async fn admin_audit_fetch(
&self,
before_id: Option<&str>,
limit: i64,
) -> Result<Vec<AdminAuditItem>> {
let admin_audits = self.admin_audits.lock().await;
if let Some(before) = before_id {
Ok(admin_audits
.iter()
.rev()
.skip_while(|(id, _)| id.as_str() <= before)
.by_ref()
.take(limit as usize)
.map(|(_, item)| item.clone())
.collect())
} else {
Ok(admin_audits
.iter()
.rev()
.by_ref()
.take(limit as usize)
.map(|(_, item)| item.clone())
.collect())
}
}
}
@@ -0,0 +1,5 @@
mod models;
mod ops;
pub use models::*;
pub use ops::*;
@@ -0,0 +1,53 @@
auto_derived_partial! {
pub struct AdminCase {
/// The case ID
#[serde(rename = "_id")]
pub id: String,
/// The case Short ID
pub short_id: String,
/// The owner of the case
pub owner_id: String,
/// The title of the case
pub title: String,
/// The status of the case (open/closed)
pub status: String,
/// When the case was closed, in iso8601
pub closed_at: Option<String>,
/// The tags for the case
pub tags: Vec<String>,
},
"PartialAdminCase"
}
impl AdminCase {
pub fn new(owner_id: &str, title: &str) -> AdminCase {
let id = ulid::Ulid::new().to_string();
let short_id = id.clone().split_off(id.len() - 7);
AdminCase {
id,
short_id,
owner_id: owner_id.to_string(),
title: title.to_string(),
status: "Open".to_string(),
closed_at: None,
tags: vec![],
}
}
pub fn merge_tags(&self, other: &[String]) -> Vec<String> {
let mut resp: Vec<String> = vec![];
// Shitty combining chain; itll only ever be like 5 items
resp.extend(self.tags.clone());
resp.extend(other.iter().filter_map(|p| {
if !self.tags.contains(p) {
Some(p.clone())
} else {
None
}
}));
resp
}
}
@@ -0,0 +1,30 @@
mod mongodb;
mod reference;
use revolt_result::Result;
use crate::models::admin_cases::{AdminCase, PartialAdminCase};
#[async_trait]
pub trait AbstractAdminCases: Sync + Send {
async fn admin_case_create(&self, case: AdminCase) -> Result<()>;
async fn admin_case_assign_report(&self, case_id: &str, report_id: &str) -> Result<()>;
async fn admin_case_edit(&self, case_id: &str, partial: &PartialAdminCase) -> Result<()>;
async fn admin_case_fetch(&self, case_id: &str) -> Result<AdminCase>;
async fn admin_case_fetch_from_shorthand(&self, short_id: &str) -> Result<AdminCase>;
/// title is fuzzy, the rest of the arguments are direct matches
/// before_id and limit are for paginating
async fn admin_case_search(
&self,
title: Option<&str>,
status: Option<&str>,
owner_id: Option<&str>,
tags: Option<Vec<String>>,
before_id: Option<&str>,
limit: i64,
) -> Result<Vec<AdminCase>>;
}
@@ -0,0 +1,85 @@
use tokio_stream::StreamExt;
use bson::Document;
use revolt_result::Result;
use crate::MongoDb;
use crate::{AdminCase, PartialAdminCase};
use super::AbstractAdminCases;
static COL: &str = "admin_cases";
#[async_trait]
impl AbstractAdminCases for MongoDb {
async fn admin_case_create(&self, case: AdminCase) -> Result<()> {
query!(self, insert_one, COL, case).map(|_| ())
}
async fn admin_case_assign_report(&self, case_id: &str, report_id: &str) -> Result<()> {
self.col::<Document>("safety_reports")
.update_one(
doc! {"_id": { "$regex": format!("{}$", report_id)}},
doc! {"case_id": case_id},
)
.await
.map(|_| ())
.map_err(|_| create_database_error!("safety_reports", "update_one"))
}
async fn admin_case_edit(&self, case_id: &str, partial: &PartialAdminCase) -> Result<()> {
query!(self, update_one_by_id, COL, case_id, partial, vec![], None).map(|_| ())
}
async fn admin_case_fetch(&self, case_id: &str) -> Result<AdminCase> {
query!(self, find_one_by_id, COL, case_id)?
.ok_or_else(|| create_database_error!("find_one", COL))
}
async fn admin_case_fetch_from_shorthand(&self, short_id: &str) -> Result<AdminCase> {
query!(self, find_one, COL, doc! {"short_id": short_id})?
.ok_or_else(|| create_database_error!("find_one", COL))
}
/// title is fuzzy, the rest of the arguments are direct matches
async fn admin_case_search(
&self,
title: Option<&str>,
status: Option<&str>,
owner_id: Option<&str>,
tags: Option<Vec<String>>,
before_id: Option<&str>,
limit: i64,
) -> Result<Vec<AdminCase>> {
let mut query = Document::new();
if let Some(title) = title {
query.insert("$text", doc! {"$search": title});
}
if let Some(status) = status {
query.insert("status", status);
}
if let Some(owner) = owner_id {
query.insert("owner_id", owner);
}
if let Some(tags) = tags {
query.insert("tags", doc! {"$elemMatch": {"$in": tags}});
}
if let Some(before) = before_id {
query.insert("_id", doc! {"$lt": before});
}
Ok(self
.col::<AdminCase>(COL)
.find(query)
.limit(limit)
.await
.map_err(|_| create_database_error!("find", COL))?
.filter_map(|f| f.ok())
.collect()
.await)
}
}
@@ -0,0 +1,112 @@
use revolt_result::Result;
use crate::ReferenceDb;
use crate::{AdminCase, PartialAdminCase};
use super::AbstractAdminCases;
#[async_trait]
impl AbstractAdminCases for ReferenceDb {
async fn admin_case_create(&self, case: AdminCase) -> Result<()> {
let mut admin_cases = self.admin_cases.lock().await;
if admin_cases.contains_key(&case.id) {
Err(create_database_error!("insert", "admin_cases"))
} else {
admin_cases.insert(case.id.to_string(), case.clone());
Ok(())
}
}
async fn admin_case_assign_report(&self, case_id: &str, report_id: &str) -> Result<()> {
let mut reports = self.safety_reports.lock().await;
if let Some(report) = reports.get_mut(report_id) {
report.case_id = Some(case_id.to_string());
Ok(())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_case_edit(&self, case_id: &str, partial: &PartialAdminCase) -> Result<()> {
let mut admin_cases = self.admin_cases.lock().await;
if let Some(case) = admin_cases.get_mut(case_id) {
case.apply_options(partial.clone());
Ok(())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_case_fetch(&self, case_id: &str) -> Result<AdminCase> {
let admin_cases = self.admin_cases.lock().await;
if let Some(case) = admin_cases.get(case_id) {
Ok(case.clone())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_case_fetch_from_shorthand(&self, short_id: &str) -> Result<AdminCase> {
let admin_cases = self.admin_cases.lock().await;
if let Some((_, case)) = admin_cases.iter().find(|(_, c)| c.short_id == short_id) {
Ok(case.clone())
} else {
Err(create_error!(NotFound))
}
}
/// title is fuzzy, the rest of the arguments are direct matches
/// before_id and limit are for paginating
async fn admin_case_search(
&self,
title: Option<&str>,
status: Option<&str>,
owner_id: Option<&str>,
tags: Option<Vec<String>>,
before_id: Option<&str>,
limit: i64,
) -> Result<Vec<AdminCase>> {
let admin_cases = self.admin_cases.lock().await;
Ok(admin_cases
.iter()
.filter(|(_, case)| {
if let Some(title) = title {
case.title.to_lowercase().contains(title)
} else {
true
}
})
.filter(|(_, case)| {
if let Some(status) = status {
case.status == status
} else {
true
}
})
.filter(|(_, case)| {
if let Some(owner) = owner_id {
case.owner_id == owner
} else {
true
}
})
.filter(|(_, case)| {
if let Some(tags) = &tags {
tags.iter().filter(|p| case.tags.contains(p)).count() > 0
} else {
true
}
})
.skip_while(|(id, _)| {
if let Some(before) = before_id {
id.as_str() <= before
} else {
false
}
})
.by_ref()
.take(limit as usize)
.map(|(_, case)| case.clone())
.collect())
}
}
@@ -0,0 +1,5 @@
mod models;
mod ops;
pub use models::*;
pub use ops::*;
@@ -0,0 +1,95 @@
use revolt_models::v0;
auto_derived_partial! {
pub struct AdminComment {
/// The comment ID
#[serde(rename = "_id")]
pub id: String,
/// The Object this comment is attached to
pub object_id: String,
/// The ID of the case this comment is attached to, if applicable
pub case_id: Option<String>,
/// The author ID
pub user_id: String,
/// When the comment was edited, if applicable, in iso8601
pub edited_at: Option<String>,
/// The content, if not a system event
pub content: Option<String>,
/// The system event that occurred, if applicable
pub system_message: Option<String>,
/// The system message target
pub system_message_target: Option<String>,
/// The system message raw context
pub system_message_context: Option<String>,
},
"PartialAdminComment"
}
impl AdminComment {
pub fn new(
object_id: &str,
user_id: &str,
content: &str,
case_id: Option<&str>,
) -> AdminComment {
let id = ulid::Ulid::new().to_string();
AdminComment {
id,
object_id: object_id.to_string(),
case_id: case_id.map(|c| c.to_string()),
user_id: user_id.to_string(),
edited_at: None,
content: Some(content.to_string()),
system_message: None,
system_message_context: None,
system_message_target: None,
}
}
pub fn new_system_message(
object_id: &str,
user_id: &str,
case_id: &str,
system_message_kind: v0::AdminAuditItemActions,
context: Option<&str>,
target: &str,
) -> AdminComment {
let id = ulid::Ulid::new().to_string();
AdminComment {
id,
object_id: object_id.to_string(),
case_id: Some(case_id.to_string()),
user_id: user_id.to_string(),
edited_at: None,
content: None,
system_message: Some(serde_json::to_string(&system_message_kind).unwrap()), // if this explodes i'll eat my hat.
system_message_context: context.map(|c| c.to_string()),
system_message_target: Some(target.to_string()),
}
}
/// Edit the comment, updating the edited_at time as well
pub fn edit(&mut self, content: &str) {
self.content = Some(content.to_string());
self.edited_at = Some(
iso8601_timestamp::Timestamp::now_utc()
.format_short()
.to_string(),
);
}
}
impl PartialAdminComment {
pub fn new() -> PartialAdminComment {
PartialAdminComment::default()
}
/// Edit the comment, updating the edited_at time as well
pub fn edit(&mut self, content: &str) {
self.content = Some(content.to_string());
self.edited_at = Some(
iso8601_timestamp::Timestamp::now_utc()
.format_short()
.to_string(),
);
}
}
@@ -0,0 +1,28 @@
mod mongodb;
mod reference;
use revolt_result::Result;
use crate::{models::admin_comments::AdminComment, PartialAdminComment};
#[async_trait]
pub trait AbstractAdminComments: Sync + Send {
async fn admin_comment_insert(&self, comment: AdminComment) -> Result<()>;
async fn admin_comment_update(
&self,
comment_id: &str,
partial: &PartialAdminComment,
) -> Result<()>;
async fn admin_comment_fetch(&self, id: &str) -> Result<AdminComment>;
/// Fetch all comments related to the case. This includes comments made on other objects.
async fn admin_comment_fetch_related_case(&self, case_id: &str) -> Result<Vec<AdminComment>>;
/// Fetch all comments on an object
async fn admin_comment_fetch_object_comments(
&self,
object_id: &str,
) -> Result<Vec<AdminComment>>;
}
@@ -0,0 +1,47 @@
use revolt_result::Result;
use crate::MongoDb;
use crate::{AdminComment, PartialAdminComment};
use super::AbstractAdminComments;
static COL: &str = "admin_comments";
#[async_trait]
impl AbstractAdminComments for MongoDb {
async fn admin_comment_insert(&self, comment: AdminComment) -> Result<()> {
query!(self, insert_one, COL, comment).map(|_| ())
}
async fn admin_comment_update(
&self,
comment_id: &str,
partial: &PartialAdminComment,
) -> Result<()> {
query!(
self,
update_one_by_id,
COL,
comment_id,
partial,
vec![],
None
)
.map(|_| ())
}
async fn admin_comment_fetch(&self, id: &str) -> Result<AdminComment> {
query!(self, find_one, COL, doc! {"_id": id})?.ok_or_else(|| create_error!(NotFound))
}
async fn admin_comment_fetch_related_case(&self, case_id: &str) -> Result<Vec<AdminComment>> {
query!(self, find, COL, doc! {"case_id": case_id})
}
async fn admin_comment_fetch_object_comments(
&self,
object_id: &str,
) -> Result<Vec<AdminComment>> {
query!(self, find, COL, doc! {"object_id": object_id})
}
}
@@ -0,0 +1,73 @@
use iso8601_timestamp::Timestamp;
use revolt_result::Result;
use crate::ReferenceDb;
use crate::{AdminComment, PartialAdminComment};
use super::AbstractAdminComments;
#[async_trait]
impl AbstractAdminComments for ReferenceDb {
async fn admin_comment_insert(&self, comment: AdminComment) -> Result<()> {
let mut admin_comments = self.admin_comments.lock().await;
if admin_comments.contains_key(&comment.id) {
Err(create_database_error!("insert", "admin_comments"))
} else {
admin_comments.insert(comment.id.to_string(), comment.clone());
Ok(())
}
}
async fn admin_comment_update(
&self,
comment_id: &str,
partial: &PartialAdminComment,
) -> Result<()> {
let mut admin_comments = self.admin_comments.lock().await;
if let Some(comment) = admin_comments.get_mut(comment_id) {
comment.apply_options(partial.clone());
comment.edited_at = Some(Timestamp::now_utc().format().to_string());
Ok(())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_comment_fetch(&self, id: &str) -> Result<AdminComment> {
let admin_comments = self.admin_comments.lock().await;
admin_comments
.get(id)
.map_or(Err(create_error!(NotFound)), |ac| Ok(ac.clone()))
}
async fn admin_comment_fetch_related_case(&self, case_id: &str) -> Result<Vec<AdminComment>> {
let admin_comments = self.admin_comments.lock().await;
Ok(admin_comments
.iter()
.filter_map(|(_, c)| {
if c.case_id.as_ref().is_some_and(|c| c == case_id) {
Some(c.clone())
} else {
None
}
})
.collect())
}
async fn admin_comment_fetch_object_comments(
&self,
object_id: &str,
) -> Result<Vec<AdminComment>> {
let admin_comments = self.admin_comments.lock().await;
Ok(admin_comments
.iter()
.filter_map(|(_, c)| {
if c.object_id == object_id {
Some(c.clone())
} else {
None
}
})
.collect())
}
}
@@ -28,6 +28,20 @@ impl AbstractMigrations for MongoDb {
init::create_database(self).await;
}
let config = revolt_config::config().await;
if config.features.admin_api_enabled {
let db = self.db();
let colls = db
.list_collection_names()
.await
.expect("Failed to fetch collection names.");
if !colls.iter().any(|x| x == "admin_audits") {
info!("You've enabled the admin api for the first time. Setting up database...");
init::create_admin_database(&db).await;
}
}
Ok(())
}
}
@@ -8,6 +8,8 @@ pub async fn create_database(db: &MongoDb) {
info!("Creating database.");
let db = db.db();
let config = revolt_config::config().await;
db.create_collection("accounts")
.await
.expect("Failed to create accounts collection.");
@@ -275,6 +277,11 @@ pub async fn create_database(db: &MongoDb) {
.await
.expect("Failed to create ratelimit_events index.");
// Only create these tables if the admin api is enabled
if config.features.admin_api_enabled {
create_admin_database(&db).await;
}
db.run_command(doc! {
"createIndexes": "accounts",
"indexes": [
@@ -361,3 +368,128 @@ pub async fn create_database(db: &MongoDb) {
info!("Created database.");
}
pub async fn create_admin_database(db: &mongodb::Database) {
db.create_collection("admin_audits")
.await
.expect("Failed to create admin_audits collection.");
db.create_collection("admin_comments")
.await
.expect("Failed to create admin_comments collection.");
db.create_collection("admin_cases")
.await
.expect("Failed to create admin_cases collection.");
db.create_collection("admin_strikes")
.await
.expect("Failed to create admin_strikes collection.");
db.create_collection("admin_tokens")
.await
.expect("Failed to create admin_tokens collection.");
db.create_collection("admin_users")
.await
.expect("Failed to create admin_users collection.");
db.run_command(doc! {
"createIndexes": "admin_comments",
"indexes": [
{
"key": {
"case_id": 1_i32,
},
"unique": false,
"name": "case_id"
},
{
"key": {
"object_id": 1_i32,
},
"unique": false,
"name": "object_id"
}
]
})
.await
.expect("Failed to create admin_comments index.");
db.run_command(doc! {
"createIndexes": "admin_cases",
"indexes": [
{
"key": {
"id": 1_i32,
},
"unique": true,
"name": "id"
},
{
"key": {
"short_id": 1_i32,
},
"unique": true,
"name": "short_id"
}
]
})
.await
.expect("Failed to create admin_cases index.");
db.run_command(doc! {
"createIndexes": "admin_users",
"indexes": [
{
"key": {
"email": 1_i32,
},
"unique": true,
"name": "case_id"
},
]
})
.await
.expect("Failed to create admin_comments index.");
db.run_command(doc! {
"createIndexes": "admin_audits",
"indexes": [
{
"key": {
"mod_id": 1_i32,
},
"unique": false,
"name": "mod_id"
},
{
"key": {
"target_id": 1_i32,
},
"unique": false,
"name": "target_id"
}
]
})
.await
.expect("Failed to create admin_audits index.");
// indexes on regular data that are only needed for admin queries
db.run_command(doc! {
"createIndexes": "server_members",
"indexes": [
{
"key": {
"joined_at": 1_i32,
},
"unique": false,
"name": "joined_at"
},
]
})
.await
.expect("Failed to create server_members::joined_at index.");
info!("Created admin database.");
}
@@ -0,0 +1,5 @@
mod models;
mod ops;
pub use models::*;
pub use ops::*;
@@ -0,0 +1,47 @@
use crate::util::basic::transform_optional_string;
auto_derived_partial! {
pub struct AdminStrike {
/// The strike ID
#[serde(rename = "_id")]
pub id: String,
/// The object receiving the strike (user/server)
pub target_id: String,
/// The moderator who gave the strike
pub mod_id: String,
/// The case the strike was made under
pub case_id: Option<String>,
/// Action associated with the strike (eg. suspension/ban)
pub associated_action: Option<String>,
/// Has the strike been removed
pub overruled: bool,
/// The user-facing reason for the strike
pub reason: String,
/// Internal context for the strike
pub mod_context: Option<String>,
},
"PartialAdminStrike"
}
impl AdminStrike {
pub fn new(
target_id: &str,
mod_id: &str,
case_id: Option<&str>,
associated_action: Option<&str>,
reason: &str,
mod_context: Option<&str>,
) -> AdminStrike {
let id = ulid::Ulid::new().to_string();
AdminStrike {
id,
target_id: target_id.to_string(),
mod_id: mod_id.to_string(),
case_id: transform_optional_string(case_id),
associated_action: transform_optional_string(associated_action),
overruled: false,
reason: reason.to_string(),
mod_context: transform_optional_string(mod_context),
}
}
}
@@ -0,0 +1,17 @@
mod mongodb;
mod reference;
use revolt_result::Result;
use crate::models::admin_strikes::models::{AdminStrike, PartialAdminStrike};
#[async_trait]
pub trait AbstractAdminStrikes: Sync + Send {
async fn admin_strike_insert(&self, strike: AdminStrike) -> Result<()>;
async fn admin_strike_update(&self, strike_id: &str, partial: PartialAdminStrike)
-> Result<()>;
async fn admin_strike_get(&self, strike_id: &str) -> Result<AdminStrike>;
async fn admin_strike_get_user(&self, user_id: &str) -> Result<Vec<AdminStrike>>;
}
@@ -0,0 +1,41 @@
use revolt_result::Result;
use crate::MongoDb;
use crate::{AdminStrike, PartialAdminStrike};
use super::AbstractAdminStrikes;
static COL: &str = "admin_strikes";
#[async_trait]
impl AbstractAdminStrikes for MongoDb {
async fn admin_strike_insert(&self, strike: AdminStrike) -> Result<()> {
query!(self, insert_one, COL, strike).map(|_| ())
}
async fn admin_strike_update(
&self,
strike_id: &str,
partial: PartialAdminStrike,
) -> Result<()> {
query!(
self,
update_one_by_id,
COL,
strike_id,
partial,
vec![],
None
)
.map(|_| ())
}
async fn admin_strike_get(&self, strike_id: &str) -> Result<AdminStrike> {
query!(self, find_one_by_id, COL, strike_id)?
.ok_or_else(|| create_database_error!("find_one", COL))?
}
async fn admin_strike_get_user(&self, user_id: &str) -> Result<Vec<AdminStrike>> {
query!(self, find, COL, doc! {"target_id": user_id})
}
}
@@ -0,0 +1,51 @@
use revolt_result::Result;
use crate::ReferenceDb;
use crate::{AdminStrike, PartialAdminStrike};
use super::AbstractAdminStrikes;
#[async_trait]
impl AbstractAdminStrikes for ReferenceDb {
async fn admin_strike_insert(&self, strike: AdminStrike) -> Result<()> {
let mut admin_strikes = self.admin_strikes.lock().await;
if admin_strikes.contains_key(&strike.id) {
Err(create_database_error!("insert", "admin_strikes"))
} else {
admin_strikes.insert(strike.id.to_string(), strike.clone());
Ok(())
}
}
async fn admin_strike_update(
&self,
strike_id: &str,
partial: PartialAdminStrike,
) -> Result<()> {
let mut admin_strikes = self.admin_strikes.lock().await;
if let Some(strike) = admin_strikes.get_mut(strike_id) {
strike.apply_options(partial);
Ok(())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_strike_get(&self, strike_id: &str) -> Result<AdminStrike> {
let admin_strikes = self.admin_strikes.lock().await;
if let Some(strike) = admin_strikes.get(strike_id) {
Ok(strike.clone())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_strike_get_user(&self, user_id: &str) -> Result<Vec<AdminStrike>> {
let admin_strikes = self.admin_strikes.lock().await;
Ok(admin_strikes
.iter()
.filter(|(_, strike)| strike.target_id == user_id)
.map(|(_, strike)| strike.clone())
.collect())
}
}
@@ -0,0 +1,12 @@
mod models;
mod ops;
#[cfg(feature = "rocket-impl")]
mod rocket;
mod schema;
#[cfg(feature = "rocket-impl")]
pub use self::rocket::*;
#[cfg(feature = "rocket-impl")]
pub use self::schema::*;
pub use models::*;
pub use ops::*;
@@ -0,0 +1,71 @@
use iso8601_timestamp::Timestamp;
use revolt_result::Result;
use crate::{AdminUser, Database};
auto_derived! {
pub struct AdminToken {
/// The token ID
#[serde(rename = "_id")]
pub id: String,
/// The user this token is attached to
pub user_id: String,
/// The token itself
pub token: String,
/// The expiry timestamp for this token, in iso6801
pub expiry: String
}
/// This struct is used to validate machine tokens when doing machine to machine communication.
pub struct AdminMachineToken {
/// Placeholder field.
pub on_behalf_of: AdminUser
}
pub enum AdminAuthorization {
AdminUser(AdminUser),
AdminMachine(AdminMachineToken),
}
}
impl AdminToken {
pub fn new(user_id: &str, expiry: Timestamp) -> AdminToken {
let id = ulid::Ulid::new().to_string();
let token = nanoid::nanoid!(64);
AdminToken {
id,
user_id: user_id.to_string(),
token,
expiry: expiry.format_short().to_string(),
}
}
}
impl AdminMachineToken {
pub async fn new_from_email(email: &str, db: &Database) -> Result<AdminMachineToken> {
println!("{}", email);
if email == "example@example.com" {
// This is basically just a workaround to make the first user.
let user_count = db.admin_user_count().await?;
println!("{}", user_count);
if user_count == 0 {
return Ok(AdminMachineToken {
on_behalf_of: AdminUser {
id: "00".to_string(),
platform_user_id: "".to_string(),
email: "example@example.com".to_string(),
active: true,
permissions: u64::MAX,
},
});
}
}
let user = db.admin_user_fetch_email(email).await?;
Ok(AdminMachineToken { on_behalf_of: user })
}
pub async fn new_from_id(user_id: &str, db: &Database) -> Result<AdminMachineToken> {
let user = db.admin_user_fetch(user_id).await?;
Ok(AdminMachineToken { on_behalf_of: user })
}
}
@@ -0,0 +1,16 @@
mod mongodb;
mod reference;
use revolt_result::Result;
use crate::models::admin_tokens::AdminToken;
#[async_trait]
pub trait AbstractAdminTokens: Sync + Send {
async fn admin_token_create(&self, token: AdminToken) -> Result<()>;
async fn admin_token_revoke(&self, token_id: &str) -> Result<()>;
async fn admin_token_authenticate(&self, token: &str) -> Result<AdminToken>;
async fn admin_token_fetch(&self, id: &str) -> Result<AdminToken>;
}
@@ -0,0 +1,34 @@
use revolt_result::Result;
use crate::AdminToken;
use crate::MongoDb;
use super::AbstractAdminTokens;
static COL: &str = "admin_tokens";
#[async_trait]
impl AbstractAdminTokens for MongoDb {
async fn admin_token_create(&self, token: AdminToken) -> Result<()> {
query!(self, insert_one, COL, token).map(|_| ())
}
async fn admin_token_revoke(&self, token_id: &str) -> Result<()> {
query!(self, delete_one_by_id, COL, token_id).map(|k| {
if k.deleted_count > 0 {
Ok(())
} else {
Err(create_error!(NotFound))
}
})?
}
async fn admin_token_authenticate(&self, token: &str) -> Result<AdminToken> {
query!(self, find_one, COL, doc! {"token": token})?
.ok_or_else(|| create_error!(InvalidCredentials))
}
async fn admin_token_fetch(&self, id: &str) -> Result<AdminToken> {
query!(self, find_one_by_id, COL, id)?.ok_or_else(|| create_error!(NotFound))
}
}
@@ -0,0 +1,53 @@
use revolt_result::Result;
use crate::AdminToken;
use crate::ReferenceDb;
use super::AbstractAdminTokens;
#[async_trait]
impl AbstractAdminTokens for ReferenceDb {
async fn admin_token_create(&self, token: AdminToken) -> Result<()> {
let mut admin_tokens = self.admin_tokens.lock().await;
if admin_tokens.contains_key(&token.id) {
Err(create_database_error!("insert", "admin_tokens"))
} else {
admin_tokens.insert(token.id.to_string(), token.clone());
Ok(())
}
}
async fn admin_token_revoke(&self, token_id: &str) -> Result<()> {
let mut admin_tokens = self.admin_tokens.lock().await;
if admin_tokens.remove(token_id).is_some() {
Ok(())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_token_authenticate(&self, token: &str) -> Result<AdminToken> {
let admin_tokens = self.admin_tokens.lock().await;
let result = admin_tokens
.iter()
.filter_map(|(_, t)| {
if t.token == token {
Some(t.clone())
} else {
None
}
})
.next();
Ok(result.ok_or_else(|| create_error!(NotFound))?)
}
async fn admin_token_fetch(&self, id: &str) -> Result<AdminToken> {
let admin_tokens = self.admin_tokens.lock().await;
let result = admin_tokens.iter().find(|tok| tok.0 == id);
Ok(result
.map(|t| t.1.clone())
.ok_or_else(|| create_error!(NotFound))?)
}
}
@@ -0,0 +1,146 @@
use iso8601_timestamp::Timestamp;
use revolt_config::{capture_internal_error, report_error};
use revolt_result::{Error, Result};
use rocket::http::Status;
use rocket::request::{self, FromRequest, Outcome, Request};
use crate::{AdminAuthorization, AdminMachineToken, AdminUser, Database};
#[rocket::async_trait]
impl<'r> FromRequest<'r> for AdminMachineToken {
type Error = Error;
async fn from_request(request: &'r Request<'_>) -> request::Outcome<Self, Self::Error> {
let user: &Option<AdminMachineToken> = request
.local_cache_async(async {
let db = request.rocket().state::<Database>().expect("`Database`");
if let Some(on_behalf_of) = request
.headers()
.get("x-admin-on-behalf-of")
.next()
.map(|x| x.to_string())
{
if let Some(token) = request
.headers()
.get("x-admin-machine")
.next()
.map(|x| x.to_string())
{
let config = revolt_config::config().await;
let token = token.to_string();
if config.api.security.admin_keys.contains(&token) {
let resp: Result<AdminMachineToken> = if on_behalf_of.contains("@") {
AdminMachineToken::new_from_email(&on_behalf_of, db).await
} else {
AdminMachineToken::new_from_id(&on_behalf_of, db).await
};
if let Ok(resp) = resp {
return Some(resp);
}
}
}
}
None
})
.await;
if let Some(user) = user {
Outcome::Success(user.clone())
} else {
Outcome::Error((Status::Unauthorized, create_error!(InvalidSession)))
}
}
}
#[rocket::async_trait]
impl<'r> FromRequest<'r> for AdminAuthorization {
type Error = Error;
async fn from_request(request: &'r Request<'_>) -> request::Outcome<Self, Self::Error> {
let machine: &Option<AdminMachineToken> = request
.local_cache_async(async {
let db = request.rocket().state::<Database>().expect("`Database`");
if let Some(token) = request
.headers()
.get("x-admin-machine")
.next()
.map(|x| x.to_string())
{
if let Some(on_behalf_of) = request
.headers()
.get("x-admin-on-behalf-of")
.next()
.map(|x| x.to_string())
{
let config = revolt_config::config().await;
let token = token.to_string();
if config.api.security.admin_keys.contains(&token) {
let resp: Result<AdminMachineToken> = if on_behalf_of.contains("@") {
AdminMachineToken::new_from_email(&on_behalf_of, db).await
} else {
AdminMachineToken::new_from_id(&on_behalf_of, db).await
};
if let Ok(resp) = resp {
return Some(resp);
}
}
}
}
None
})
.await;
if let Some(machine) = machine {
if !machine.on_behalf_of.active {
Outcome::Error((Status::Unauthorized, create_error!(LockedOut)))
} else {
Outcome::Success(AdminAuthorization::AdminMachine(machine.clone()))
}
} else {
let user: &Option<AdminUser> = request
.local_cache_async(async {
let db = request.rocket().state::<Database>().expect("`Database`");
let token = request
.headers()
.get("x-admin-user")
.next()
.map(|x| x.to_string());
if let Some(token) = token {
if let Ok(admin_token) = db.admin_token_authenticate(&token).await {
if let Some(expiry) = Timestamp::parse(&admin_token.expiry) {
if expiry < Timestamp::now_utc() {
if let Err(e) = db.admin_token_revoke(&admin_token.id).await {
capture_internal_error!(e);
}
return None;
} else if let Ok(user) =
db.admin_user_fetch(&admin_token.user_id).await
{
if !user.active {
return None;
}
return Some(user);
}
}
}
}
None
})
.await;
if let Some(user) = user {
if !user.active {
Outcome::Error((Status::Unauthorized, create_error!(LockedOut)))
} else {
Outcome::Success(AdminAuthorization::AdminUser(user.clone()))
}
} else {
Outcome::Error((Status::Unauthorized, create_error!(InvalidSession)))
}
}
}
}
@@ -0,0 +1,59 @@
use revolt_okapi::openapi3::{SecurityScheme, SecuritySchemeData};
use revolt_rocket_okapi::{
gen::OpenApiGenerator,
request::{OpenApiFromRequest, RequestHeaderInput},
};
use crate::{AdminAuthorization, AdminMachineToken};
impl OpenApiFromRequest<'_> for AdminAuthorization {
fn from_request_input(
_gen: &mut OpenApiGenerator,
_name: String,
_required: bool,
) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
let mut requirements = schemars::Map::new();
requirements.insert("Admin Token".to_owned(), vec![]);
Ok(RequestHeaderInput::Security(
"Admin Token".to_owned(),
SecurityScheme {
data: SecuritySchemeData::ApiKey {
name: "x-admin-user".to_owned(),
location: "header".to_owned(),
},
description: Some("Used to authenticate as an admin user.
Can instead use an x-admin-machine token with an x-on-behalf-of containing the userid or email of
the user the machine is performing the action for.".to_owned()),
extensions: schemars::Map::new(),
},
requirements,
))
}
}
impl OpenApiFromRequest<'_> for AdminMachineToken {
fn from_request_input(
_gen: &mut OpenApiGenerator,
_name: String,
_required: bool,
) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
let mut requirements = schemars::Map::new();
requirements.insert("Admin Machine Token".to_owned(), vec![]);
Ok(RequestHeaderInput::Security(
"Admin Machine Token".to_owned(),
SecurityScheme {
data: SecuritySchemeData::ApiKey {
name: "x-admin-machine".to_owned(),
location: "header".to_owned(),
},
description: Some("Used by machines to authenticate on behalf of a user.
Machines are trusted devices that authenticate as other users via providing the x-on-behalf-of header
with an admin user's ID or email.".to_owned()),
extensions: schemars::Map::new(),
},
requirements,
))
}
}
@@ -0,0 +1,9 @@
mod models;
mod ops;
#[cfg(feature = "rocket-impl")]
mod rocket;
#[cfg(feature = "rocket-impl")]
pub use self::rocket::*;
pub use models::*;
pub use ops::*;
@@ -0,0 +1,69 @@
use revolt_result::Result;
use crate::Database;
auto_derived_partial! {
pub struct AdminUser {
/// The ID of the user
#[serde(rename = "_id")]
pub id: String,
/// The user's revolt ID.
pub platform_user_id: String,
/// The user's email
pub email: String,
/// Whether the user is active or not (ie. can they use the api)
pub active: bool,
/// The permissions of the user
pub permissions: u64,
},
"PartialAdminUser"
}
impl AdminUser {
pub fn new(email: &str, platform_user_id: &str, permissions: u64) -> AdminUser {
let id = ulid::Ulid::new().to_string();
AdminUser {
id,
platform_user_id: platform_user_id.to_string(),
email: email.to_string(),
active: true,
permissions,
}
}
pub async fn find_by_id(id: &str, db: &Database) -> Result<AdminUser> {
return db.admin_user_fetch(id).await;
}
pub async fn find_by_email(email: &str, db: &Database) -> Result<AdminUser> {
return db.admin_user_fetch_email(email).await;
}
}
pub struct AdminUserPermissionFlagsValue(pub u64);
impl AdminUserPermissionFlagsValue {
pub fn has(&self, flag: revolt_models::v0::AdminUserPermissionFlags) -> bool {
self.has_value(flag as u64)
}
pub fn has_value(&self, bit: u64) -> bool {
let mask = 1 << bit;
self.0 & mask == mask
}
pub fn set(
&mut self,
flag: revolt_models::v0::AdminUserPermissionFlags,
toggle: bool,
) -> &mut Self {
self.set_value(flag as u64, toggle)
}
pub fn set_value(&mut self, bit: u64, toggle: bool) -> &mut Self {
if toggle {
self.0 |= 1 << bit;
} else {
self.0 &= !(1 << bit);
}
self
}
}
@@ -0,0 +1,20 @@
mod mongodb;
mod reference;
use revolt_result::Result;
use crate::models::admin_users::models::{AdminUser, PartialAdminUser};
#[async_trait]
pub trait AbstractAdminUsers: Sync + Send {
async fn admin_user_insert(&self, user: AdminUser) -> Result<()>;
async fn admin_user_update(&self, user_id: &str, partial: PartialAdminUser) -> Result<()>;
async fn admin_user_fetch(&self, user_id: &str) -> Result<AdminUser>;
async fn admin_user_fetch_email(&self, email: &str) -> Result<AdminUser>;
async fn admin_user_list(&self) -> Result<Vec<AdminUser>>;
async fn admin_user_count(&self) -> Result<u16>;
}
@@ -0,0 +1,35 @@
use revolt_result::Result;
use crate::MongoDb;
use crate::{AdminUser, PartialAdminUser};
use super::AbstractAdminUsers;
static COL: &str = "admin_users";
#[async_trait]
impl AbstractAdminUsers for MongoDb {
async fn admin_user_insert(&self, user: AdminUser) -> Result<()> {
query!(self, insert_one, COL, user).map(|_| ())
}
async fn admin_user_update(&self, user_id: &str, partial: PartialAdminUser) -> Result<()> {
query!(self, update_one_by_id, COL, user_id, partial, vec![], None).map(|_| ())
}
async fn admin_user_fetch(&self, user_id: &str) -> Result<AdminUser> {
query!(self, find_one_by_id, COL, user_id)?.ok_or_else(|| create_error!(NotFound))
}
async fn admin_user_fetch_email(&self, email: &str) -> Result<AdminUser> {
query!(self, find_one, COL, doc! {"email": email})?.ok_or_else(|| create_error!(NotFound))
}
async fn admin_user_list(&self) -> Result<Vec<AdminUser>> {
query!(self, find, COL, doc! {})
}
async fn admin_user_count(&self) -> Result<u16> {
query!(self, count_documents, COL, doc! {}).map(|resp| resp as u16)
}
}
@@ -0,0 +1,56 @@
use revolt_result::Result;
use crate::ReferenceDb;
use crate::{AdminUser, PartialAdminUser};
use super::AbstractAdminUsers;
#[async_trait]
impl AbstractAdminUsers for ReferenceDb {
async fn admin_user_insert(&self, user: AdminUser) -> Result<()> {
let mut admin_users = self.admin_users.lock().await;
if admin_users.contains_key(&user.id) {
Err(create_database_error!("insert", "admin_users"))
} else {
admin_users.insert(user.id.to_string(), user.clone());
Ok(())
}
}
async fn admin_user_update(&self, user_id: &str, partial: PartialAdminUser) -> Result<()> {
let mut admin_users = self.admin_users.lock().await;
if let Some(existing_user) = admin_users.get_mut(user_id) {
existing_user.apply_options(partial);
Ok(())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_user_fetch(&self, user_id: &str) -> Result<AdminUser> {
let admin_users = self.admin_users.lock().await;
if let Some(user) = admin_users.get(user_id) {
Ok(user.clone())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_user_fetch_email(&self, email: &str) -> Result<AdminUser> {
let admin_users = self.admin_users.lock().await;
if let Some((_, user)) = admin_users.iter().filter(|(_, p)| p.email == email).next() {
Ok(user.clone())
} else {
Err(create_error!(NotFound))
}
}
async fn admin_user_list(&self) -> Result<Vec<AdminUser>> {
let admin_users = self.admin_users.lock().await;
Ok(admin_users.iter().map(|(_, u)| u).cloned().collect())
}
async fn admin_user_count(&self) -> Result<u16> {
Ok(self.admin_users.lock().await.len() as u16)
}
}
@@ -0,0 +1,43 @@
use rocket::http::Status;
use rocket::request::{self, FromRequest, Outcome, Request};
use revolt_result::Error;
use crate::{AdminUser, Database};
#[rocket::async_trait]
impl<'r> FromRequest<'r> for AdminUser {
type Error = Error;
async fn from_request(request: &'r Request<'_>) -> request::Outcome<Self, Self::Error> {
let user: &Option<AdminUser> = request
.local_cache_async(async {
let db = request.rocket().state::<Database>().expect("`Database`");
let token = request
.headers()
.get("x-admin-user")
.next()
.map(|x| x.to_string());
if let Some(token) = token {
if let Ok(admin_token) = db.admin_token_authenticate(&token).await {
if let Ok(user) = db.admin_user_fetch(&admin_token.user_id).await {
return Some(user);
}
}
}
None
})
.await;
if let Some(user) = user {
if !user.active {
Outcome::Error((Status::Unauthorized, create_error!(LockedOut)))
} else {
Outcome::Success(user.clone())
}
} else {
Outcome::Error((Status::Unauthorized, create_error!(InvalidCredentials)))
}
}
}
@@ -12,7 +12,14 @@ static COL: &str = "channel_invites";
impl AbstractChannelInvites for MongoDb {
/// Insert a new invite into the database
async fn insert_invite(&self, invite: &Invite) -> Result<()> {
query!(self, insert_one, COL, &invite).map(|_| ())
self.insert_one(COL, &invite).await.or_else(|e| {
if e.to_string().contains("duplicate") {
return Err(create_error!(InviteExists));
} else {
return Err(create_database_error!("insert_one", COL));
}
})?;
Ok(())
}
/// Fetch an invite by the code
@@ -340,7 +340,7 @@ impl Channel {
pub async fn add_user_to_group(
&mut self,
db: &Database,
amqp: Option<&AMQP>,
amqp: &AMQP,
user: &User,
by_id: &str,
) -> Result<()> {
@@ -377,7 +377,7 @@ impl Channel {
.into_message(id.to_string())
.send(
db,
amqp,
Some(amqp),
MessageAuthor::System {
username: &user.username,
avatar: user.avatar.as_ref().map(|file| file.id.as_ref()),
@@ -665,7 +665,7 @@ impl Channel {
pub async fn remove_user_from_group(
&self,
db: &Database,
amqp: Option<&AMQP>,
amqp: &AMQP,
user: &User,
by_id: Option<&str>,
silent: bool,
@@ -697,7 +697,7 @@ impl Channel {
.into_message(id.to_string())
.send(
db,
amqp,
Some(amqp),
MessageAuthor::System {
username: name,
avatar: None,
@@ -710,7 +710,7 @@ impl Channel {
.await
.ok();
} else {
return self.delete(db, amqp).await;
return self.delete(db).await;
}
}
@@ -737,7 +737,7 @@ impl Channel {
.into_message(id.to_string())
.send(
db,
amqp,
Some(amqp),
MessageAuthor::System {
username: &user.username,
avatar: user.avatar.as_ref().map(|file| file.id.as_ref()),
@@ -759,22 +759,20 @@ impl Channel {
}
/// Delete a channel
pub async fn delete(&self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
pub async fn delete(&self, db: &Database) -> 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?;
if let Some(amqp) = amqp {
if let Err(e) = amqp.delete_channel_search(self.id().to_string()).await {
log::error!("Error pushing message to RabbitMQ: {e}");
revolt_config::capture_error(&e);
}
}
Ok(())
db.delete_channel(self).await
}
/// Wipe a channel's messages
pub async fn wipe(&self, db: &Database) -> Result<()> {
let id = self.id().to_string();
EventV1::ChannelWipe { id: id.clone() }.p(id).await;
db.wipe_channel(self).await
}
}
@@ -54,5 +54,6 @@ pub trait AbstractChannels: Sync + Send {
async fn remove_user_from_groups(&self, channel_ids: Vec<String>, user_id: &str) -> Result<()>;
// Delete a channel
async fn delete_channel(&self, channel_id: &Channel) -> Result<()>;
async fn delete_channel(&self, channel_id: &Channel) -> Result<()>; // Wipe a channel's messages
async fn wipe_channel(&self, channel: &Channel) -> Result<()>;
}
@@ -304,6 +304,23 @@ impl AbstractChannels for MongoDb {
// Delete the channel itself
query!(self, delete_one_by_id, COL, channel.id()).map(|_| ())
}
// Wipe a channel's messages
async fn wipe_channel(&self, channel: &Channel) -> Result<()> {
let id = channel.id().to_string();
// Delete invites and unreads.
self.delete_associated_channel_objects(Bson::String(id.to_string()))
.await?;
// Delete messages.
self.delete_bulk_messages(doc! {
"channel": &id
})
.await?;
Ok(())
}
}
impl MongoDb {
@@ -182,4 +182,10 @@ impl AbstractChannels for ReferenceDb {
Err(create_error!(NotFound))
}
}
async fn wipe_channel(&self, channel: &Channel) -> Result<()> {
let mut messages = self.messages.lock().await;
messages.retain(|_, message| message.channel != channel.id());
Ok(())
}
}
+86 -164
View File
@@ -1,6 +1,6 @@
use indexmap::{IndexMap, IndexSet};
use iso8601_timestamp::Timestamp;
use revolt_config::{capture_error, config, FeaturesLimits};
use revolt_config::{config, FeaturesLimits};
use revolt_models::v0::{
self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort,
MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text,
@@ -18,7 +18,7 @@ use crate::{
bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
permissions::DatabasePermissionQuery,
},
Channel, Database, Emoji, File, Member, User, AMQP,
Channel, Database, Emoji, File, User, AMQP,
};
#[cfg(feature = "tasks")]
@@ -210,13 +210,6 @@ auto_derived!(
pub enum FieldsMessage {
Pinned,
}
/// Message along with the user for the author
pub struct MessageWithUser {
#[serde(flatten)]
pub message: Message,
pub user: Option<User>,
}
);
pub struct MessageFlagsValue(pub u32);
@@ -279,8 +272,8 @@ impl Message {
channel: Channel,
data: DataMessageSend,
author: MessageAuthor<'_>,
user: Option<User>,
member: Option<Member>,
user: Option<v0::User>,
member: Option<v0::Member>,
limits: FeaturesLimits,
mut idempotency: IdempotencyKey,
generate_embeds: bool,
@@ -617,10 +610,10 @@ impl Message {
/// Send a message without any notifications
pub async fn send_without_notifications(
&self,
&mut self,
db: &Database,
user: Option<User>,
member: Option<Member>,
user: Option<v0::User>,
member: Option<v0::Member>,
is_dm: bool,
generate_embeds: bool,
// This determines if this function should queue the mentions task or if somewhere else will.
@@ -630,18 +623,9 @@ impl Message {
db.insert_message(self).await?;
// Fan out events
EventV1::Message(self.clone().into_model(
match user {
Some(user) => {
let is_online = revolt_presence::is_online(&user.id).await;
Some(user.into_known_static(is_online).await)
}
None => None,
},
member.map(Into::into),
))
.p(self.channel.to_string())
.await;
EventV1::Message(self.clone().into_model(user, member))
.p(self.channel.to_string())
.await;
// Update last_message_id
#[cfg(feature = "tasks")]
@@ -687,10 +671,10 @@ impl Message {
pub async fn send(
&mut self,
db: &Database,
amqp: Option<&AMQP>, // this is optional mostly for tests.
_amqp: Option<&AMQP>, // this is optional mostly for tests.
author: MessageAuthor<'_>,
user: Option<User>,
member: Option<Member>,
user: Option<v0::User>,
member: Option<v0::Member>,
channel: &Channel,
generate_embeds: bool,
) -> Result<()> {
@@ -720,17 +704,7 @@ impl Message {
messages: vec![(
Some(
PushNotification::from(
self.clone().into_model(
match user.clone() {
Some(user) => {
let is_online =
revolt_presence::is_online(&user.id).await;
Some(user.into_known_static(is_online).await)
}
None => None,
},
member.map(Into::into),
),
self.clone().into_model(user, member),
Some(author.clone()),
channel.to_owned().into(),
)
@@ -756,13 +730,6 @@ impl Message {
.await;
}
if let Some(amqp) = amqp {
if let Err(e) = amqp.new_message_search(self.clone(), user).await {
log::error!("Error pushing message to RabbitMQ: {e}");
capture_error(&e);
}
}
Ok(())
}
@@ -815,7 +782,6 @@ impl Message {
pub async fn update(
&mut self,
db: &Database,
amqp: Option<&AMQP>,
partial: PartialMessage,
remove: Vec<FieldsMessage>,
) -> Result<()> {
@@ -837,74 +803,9 @@ impl Message {
.p(self.channel.clone())
.await;
if let Some(amqp) = amqp {
if let Err(e) = amqp
.edit_message_search(self.clone(), self.fetch_author(db).await)
.await
{
log::error!("Error pushing message to RabbitMQ: {e}");
capture_error(&e);
}
}
Ok(())
}
pub async fn fetch_users(
db: &Database,
messages: &[Message],
server_id: Option<&str>,
) -> Result<(Vec<User>, Vec<Member>)> {
let user_ids = messages
.iter()
.flat_map(|m| {
let mut users = vec![m.author.clone()];
if let Some(system) = &m.system {
match system {
SystemMessage::ChannelDescriptionChanged { by } => users.push(by.clone()),
SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
users.push(from.clone());
users.push(to.clone())
}
SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
SystemMessage::UserAdded { by, id, .. }
| SystemMessage::UserRemove { by, id, .. } => {
users.push(by.clone());
users.push(id.clone());
}
SystemMessage::UserBanned { id, .. }
| SystemMessage::UserKicked { id, .. }
| SystemMessage::UserJoined { id, .. }
| SystemMessage::UserLeft { id, .. } => {
users.push(id.clone());
}
SystemMessage::Text { .. } => {}
SystemMessage::MessagePinned { by, .. } => {
users.push(by.clone());
}
SystemMessage::MessageUnpinned { by, .. } => {
users.push(by.clone());
}
SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
}
}
users
})
.collect::<HashSet<String>>()
.into_iter()
.collect::<Vec<String>>();
let users = db.fetch_users(&user_ids).await?;
let members = if let Some(server_id) = server_id {
db.fetch_members(server_id, &user_ids).await?
} else {
Vec::new()
};
Ok((users, members))
}
/// Helper function to fetch many messages with users
pub async fn fetch_with_users(
db: &Database,
@@ -913,55 +814,93 @@ impl Message {
include_users: Option<bool>,
server_id: Option<&str>,
) -> Result<BulkMessageResponse> {
let messages = db.fetch_messages(query).await?;
let messages: Vec<v0::Message> = db
.fetch_messages(query)
.await?
.into_iter()
.map(|msg| msg.into_model(None, None))
.collect();
if let Some(true) = include_users {
let (users, members) = Message::fetch_users(db, &messages, server_id).await?;
let user_ids = messages
.iter()
.flat_map(|m| {
let mut users = vec![m.author.clone()];
if let Some(system) = &m.system {
match system {
v0::SystemMessage::ChannelDescriptionChanged { by } => {
users.push(by.clone())
}
v0::SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
v0::SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
users.push(from.clone());
users.push(to.clone())
}
v0::SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
v0::SystemMessage::UserAdded { by, id, .. }
| v0::SystemMessage::UserRemove { by, id, .. } => {
users.push(by.clone());
users.push(id.clone());
}
v0::SystemMessage::UserBanned { id, .. }
| v0::SystemMessage::UserKicked { id, .. }
| v0::SystemMessage::UserJoined { id, .. }
| v0::SystemMessage::UserLeft { id, .. } => {
users.push(id.clone());
}
v0::SystemMessage::Text { .. } => {}
v0::SystemMessage::MessagePinned { by, .. } => {
users.push(by.clone());
}
v0::SystemMessage::MessageUnpinned { by, .. } => {
users.push(by.clone());
}
v0::SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
}
}
users
})
.collect::<HashSet<String>>()
.into_iter()
.collect::<Vec<String>>();
let users = User::fetch_many_ids_as_mutuals(db, perspective, &user_ids).await?;
Ok(BulkMessageResponse::MessagesAndUsers {
messages: messages
.into_iter()
.map(|msg| msg.into_model(None, None))
.collect(),
users: User::into_mutuals(perspective, users).await,
members: Some(members.into_iter().map(Into::into).collect()),
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 {
Ok(BulkMessageResponse::JustMessages(
messages
.into_iter()
.map(|msg| msg.into_model(None, None))
.collect(),
))
Ok(BulkMessageResponse::JustMessages(messages))
}
}
/// Append content to message
pub async fn append(
db: &Database,
amqp: Option<&AMQP>,
id: String,
channel: String,
append: AppendMessage,
) -> Result<()> {
if let Some(message) = db.append_message(&id, &append).await? {
if let Some(amqp) = amqp {
let author = message.fetch_author(db).await;
db.append_message(&id, &append).await?;
if let Err(e) = amqp.edit_message_search(message, author).await {
log::error!("Error pushing message to RabbitMQ: {e}");
capture_error(&e);
}
}
EventV1::MessageAppend {
id,
channel: channel.to_string(),
append: append.into(),
}
.p(channel)
.await;
EventV1::MessageAppend {
id,
channel: channel.to_string(),
append: append.into(),
}
.p(channel)
.await;
Ok(())
}
@@ -1059,10 +998,9 @@ impl Message {
}
/// Delete a message
pub async fn delete(&self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
pub async fn delete(self, db: &Database) -> Result<()> {
let file_ids: Vec<String> = self
.attachments
.as_ref()
.map(|files| files.iter().map(|file| file.id.to_string()).collect())
.unwrap_or_default();
@@ -1072,20 +1010,12 @@ 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.clone(),
id: self.id,
channel: self.channel.clone(),
}
.p(self.channel.clone())
.p(self.channel)
.await;
Ok(())
}
@@ -1186,14 +1116,6 @@ impl Message {
FieldsMessage::Pinned => self.pinned = None,
}
}
pub async fn fetch_author(&self, db: &Database) -> Option<User> {
if self.webhook.is_some() {
None
} else {
db.fetch_user(&self.author).await.ok()
}
}
}
impl SystemMessage {
@@ -1,11 +1,8 @@
use revolt_result::Result;
use std::collections::HashMap;
use std::time::SystemTime;
use revolt_result::Result;
use crate::{
util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
MessageWithUser, PartialMessage,
};
use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
#[cfg(feature = "mongodb")]
mod mongodb;
@@ -26,15 +23,10 @@ pub trait AbstractMessages: Sync + Send {
async fn fetch_messages_by_id(&self, ids: &[String]) -> Result<Vec<Message>>;
/// Update a given message with new information
async fn update_message(
&self,
id: &str,
message: &PartialMessage,
remove: Vec<FieldsMessage>,
) -> Result<()>;
async fn update_message(&self, id: &str, message: &PartialMessage, remove: Vec<FieldsMessage>) -> Result<()>;
/// Append information to a given message
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>>;
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()>;
/// Add a new reaction to a message
async fn add_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()>;
@@ -56,11 +48,8 @@ pub trait AbstractMessages: Sync + Send {
&self,
channels: &[String],
author: &str,
since: SystemTime,
since: SystemTime
) -> Result<HashMap<String, Vec<String>>>;
/// Fetches all messages along with their author from every message in decending order
async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>>;
async fn delete_messages_by_user(&self, user_id: &str) -> Result<()>;
}
@@ -1,7 +1,7 @@
use bson::{to_bson, Document};
use futures::try_join;
use futures::StreamExt;
use mongodb::options::{FindOptions, ReadConcern, ReturnDocument};
use mongodb::options::FindOptions;
use revolt_models::v0::MessageSort;
use revolt_result::Result;
use std::collections::{HashMap, HashSet};
@@ -9,8 +9,8 @@ use std::time::SystemTime;
use ulid::Ulid;
use crate::{
util::ChunkedDatabaseGenerator, AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath,
Message, MessageQuery, MessageTimePeriod, MessageWithUser, MongoDb, PartialMessage,
AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath, Message, MessageQuery,
MessageTimePeriod, MongoDb, PartialMessage,
};
use super::AbstractMessages;
@@ -205,7 +205,7 @@ impl AbstractMessages for MongoDb {
}
/// Append information to a given message
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>> {
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
let mut query = doc! {};
if let Some(embeds) = &append.embeds {
@@ -223,18 +223,18 @@ impl AbstractMessages for MongoDb {
}
if query.is_empty() {
return Ok(None);
return Ok(());
}
self.col::<Message>(COL)
.find_one_and_update(
self.col::<Document>(COL)
.update_one(
doc! {
"_id": id
},
query,
)
.return_document(ReturnDocument::After)
.await
.map(|_| ())
.map_err(|_| create_database_error!("update_one", COL))
}
@@ -417,53 +417,6 @@ impl AbstractMessages for MongoDb {
Ok(deleted_messages)
}
/// Fetches all messages along with their author from every message in decending order
async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>> {
let mut session = self
.start_session()
.await
.map_err(|_| create_database_error!("start_session", COL))?;
session
.start_transaction()
.read_concern(ReadConcern::snapshot())
.await
.map_err(|_| create_database_error!("start_transaction", COL))?;
let cursor = self
.col::<Message>(COL)
.aggregate([
doc! {
"$lookup": {
"from": "users",
"localField": "author",
"foreignField": "_id",
"as": "user"
}
},
doc! {
"$set": {
"user": {
"$first": "$user"
}
}
},
doc! {
"$sort": {
"_id": -1
}
},
])
.with_type::<MessageWithUser>()
.session(&mut session)
.batch_size(1000)
.await
.inspect_err(|e| log::error!("{e}"))
.map_err(|_| create_database_error!("aggregate", COL))?;
Ok(ChunkedDatabaseGenerator::new_mongo(session, cursor))
}
async fn delete_messages_by_user(&self, user_id: &str) -> Result<()> {
self.delete_bulk_messages(doc! {
"author": user_id,
@@ -1,6 +1,6 @@
use crate::{
util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
MessageWithUser, PartialMessage, ReferenceDb,
AppendMessage, FieldsMessage, Message, MessageQuery,
PartialMessage, ReferenceDb,
};
use futures::future::try_join_all;
use indexmap::IndexSet;
@@ -215,7 +215,7 @@ impl AbstractMessages for ReferenceDb {
}
/// Append information to a given message
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>> {
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
let mut messages = self.messages.lock().await;
if let Some(message_data) = messages.get_mut(id) {
if let Some(embeds) = &append.embeds {
@@ -226,11 +226,9 @@ impl AbstractMessages for ReferenceDb {
message_data.embeds = Some(embeds.clone());
}
}
Ok(Some(message_data.clone()))
} else {
Ok(None)
}
Ok(())
} else {
Err(create_error!(NotFound))
}
@@ -355,24 +353,6 @@ impl AbstractMessages for ReferenceDb {
Ok(deleted_messages)
}
/// Fetches all messages along with their author from every message in decending order
async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>> {
let users = self.users.lock().await;
Ok(ChunkedDatabaseGenerator::new_reference(
self.messages
.lock()
.await
.values()
.cloned()
.map(|message| MessageWithUser {
user: users.get(&message.author).cloned(),
message,
})
.collect(),
))
}
async fn delete_messages_by_user(&self, user_id: &str) -> Result<()> {
let mut messages = self.messages.lock().await;
+18
View File
@@ -1,4 +1,10 @@
mod admin_audits;
mod admin_cases;
mod admin_comments;
mod admin_migrations;
mod admin_strikes;
mod admin_tokens;
mod admin_users;
mod bots;
mod channel_invites;
mod channel_unreads;
@@ -22,7 +28,13 @@ mod account_invites;
mod sessions;
mod mfa_tickets;
pub use admin_audits::*;
pub use admin_cases::*;
pub use admin_comments::*;
pub use admin_migrations::*;
pub use admin_strikes::*;
pub use admin_tokens::*;
pub use admin_users::*;
pub use bots::*;
pub use channel_invites::*;
pub use channel_unreads::*;
@@ -54,6 +66,12 @@ use crate::MongoDb;
pub trait AbstractDatabase:
Sync
+ Send
+ admin_audits::AbstractAdminAudits
+ admin_cases::AbstractAdminCases
+ admin_comments::AbstractAdminComments
+ admin_strikes::AbstractAdminStrikes
+ admin_users::AbstractAdminUsers
+ admin_tokens::AbstractAdminTokens
+ admin_migrations::AbstractMigrations
+ bots::AbstractBots
+ channels::AbstractChannels
@@ -18,5 +18,7 @@ auto_derived!(
/// Additional notes included on the report
#[serde(default)]
pub notes: String,
/// The case this report is assigned to
pub case_id: Option<String>,
}
);
@@ -110,6 +110,7 @@ impl Member {
server: &Server,
user: &User,
channels: Option<Vec<Channel>>,
create_alert: bool,
) -> Result<(Member, Vec<Channel>)> {
if db.fetch_ban(&server.id, &user.id).await.is_ok() {
return Err(create_error!(Banned));
@@ -188,18 +189,20 @@ impl Member {
.private(user.id.clone())
.await;
if let Some(id) = server
.system_messages
.as_ref()
.and_then(|x| x.user_joined.as_ref())
{
SystemMessage::UserJoined {
id: user.id.clone(),
if create_alert {
if let Some(id) = server
.system_messages
.as_ref()
.and_then(|x| x.user_joined.as_ref())
{
SystemMessage::UserJoined {
id: user.id.clone(),
}
.into_message(id.to_string())
.send_without_notifications(db, None, None, false, false, false)
.await
.ok();
}
.into_message(id.to_string())
.send_without_notifications(db, None, None, false, false, false)
.await
.ok();
}
Ok((member, channels))
@@ -349,8 +352,10 @@ mod tests {
.unwrap()
.0;
Member::create(&db, &server, &owner, None).await.unwrap();
let mut kickable_member = Member::create(&db, &server, &kickable_user, None)
Member::create(&db, &server, &owner, None, false)
.await
.unwrap();
let mut kickable_member = Member::create(&db, &server, &kickable_user, None, false)
.await
.unwrap()
.0;
@@ -374,7 +379,7 @@ mod tests {
.await
.unwrap();
let kickable_member = Member::create(&db, &server, &kickable_user, None)
let kickable_member = Member::create(&db, &server, &kickable_user, None, false)
.await
.unwrap()
.0;
@@ -132,4 +132,18 @@ pub trait AbstractServerMembers: Sync + Send {
/// Removes a user from every server they are in
async fn clear_memberships(&self, user_id: &str) -> Result<()>;
/// Fetch all participants of a server.
async fn fetch_server_participants(
&self,
server_id: &str,
) -> Result<Vec<(crate::User, Option<Member>)>>;
/// Fetch all members of a server
async fn fetch_server_members(
&self,
server_id: &str,
page_size: u8,
after: Option<usize>,
) -> Result<Vec<(crate::User, Member)>>;
}
@@ -1,10 +1,12 @@
use bson::Document;
use std::collections::HashMap;
use futures::StreamExt;
use iso8601_timestamp::Timestamp;
use mongodb::options::ReadConcern;
use revolt_result::Result;
use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
use crate::{AbstractUsers, FieldsMember, Member, MemberCompositeKey, PartialMember};
use crate::{IntoDocumentPath, MongoDb};
use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
@@ -337,6 +339,88 @@ impl AbstractServerMembers for MongoDb {
.map(|_| ())
.map_err(|_| create_database_error!("delete_many", COL))
}
async fn fetch_server_participants(
&self,
server_id: &str,
) -> Result<Vec<(crate::User, Option<Member>)>> {
let server: crate::Server = query!(self, find_one, "servers", doc! {"_id": server_id})?
.ok_or_else(|| create_database_error!("find", "servers"))?;
let ids: Vec<String> = self
.db()
.collection::<crate::Message>("messages")
.distinct("author", doc! {"channel": {"$in": server.channels}})
.await
.map_err(|_| create_database_error!("distinct", "messages"))?
.iter()
.map(|b| bson::from_bson::<String>(b.clone()).unwrap()) // json encoded string for some logic-defying reason
.collect();
println!("{:?}", &ids);
let users = self.fetch_users(&ids).await?;
let members = self
.fetch_members(server_id, &ids)
.await?
.iter()
.map(|f| (f.id.user.clone(), f.clone()))
.collect::<HashMap<String, Member>>();
Ok(users
.iter()
.map(|u| (u.clone(), members.get(&u.id).cloned()))
.collect())
}
async fn fetch_server_members(
&self,
server_id: &str,
page_size: u8,
after: Option<usize>,
) -> Result<Vec<(crate::User, Member)>> {
let mut members_stream = if let Some(after) = after {
self.col::<crate::Member>(COL)
.find(doc! {"_id.server": server_id, "joined_at": {"$gt": after as i64}})
.sort(doc! {"joined_at": -1})
.limit(page_size as i64)
.await
.map_err(|_| create_database_error!("find", COL))?
} else {
self.col::<crate::Member>(COL)
.find(doc! {"_id.server": server_id})
.sort(doc! {"joined_at": -1})
.limit(page_size as i64)
.await
.map_err(|_| create_database_error!("find", COL))?
};
let mut members = vec![];
if members_stream.advance().await.is_ok_and(|f| f) {
loop {
if let Ok(x) = members_stream.deserialize_current() {
members.push(x);
if let Ok(r) = members_stream.advance().await {
if !r {
break;
}
} else {
break;
}
}
}
}
let user_ids: Vec<String> = members.iter().map(|m| m.id.user.clone()).collect();
let mut set = HashMap::new();
self.fetch_users(&user_ids).await?.iter().for_each(|f| {
set.insert(f.id.clone(), f.clone());
});
return Ok(members
.iter()
.map(|f| (set.remove(&f.id.user).unwrap(), f.clone()))
.collect());
}
}
impl IntoDocumentPath for FieldsMember {
@@ -1,3 +1,6 @@
use std::collections::HashSet;
use iso8601_timestamp::Timestamp;
use revolt_result::Result;
use crate::ReferenceDb;
@@ -209,4 +212,91 @@ impl AbstractServerMembers for ReferenceDb {
Ok(())
}
async fn fetch_server_participants(
&self,
server_id: &str,
) -> Result<Vec<(crate::User, Option<Member>)>> {
let servers = self.servers.lock().await;
let server = servers.get(server_id).ok_or(create_error!(NotFound))?;
let mut hash = HashSet::new();
server.channels.iter().for_each(|c| {
hash.insert(c.clone());
});
drop(servers);
let messages = self.messages.lock().await;
let userids = messages
.iter()
.filter(|(_, message)| hash.contains(&message.channel))
.map(|(_, message)| message.author.clone())
.collect::<HashSet<String>>();
drop(messages);
let users = self.users.lock().await;
let members = self.server_members.lock().await;
let resp_users: Vec<(crate::User, Option<crate::Member>)> = users
.iter()
.filter(|(uid, _)| userids.contains(*uid))
.map(|(uid, user)| {
let key = MemberCompositeKey {
server: server_id.to_string(),
user: uid.clone(),
};
// fuck it we ball
(user.clone(), members.get(&key).map(|f| f.clone()))
})
.collect();
Ok(resp_users)
}
/// Fetch all members of a server
async fn fetch_server_members(
&self,
server_id: &str,
page_size: u8,
after: Option<usize>,
) -> Result<Vec<(crate::User, Member)>> {
let members = self.server_members.lock().await;
let users = self.users.lock().await;
let mut server_members: Vec<&Member> = members
.iter()
.filter(|(_, f)| f.id.server == server_id)
.map(|(_, m)| m)
.collect();
server_members.sort_by(|a, b| a.joined_at.cmp(&b.joined_at));
let iterator = server_members.iter();
let mut resp: Vec<&&Member> = if let Some(after) = after {
iterator
.skip_while(|f| {
f.joined_at
.duration_since(Timestamp::UNIX_EPOCH)
.whole_milliseconds()
<= after as i128
})
.collect()
} else {
iterator.collect()
};
resp.truncate(page_size as usize);
Ok(resp
.iter()
.map(|f| {
(
users.get(&f.id.user).unwrap().clone(),
f.clone().clone().clone(),
)
})
.collect())
}
}
@@ -1,12 +1,11 @@
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::{AMQP, Channel, Database, File, User, events::client::EventV1};
use crate::{events::client::EventV1, Channel, Database, File, User};
auto_derived_partial!(
/// Server
@@ -215,16 +214,7 @@ impl Server {
}
/// Delete a server
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);
}
}
}
pub async fn delete(self, db: &Database) -> Result<()> {
EventV1::ServerDelete {
id: self.id.clone(),
}
+30 -21
View File
@@ -381,17 +381,6 @@ impl User {
)
}
pub async fn into_mutuals(perspective: &User, users: Vec<User>) -> Vec<v0::User> {
let online_ids =
filter_online(&users.iter().map(|user| user.id.clone()).collect::<Vec<_>>()).await;
join_all(users.into_iter().map(|user| async {
let is_online = online_ids.contains(&user.id);
user.into_known(perspective, is_online).await
}))
.await
}
/// Find a free discriminator for a given username
pub async fn find_discriminator(
db: &Database,
@@ -542,7 +531,7 @@ impl User {
pub async fn add_friend(
&mut self,
db: &Database,
amqp: Option<&AMQP>,
amqp: &AMQP,
target: &mut User,
) -> Result<()> {
match self.relationship_with(&target.id) {
@@ -552,10 +541,8 @@ impl User {
RelationshipStatus::Blocked => Err(create_error!(Blocked)),
RelationshipStatus::BlockedOther => Err(create_error!(BlockedByOther)),
RelationshipStatus::Incoming => {
if let Some(amqp) = amqp {
// Accept incoming friend request
_ = amqp.friend_request_accepted(self, target).await;
};
// Accept incoming friend request
_ = amqp.friend_request_accepted(self, target).await;
self.apply_relationship(
db,
@@ -585,9 +572,7 @@ impl User {
}));
}
if let Some(amqp) = amqp {
_ = amqp.friend_request_received(target, self).await;
};
_ = amqp.friend_request_received(target, self).await;
// Send the friend request
self.apply_relationship(
@@ -882,13 +867,13 @@ impl User {
/// - deletes owned bots, servers and messages
/// - removes user from all groups
/// - clears relationships
pub async fn delete(&mut self, db: &Database, amqp: Option<&AMQP>) -> Result<()> {
pub async fn delete(&mut self, db: &Database) -> Result<()> {
for bot in db.fetch_bots_by_user(&self.id).await? {
bot.delete(db).await?;
}
for server in db.fetch_owned_servers(&self.id).await? {
server.delete(db, amqp).await?;
server.delete(db).await?;
}
self.remove_from_all_groups(db).await?;
@@ -901,6 +886,30 @@ impl User {
}
}
pub struct UserFlagsValue(pub u32);
impl UserFlagsValue {
pub fn has(&self, flag: UserFlags) -> bool {
self.has_value(flag as u32)
}
pub fn has_value(&self, bit: u32) -> bool {
let mask = 1 << bit;
self.0 & mask == mask
}
pub fn set(&mut self, flag: UserFlags, toggle: bool) -> &mut Self {
self.set_value(flag as u32, toggle)
}
pub fn set_value(&mut self, bit: u32, toggle: bool) -> &mut Self {
if toggle {
self.0 |= 1 << bit;
} else {
self.0 &= !(1 << bit);
}
self
}
}
#[cfg(test)]
mod tests {
use crate::User;
+1 -1
View File
@@ -16,7 +16,7 @@ pub fn start_workers(db: Database, amqp: AMQP) {
for _ in 0..WORKER_COUNT {
task::spawn(ack::worker(db.clone(), amqp.clone()));
task::spawn(last_message_id::worker(db.clone()));
task::spawn(process_embeds::worker(db.clone(), amqp.clone()));
task::spawn(process_embeds::worker(db.clone()));
}
}
@@ -1,4 +1,4 @@
use crate::{AMQP, AppendMessage, Database, models::Message};
use crate::{models::Message, AppendMessage, Database};
use futures::future::join_all;
use linkify::{LinkFinder, LinkKind};
@@ -41,7 +41,7 @@ pub async fn queue(channel: String, id: String, content: String) {
}
/// Start a new worker
pub async fn worker(db: Database, amqp: AMQP) {
pub async fn worker(db: Database) {
let semaphore = Arc::new(Semaphore::new(
config().await.api.workers.max_concurrent_connections,
));
@@ -49,7 +49,6 @@ pub async fn worker(db: Database, amqp: AMQP) {
loop {
let task = Q.pop().await;
let db = db.clone();
let amqp = amqp.clone();
let semaphore = semaphore.clone();
spawn(async move {
@@ -65,7 +64,6 @@ pub async fn worker(db: Database, amqp: AMQP) {
if let Ok(embeds) = embeds {
if let Err(err) = Message::append(
&db,
Some(&amqp),
task.id,
task.channel,
AppendMessage {
+3
View File
@@ -0,0 +1,3 @@
pub fn transform_optional_string(s: Option<&str>) -> Option<String> {
s.map(|f| f.to_string())
}
@@ -612,6 +612,7 @@ impl From<crate::Report> for Report {
additional_context: value.additional_context,
status: value.status,
notes: value.notes,
case_id: value.case_id,
}
}
}
@@ -1525,3 +1526,98 @@ impl From<WebPushSubscription> for crate::WebPushSubscription {
}
}
}
impl From<crate::AdminUser> for AdminUser {
fn from(value: crate::AdminUser) -> Self {
AdminUser {
id: value.id,
platform_user_id: value.platform_user_id,
email: value.email,
active: value.active,
permissions: value.permissions,
revolt_user: None,
}
}
}
impl From<AdminUser> for crate::AdminUser {
fn from(value: AdminUser) -> Self {
crate::AdminUser {
id: value.id,
platform_user_id: value.platform_user_id,
email: value.email,
active: value.active,
permissions: value.permissions,
}
}
}
impl From<AdminUserEdit> for crate::PartialAdminUser {
fn from(value: AdminUserEdit) -> Self {
crate::PartialAdminUser {
id: None,
platform_user_id: value.platform_user_id,
email: value.email,
active: value.active,
permissions: value.permissions,
}
}
}
impl From<AdminToken> for crate::AdminToken {
fn from(value: AdminToken) -> Self {
crate::AdminToken {
id: value.id,
user_id: value.user_id,
token: value.token,
expiry: value.expiry.format_short().to_string(),
}
}
}
impl From<crate::AdminToken> for AdminToken {
fn from(value: crate::AdminToken) -> Self {
AdminToken {
id: value.id,
user_id: value.user_id,
token: value.token,
expiry: Timestamp::parse(&value.expiry).unwrap(), // if it's invalid it shouldn't have made it to this point.
}
}
}
impl From<crate::AdminComment> for AdminComment {
fn from(value: crate::AdminComment) -> Self {
AdminComment {
id: value.id,
case_id: value.case_id,
object_id: value.object_id,
user_id: value.user_id,
edited_at: value.edited_at.map(|f| Timestamp::parse(&f).unwrap()),
content: value.content,
system_message: value
.system_message
.map(|f| serde_json::from_str(&f).unwrap()),
system_message_target: value.system_message_target,
system_message_context: value.system_message_context,
}
}
}
impl From<AdminComment> for crate::AdminComment {
fn from(value: AdminComment) -> Self {
crate::AdminComment {
id: value.id,
object_id: value.object_id,
case_id: value.case_id,
user_id: value.user_id,
edited_at: value.edited_at.map(|f| f.format_short().to_string()),
content: value.content,
system_message: value
.system_message
.map(|f| serde_json::to_string(&f).unwrap()), // i'll eat my hat if this fails
system_message_target: value.system_message_target,
system_message_context: value.system_message_context,
}
}
}
+2 -1
View File
@@ -1,4 +1,5 @@
pub mod acker;
pub mod basic;
pub mod bridge;
pub mod bulk_permissions;
pub mod captcha;
@@ -13,5 +14,5 @@ pub mod reference;
pub mod shield;
pub mod test_fixtures;
pub use chunked::ChunkedDatabaseGenerator;
pub use funcs::*;
pub use chunked::ChunkedDatabaseGenerator;
+15 -2
View File
@@ -2,7 +2,7 @@ use std::str::FromStr;
use revolt_result::Result;
#[cfg(feature = "rocket-impl")]
use rocket::request::FromParam;
use rocket::{form::FromFormField, request::FromParam};
#[cfg(feature = "rocket-impl")]
use schemars::{
schema::{InstanceType, Schema, SchemaObject, SingleOrVec},
@@ -10,7 +10,8 @@ use schemars::{
};
use crate::{
Bot, Channel, Database, Emoji, Invite, Member, Message, Server, ServerBan, User, Webhook,
AdminToken, Bot, Channel, Database, Emoji, Invite, Member, Message, Server, ServerBan, User,
Webhook,
};
/// Reference to some object in the database
@@ -25,6 +26,11 @@ impl<'a> Reference<'a> {
Reference { id }
}
/// Fetch Admin Token from Ref
pub async fn as_admin_token(&self, db: &Database) -> Result<AdminToken> {
db.admin_token_fetch(&self.id).await
}
/// Fetch ban from Ref
pub async fn as_ban(&self, db: &Database, server: &str) -> Result<ServerBan> {
db.fetch_ban(server, self.id).await
@@ -126,3 +132,10 @@ impl<'a> JsonSchema for Reference<'a> {
})
}
}
#[cfg(feature = "rocket-impl")]
impl<'r> FromFormField<'r> for Reference<'r> {
fn from_value(field: rocket::form::ValueField<'r>) -> rocket::form::Result<'r, Self> {
Ok(Reference::from_unchecked(field.value.into()))
}
}
+399
View File
@@ -0,0 +1,399 @@
use iso8601_timestamp::Timestamp;
use crate::v0::{self, Report, User};
auto_derived! {
#[derive(Default)]
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminAuditItemCreate {
/// The ID of the mod who performed the action
#[cfg_attr(feature="serde", serde(rename="mod"))]
pub mod_id: String,
/// The action performed (previously 'permission')
pub action: String,
/// The relevant case ID, if applicable
#[cfg_attr(feature="serde", serde(rename="case"))]
pub case_id: String,
/// The id of the target (any object) the action was taken against, if applicable
#[cfg_attr(feature="serde", serde(rename="target"))]
pub target_id: Option<String>,
/// The context attached to the action, if applicable (eg. search phrases)
pub context: Option<String>,
}
pub struct AdminAuditItem {
/// The audit item ID
pub id: String,
/// The ID of the mod who performed the action
#[cfg_attr(feature="serde", serde(rename="mod"))]
pub mod_id: String,
/// The action performed (previously 'permission')
pub action: AdminAuditItemActions,
/// The relevant case ID, if applicable
#[cfg_attr(feature="serde", serde(rename="case"))]
pub case_id: String,
/// The id of the target (any object) the action was taken against, if applicable
#[cfg_attr(feature="serde", serde(rename="target"))]
pub target_id: Option<String>,
/// The context attached to the action, if applicable (eg. search phrases)
pub context: Option<String>,
/// The time the action occurred at.
pub timestamp: Timestamp,
}
#[derive(Default)]
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminCommentCreate {
/// The ID of the case this comment is attached to, if applicable. This should be the 7 character short code.
#[cfg_attr(feature="serde", serde(rename="case"))]
pub case_id: Option<String>,
/// The ID of the object this comment is attached to. Use full case ULID when creating a case comment.
#[cfg_attr(feature="serde", serde(rename="object"))]
pub object_id: String,
/// The content
#[cfg_attr(feature = "validator", validate(length(min = 1, max = 2000)))]
pub content: String
}
#[derive(Default)]
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminCommentEdit {
/// The new content
#[cfg_attr(feature = "validator", validate(length(min = 1, max = 2000)))]
pub content: String
}
pub struct AdminComment {
/// The comment ID
pub id: String,
/// The ID of the case this comment is attached to, if applicable. This should be the 7 character short code
#[cfg_attr(feature="serde", serde(rename="case"))]
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub case_id: Option<String>,
/// The object this comment is attached to. If this is referencing a case this is the full case ID
#[cfg_attr(feature="serde", serde(rename="object"))]
pub object_id: String,
/// The user who posted the comment/action
#[cfg_attr(feature="serde", serde(rename="user"))]
pub user_id: String,
/// When the comment was edited, if applicable, in iso8601
pub edited_at: Option<Timestamp>,
/// The content, if this is not a system event
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub content: Option<String>,
/// The system event that happened, if this is not a comment (only applicable for cases)
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub system_message: Option<AdminAuditItemActions>,
/// The system message target
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub system_message_target: Option<String>,
/// The system message raw context
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub system_message_context: Option<String>,
}
#[derive(Default)]
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminCaseCreate {
/// The owner of the case. Defaults to the creator
#[cfg_attr(feature="serde", serde(rename="owner"))]
pub owner: Option<String>,
/// The title of the case. If not provided, a default will be generated from the report(s) assigned
pub title: Option<String>,
/// The report IDs to initially attach to this case (and generate a default title from)
#[cfg_attr(feature = "validator", validate(length(min = 1)))]
pub initial_reports: Vec<String>
}
#[derive(Default)]
pub struct AdminCaseEdit {
/// The new owner of the case.
#[cfg_attr(feature="serde", serde(rename="owner"))]
pub owner_id: Option<String>,
/// The new title of the case.
pub title: Option<String>,
/// Report IDs to add to the case.
pub add_reports: Option<Vec<String>>,
/// Report IDs to remove from the case.
pub remove_reports: Option<Vec<String>>
}
pub struct AdminCase {
/// The case ID
pub id: String,
/// The case Short ID
pub short_id: String,
/// The owner of the case
#[cfg_attr(feature="serde", serde(rename="owner"))]
pub owner_id: String,
/// The title of the case
pub title: String,
/// The status of the case (open/closed)
pub status: String,
/// When the case was closed
pub closed_at: Option<Timestamp>,
/// The tags for the case
pub tags: Vec<String>,
/// The reports assigned to this case
pub reports: Vec<Report>
}
pub struct AdminStrike {
/// The strike ID
pub id: String,
/// The object receiving the strike (user/server)
#[cfg_attr(feature="serde", serde(rename="target"))]
pub target_id: String,
/// The moderator who gave the strike
#[cfg_attr(feature="serde", serde(rename="mod"))]
pub mod_id: String,
/// The case the strike was made under
#[cfg_attr(feature="serde", serde(rename="case"))]
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub case_id: Option<String>,
/// Action associated with the strike (eg. suspension/ban)
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub associated_action: Option<String>,
/// Has the strike been removed
#[cfg_attr(feature="serde", serde(skip_serializing_if="crate::if_false"))]
pub overruled: bool,
/// The user-facing reason for the strike
pub reason: String,
/// Internal context for the strike
#[cfg_attr(feature="serde", serde(skip_serializing_if="Option::is_none"))]
pub mod_context: Option<String>,
}
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminStrikeCreate {
/// The object receiving the strike (user/server)
#[cfg_attr(feature="serde", serde(rename="target"))]
pub target_id: String,
/// The case the strike was made under
#[cfg_attr(feature="serde", serde(rename="case"))]
pub case_id: Option<String>,
/// Action associated with the strike (eg. suspension/ban)
#[cfg_attr(feature = "validator", validate(length(min = 1, max = 25)))]
pub associated_action: Option<String>,
/// The user-facing reason for the strike
#[cfg_attr(feature = "validator", validate(length(max = 2000)))]
pub reason: String,
/// Internal context for the strike
#[cfg_attr(feature = "validator", validate(length(max = 2000)))]
pub mod_context: Option<String>,
}
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminStrikeEdit {
/// The case the strike was made under
#[cfg_attr(feature="serde", serde(rename="case"))]
pub case_id: Option<String>,
/// Action associated with the strike (eg. suspension/ban)
#[cfg_attr(feature = "validator", validate(length(min = 1, max = 25)))]
pub associated_action: Option<String>,
/// The user-facing reason for the strike
#[cfg_attr(feature = "validator", validate(length(max = 2000)))]
pub reason: Option<String>,
/// Internal context for the strike
#[cfg_attr(feature = "validator", validate(length(max = 2000)))]
pub mod_context: Option<String>,
}
pub struct AdminToken {
/// The token ID
pub id: String,
/// The user this token is attached to
#[cfg_attr(feature = "serde", serde(rename="user"))]
pub user_id: String,
/// The token itself
pub token: String,
/// The expiry timestamp for this token, in iso6801
pub expiry: Timestamp
}
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminTokenCreate {
/// The expiry timestamp for this token, in iso6801. Max 30 days from current time.
pub expiry: Timestamp,
}
pub struct AdminUser {
/// The ID of the user
pub id: String,
/// The user's revolt ID.
pub platform_user_id: String,
/// The user's email
pub email: String,
/// Whether the user is active or not (ie. can they use the api)
pub active: bool,
/// The permissions of the user
pub permissions: u64,
/// The Revolt user attached to this user. Use this for data like names and avatars.
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub revolt_user: Option<User>
}
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminUserCreate {
/// The user's revolt ID.
#[cfg_attr(feature = "validator", validate(length(min = 10, max = 20)))]
pub platform_user_id: String,
/// The user's email
#[cfg_attr(feature = "validator", validate(email))]
pub email: String,
/// Whether the user is active or not (ie. can they use the api)
pub active: bool,
/// The permissions of the user
pub permissions: u64,
}
#[cfg_attr(feature = "validator", derive(validator::Validate))]
pub struct AdminUserEdit {
/// The user's revolt ID.
#[cfg_attr(feature = "validator", validate(length(min = 10, max = 20)))]
pub platform_user_id: Option<String>,
/// The user's email
#[cfg_attr(feature = "validator", validate(email))]
pub email: Option<String>,
/// Whether the user is active or not (ie. can they use the api)
pub active: Option<bool>,
/// The permissions of the user
pub permissions: Option<u64>,
}
// Flags
#[repr(u32)]
pub enum AdminUserPermissionFlags {
Comments = 0,
ManageAdminUsers = 1,
CreateTokens = 2,
ViewUsers = 3,
ManageUsers = 4,
ManageUsersSenstiveInfo = 5,
ViewServers = 6,
ManageServers = 7,
ViewDMChannels = 8,
ManageDMChannels = 9,
ViewCases = 10,
ManageOwnCases = 11,
ManageOtherCases = 12,
ViewReports = 13,
Search = 14,
ManageNotes = 15,
Discover = 16,
ManageAccounts = 17,
ManageChannels = 18,
}
pub enum AdminAuditItemActions {
// Meta
CreateAdminUser,
EditAdminUser,
CreateToken,
RevokeToken,
// Comments
CommentCreate,
CommentEdit,
CommentFetchForObject,
// Servers
ServerFetch,
ServerFetchParticipants,
ServerFetchMembers,
ServerAddMember,
ServerChangeOwner,
ServerCreateInvite,
ServerDeleteInvite,
ServerDeleteAllInvites,
ServerDelete,
ServerEdit,
ServerRemoveMember,
ServerSetFlags,
ServerInstanceBanAllMembers,
ServerBanMember,
ServerUnbanMember,
// Accounts
DeleteAccount,
DisableAccount,
EmailChange,
EmailVerify,
EnableAccount,
// Channels
DeleteChannel,
EditChannel,
WipeChannel
}
// Joiner payloads
pub struct AdminServerResponse {
pub server: v0::Server,
pub owner: v0::User,
pub comments: Vec<v0::AdminComment>,
}
pub struct AdminServerParticipantsResponse {
pub users: Vec<v0::User>,
pub members: Vec<v0::Member>,
pub sort_strategy: String
}
pub struct AdminMemberWithUserAndOffsetResponse {
pub after: Option<usize>,
pub users: Vec<v0::MemberWithUserResponse>
}
}
impl AdminAuditItemActions {
/// Does this action create a case comment?
pub fn makes_comment(&self) -> bool {
match self {
AdminAuditItemActions::CreateAdminUser => false,
AdminAuditItemActions::EditAdminUser => false,
AdminAuditItemActions::CreateToken => false,
AdminAuditItemActions::RevokeToken => false,
AdminAuditItemActions::CommentCreate => false,
AdminAuditItemActions::CommentEdit => false,
AdminAuditItemActions::CommentFetchForObject => false,
AdminAuditItemActions::ServerFetch => false,
AdminAuditItemActions::ServerFetchParticipants => false,
AdminAuditItemActions::ServerAddMember => true,
AdminAuditItemActions::ServerChangeOwner => true,
AdminAuditItemActions::ServerCreateInvite => true,
AdminAuditItemActions::ServerDeleteInvite => true,
AdminAuditItemActions::ServerDeleteAllInvites => true,
AdminAuditItemActions::ServerDelete => true,
AdminAuditItemActions::ServerEdit => true,
AdminAuditItemActions::ServerRemoveMember => true,
AdminAuditItemActions::ServerSetFlags => true,
AdminAuditItemActions::ServerInstanceBanAllMembers => true,
AdminAuditItemActions::ServerBanMember => true,
AdminAuditItemActions::ServerUnbanMember => true,
AdminAuditItemActions::ServerFetchMembers => false,
AdminAuditItemActions::DeleteAccount => true,
AdminAuditItemActions::DisableAccount => true,
AdminAuditItemActions::EmailChange => true,
AdminAuditItemActions::EmailVerify => true,
AdminAuditItemActions::EnableAccount => true,
AdminAuditItemActions::DeleteChannel => true,
AdminAuditItemActions::EditChannel => true,
AdminAuditItemActions::WipeChannel => true,
}
}
}
impl std::fmt::Display for AdminAuditItemActions {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", self)
}
}
+3 -3
View File
@@ -1,3 +1,4 @@
mod admin;
mod bots;
mod channel_invites;
mod channel_unreads;
@@ -14,11 +15,11 @@ mod server_members;
mod servers;
mod user_settings;
mod users;
mod search;
mod accounts;
mod mfa_tickets;
mod sessions;
pub use admin::*;
pub use bots::*;
pub use channel_invites::*;
pub use channel_unreads::*;
@@ -35,7 +36,6 @@ pub use server_members::*;
pub use servers::*;
pub use user_settings::*;
pub use users::*;
pub use search::*;
pub use accounts::*;
pub use mfa_tickets::*;
pub use sessions::*;
pub use sessions::*;
@@ -8,6 +8,8 @@ auto_derived!(
pub id: String,
/// Id of the user creating this report
pub author_id: String,
/// The case this report is attached to
pub case_id: Option<String>,
/// Reported content
pub content: ReportedContent,
/// Additional report context
-74
View File
@@ -1,74 +0,0 @@
use iso8601_timestamp::Timestamp;
use std::collections::HashSet;
auto_derived!(
/// Options for searching messages in a server or channel
pub struct DataChannelMessagesSearch {
/// Channel to search in
pub channel: Option<String>,
/// Server to search in
pub server: Option<String>,
/// Filter options
pub filters: Option<DataChannelMessagesSearchFilters>,
/// What index to start the search at
pub offset: Option<u64>,
/// Max amount of messages to return
pub limit: Option<u64>,
/// Sort order
pub sort: Option<SortOrder>,
}
/// Message search filters
pub struct DataChannelMessagesSearchFilters {
/// Message content
pub content: Option<String>,
/// Specific user
pub author: Option<HashSet<String>>,
/// Mentions a user
pub mentions: Option<HashSet<String>>,
/// Mentions a role
pub role_mentions: Option<HashSet<String>>,
/// Send before a specific date
pub before_date: Option<Timestamp>,
/// Sent after a specific date
pub after_date: Option<Timestamp>,
/// What type of user sent the message
pub author_type: Option<HashSet<AuthorType>>,
/// Whether the message is pinned or not
pub pinned: Option<bool>,
/// Require message to have a specific component type
pub components: Option<HashSet<MessageComponent>>,
}
/// Message author type
#[derive(Copy, Hash)]
pub enum AuthorType {
User,
Bot,
Webhook,
}
/// Message component
#[derive(Copy, Hash)]
pub enum MessageComponent {
Image,
Video,
Link,
File,
Embed,
}
/// Message sort order
#[derive(Copy, Default)]
#[cfg_attr(feature = "serde", serde(rename_all = "lowercase"))]
pub enum SortOrder {
Asc,
#[default]
Desc,
}
);
@@ -159,4 +159,9 @@ auto_derived!(
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsMember>,
}
pub struct MemberWithUserResponse {
pub user: crate::v0::User,
pub member: Member,
}
);
+2
View File
@@ -205,6 +205,8 @@ auto_derived!(
Banned = 4,
/// User was marked as spam and removed from platform
Spam = 8,
/// User is a global moderator
GlobalModerator = 16,
}
/// New user profile data
+6
View File
@@ -39,9 +39,11 @@ impl IntoResponse for Error {
ErrorType::GroupTooLarge { .. } => StatusCode::FORBIDDEN,
ErrorType::AlreadyInGroup => StatusCode::CONFLICT,
ErrorType::NotInGroup => StatusCode::NOT_FOUND,
ErrorType::NotAMember => StatusCode::BAD_REQUEST,
ErrorType::AlreadyPinned => StatusCode::BAD_REQUEST,
ErrorType::NotPinned => StatusCode::BAD_REQUEST,
ErrorType::InSlowmode { retry_after: _ } => StatusCode::TOO_MANY_REQUESTS,
ErrorType::InviteExists => StatusCode::BAD_REQUEST,
ErrorType::CantCreateServers => StatusCode::FORBIDDEN,
ErrorType::UnknownServer => StatusCode::NOT_FOUND,
@@ -120,6 +122,10 @@ impl IntoResponse for Error {
ErrorType::DisallowedMFAMethod => StatusCode::BAD_REQUEST,
ErrorType::OperationFailed => StatusCode::INTERNAL_SERVER_ERROR,
ErrorType::IncorrectData { .. } => StatusCode::BAD_REQUEST,
ErrorType::ImATeaPot => StatusCode::IM_A_TEAPOT,
ErrorType::PrivilegedAccount => StatusCode::FORBIDDEN,
};
(status, Json(&self)).into_response()
+8
View File
@@ -106,12 +106,14 @@ pub enum ErrorType {
InSlowmode {
retry_after: u64,
},
InviteExists,
// ? Server related errors
CantCreateServers,
UnknownServer,
InvalidRole,
Banned,
NotAMember,
TooManyServers {
max: usize,
},
@@ -211,6 +213,12 @@ pub enum ErrorType {
TotpAlreadyEnabled,
DisallowedMFAMethod,
// ? Admin API relate errors
PrivilegedAccount,
// :)
ImATeaPot,
}
#[macro_export]
+6
View File
@@ -41,10 +41,12 @@ impl<'r> Responder<'r, 'static> for Error {
ErrorType::GroupTooLarge { .. } => Status::Forbidden,
ErrorType::AlreadyInGroup => Status::Conflict,
ErrorType::NotInGroup => Status::NotFound,
ErrorType::NotAMember => Status::BadRequest,
ErrorType::AlreadyPinned => Status::BadRequest,
ErrorType::NotPinned => Status::BadRequest,
ErrorType::InSlowmode { retry_after: _ } => Status::TooManyRequests,
ErrorType::InvalidFlagValue => Status::BadRequest,
ErrorType::InviteExists => Status::BadRequest,
ErrorType::CantCreateServers => Status::Forbidden,
ErrorType::UnknownServer => Status::NotFound,
@@ -125,6 +127,10 @@ impl<'r> Responder<'r, 'static> for Error {
ErrorType::DisallowedMFAMethod => Status::BadRequest,
ErrorType::OperationFailed => Status::InternalServerError,
ErrorType::IncorrectData { .. } => Status::BadRequest,
ErrorType::ImATeaPot => Status::ImATeapot,
ErrorType::PrivilegedAccount => Status::Forbidden,
};
// Serialize the error data structure into JSON.
-19
View File
@@ -1,19 +0,0 @@
[package]
name = "revolt-search"
version = "0.13.7"
edition = "2024"
license = "MIT"
publish = false
[dependencies]
revolt-config = { workspace = true }
revolt-database = { workspace = true }
revolt-models = { workspace = true }
elasticsearch = { workspace = true }
serde_json = { workspace = true }
ulid = { workspace = true }
iso8601-timestamp = { workspace = true, features = ["serde", "bson"] }
futures = { workspace = true }
elasticsearch-dsl = { workspace = true }
serde = { workspace = true }
linkify = { workspace = true }
-413
View File
@@ -1,413 +0,0 @@
use std::fmt::Display;
use elasticsearch::{
BulkOperation, BulkParts, CreateParts, DeleteByQueryParts, DeleteParts, Elasticsearch,
IndexParts, SearchParts,
auth::Credentials,
http::{
response::Exception,
transport::{SingleNodeConnectionPool, TransportBuilder},
},
indices::{IndicesCreateParts, IndicesDeleteParts},
};
use elasticsearch_dsl::{FieldSort, Query, Search, SearchResponse, Sort};
use linkify::{LinkFinder, LinkKind};
use revolt_database::{Database, Message, MessageWithUser, User};
use serde_json::{Map, Value, json, to_value};
pub use elasticsearch;
use crate::{AuthorType, MessageComponent, SearchTerms};
/// Elasticsearch errors
#[derive(Debug)]
pub enum Error {
Http(elasticsearch::Error),
Exception(Exception),
}
impl From<elasticsearch::Error> for Error {
fn from(value: elasticsearch::Error) -> Self {
Self::Http(value)
}
}
impl From<Exception> for Error {
fn from(value: Exception) -> Self {
Self::Exception(value)
}
}
impl std::error::Error for Error {}
impl Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Error::Http(error) => write!(f, "Http error: {error}"),
Error::Exception(exception) => write!(f, "Elasticsearch error: {exception:?}"),
}
}
}
/// Higher level elasticsearch API more fit for our specific usecase
#[derive(Debug, Clone)]
pub struct ElasticsearchClient {
pub inner: Elasticsearch,
}
impl ElasticsearchClient {
pub fn new(host: &str, port: u16, key: String) -> Self {
let pool =
SingleNodeConnectionPool::new(format!("{host}:{port}").as_str().try_into().unwrap());
let transport = TransportBuilder::new(pool)
.auth(Credentials::EncodedApiKey(key))
.build()
.unwrap();
let inner = Elasticsearch::new(transport);
Self { inner }
}
/// Delete messages index along with all documents
pub async fn delete_indexes(&self) -> Result<(), Error> {
let exception = self
.inner
.indices()
.delete(IndicesDeleteParts::Index(&["messages"]))
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
/// Create the messages index
pub async fn setup_indexes(&self) -> Result<(), Error> {
let exception = self
.inner
.indices()
.create(IndicesCreateParts::Index("messages"))
.body(json!({
"mappings": {
"properties": {
"content": {"type": "text"},
"author": {"type": "keyword"},
"author_type": {"type": "keyword"},
"channel": {"type": "keyword"},
"mentions": {"type": "keyword"},
"role_mentions": {"type": "keyword"},
"pinned": {"type": "boolean"},
"embeds": {
"properties": {}
},
"attachments": {
"type": "nested",
"dynamic": false,
"properties": {
"metadata.type": {
"type": "keyword"
}
}
},
"has_link": { "type": "boolean" },
}
}
}))
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
/// Performs a search for messages, returns a vec of message ids
pub async fn search(&self, terms: SearchTerms) -> Result<Vec<String>, Error> {
let mut query = Query::bool().filter(Query::terms("channel", terms.channels));
if let Some(content) = terms.filters.content {
query = query.filter(Query::r#match("content", content))
}
if let Some(author) = terms.filters.author {
query = query.filter(Query::terms("author", author))
}
if let Some(mentions) = terms.filters.mentions {
query = query.filter(Query::terms("mentions", mentions))
}
if let Some(author_type) = terms.filters.author_type {
query = query.filter(Query::terms("author_type", author_type))
}
if let Some(pinned) = terms.filters.pinned {
if pinned {
query = query.filter(Query::exists("pinned"))
} else {
query = query.filter(Query::bool().must_not(Query::exists("pinned")))
}
}
if let Some(components) = terms.filters.components {
let mut components_query = Query::bool();
let mut attachments_query = Query::bool();
for component in components {
match component {
MessageComponent::Image => {
attachments_query = attachments_query
.should(Query::term("attachments.metadata.type", "Image"))
}
MessageComponent::Video => {
attachments_query = attachments_query
.should(Query::term("attachments.metadata.type", "Video"))
}
MessageComponent::Link => {
components_query = components_query.should(Query::exists("has_link"))
}
MessageComponent::File => {
attachments_query = attachments_query.should(Query::exists("attachments"))
}
MessageComponent::Embed => {
components_query = components_query.should(Query::exists("embeds"))
}
};
}
query = query
.filter(components_query.should(Query::nested("attachments", attachments_query)));
}
let search = Search::new()
.query(query)
.stats(false)
.sort(Sort::FieldSort(
FieldSort::new("_id".to_string()).order(terms.sort.unwrap_or_default().into()),
));
let response = self
.inner
.search(SearchParts::Index(&["messages"]))
.stored_fields(&[])
.body(search)
.size(terms.limit.unwrap_or(100) as i64)
.from(terms.offset.unwrap_or(0) as i64)
.send()
.await?;
if response.status_code().is_success() {
let messages = response.json::<SearchResponse>().await?;
Ok(messages.hits.hits.into_iter().map(|hit| hit.id).collect())
} else {
Err(response
.exception()
.await?
.expect("No exception with error response.")
.into())
}
}
/// Creates a source for a message which can be stored and indexed into elasticsearch
fn create_message_source(
&self,
_db: &Database,
message: Message,
author: Option<User>,
) -> Value {
let mut map = Map::new();
map.insert("channel".to_string(), Value::String(message.channel));
map.insert("author".to_string(), Value::String(message.author));
if let Some(content) = message.content {
// Is there a better way to handle this? can elasticsearch index links itself?
// Maybe in the future store the domains and be able to filter by that as well
let mut finder = LinkFinder::new();
finder.kinds(&[LinkKind::Url]);
if finder.links(&content).next().is_some() {
map.insert("has_link".to_string(), Value::Bool(true));
}
map.insert("content".to_string(), Value::String(content));
}
if let Some(attachments) = message.attachments {
// TODO: fetch the file metadata from FileHash because of File.metadata deprecation
// let metadata = attachment.as_hash(db).await.expect("Failed to fetch FileHash").metadata;
map.insert(
"attachments".to_string(),
serde_json::to_value(attachments).unwrap(),
);
}
if let Some(embeds) = message.embeds {
map.insert("embeds".to_string(), serde_json::to_value(embeds).unwrap());
}
if let Some(mentions) = message.mentions {
map.insert(
"mentions".to_string(),
serde_json::to_value(mentions).unwrap(),
);
}
if let Some(role_mentions) = message.role_mentions {
map.insert(
"role_mentions".to_string(),
serde_json::to_value(role_mentions).unwrap(),
);
}
if let Some(pinned) = message.pinned {
map.insert("pinned".to_string(), Value::Bool(pinned));
}
// This will turn bot author type to user author type if this is ran on a deleted message,
// due to the author not existing anymore so fetching will fail, this is probably niche enough
// to not really matter, might try fix in the future.
map.insert(
"author_type".to_string(),
to_value(if message.webhook.is_some() {
AuthorType::Webhook
} else if author.is_some_and(|user| user.bot.is_some()) {
AuthorType::Bot
} else {
AuthorType::User
})
.unwrap(),
);
Value::Object(map)
}
/// Bulk uploads and indexes messages to elasticsearch
pub async fn bulk_index_messages(
&self,
db: &Database,
messages: Vec<MessageWithUser>,
) -> Result<(), Error> {
let mut ops = Vec::<BulkOperation<Value>>::new();
for message in messages {
let id = message.message.id.clone();
let source = self.create_message_source(db, message.message, message.user);
ops.push(BulkOperation::create(source).id(id).into());
}
let exception = self
.inner
.bulk(BulkParts::Index("messages"))
.body(ops)
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
/// Uploads and indexes a single message to elasticsearch
pub async fn index_message(
&self,
db: &Database,
message: Message,
author: Option<User>,
) -> Result<(), Error> {
let id = message.id.clone();
let source = self.create_message_source(db, message, author);
let exception = self
.inner
.create(CreateParts::IndexId("messages", &id))
.body(source)
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
/// Updates or upserts an existing message to elasticsearch
pub async fn edit_message(
&self,
db: &Database,
message: Message,
author: Option<User>,
) -> Result<(), Error> {
let id = message.id.clone();
let source = self.create_message_source(db, message, author);
let exception = self
.inner
.index(IndexParts::IndexId("messages", &id))
.body(source)
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
/// Deletes a message from elasticsearch
pub async fn delete_message(&self, message_id: &str) -> Result<(), Error> {
let exception = self
.inner
.delete(DeleteParts::IndexId("messages", message_id))
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
/// Deletes all messages in a channel from elasticsearch
pub async fn delete_channel(&self, channel_id: &str) -> Result<(), Error> {
let exception = self
.inner
.delete_by_query(DeleteByQueryParts::Index(&["messages"]))
.body(Search::new().query(Query::term("channel", channel_id)))
.send()
.await?
.exception()
.await?;
if let Some(exception) = exception {
Err(exception.into())
} else {
Ok(())
}
}
}
-5
View File
@@ -1,5 +0,0 @@
mod client;
mod types;
pub use client::*;
pub use types::*;
-134
View File
@@ -1,134 +0,0 @@
use std::collections::HashSet;
use iso8601_timestamp::Timestamp;
use revolt_models::v0;
use serde::Serialize;
/// Message author type
#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Hash)]
pub enum AuthorType {
User,
Bot,
Webhook,
}
/// Message component
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum MessageComponent {
Image,
Video,
Link,
File,
Embed,
}
/// Message search filters
#[derive(Debug, Clone, Default, PartialEq)]
pub struct SearchFilters {
/// Message content
pub content: Option<String>,
/// Specific user
pub author: Option<HashSet<String>>,
/// Mentions a user
pub mentions: Option<HashSet<String>>,
/// Mentions a role
pub role_mentions: Option<HashSet<String>>,
/// Send before a specific date
pub before_date: Option<Timestamp>,
/// Sent after a specific date
pub after_date: Option<Timestamp>,
/// What type of user sent the message
pub author_type: Option<HashSet<AuthorType>>,
/// Whether the message is pinned or not
pub pinned: Option<bool>,
/// Require message to have a specific component type
pub components: Option<HashSet<MessageComponent>>,
}
/// Message sort order
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum SortOrder {
Asc,
#[default]
Desc,
}
/// Options for searching messages in a server or channel
#[derive(Debug, Clone, PartialEq)]
pub struct SearchTerms {
/// Channels to search in
pub channels: Vec<String>,
/// Filter options
pub filters: SearchFilters,
/// What index to start the search at
pub offset: Option<u64>,
/// Max amount of messages to return
pub limit: Option<u64>,
/// Sort order
pub sort: Option<SortOrder>,
}
impl From<v0::AuthorType> for AuthorType {
fn from(value: v0::AuthorType) -> Self {
match value {
v0::AuthorType::User => AuthorType::User,
v0::AuthorType::Bot => AuthorType::Bot,
v0::AuthorType::Webhook => AuthorType::Webhook,
}
}
}
impl From<v0::MessageComponent> for MessageComponent {
fn from(value: v0::MessageComponent) -> Self {
match value {
v0::MessageComponent::Image => MessageComponent::Image,
v0::MessageComponent::Video => MessageComponent::Video,
v0::MessageComponent::Link => MessageComponent::Link,
v0::MessageComponent::File => MessageComponent::File,
v0::MessageComponent::Embed => MessageComponent::Embed,
}
}
}
impl From<v0::SortOrder> for SortOrder {
fn from(value: v0::SortOrder) -> Self {
match value {
v0::SortOrder::Asc => SortOrder::Asc,
v0::SortOrder::Desc => SortOrder::Desc,
}
}
}
impl From<v0::DataChannelMessagesSearchFilters> for SearchFilters {
fn from(value: v0::DataChannelMessagesSearchFilters) -> Self {
Self {
content: value.content,
author: value.author,
mentions: value.mentions,
role_mentions: value.role_mentions,
before_date: value.before_date,
after_date: value.after_date,
author_type: value
.author_type
.map(|types| types.into_iter().map(Into::into).collect()),
pinned: value.pinned,
components: value
.components
.map(|types| types.into_iter().map(Into::into).collect()),
}
}
}
impl From<SortOrder> for elasticsearch_dsl::SortOrder {
fn from(value: SortOrder) -> Self {
match value {
SortOrder::Asc => elasticsearch_dsl::SortOrder::Asc,
SortOrder::Desc => elasticsearch_dsl::SortOrder::Desc,
}
}
}
@@ -1,10 +1,10 @@
use std::time::Duration;
use revolt_database::{Database, AMQP};
use revolt_database::Database;
use revolt_result::Result;
use tokio::time::sleep;
pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
loop {
let accounts = db.fetch_accounts_due_for_deletion().await?;
let count = accounts.len();
@@ -12,7 +12,7 @@ pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
for mut account in accounts {
let mut user = db.fetch_user(&account.id).await?;
user.delete(&db, Some(&amqp)).await?;
user.delete(&db).await?;
account.mark_deleted(&db).await?;
}
@@ -1,9 +1,10 @@
use std::sync::Arc;
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
@@ -15,7 +16,11 @@ pub struct AckConsumer {
#[async_trait]
impl Consumer for AckConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -1,10 +1,11 @@
use std::{collections::HashMap, sync::Arc};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
@@ -16,7 +17,11 @@ pub struct DmCallConsumer {
#[async_trait]
impl Consumer for DmCallConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -1,10 +1,11 @@
use std::{collections::HashMap, sync::Arc};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
@@ -16,7 +17,11 @@ pub struct FRAcceptedConsumer {
#[async_trait]
impl Consumer for FRAcceptedConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -1,10 +1,11 @@
use std::{collections::HashMap, sync::Arc};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
@@ -16,7 +17,11 @@ pub struct FRReceivedConsumer {
#[async_trait]
impl Consumer for FRReceivedConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -1,10 +1,11 @@
use std::{collections::HashMap, sync::Arc};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
@@ -16,7 +17,11 @@ pub struct GenericConsumer {
#[async_trait]
impl Consumer for GenericConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -4,13 +4,13 @@ use std::{
sync::Arc,
};
use crate::utils::render_notification_content;
use crate::utils::{render_notification_content, Consumer};
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use revolt_database::{
amqp::consumer::Consumer, events::rabbit::*,
util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member, MessageFlagsValue,
events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
MessageFlagsValue,
};
use revolt_models::v0::{MessageFlags, PushNotification};
use revolt_result::ToRevoltError;
@@ -29,7 +29,11 @@ impl MassMessageConsumer {
push: &PushNotification,
users: &[String],
) -> Result<()> {
if let Ok(sessions) = self.db.fetch_sessions_with_subscription(users).await {
if let Ok(sessions) = self
.db
.fetch_sessions_with_subscription(users)
.await
{
let config = revolt_config::config().await;
for session in sessions {
if let Some(sub) = session.subscription {
@@ -68,7 +72,11 @@ impl MassMessageConsumer {
#[async_trait]
impl Consumer for MassMessageConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -1,11 +1,11 @@
use std::{collections::HashMap, sync::Arc};
use crate::utils::render_notification_content;
use crate::utils::{render_notification_content, Consumer};
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
@@ -17,7 +17,11 @@ pub struct MessageConsumer {
#[async_trait]
impl Consumer for MessageConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<Channel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
@@ -1,5 +1,6 @@
use std::{borrow::Cow, collections::BTreeMap, io::Cursor, sync::Arc};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use base64::{
@@ -14,7 +15,7 @@ use revolt_a2::{
},
Client, ClientConfig, Endpoint, Error, ErrorBody, ErrorReason, Priority, PushType, Response,
};
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
use revolt_models::v0::{Channel, Message, PushNotification};
use serde::Serialize;
@@ -122,7 +123,11 @@ impl ApnsOutboundConsumer {
#[async_trait]
impl Consumer for ApnsOutboundConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<AMQPChannel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
) -> Self {
let config = revolt_config::config().await;
if config.pushd.apn.pkcs8.is_empty()
@@ -1,5 +1,6 @@
use std::{collections::HashMap, sync::Arc, time::Duration};
use crate::utils::Consumer;
use anyhow::{bail, Result};
use async_trait::async_trait;
use fcm_v1::{
@@ -9,7 +10,7 @@ use fcm_v1::{
};
use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
use revolt_config::config;
use revolt_database::{amqp::consumer::Consumer, events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, Database};
use serde_json::Value;
/// Custom notification data
@@ -125,7 +126,11 @@ pub struct FcmOutboundConsumer {
#[async_trait]
impl Consumer for FcmOutboundConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<AMQPChannel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
) -> Self {
let config = revolt_config::config().await;
Self {
@@ -1,5 +1,7 @@
use std::{collections::HashMap, sync::Arc};
use crate::utils::Consumer;
use anyhow::{anyhow, bail, Result};
use async_trait::async_trait;
use base64::{
@@ -7,9 +9,7 @@ use base64::{
Engine as _,
};
use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
use revolt_database::{
amqp::consumer::Consumer, events::rabbit::*, util::format_display_name, Database,
};
use revolt_database::{events::rabbit::*, util::format_display_name, Database};
use web_push::{
ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
WebPushClient, WebPushError, WebPushMessageBuilder,
@@ -27,7 +27,11 @@ pub struct VapidOutboundConsumer {
#[async_trait]
impl Consumer for VapidOutboundConsumer {
async fn create(db: Database, connection: Arc<Connection>, channel: Arc<AMQPChannel>, _: ()) -> Self {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
) -> Self {
let config = revolt_config::config().await;
if config.pushd.vapid.private_key.is_empty() || config.pushd.vapid.public_key.is_empty() {

Some files were not shown because too many files have changed in this diff Show More