chore: ignore WS protocol errors

refactor: one channel per consumer
refactor: move lapin code into revolt-broker crate
2025-06-28 18:55:15 +01:00 · 2025-06-28 18:26:51 +01:00 · 2025-06-28 17:23:44 +01:00 · 2025-06-28 13:35:18 +01:00
183 changed files with 2155 additions and 1692 deletions
@@ -97,6 +97,54 @@ version = "0.2.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
 [[package]]
 name = "amq-protocol"
 version = "8.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a56406d1cfd9933844a083206bf10fe3748c753bf43c64d4f4453d87181c14f6"
 dependencies = [
 "amq-protocol-tcp",
 "amq-protocol-types",
 "amq-protocol-uri",
 "cookie-factory",
 "nom 8.0.0",
 "serde",
 ]
 [[package]]
 name = "amq-protocol-tcp"
 version = "8.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c95840a64f5529882346cf7710d0c8d6b6e6fb4ecd3fed0883d1ae97c42efc14"
 dependencies = [
 "amq-protocol-uri",
 "tcp-stream",
 "tracing",
 ]
 [[package]]
 name = "amq-protocol-types"
 version = "8.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6a7bc568496f6fd598f44bf5c7e9f8157111634a54c114cb9c087e05d282c91a"
 dependencies = [
 "cookie-factory",
 "nom 8.0.0",
 "serde",
 "serde_json",
 ]
 [[package]]
 name = "amq-protocol-uri"
 version = "8.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a1d80bcd4af9f0fd2bad6def25c66d55130178749cc98000fd3de2b44991b5ed"
 dependencies = [
 "amq-protocol-types",
 "percent-encoding",
 "url",
 ]
 [[package]]
 name = "amqp_serde"
 version = "0.4.2"
@@ -181,6 +229,45 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
 [[package]]
 name = "asn1-rs"
 version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56624a96882bb8c26d61312ae18cb45868e5a9992ea73c58e45c3101e56a1e60"
 dependencies = [
 "asn1-rs-derive",
 "asn1-rs-impl",
 "displaydoc",
 "nom 7.1.3",
 "num-traits",
 "rusticata-macros",
 "thiserror 2.0.12",
 "time",
 ]
 [[package]]
 name = "asn1-rs-derive"
 version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c"
 dependencies = [
 "proc-macro2",
 "quote 1.0.40",
 "syn 2.0.101",
 "synstructure 0.13.2",
 ]
 [[package]]
 name = "asn1-rs-impl"
 version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7"
 dependencies = [
 "proc-macro2",
 "quote 1.0.40",
 "syn 2.0.101",
 ]
 [[package]]
 name = "async-attributes"
 version = "1.1.2"
@@ -245,6 +332,31 @@ dependencies = [
 "tokio 1.45.1",
 ]
 [[package]]
 name = "async-global-executor"
 version = "3.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "13f937e26114b93193065fd44f507aa2e9169ad0cdabbb996920b1fe1ddea7ba"
 dependencies = [
 "async-channel 2.3.1",
 "async-executor",
 "async-io",
 "async-lock 3.4.0",
 "blocking",
 "futures-lite 2.6.0",
 ]
 [[package]]
 name = "async-global-executor-trait"
 version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9af57045d58eeb1f7060e7025a1631cbc6399e0a1d10ad6735b3d0ea7f8346ce"
 dependencies = [
 "async-global-executor 3.1.0",
 "async-trait",
 "executor-trait",
 ]
 [[package]]
 name = "async-io"
 version = "2.4.1"
@@ -303,6 +415,18 @@ dependencies = [
 "tracing",
 ]
 [[package]]
 name = "async-reactor-trait"
 version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "35ce4c5a0627f1b37c346233c34ef2263e6a58812f7c071b71d3712511421422"
 dependencies = [
 "async-io",
 "async-trait",
 "futures-core",
 "reactor-trait",
 ]
 [[package]]
 name = "async-recursion"
 version = "1.1.1"
@@ -340,7 +464,7 @@ checksum = "730294c1c08c2e0f85759590518f6333f0d5a0a766a27d519c1b244c3dfd8a24"
 dependencies = [
 "async-attributes",
 "async-channel 1.9.0",
- "async-global-executor",
+ "async-global-executor 2.4.1",
 "async-io",
 "async-lock 3.4.0",
 "async-process",
@@ -510,7 +634,7 @@ dependencies = [
 "anyhow",
 "arrayvec",
 "log",
- "nom",
+ "nom 7.1.3",
 "num-rational",
 "v_frame",
 ]
@@ -1256,6 +1380,15 @@ dependencies = [
 "generic-array 0.14.7",
 ]
 [[package]]
 name = "block-padding"
 version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93"
 dependencies = [
 "generic-array 0.14.7",
 ]
 [[package]]
 name = "blocking"
 version = "1.6.1"
@@ -1386,6 +1519,15 @@ version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a2698f953def977c68f935bb0dfa959375ad4638570e969e2f1e9f433cbf1af6"
 [[package]]
 name = "cbc"
 version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6"
 dependencies = [
 "cipher",
 ]
 [[package]]
 name = "cc"
 version = "1.2.26"
@@ -1403,7 +1545,7 @@ version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
 dependencies = [
- "nom",
+ "nom 7.1.3",
 ]
 [[package]]
@@ -1487,6 +1629,18 @@ dependencies = [
 "cc",
 ]
 [[package]]
 name = "cms"
 version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b77c319abfd5219629c45c34c89ba945ed3c5e49fcde9d16b6c3885f118a730"
 dependencies = [
 "const-oid 0.9.6",
 "der 0.7.10",
 "spki 0.7.3",
 "x509-cert",
 ]
 [[package]]
 name = "coarsetime"
 version = "0.1.36"
@@ -1536,7 +1690,7 @@ dependencies = [
 "async-trait",
 "json5",
 "lazy_static",
- "nom",
+ "nom 7.1.3",
 "pathdiff",
 "ron",
 "rust-ini",
@@ -2014,7 +2168,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "79b71cca7d95d7681a4b3b9cdf63c8dbc3730d0584c2c74e31416d64a90493f4"
 dependencies = [
 "const-oid 0.6.2",
- "der_derive",
+ "der_derive 0.4.1",
 ]
 [[package]]
@@ -2035,10 +2189,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
 dependencies = [
 "const-oid 0.9.6",
 "der_derive 0.7.3",
 "flagset",
 "pem-rfc7468 0.7.0",
 "zeroize",
 ]
 [[package]]
 name = "der-parser"
 version = "10.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6"
 dependencies = [
 "asn1-rs",
 "displaydoc",
 "nom 7.1.3",
 "num-bigint",
 "num-traits",
 "rusticata-macros",
 ]
 [[package]]
 name = "der_derive"
 version = "0.4.1"
@@ -2051,6 +2221,17 @@ dependencies = [
 "synstructure 0.12.6",
 ]
 [[package]]
 name = "der_derive"
 version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18"
 dependencies = [
 "proc-macro2",
 "quote 1.0.40",
 "syn 2.0.101",
 ]
 [[package]]
 name = "deranged"
 version = "0.4.0"
@@ -2096,6 +2277,15 @@ dependencies = [
 "syn 2.0.101",
 ]
 [[package]]
 name = "des"
 version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ffdd80ce8ce993de27e9f063a444a4d53ce8e8db4c1f00cc03af5ad5a9867a1e"
 dependencies = [
 "cipher",
 ]
 [[package]]
 name = "devise"
 version = "0.4.2"
@@ -2384,7 +2574,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cea14ef9355e3beab063703aa9dab15afd25f0667c341310c1e5274bb1d0da18"
 dependencies = [
 "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 [[package]]
@@ -2414,6 +2604,15 @@ dependencies = [
 "pin-project-lite 0.2.16",
 ]
 [[package]]
 name = "executor-trait"
 version = "2.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "13c39dff9342e4e0e16ce96be751eb21a94e94a87bb2f6e63ad1961c2ce109bf"
 dependencies = [
 "async-trait",
 ]
 [[package]]
 name = "exr"
 version = "1.73.0"
@@ -2522,6 +2721,12 @@ dependencies = [
 "winapi",
 ]
 [[package]]
 name = "flagset"
 version = "0.4.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe"
 [[package]]
 name = "flate2"
 version = "1.1.1"
@@ -2541,6 +2746,17 @@ dependencies = [
 "num-traits",
 ]
 [[package]]
 name = "flume"
 version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
 dependencies = [
 "futures-core",
 "futures-sink",
 "spin",
 ]
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -3747,6 +3963,7 @@ version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
 dependencies = [
 "block-padding",
 "generic-array 0.14.7",
 ]
@@ -3806,7 +4023,7 @@ checksum = "e04d7f318608d35d4b61ddd75cbdaee86b023ebe2bd5a66ee0915f0bf93095a9"
 dependencies = [
 "hermit-abi 0.5.1",
 "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 [[package]]
@@ -4111,6 +4328,25 @@ dependencies = [
 "log",
 ]
 [[package]]
 name = "lapin"
 version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "792d823427d788859cf36d471a9983e52e07e1267fb50ec07ecb240ef4ecd8af"
 dependencies = [
 "amq-protocol",
 "async-global-executor-trait",
 "async-reactor-trait",
 "async-trait",
 "executor-trait",
 "flume",
 "futures-core",
 "futures-io",
 "reactor-trait",
 "tracing",
 "waker-fn",
 ]
 [[package]]
 name = "lazy_static"
 version = "1.5.0"
@@ -4148,7 +4384,7 @@ dependencies = [
 "idna 0.3.0",
 "mime",
 "native-tls",
- "nom",
+ "nom 7.1.3",
 "once_cell",
 "quoted_printable",
 "socket2 0.4.10",
@@ -4655,7 +4891,7 @@ dependencies = [
 "md-5",
 "mongodb-internal-macros",
 "once_cell",
- "pbkdf2",
+ "pbkdf2 0.11.0",
 "percent-encoding",
 "rand 0.8.5",
 "rustc_version_runtime",
@@ -4758,6 +4994,15 @@ dependencies = [
 "minimal-lexical",
 ]
 [[package]]
 name = "nom"
 version = "8.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405"
 dependencies = [
 "memchr",
 ]
 [[package]]
 name = "noop_proc_macro"
 version = "0.3.0"
@@ -4938,6 +5183,15 @@ dependencies = [
 "memchr",
 ]
 [[package]]
 name = "oid-registry"
 version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7"
 dependencies = [
 "asn1-rs",
 ]
 [[package]]
 name = "once_cell"
 version = "1.21.3"
@@ -5027,6 +5281,28 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
 [[package]]
 name = "p12-keystore"
 version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3cae83056e7cb770211494a0ecf66d9fa7eba7d00977e5bb91f0e925b40b937f"
 dependencies = [
 "cbc",
 "cms",
 "der 0.7.10",
 "des",
 "hex",
 "hmac",
 "pkcs12",
 "pkcs5",
 "rand 0.9.1",
 "rc2",
 "sha1",
 "sha2",
 "thiserror 2.0.12",
 "x509-parser",
 ]
 [[package]]
 name = "p256"
 version = "0.11.1"
@@ -5112,6 +5388,16 @@ dependencies = [
 "digest",
 ]
 [[package]]
 name = "pbkdf2"
 version = "0.12.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2"
 dependencies = [
 "digest",
 "hmac",
 ]
 [[package]]
 name = "pear"
 version = "0.2.9"
@@ -5382,6 +5668,36 @@ dependencies = [
 "zeroize",
 ]
 [[package]]
 name = "pkcs12"
 version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "695b3df3d3cc1015f12d70235e35b6b79befc5fa7a9b95b951eab1dd07c9efc2"
 dependencies = [
 "cms",
 "const-oid 0.9.6",
 "der 0.7.10",
 "digest",
 "spki 0.7.3",
 "x509-cert",
 "zeroize",
 ]
 [[package]]
 name = "pkcs5"
 version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e847e2c91a18bfa887dd028ec33f2fe6f25db77db3619024764914affe8b69a6"
 dependencies = [
 "aes",
 "cbc",
 "der 0.7.10",
 "pbkdf2 0.12.2",
 "scrypt",
 "sha2",
 "spki 0.7.3",
 ]
 [[package]]
 name = "pkcs8"
 version = "0.9.0"
@@ -5936,6 +6252,15 @@ dependencies = [
 "crossbeam-utils",
 ]
 [[package]]
 name = "rc2"
 version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "62c64daa8e9438b84aaae55010a93f396f8e60e3911590fcba770d04643fc1dd"
 dependencies = [
 "cipher",
 ]
 [[package]]
 name = "rdrand"
 version = "0.4.0"
@@ -5945,6 +6270,18 @@ dependencies = [
 "rand_core 0.3.1",
 ]
 [[package]]
 name = "reactor-trait"
 version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b2eb4ff39ed91c79034b2856a73b07bc8afd752c981bc4433bcbeff531e275a"
 dependencies = [
 "async-io",
 "async-trait",
 "futures-core",
 "futures-io",
 ]
 [[package]]
 name = "redis"
 version = "0.23.3"
@@ -5993,7 +6330,7 @@ dependencies = [
 "cookie-factory",
 "crc16",
 "log",
- "nom",
+ "nom 7.1.3",
 ]
 [[package]]
@@ -6228,12 +6565,15 @@ dependencies = [
 "bincode",
 "fred",
 "futures",
 "lapin",
 "log",
 "lru 0.7.8",
 "lru_time_cache",
 "once_cell",
 "querystring",
 "rand 0.6.5",
 "redis-kiss",
 "revolt-broker",
 "revolt-config",
 "revolt-database",
 "revolt-models",
@@ -6247,6 +6587,19 @@ dependencies = [
 "ulid 0.5.0",
 ]
 [[package]]
 name = "revolt-broker"
 version = "0.8.8"
 dependencies = [
 "async-std",
 "lapin",
 "log",
 "rand 0.9.1",
 "revolt-config",
 "rmp-serde",
 "serde",
 ]
 [[package]]
 name = "revolt-config"
 version = "0.8.8"
@@ -6291,10 +6644,12 @@ dependencies = [
 "bson",
 "deadqueue",
 "decancer",
 "fcm_v1",
 "futures",
 "indexmap 1.9.3",
 "isahc",
 "iso8601-timestamp",
 "lapin",
 "linkify 0.8.1",
 "log",
 "lru 0.11.1",
@@ -6304,15 +6659,18 @@ dependencies = [
 "rand 0.8.5",
 "redis-kiss",
 "regex",
 "revolt-broker",
 "revolt-config",
 "revolt-models",
 "revolt-parser",
 "revolt-permissions",
 "revolt-presence",
 "revolt-result",
 "revolt_a2",
 "revolt_okapi",
 "revolt_optional_struct",
 "revolt_rocket_okapi",
 "rmp-serde",
 "rocket",
 "schemars",
 "serde",
@@ -6321,6 +6679,7 @@ dependencies = [
 "unicode-segmentation",
 "url-escape",
 "validator 0.16.1",
 "web-push",
 ]
 [[package]]
@@ -6338,6 +6697,7 @@ dependencies = [
 "futures",
 "impl_ops",
 "iso8601-timestamp",
 "lapin",
 "lettre",
 "linkify 0.6.0",
 "log",
@@ -6349,6 +6709,7 @@ dependencies = [
 "redis-kiss",
 "regex",
 "reqwest 0.11.27",
 "revolt-broker",
 "revolt-config",
 "revolt-database",
 "revolt-models",
@@ -6356,6 +6717,7 @@ dependencies = [
 "revolt-presence",
 "revolt-result",
 "revolt_rocket_okapi",
 "rmp-serde",
 "rocket",
 "rocket_authifier",
 "rocket_cors",
@@ -6511,7 +6873,6 @@ dependencies = [
 "revolt_rocket_okapi",
 "rocket",
 "schemars",
 "sentry",
 "serde",
 "serde_json",
 "utoipa",
@@ -6890,6 +7251,15 @@ dependencies = [
 "semver",
 ]
 [[package]]
 name = "rusticata-macros"
 version = "4.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632"
 dependencies = [
 "nom 7.1.3",
 ]
 [[package]]
 name = "rustix"
 version = "0.38.44"
@@ -6900,7 +7270,7 @@ dependencies = [
 "errno",
 "libc",
 "linux-raw-sys 0.4.15",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 [[package]]
@@ -6913,7 +7283,7 @@ dependencies = [
 "errno",
 "libc",
 "linux-raw-sys 0.9.4",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 [[package]]
@@ -6956,6 +7326,19 @@ dependencies = [
 "zeroize",
 ]
 [[package]]
 name = "rustls-connector"
 version = "0.21.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b68f41384a0c195494f07c9bf5d020dcfca869d3b1492c9391bbc267bd9904fd"
 dependencies = [
 "log",
 "rustls 0.23.27",
 "rustls-native-certs 0.8.1",
 "rustls-pki-types",
 "rustls-webpki 0.103.3",
 ]
 [[package]]
 name = "rustls-native-certs"
 version = "0.6.3"
@@ -7070,6 +7453,15 @@ version = "1.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f"
 [[package]]
 name = "salsa20"
 version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213"
 dependencies = [
 "cipher",
 ]
 [[package]]
 name = "schannel"
 version = "0.1.27"
@@ -7132,6 +7524,17 @@ dependencies = [
 "tendril",
 ]
 [[package]]
 name = "scrypt"
 version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f"
 dependencies = [
 "pbkdf2 0.12.2",
 "salsa20",
 "sha2",
 ]
 [[package]]
 name = "sct"
 version = "0.7.1"
@@ -7723,6 +8126,9 @@ name = "spin"
 version = "0.9.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
 dependencies = [
 "lock_api",
 ]
 [[package]]
 name = "spki"
@@ -8027,6 +8433,18 @@ version = "0.12.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1"
 [[package]]
 name = "tcp-stream"
 version = "0.30.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b0217559d8c5a883d66838180671a74007e785689364ebe9627d517af936eaa9"
 dependencies = [
 "cfg-if",
 "p12-keystore",
 "rustls-connector",
 "rustls-pemfile 2.2.0",
 ]
 [[package]]
 name = "tempfile"
 version = "3.20.0"
@@ -8037,7 +8455,7 @@ dependencies = [
 "getrandom 0.3.3",
 "once_cell",
 "rustix 1.0.7",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 [[package]]
@@ -9234,7 +9652,7 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.59.0",
 ]
 [[package]]
@@ -9638,6 +10056,34 @@ dependencies = [
 "tap",
 ]
 [[package]]
 name = "x509-cert"
 version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94"
 dependencies = [
 "const-oid 0.9.6",
 "der 0.7.10",
 "spki 0.7.3",
 ]
 [[package]]
 name = "x509-parser"
 version = "0.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4569f339c0c402346d4a75a9e39cf8dad310e287eef1ff56d4c68e5067f53460"
 dependencies = [
 "asn1-rs",
 "data-encoding",
 "der-parser",
 "lazy_static",
 "nom 7.1.3",
 "oid-registry",
 "rusticata-macros",
 "thiserror 2.0.12",
 "time",
 ]
 [[package]]
 name = "xmlparser"
 version = "0.13.6"
@@ -26,7 +26,3 @@ disallowed-methods = [
    # Prefer to use Object::delete(&self)
    "revolt_database::models::bots::ops::AbstractBots::delete_bot",
 ]
 disallowed-types = [
    "rocket::serde::json::Json",
 ]
@@ -34,7 +34,7 @@ services:
      - minio
    entrypoint: >
      /bin/sh -c "while ! /usr/bin/mc ready minio; do
-        /usr/bin/mc alias set minio http://minio:9000 minioautumn minioautumn;
+        /usr/bin/mc config host add minio http://minio:9000 minioautumn minioautumn;
        echo 'Waiting minio...' && sleep 1;
      done; /usr/bin/mc mb minio/revolt-uploads; exit 0;"
@@ -9,6 +9,7 @@ edition = "2021"
 [dependencies]
 # util
 log = "*"
 rand = "*"
 sentry = "0.31.5"
 lru = "0.7.6"
 ulid = "0.5.0"
@@ -37,7 +38,8 @@ async-std = { version = "1.8.0", features = [
 # core
 authifier = { version = "1.0.15" }
-revolt-result = { path = "../core/result", features = ["sentry"] }
+revolt-result = { path = "../core/result" }
 revolt-broker = { path = "../core/broker" }
 revolt-models = { path = "../core/models" }
 revolt-config = { path = "../core/config" }
 revolt-database = { path = "../core/database" }
@@ -46,3 +48,6 @@ revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
 # redis
 fred = { version = "8.0.1", features = ["subscriber-client"] }
 # rabbit
 lapin = { version = "3.0.0" }
@@ -0,0 +1,129 @@
 use async_std::{net::TcpStream, sync::Mutex};
 use async_tungstenite::WebSocketStream;
 use futures::{join, SinkExt, StreamExt, TryStreamExt};
 use revolt_config::report_internal_error;
 use revolt_database::{
    events::{client::EventV1, server::ClientMessage},
    iso8601_timestamp::Timestamp,
    Database, User, UserHint,
 };
 use revolt_presence::{create_session, delete_session};
 use revolt_result::create_error;
 use crate::{
    client::{
        subscriber::client_subscriber,
        worker::{client_worker, WorkerRef},
    },
    config::ProtocolConfiguration,
    events::state::State,
 };
 /// Core event loop of gateway clients
 pub async fn client_core(
    db: &'static Database,
    ws: WebSocketStream<TcpStream>,
    mut config: ProtocolConfiguration,
 ) {
    // Split the socket for simultaneously read and write.
    let (mut write, mut read) = ws.split();
    // If the user has not provided authentication, request information.
    if config.get_session_token().is_none() {
        while let Ok(Some(message)) = read.try_next().await {
            if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
                config.set_session_token(token);
                break;
            }
        }
    }
    // Try to authenticate the user.
    let Some(token) = config.get_session_token().as_ref() else {
        write
            .send(config.encode(&EventV1::Error {
                data: create_error!(InvalidSession),
            }))
            .await
            .ok();
        return;
    };
    let (user, session_id) = match User::from_token(db, token, UserHint::Any).await {
        Ok(user) => user,
        Err(err) => {
            write
                .send(config.encode(&EventV1::Error { data: err }))
                .await
                .ok();
            return;
        }
    };
    info!(
        "Authenticated user {}#{}",
        user.username, user.discriminator
    );
    db.update_session_last_seen(&session_id, Timestamp::now_utc())
        .await
        .ok();
    // Create local state.
    let mut state = State::from(user, session_id);
    let user_id = state.cache.user_id.clone();
    // Notify socket we have authenticated.
    if report_internal_error!(write.send(config.encode(&EventV1::Authenticated)).await).is_err() {
        return;
    }
    // Download required data to local cache and send Ready payload.
    let ready_payload = match report_internal_error!(
        state
            .generate_ready_payload(db, config.get_ready_payload_fields())
            .await
    ) {
        Ok(ready_payload) => ready_payload,
        Err(_) => return,
    };
    if report_internal_error!(write.send(config.encode(&ready_payload)).await).is_err() {
        return;
    }
    // Create presence session.
    let (first_session, session_id) = create_session(&user_id, 0).await;
    // If this was the first session, notify other users that we just went online.
    if first_session {
        state.broadcast_presence_change(true).await;
    }
    {
        let worker_ref = WorkerRef::from(&state);
        let write = Mutex::new(write);
        let (reload, reloaded) = async_channel::bounded(1);
        let (cancel_1, cancelled_1) = async_channel::bounded(1);
        let (cancel_2, cancelled_2) = async_channel::bounded(1);
        join!(
            async {
                client_subscriber(&write, cancelled_1, reloaded, &config, db, &mut state).await;
                cancel_2.send(()).await.ok();
            },
            async {
                client_worker(read, &write, cancelled_2, reload, &config, worker_ref).await;
                cancel_1.send(()).await.ok();
            }
        );
    }
    // Clean up presence session.
    let last_session = delete_session(&user_id, session_id).await;
    // If this was the last session, notify other users that we just went offline.
    if last_session {
        state.broadcast_presence_change(false).await;
    }
 }
@@ -0,0 +1,3 @@
 pub mod core;
 pub mod subscriber;
 pub mod worker;
@@ -0,0 +1,112 @@
 use async_channel::Receiver;
 use async_std::{net::TcpStream, sync::Mutex};
 use async_tungstenite::WebSocketStream;
 use authifier::AuthifierEvent;
 use futures::{pin_mut, select, stream::SplitSink, FutureExt, SinkExt};
 use revolt_broker::event_stream;
 use revolt_database::{events::client::EventV1, Database};
 use sentry::Level;
 use crate::{
    config::ProtocolConfiguration,
    events::state::{State, SubscriptionStateChange},
 };
 /// Event subscriber loop
 pub async fn client_subscriber(
    write: &Mutex<SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>>,
    cancelled: Receiver<()>,
    reloaded: Receiver<()>,
    protocol_config: &ProtocolConfiguration,
    db: &'static Database,
    state: &mut State,
 ) {
    let mut consumer = event_stream::Consumer::new().await;
    consumer.set_topics(state.subscribed.read().await.clone());
    let mut cancel = false;
    loop {
        // Reload consumer if subscriptions change
        if !matches!(state.apply_state().await, SubscriptionStateChange::None) {
            consumer.set_topics(state.subscribed.read().await.clone());
        }
        // Read incoming events
        loop {
            let reloaded = reloaded.recv().fuse();
            let cancelled = cancelled.recv().fuse();
            let delivery = consumer.next().fuse();
            pin_mut!(delivery, reloaded, cancelled);
            select! {
                _ = reloaded => {
                    break;
                }
                _ = cancelled => {
                    cancel = true;
                    break;
                }
                event = delivery => {
                    if let Some(mut event) = event {
                        // Handle the event
                        if let EventV1::Auth(auth) = &event {
                            if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
                                if &state.session_id == session_id {
                                    event = EventV1::Logout;
                                }
                            } else if let AuthifierEvent::DeleteAllSessions {
                                exclude_session_id, ..
                            } = auth
                            {
                                if let Some(excluded) = exclude_session_id {
                                    if &state.session_id != excluded {
                                        event = EventV1::Logout;
                                    }
                                } else {
                                    event = EventV1::Logout;
                                }
                            }
                        } else {
                            let should_send = state.handle_incoming_event_v1(db, &mut event).await;
                            if !should_send {
                                continue;
                            }
                        }
                        let result = write.lock().await.send(protocol_config.encode(&event)).await;
                        if let Err(e) = result {
                            use async_tungstenite::tungstenite::Error;
                            if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
                                let err = format!("Error while sending an event: {e:?}");
                                warn!("{}", err);
                                sentry::capture_message(&err, Level::Warning);
                            }
                            cancel = true;
                            break;
                        }
                        if let EventV1::Logout = event {
                            info!("User {} received log out event!", state.user_id);
                            cancel = true;
                            break;
                        }
                        break;
                    } else {
                        cancel = true;
                        break;
                    }
                }
            }
        }
        // Break out if cancelled
        if cancel {
            break;
        }
    }
    consumer.dispose_channel().await;
 }
@@ -0,0 +1,124 @@
 use std::{collections::HashSet, sync::Arc};
 use async_channel::{Receiver, Sender};
 use async_std::{
    net::TcpStream,
    sync::{Mutex, RwLock},
 };
 use async_tungstenite::WebSocketStream;
 use futures::{
    pin_mut, select,
    stream::{SplitSink, SplitStream},
    FutureExt, SinkExt, TryStreamExt,
 };
 use revolt_database::events::{client::EventV1, server::ClientMessage};
 use sentry::Level;
 use crate::{config::ProtocolConfiguration, events::state::State};
 pub struct WorkerRef {
    user_id: String,
    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
    subscribed: Arc<RwLock<HashSet<String>>>,
 }
 impl WorkerRef {
    pub fn from(state: &State) -> WorkerRef {
        WorkerRef {
            user_id: state.user_id.clone(),
            active_servers: state.active_servers.clone(),
            subscribed: state.subscribed.clone(),
        }
    }
 }
 /// Incoming message handling
 pub async fn client_worker(
    mut read: SplitStream<WebSocketStream<TcpStream>>,
    write: &Mutex<SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>>,
    cancelled: Receiver<()>,
    reload: Sender<()>,
    config: &ProtocolConfiguration,
    state: WorkerRef,
 ) {
    loop {
        let read = read.try_next().fuse();
        let cancelled = cancelled.recv().fuse();
        pin_mut!(read, cancelled);
        select! {
            _ = cancelled => { return; },
            msg = read => {
                let msg = match msg {
                    Ok(Some(msg)) => msg,
                    Ok(None) => {
                        warn!("Received a None message!");
                        return;
                    }
                    Err(e) => {
                        use async_tungstenite::tungstenite::Error;
                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed | Error::Protocol(_)) {
                            let err = format!("Error while reading an event: {e:?}");
                            warn!("{}", err);
                            sentry::capture_message(&err, Level::Warning);
                        }
                        return;
                    }
                };
                let Ok(payload) = config.decode(&msg) else {
                    continue;
                };
                match payload {
                    ClientMessage::BeginTyping { channel } => {
                        if !state.subscribed.read().await.contains(&channel) {
                            continue;
                        }
                        EventV1::ChannelStartTyping {
                            id: channel.clone(),
                            user: state.user_id.clone(),
                        }
                        .p(channel.clone())
                        .await;
                    }
                    ClientMessage::EndTyping { channel } => {
                        if !state.subscribed.read().await.contains(&channel) {
                            continue;
                        }
                        EventV1::ChannelStopTyping {
                            id: channel.clone(),
                            user: state.user_id.clone(),
                        }
                        .p(channel.clone())
                        .await;
                    }
                    ClientMessage::Subscribe { server_id } => {
                        let mut servers = state.active_servers.lock().await;
                        let has_item = servers.contains_key(&server_id);
                        servers.insert(server_id, ());
                        if !has_item {
                            // Poke the listener to adjust subscriptions
                            reload.send(()).await.ok();
                        }
                    }
                    ClientMessage::Ping { data, responded } => {
                        if responded.is_none() {
                            write
                                .lock()
                                .await
                                .send(config.encode(&EventV1::Pong { data }))
                                .await
                                .ok();
                        }
                    }
                    _ => {}
                }
            }
        }
    }
 }
@@ -1,7 +1,7 @@
 use async_tungstenite::tungstenite::{handshake, Message};
 use futures::channel::oneshot::Sender;
 use revolt_database::events::client::ReadyPayloadFields;
-use revolt_result::{create_error, Result, ToRevoltError};
+use revolt_result::{create_error, Result};
 use serde::{Deserialize, Serialize};
 /// Enumeration of supported protocol formats
@@ -38,22 +38,16 @@ impl ProtocolConfiguration {
        match self.format {
            ProtocolFormat::Json => {
                if let Message::Text(text) = msg {
-                    // Log the error in-case we make a breaking change to the payload
+                    serde_json::from_str(text).map_err(|_| create_error!(InternalError))
                    serde_json::from_str(text)
                        .capture_error()
                        .map_err(|_| create_error!(UnprocessableEntity))
                } else {
-                    Err(create_error!(UnprocessableEntity))
+                    Err(create_error!(InternalError))
                }
            }
            ProtocolFormat::Msgpack => {
                if let Message::Binary(buf) = msg {
-                    rmp_serde::from_slice(buf)
+                    rmp_serde::from_slice(buf).map_err(|_| create_error!(InternalError))
                        .capture_error()
                        .map_err(|_| create_error!(UnprocessableEntity))
                } else {
-                    Err(create_error!(UnprocessableEntity))
+                    Err(create_error!(InternalError))
                }
            }
        }
@@ -66,6 +66,7 @@ impl Default for Cache {
 pub struct State {
    pub cache: Cache,
    pub user_id: String,
    pub session_id: String,
    pub private_topic: String,
    pub state: SubscriptionStateChange,
@@ -87,6 +88,7 @@ impl State {
            ..Default::default()
        };
        let user_id = user.id.clone();
        cache.users.insert(user.id.clone(), user);
        State {
@@ -96,6 +98,7 @@ impl State {
                Duration::from_secs(900),
                5,
            ))),
            user_id,
            session_id,
            private_topic,
            state: SubscriptionStateChange::Reset,
@@ -6,10 +6,11 @@ use revolt_presence::clear_region;
 #[macro_use]
 extern crate log;
-pub mod config;
+pub mod client;
 pub mod events;
 mod config;
 mod database;
 mod events;
 mod websocket;
 #[async_std::main]
@@ -1,39 +1,11 @@
-use std::{collections::HashSet, net::SocketAddr, sync::Arc};
+use std::net::SocketAddr;
-use async_tungstenite::WebSocketStream;
+use async_std::net::TcpStream;
-use authifier::AuthifierEvent;
+use futures::channel::oneshot;
-use fred::{
+use revolt_database::Database;
    error::RedisErrorKind,
    interfaces::{ClientLike, EventInterface, PubsubInterface},
    types::RedisConfig,
 };
 use futures::{
    channel::oneshot,
    join, pin_mut, select,
    stream::{SplitSink, SplitStream},
    FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
 use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
 use revolt_database::{
    events::{client::EventV1, server::ClientMessage},
    iso8601_timestamp::Timestamp,
    Database, User, UserHint,
 };
 use revolt_presence::{create_session, delete_session};
-use async_std::{
+use crate::client::core::client_core;
-    net::TcpStream,
+use crate::config::WebsocketHandshakeCallback;
    sync::{Mutex, RwLock},
    task::spawn,
 };
 use revolt_result::{create_error, ToRevoltError};
 use sentry::Level;
 use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
 type WsReader = SplitStream<WebSocketStream<TcpStream>>;
 type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
 /// Start a new WebSocket client worker given access to the database,
 /// the relevant TCP stream and the remote address of the client.
@@ -53,7 +25,7 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
    };
    // Verify we've received a valid config, otherwise we should just drop the connection.
-    let Ok(mut config) = receiver.await else {
+    let Ok(config) = receiver.await else {
        return;
    };
@@ -63,447 +35,5 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
        config.get_protocol_format()
    );
-    // Split the socket for simultaneously read and write.
+    client_core(db, ws, config).await;
    let (mut write, mut read) = ws.split();
    // If the user has not provided authentication, request information.
    if config.get_session_token().is_none() {
        while let Ok(Some(message)) = read.try_next().await {
            if let Ok(ClientMessage::Authenticate { token }) = config.decode(&message) {
                config.set_session_token(token);
                break;
            }
        }
    }
    // Try to authenticate the user.
    let Some(token) = config.get_session_token().as_ref() else {
        write
            .send(config.encode(&EventV1::Error {
                data: create_error!(InvalidSession),
            }))
            .await
            .ok();
        return;
    };
    let (user, session_id) = match User::from_token(db, token, UserHint::Any).await {
        Ok(user) => user,
        Err(err) => {
            write
                .send(config.encode(&EventV1::Error { data: err }))
                .await
                .ok();
            return;
        }
    };
    info!("User {addr:?} authenticated as @{}", user.username);
    db.update_session_last_seen(&session_id, Timestamp::now_utc())
        .await
        .ok();
    // Create local state.
    let mut state = State::from(user, session_id);
    let user_id = state.cache.user_id.clone();
    // Notify socket we have authenticated.
    if write.send(config.encode(&EventV1::Authenticated)).await.to_internal_error().is_err() {
        return;
    }
    // Download required data to local cache and send Ready payload.
    let ready_payload = match state
            .generate_ready_payload(db, config.get_ready_payload_fields())
            .await
            .to_internal_error()
    {
        Ok(ready_payload) => ready_payload,
        Err(_) => return,
    };
    if write.send(config.encode(&ready_payload)).await.to_internal_error().is_err() {
        return;
    }
    // Create presence session.
    let (first_session, session_id) = create_session(&user_id, 0).await;
    // If this was the first session, notify other users that we just went online.
    if first_session {
        state.broadcast_presence_change(true).await;
    }
    {
        // Setup channels and mutexes
        let write = Mutex::new(write);
        let subscribed = state.subscribed.clone();
        let active_servers = state.active_servers.clone();
        let (topic_signal_s, topic_signal_r) = async_channel::unbounded();
        // TODO: this needs to be rewritten
        // Create channels through which the tasks can signal to each other they need to clean up
        let (kill_signal_1_s, kill_signal_1_r) = async_channel::bounded(1);
        let (kill_signal_2_s, kill_signal_2_r) = async_channel::bounded(1);
        // Create a PubSub connection to poll on.
        let listener = listener_with_kill_signal(
            db,
            &mut state,
            addr,
            &config,
            topic_signal_r,
            kill_signal_1_r,
            &write,
            kill_signal_2_s,
        );
        // Read from WebSocket stream.
        let worker = worker_with_kill_signal(
            addr,
            subscribed,
            active_servers,
            user_id.clone(),
            &config,
            topic_signal_s,
            kill_signal_2_r,
            read,
            &write,
            kill_signal_1_s,
        );
        join!(listener, worker);
    }
    // Clean up presence session.
    let last_session = delete_session(&user_id, session_id).await;
    // If this was the last session, notify other users that we just went offline.
    if last_session {
        state.broadcast_presence_change(false).await;
    }
 }
 #[allow(clippy::too_many_arguments)]
 async fn listener_with_kill_signal(
    db: &'static Database,
    state: &mut State,
    addr: SocketAddr,
    config: &ProtocolConfiguration,
    topic_signal_r: async_channel::Receiver<()>,
    kill_signal_r: async_channel::Receiver<()>,
    write: &Mutex<WsWriter>,
    kill_signal_s: async_channel::Sender<()>,
 ) {
    listener(
        db,
        state,
        addr,
        config,
        topic_signal_r,
        kill_signal_r,
        write,
    )
    .await;
    kill_signal_s.send(()).await.ok();
 }
 async fn listener(
    db: &'static Database,
    state: &mut State,
    addr: SocketAddr,
    config: &ProtocolConfiguration,
    topic_signal_r: async_channel::Receiver<()>,
    kill_signal_r: async_channel::Receiver<()>,
    write: &Mutex<WsWriter>,
 ) {
    let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
    let subscriber = match fred::types::Builder::from_config(redis_config)
        .build_subscriber_client()
        .to_internal_error()
    {
        Ok(subscriber) => subscriber,
        Err(_) => return,
    };
    if subscriber.init().await.to_internal_error().is_err() {
        return;
    }
    // Handle Redis connection dropping
    let (clean_up_s, clean_up_r) = async_channel::bounded(1);
    let clean_up_s = Arc::new(Mutex::new(clean_up_s));
    subscriber.on_error(move |err| {
        if let RedisErrorKind::Canceled = err.kind() {
            let clean_up_s = clean_up_s.clone();
            spawn(async move {
                clean_up_s.lock().await.send(()).await.ok();
            });
        }
        Ok(())
    });
    let mut message_rx = subscriber.message_rx();
    'out: loop {
        // Check for state changes for subscriptions.
        match state.apply_state().await {
            SubscriptionStateChange::Reset => {
                if subscriber.unsubscribe_all().await.to_internal_error().is_err() {
                    break 'out;
                }
                let subscribed = state.subscribed.read().await;
                for id in subscribed.iter() {
                    if subscriber.subscribe(id).await.to_internal_error().is_err() {
                        break 'out;
                    }
                }
                #[cfg(debug_assertions)]
                info!("{addr:?} has reset their subscriptions");
            }
            SubscriptionStateChange::Change { add, remove } => {
                for id in remove {
                    #[cfg(debug_assertions)]
                    info!("{addr:?} unsubscribing from {id}");
                    if subscriber.unsubscribe(id).await.to_internal_error().is_err() {
                        break 'out;
                    }
                }
                for id in add {
                    #[cfg(debug_assertions)]
                    info!("{addr:?} subscribing to {id}");
                    if subscriber.subscribe(id).await.to_internal_error().is_err() {
                        break 'out;
                    }
                }
            }
            SubscriptionStateChange::None => {}
        }
        let t1 = message_rx.recv().fuse();
        let t2 = topic_signal_r.recv().fuse();
        let t3 = kill_signal_r.recv().fuse();
        let t4 = clean_up_r.recv().fuse();
        pin_mut!(t1, t2, t3, t4);
        select! {
            _ = t4 => {
                break 'out;
            },
            _ = t3 => {
                break 'out;
            },
            _ = t2 => {},
            message = t1 => {
                // Handle incoming events.
                let message = match message.to_internal_error() {
                    Ok(message) => message,
                    Err(_) => break 'out
                };
                let event = match *REDIS_PAYLOAD_TYPE {
                    PayloadType::Json => message
                        .value
                        .as_str()
                        .and_then(|s| serde_json::from_str::<EventV1>(s.as_ref()).to_internal_error().ok()),
                    PayloadType::Msgpack => message
                        .value
                        .as_bytes()
                        .and_then(|b| rmp_serde::from_slice::<EventV1>(b).to_internal_error().ok()),
                    PayloadType::Bincode => message
                        .value
                        .as_bytes()
                        .and_then(|b| bincode::deserialize::<EventV1>(b).to_internal_error().ok()),
                };
                let Some(mut event) = event else {
                    let err = format!(
                        "Failed to deserialise event for {}: `{:?}`",
                        message.channel,
                        message
                            .value
                    );
                    error!("{}", err);
                    sentry::capture_message(&err, Level::Error);
                    break 'out;
                };
                if let EventV1::Auth(auth) = &event {
                    if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
                        if &state.session_id == session_id {
                            event = EventV1::Logout;
                        }
                    } else if let AuthifierEvent::DeleteAllSessions {
                        exclude_session_id, ..
                    } = auth
                    {
                        if let Some(excluded) = exclude_session_id {
                            if &state.session_id != excluded {
                                event = EventV1::Logout;
                            }
                        } else {
                            event = EventV1::Logout;
                        }
                    }
                } else {
                    let should_send = state.handle_incoming_event_v1(db, &mut event).await;
                    if !should_send {
                        continue;
                    }
                }
                let result = write.lock().await.send(config.encode(&event)).await;
                if let Err(e) = result {
                    use async_tungstenite::tungstenite::Error;
                    if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
                        let err = format!("Error while sending an event to {addr:?}: {e:?}");
                        warn!("{}", err);
                        sentry::capture_message(&err, Level::Warning);
                    }
                    break 'out;
                }
                if let EventV1::Logout = event {
                    info!("User {addr:?} received log out event!");
                    break 'out;
                }
            }
        }
    }
    subscriber.quit().await.to_internal_error().ok();
 }
 #[allow(clippy::too_many_arguments)]
 async fn worker_with_kill_signal(
    addr: SocketAddr,
    subscribed: Arc<RwLock<HashSet<String>>>,
    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
    user_id: String,
    config: &ProtocolConfiguration,
    topic_signal_s: async_channel::Sender<()>,
    kill_signal_r: async_channel::Receiver<()>,
    read: WsReader,
    write: &Mutex<WsWriter>,
    kill_signal_s: async_channel::Sender<()>,
 ) {
    worker(
        addr,
        subscribed,
        active_servers,
        user_id,
        config,
        topic_signal_s,
        kill_signal_r,
        read,
        write,
    )
    .await;
    kill_signal_s.send(()).await.ok();
 }
 #[allow(clippy::too_many_arguments)]
 async fn worker(
    addr: SocketAddr,
    subscribed: Arc<RwLock<HashSet<String>>>,
    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
    user_id: String,
    config: &ProtocolConfiguration,
    topic_signal_s: async_channel::Sender<()>,
    kill_signal_r: async_channel::Receiver<()>,
    mut read: WsReader,
    write: &Mutex<WsWriter>,
 ) {
    loop {
        let t1 = read.try_next().fuse();
        let t2 = kill_signal_r.recv().fuse();
        pin_mut!(t1, t2);
        select! {
            _ = t2 => {
                return;
            },
            result = t1 => {
                let msg = match result {
                    Ok(Some(msg)) => msg,
                    Ok(None) => {
                        warn!("Received a None message!");
                        sentry::capture_message("Received a None message!", Level::Warning);
                        return;
                    }
                    Err(e) => {
                        use async_tungstenite::tungstenite::Error;
                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
                            let err = format!("Error while reading an event from {addr:?}: {e:?}");
                            warn!("{}", err);
                            sentry::capture_message(&err, Level::Warning);
                        }
                        return;
                    }
                };
                let Ok(payload) = config.decode(&msg) else {
                    continue;
                };
                match payload {
                    ClientMessage::BeginTyping { channel } => {
                        if !subscribed.read().await.contains(&channel) {
                            continue;
                        }
                        EventV1::ChannelStartTyping {
                            id: channel.clone(),
                            user: user_id.clone(),
                        }
                        .p(channel.clone())
                        .await;
                    }
                    ClientMessage::EndTyping { channel } => {
                        if !subscribed.read().await.contains(&channel) {
                            continue;
                        }
                        EventV1::ChannelStopTyping {
                            id: channel.clone(),
                            user: user_id.clone(),
                        }
                        .p(channel.clone())
                        .await;
                    }
                    ClientMessage::Subscribe { server_id } => {
                        let mut servers = active_servers.lock().await;
                        let has_item = servers.contains_key(&server_id);
                        servers.insert(server_id, ());
                        if !has_item {
                            // Poke the listener to adjust subscriptions
                            topic_signal_s.send(()).await.ok();
                        }
                    }
                    ClientMessage::Ping { data, responded } => {
                        if responded.is_none() {
                            write
                                .lock()
                                .await
                                .send(config.encode(&EventV1::Pong { data }))
                                .await
                                .ok();
                        }
                    }
                    _ => {}
                }
            }
        }
    }
 }
@@ -0,0 +1,25 @@
 [package]
 name = "revolt-broker"
 version = "0.8.8"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
 description = "Revolt Backend: Event Broker"
 [dependencies]
 # Utility
 log = "0.4"
 rand = "0.9.1"
 # RabbitMQ/AMQP client
 lapin = "3.0.0"
 # Async runtime
 async-std = { version = "1.8.0" }
 # Serialisation
 serde = "1"
 rmp-serde = "1.3.0"
 # Core
 revolt-config = { version = "0.8.8", path = "../config" }
@@ -0,0 +1,180 @@
 use std::{
    collections::{HashMap, HashSet},
    sync::Arc,
 };
 use async_std::stream::StreamExt;
 use lapin::{
    Channel, Connection,
    options::BasicAckOptions,
    types::{AMQPValue, FieldArray, FieldTable, LongLongInt},
 };
 use log::info;
 use rand::Rng;
 use revolt_config::{capture_internal_error, config};
 use serde::de::DeserializeOwned;
 use crate::event_stream::{create_channel, get_connection};
 pub struct Consumer {
    #[allow(dead_code)]
    conn: Arc<Connection>,
    channel: Channel,
    tag: String,
    topics: HashSet<String>,
    topics_changed: bool,
    consumer: Option<lapin::Consumer>,
    offset: Option<LongLongInt>,
 }
 impl Consumer {
    /// Create a new event stream consumer
    pub async fn new() -> Consumer {
        let config = config().await;
        let conn = get_connection().await;
        let channel = create_channel(&conn, config.rabbit.event_stream).await;
        Consumer {
            conn,
            channel,
            tag: rand::rng()
                .sample_iter::<char, _>(&rand::distr::StandardUniform)
                .take(32)
                .collect(),
            topics: HashSet::new(),
            topics_changed: false,
            consumer: None,
            offset: None,
        }
    }
    /// Update the set of topics
    pub fn set_topics(&mut self, topics: HashSet<String>) {
        self.topics = topics;
        self.topics_changed = true;
    }
    /// Get the current consumer
    pub async fn ensure_consumer(&mut self) {
        if self.topics_changed {
            info!("Topics changed, disposing the consumer.");
            self.dispose_consumer().await;
            self.topics_changed = false;
        }
        if self.consumer.is_none() {
            info!("Creating a new consumer, tag={}", self.tag);
            let config = config().await;
            // Build arguments for consumer
            let mut args: FieldTable = Default::default();
            // Configure stream filter to select topics we are listening for
            {
                let mut filter: FieldArray = Default::default();
                for topic in &self.topics {
                    filter.push(AMQPValue::LongString(topic.as_str().into()));
                }
                args.insert("x-stream-filter".into(), AMQPValue::FieldArray(filter));
            }
            // Set stream offset if applicable
            if let Some(offset) = self.offset {
                args.insert("x-stream-offset".into(), AMQPValue::LongLongInt(offset));
            }
            // Create the consumer
            self.consumer = Some(
                self.channel
                    .basic_consume(
                        &config.rabbit.event_stream.queue,
                        &self.tag,
                        Default::default(),
                        args,
                    )
                    .await
                    .unwrap(),
            );
        }
    }
    /// Close the active consumer if one exists
    pub async fn dispose_consumer(&mut self) {
        if let Some(consumer) = self.consumer.as_ref() {
            if consumer.state().is_active() {
                if let Err(err) = self
                    .channel
                    .basic_cancel(&self.tag, Default::default())
                    .await
                {
                    eprintln!("Failed to close consumer! {:?}", err);
                }
                // is this necessary?
                // else {
                // Read the consumer to the end
                //     while let Some(delivery) = consumer.next().await {
                //         let delivery = delivery.expect("error in consumer");
                //         delivery.ack(BasicAckOptions::default()).await.expect("ack");
                //     }
                // }
            }
            self.consumer = None;
        }
    }
    /// Close the active channel
    pub async fn dispose_channel(&mut self) {
        // Close the channel -- don't do this actually
        capture_internal_error!(self.channel.close(0, "closing channel").await);
    }
    /// Get the next item
    pub async fn next<T: DeserializeOwned>(&mut self) -> Option<T> {
        self.ensure_consumer().await;
        let consumer = self.consumer.as_mut().unwrap();
        while let Some(Ok(delivery)) = consumer.next().await {
            // Acknowledgement is required
            delivery.ack(BasicAckOptions::default()).await.expect("ack");
            // Parse the delivery headers
            let headers: HashMap<String, AMQPValue> = delivery
                .properties
                .headers()
                .as_ref()
                .map(|table| {
                    table
                        .into_iter()
                        .map(|(k, v)| (k.to_string(), v.clone()))
                        .collect()
                })
                .unwrap_or_default();
            // Keep track of the current offset
            let stream_offset = headers
                .get("x-stream-offset")
                .expect("`x-stream-offset` not present in message!");
            self.offset = Some(stream_offset.as_long_long_int().unwrap() + 1);
            // Client-side topic filtering (broker uses Bloom filter so may have false-positives)
            let filter_value = headers
                .get("x-stream-filter-value")
                .expect("`x-stream-filter-value` not present in message!")
                .as_long_string()
                .expect("`string`")
                .to_string();
            if self.topics.contains(&filter_value) {
                // Deserialise the data
                return Some(rmp_serde::from_slice(&delivery.data).expect("`data`"));
            }
        }
        None
    }
 }
@@ -0,0 +1,7 @@
 mod consumer;
 mod pool;
 mod publish;
 pub use consumer::Consumer;
 pub use pool::{create_channel, get_connection};
 pub use publish::publish_event;
@@ -0,0 +1,99 @@
 use async_std::sync::Mutex;
 use lapin::{
    Connection,
    options::QueueDeclareOptions,
    types::{AMQPValue, FieldTable},
 };
 use log::{debug, warn};
 use revolt_config::{RabbitEventStream, config};
 use std::sync::Arc;
 use crate::create_client;
 /// Get a handle to the event stream
 pub async fn get_connection() -> Arc<Connection> {
    let config = config().await;
    static CONNECTIONS: Mutex<Vec<Arc<Connection>>> = Mutex::new(Vec::new());
    let mut connections = CONNECTIONS.lock().await;
    connections.retain(|item| {
        if item.status().connected() {
            true
        } else {
            warn!(
                "Dropping connection with status {:?}",
                item.status().state()
            );
            false
        }
    });
    debug!(
        "Connections: {}, Clients: {:?}",
        connections.len(),
        connections
            .iter()
            .map(Arc::strong_count)
            .collect::<Vec<usize>>()
    );
    for conn in connections.iter() {
        if Arc::strong_count(conn) < config.rabbit.event_stream.channels_per_conn {
            return conn.clone();
        }
    }
    let conn = Arc::new(create_client().await);
    connections.push(conn.clone());
    conn
 }
 /// Create a channel
 pub async fn create_channel(
    conn: &lapin::Connection,
    event_stream: RabbitEventStream,
 ) -> lapin::Channel {
    let channel = conn.create_channel().await.unwrap();
    let mut args: FieldTable = Default::default();
    args.insert(
        // set queue type to stream
        "x-queue-type".into(),
        AMQPValue::LongString("stream".into()),
    );
    args.insert(
        // max. size of the stream
        "x-max-length-bytes".into(),
        AMQPValue::LongLongInt(event_stream.stream_max_length_bytes),
    );
    args.insert(
        // size of the Bloom filter
        "x-stream-filter-size-bytes".into(),
        AMQPValue::LongLongInt(event_stream.filter_size_bytes),
    );
    channel
        .queue_declare(
            &event_stream.queue,
            QueueDeclareOptions {
                durable: true,
                ..Default::default()
            },
            args,
        )
        .await
        .unwrap();
    channel
        .basic_qos(event_stream.qos_prefetch, Default::default())
        .await
        .unwrap();
    channel
 }
@@ -0,0 +1,37 @@
 use lapin::{
    Error,
    protocol::basic::AMQPProperties,
    publisher_confirm::PublisherConfirm,
    types::{AMQPValue, FieldTable},
 };
 use revolt_config::config;
 use serde::Serialize;
 use crate::event_stream::{create_channel, get_connection};
 /// Publish an event to the message broker
 pub async fn publish_event<T: Serialize>(
    channel: &str,
    data: &T,
 ) -> Result<PublisherConfirm, Error> {
    let config = config().await;
    let mut headers: FieldTable = Default::default();
    headers.insert(
        "x-stream-filter-value".into(),
        AMQPValue::LongString(channel.into()),
    );
    let conn = get_connection().await;
    create_channel(&conn, config.rabbit.event_stream.clone())
        .await
        .basic_publish(
            &config.rabbit.event_stream.exchange,
            &config.rabbit.event_stream.queue,
            Default::default(),
            &rmp_serde::to_vec_named(data).unwrap(),
            AMQPProperties::default().with_headers(headers),
        )
        .await
 }
@@ -0,0 +1,18 @@
 use revolt_config::config;
 pub mod event_stream;
 /// Create a lapin client
 pub async fn create_client() -> lapin::Connection {
    let config = config().await;
    lapin::Connection::connect(
        &format!(
            "amqp://{}:{}@{}:{}/%2f",
            config.rabbit.username, config.rabbit.password, config.rabbit.host, config.rabbit.port
        ),
        Default::default(),
    )
    .await
    .unwrap()
 }
@@ -11,9 +11,8 @@ description = "Revolt Backend: Configuration"
 [features]
 anyhow = ["dep:sentry-anyhow"]
 report-macros = ["revolt-result"]
 sentry = ["dep:sentry"]
 test = ["async-std"]
-default = ["test", "sentry"]
+default = ["test", "anyhow"]
 [dependencies]
 # Utility
@@ -33,7 +32,7 @@ log = "0.4.14"
 pretty_env_logger = "0.4.0"
 # Sentry
-sentry = { version = "0.31.5", optional = true }
+sentry = "0.31.5"
 sentry-anyhow = { version = "0.38.1", optional = true }
 # Core
@@ -28,6 +28,20 @@ port = 5672
 username = "rabbituser"
 password = "rabbitpass"
 [rabbit.event_stream]
 # Configuration for event brokerage
 # Using default/direct exchange
 exchange = ""
 queue = "revolt.events"
 # Number of channels that can be opened per single TCP connection
 channels_per_conn = 128
 # Maximum size of the stream
 stream_max_length_bytes = 5_000_000_000
 # Size of the Bloom filter
 filter_size_bytes = 26
 # Number of messages to prefetch
 qos_prefetch = 100
 [api]
 [api.registration]
@@ -6,12 +6,10 @@ use futures_locks::RwLock;
 use once_cell::sync::Lazy;
 use serde::Deserialize;
 #[cfg(feature = "sentry")]
 pub use sentry::{capture_error, capture_message, Level};
 #[cfg(feature = "anyhow")]
 pub use sentry_anyhow::capture_anyhow;
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! report_error {
    ( $expr: expr, $error: ident $( $tt:tt )? ) => {
@@ -26,7 +24,7 @@ macro_rules! report_error {
    };
 }
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! capture_internal_error {
    ( $expr: expr ) => {
@@ -37,7 +35,7 @@ macro_rules! capture_internal_error {
    };
 }
-#[cfg(all(feature = "report-macros", feature = "sentry"))]
+#[cfg(feature = "report-macros")]
 #[macro_export]
 macro_rules! report_internal_error {
    ( $expr: expr ) => {
@@ -110,12 +108,24 @@ pub struct Database {
    pub redis: String,
 }
 #[derive(Deserialize, Debug, Clone)]
 pub struct RabbitEventStream {
    pub exchange: String,
    pub queue: String,
    pub channels_per_conn: usize,
    pub stream_max_length_bytes: i64,
    pub filter_size_bytes: i64,
    pub qos_prefetch: u16,
 }
 #[derive(Deserialize, Debug, Clone)]
 pub struct Rabbit {
    pub host: String,
    pub port: u16,
    pub username: String,
    pub password: String,
    pub event_stream: RabbitEventStream,
 }
 #[derive(Deserialize, Debug, Clone)]
@@ -421,7 +431,6 @@ pub async fn config() -> Settings {
 }
 /// Configure logging and common Rust variables
 #[cfg(feature = "sentry")]
 pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry::ClientInitGuard> {
    if std::env::var("RUST_LOG").is_err() {
        std::env::set_var("RUST_LOG", "info");
@@ -447,7 +456,6 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry:
    }
 }
 #[cfg(feature = "sentry")]
 #[macro_export]
 macro_rules! configure {
    ($application: ident) => {
@@ -10,12 +10,12 @@ description = "Revolt Backend: Database Implementation"
 [features]
 # Databases
-mongodb = ["dep:mongodb", "bson", "authifier/database-mongodb"]
+mongodb = ["dep:mongodb", "bson"]
 # ... Other
 tasks = ["isahc", "linkify", "url-escape"]
-async-std-runtime = ["async-std", "authifier/async-std-runtime"]
+async-std-runtime = ["async-std"]
-rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi", "authifier/rocket_impl"]
+rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi"]
 axum-impl = ["axum"]
 redis-is-patched = ["revolt-presence/redis-is-patched"]
@@ -37,6 +37,7 @@ revolt-permissions = { version = "0.8.8", path = "../permissions", features = [
    "bson",
 ] }
 revolt-parser = { version = "0.8.8", path = "../parser" }
 revolt-broker = { version = "0.8.8", path = "../broker" }
 # Utility
 log = "0.4"
@@ -56,6 +57,7 @@ isahc = { optional = true, version = "1.7", features = ["json"] }
 # Serialisation
 serde_json = "1"
 rmp-serde = "1.0.0"
 revolt_optional_struct = "0.2.0"
 serde = { version = "1", features = ["derive"] }
 iso8601-timestamp = { version = "0.2.10", features = ["serde", "bson"] }
@@ -91,8 +93,14 @@ rocket = { version = "0.5.1", default-features = false, features = [
 revolt_okapi = { version = "0.9.1", optional = true }
 revolt_rocket_okapi = { version = "0.10.0", optional = true }
 # Notifications
 fcm_v1 = "0.3.0"
 web-push = "0.10.0"
 revolt_a2 = { version = "0.10", default-features = false, features = ["ring"] }
 # Authifier
-authifier = { version = "1.0.15" }
+authifier = { version = "1.0.15", features = ["rocket_impl"] }
 # RabbitMQ
 amqprs = { version = "1.7.0" }
 lapin = { version = "3.0.0" }
@@ -1,4 +1,3 @@
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -14,7 +13,6 @@ use authifier::Authifier;
 use rand::Rng;
 use revolt_config::config;
 #[cfg(feature = "mongodb")]
 pub use self::mongodb::*;
 pub use self::reference::*;
@@ -27,10 +25,8 @@ pub enum DatabaseInfo {
    /// Use the mock database
    Reference,
    /// Connect to MongoDB
    #[cfg(feature = "mongodb")]
    MongoDb { uri: String, database_name: String },
    /// Use existing MongoDB connection
    #[cfg(feature = "mongodb")]
    MongoDbFromClient(::mongodb::Client, String),
 }
@@ -40,7 +36,6 @@ pub enum Database {
    /// Mock database
    Reference(ReferenceDb),
    /// MongoDB database
    #[cfg(feature = "mongodb")]
    MongoDb(MongoDb),
 }
@@ -50,7 +45,7 @@ impl DatabaseInfo {
    pub async fn connect(self) -> Result<Database, String> {
        let config = config().await;
-        match self {
+        Ok(match self {
            DatabaseInfo::Auto => {
                if std::env::var("TEST_DB").is_ok() {
                    DatabaseInfo::Test(format!(
@@ -58,20 +53,16 @@ impl DatabaseInfo {
                        rand::thread_rng().gen_range(1_000_000..10_000_000)
                    ))
                    .connect()
-                    .await
+                    .await?
                } else if !config.database.mongodb.is_empty() {
-                    #[cfg(feature = "mongodb")]
+                    DatabaseInfo::MongoDb {
                    return DatabaseInfo::MongoDb {
                        uri: config.database.mongodb,
                        database_name: "revolt".to_string(),
                    }
                    .connect()
-                    .await;
+                    .await?
                    #[cfg(not(feature = "mongodb"))]
                    return Err("MongoDB not enabled.".to_string())
                } else {
-                    DatabaseInfo::Reference.connect().await
+                    DatabaseInfo::Reference.connect().await?
                }
            }
            DatabaseInfo::Test(database_name) => {
@@ -79,36 +70,30 @@ impl DatabaseInfo {
                    .expect("`TEST_DB` environment variable should be set to REFERENCE or MONGODB")
                    .as_str()
                {
-                    "REFERENCE" => DatabaseInfo::Reference.connect().await,
+                    "REFERENCE" => DatabaseInfo::Reference.connect().await?,
                    "MONGODB" => {
-                        #[cfg(feature = "mongodb")]
+                        DatabaseInfo::MongoDb {
                        return DatabaseInfo::MongoDb {
                            uri: config.database.mongodb,
                            database_name,
                        }
                        .connect()
-                        .await;
+                        .await?
                        #[cfg(not(feature = "mongodb"))]
                        return Err("MongoDB not enabled.".to_string())
                    }
                    _ => unreachable!("must specify REFERENCE or MONGODB"),
                }
            }
-            DatabaseInfo::Reference => Ok(Database::Reference(Default::default())),
+            DatabaseInfo::Reference => Database::Reference(Default::default()),
            #[cfg(feature = "mongodb")]
            DatabaseInfo::MongoDb { uri, database_name } => {
                let client = ::mongodb::Client::with_uri_str(uri)
                    .await
                    .map_err(|_| "Failed to init db connection.".to_string())?;
-                Ok(Database::MongoDb(MongoDb(client, database_name)))
+                Database::MongoDb(MongoDb(client, database_name))
            }
            #[cfg(feature = "mongodb")]
            DatabaseInfo::MongoDbFromClient(client, database_name) => {
-                Ok(Database::MongoDb(MongoDb(client, database_name)))
+                Database::MongoDb(MongoDb(client, database_name))
            }
-        }
+        })
    }
 }
@@ -234,16 +219,12 @@ impl Database {
        Authifier {
            database: match self {
                Database::Reference(_) => Default::default(),
                #[cfg(feature = "mongodb")]
                Database::MongoDb(MongoDb(client, _)) => authifier::Database::MongoDb(
                    authifier::database::MongoDb(client.database("revolt")),
                ),
            },
            config: auth_config,
            #[cfg(feature = "tasks")]
            event_channel: Some(crate::tasks::authifier_relay::sender()),
            #[cfg(not(feature = "tasks"))]
            event_channel: None,
        }
    }
 }
@@ -10,6 +10,7 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 database_derived!(
    #[cfg(feature = "mongodb")]
    /// MongoDB implementation
    pub struct MongoDb(pub ::mongodb::Client, pub String);
 );
@@ -1,4 +1,5 @@
 use authifier::AuthifierEvent;
 use revolt_broker::event_stream;
 use revolt_result::Error;
 use serde::{Deserialize, Serialize};
@@ -253,14 +254,16 @@ pub enum EventV1 {
 impl EventV1 {
    /// Publish helper wrapper
    pub async fn p(self, channel: String) {
        #[cfg(not(debug_assertions))]
        redis_kiss::p(channel, self).await;
        #[cfg(debug_assertions)]
        info!("Publishing event to {channel}: {self:?}");
        let result = event_stream::publish_event(&channel, &self).await;
        #[cfg(not(debug_assertions))]
        result.ok();
        #[cfg(debug_assertions)]
-        redis_kiss::publish(channel, self).await.unwrap();
+        result.unwrap();
    }
    /// Publish user event
@@ -25,9 +25,6 @@ pub use mongodb;
 #[macro_use]
 extern crate bson;
 #[cfg(not(feature = "async-std-runtime"))]
 compile_error!("async-std-runtime feature must be enabled.");
 #[macro_export]
 #[cfg(debug_assertions)]
 macro_rules! query {
@@ -106,7 +103,6 @@ pub mod util;
 pub use models::*;
 pub mod events;
 #[cfg(feature = "tasks")]
 pub mod tasks;
 mod amqp;
@@ -1,4 +1,3 @@
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::{Bot, FieldsBot, PartialBot};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::Invite;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::ChannelUnread;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::{FieldsWebhook, PartialWebhook, Webhook};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -8,13 +8,10 @@ use serde::{Deserialize, Serialize};
 use ulid::Ulid;
 use crate::{
-    events::client::EventV1, Database, File, PartialServer,
+    events::client::EventV1, tasks::ack::AckEvent, Database, File, IntoDocumentPath, PartialServer,
    Server, SystemMessage, User, AMQP,
 };
 #[cfg(feature = "mongodb")]
 use crate::IntoDocumentPath;
 auto_derived!(
    #[serde(tag = "channel_type")]
    pub enum Channel {
@@ -649,11 +646,10 @@ impl Channel {
        .private(user.to_string())
        .await;
        #[cfg(feature = "tasks")]
        crate::tasks::ack::queue_ack(
            self.id().to_string(),
            user.to_string(),
-            crate::tasks::ack::AckEvent::AckMessage {
+            AckEvent::AckMessage {
                id: message.to_string(),
            },
        )
@@ -770,7 +766,6 @@ impl Channel {
    }
 }
 #[cfg(feature = "mongodb")]
 impl IntoDocumentPath for FieldsChannel {
    fn as_path(&self) -> Option<&'static str> {
        Some(match self {
@@ -1,7 +1,5 @@
 use crate::{revolt_result::Result, Channel, FieldsChannel, PartialChannel};
 use revolt_permissions::OverrideField;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::Emoji;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::FileHash;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -4,7 +4,6 @@ use crate::File;
 use super::FileUsedFor;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -1,7 +1,7 @@
 use bson::to_document;
 use bson::Document;
 use revolt_config::report_internal_error;
 use revolt_result::Result;
 use revolt_result::ToRevoltError;
 use crate::File;
 use crate::FileUsedFor;
@@ -106,7 +106,7 @@ impl AbstractAttachments for MongoDb {
                },
                doc! {
                    "$set": {
-                        "used_for": to_document(&used_for).to_internal_error()?,
+                        "used_for": report_internal_error!(to_document(&used_for))?,
                        "uploader_id": uploader_id
                    }
                },
@@ -10,9 +10,11 @@ use revolt_models::v0::{
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue};
 use revolt_result::{ErrorType, Result};
 use ulid::Ulid;
 use validator::Validate;
 use crate::{
    events::client::EventV1,
    tasks::{self, ack::AckEvent},
    util::{
        bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
        permissions::DatabasePermissionQuery,
@@ -20,9 +22,6 @@ use crate::{
    Channel, Database, Emoji, File, User, AMQP,
 };
 #[cfg(feature = "tasks")]
 use crate::tasks::{self, ack::AckEvent};
 auto_derived_partial!(
    /// Message
    pub struct Message {
@@ -488,28 +487,31 @@ impl Message {
                | Channel::VoiceChannel { ref server, .. } => {
                    let mentions_vec = Vec::from_iter(user_mentions.iter().cloned());
-                    let valid_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await?;
+                    let valid_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await;
                    if let Ok(valid_members) = valid_members {
                        let valid_mentions = HashSet::<&String, RandomState>::from_iter(
                            valid_members.iter().map(|m| &m.id.user),
                        );
-                    let valid_mentions = HashSet::<&String, RandomState>::from_iter(
+                        user_mentions.retain(|m| valid_mentions.contains(m)); // quick pass, validate mentions are in the server
                        valid_members.iter().map(|m| &m.id.user),
                    );
-                    user_mentions.retain(|m| valid_mentions.contains(m)); // quick pass, validate mentions are in the server
+                        if !user_mentions.is_empty() {
                            // if there are still mentions, drill down to a channel-level
                            let member_channel_view_perms =
                                BulkDatabasePermissionQuery::from_server_id(db, server)
                                    .await
                                    .channel(&channel)
                                    .members(&valid_members)
                                    .members_can_see_channel()
                                    .await;
-                    if !user_mentions.is_empty() {
+                            user_mentions
-                        // if there are still mentions, drill down to a channel-level
+                                .retain(|m| *member_channel_view_perms.get(m).unwrap_or(&false));
-                        let member_channel_view_perms =
+                        }
-                            BulkDatabasePermissionQuery::from_server_id(db, server)
+                    } else {
-                                .await
+                        revolt_config::capture_error(&valid_members.unwrap_err());
-                                .channel(&channel)
+                        return Err(create_error!(InternalError));
                                .members(&valid_members)
                                .members_can_see_channel()
                                .await;
                        user_mentions
                            .retain(|m| *member_channel_view_perms.get(m).unwrap_or(&false));
                    }
                }
                Channel::SavedMessages { .. } => {
                    user_mentions.clear();
@@ -614,11 +616,9 @@ impl Message {
            .await;
        // Update last_message_id
        #[cfg(feature = "tasks")]
        tasks::last_message_id::queue(self.channel.to_string(), self.id.to_string(), is_dm).await;
        // Add mentions for affected users
        #[cfg(feature = "tasks")]
        if !mentions_elsewhere {
            if let Some(mentions) = &self.mentions {
                tasks::ack::queue_message(
@@ -637,7 +637,6 @@ impl Message {
        }
        // Generate embeds
        #[cfg(feature = "tasks")]
        if generate_embeds {
            if let Some(content) = &self.content {
                tasks::process_embeds::queue(
@@ -674,12 +673,10 @@ impl Message {
        )
        .await?;
        if !self.has_suppressed_notifications()
            && (self.mentions.is_some() || self.contains_mass_push_mention())
        {
            // send Push notifications
            #[cfg(feature = "tasks")]
            tasks::ack::queue_message(
                self.channel.to_string(),
                AckEvent::ProcessMessage {
@@ -713,6 +710,12 @@ impl Message {
    /// Create text embed from sendable embed
    pub async fn create_embed(&self, db: &Database, embed: SendableEmbed) -> Result<Embed> {
        embed.validate().map_err(|error| {
            create_error!(FailedValidation {
                error: error.to_string()
            })
        })?;
        let media = if let Some(id) = embed.media {
            Some(File::use_attachment(db, &id, &self.id, &self.author).await?)
        } else {
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -38,10 +38,7 @@ pub use servers::*;
 pub use user_settings::*;
 pub use users::*;
-use crate::{Database, ReferenceDb};
+use crate::{Database, MongoDb, ReferenceDb};
 #[cfg(feature = "mongodb")]
 use crate::MongoDb;
 pub trait AbstractDatabase:
    Sync
@@ -69,8 +66,6 @@ pub trait AbstractDatabase:
 }
 impl AbstractDatabase for ReferenceDb {}
 #[cfg(feature = "mongodb")]
 impl AbstractDatabase for MongoDb {}
 impl std::ops::Deref for Database {
@@ -79,7 +74,6 @@ impl std::ops::Deref for Database {
    fn deref(&self) -> &Self::Target {
        match &self {
            Database::Reference(dummy) => dummy,
            #[cfg(feature = "mongodb")]
            Database::MongoDb(mongo) => mongo,
        }
    }
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::PolicyChange;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -1,8 +1,6 @@
 use std::time::Duration;
 use crate::{revolt_result::Result, RatelimitEvent, RatelimitEventType};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::Report;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::Snapshot;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::{MemberCompositeKey, ServerBan};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -1,20 +1,16 @@
-#[cfg(feature = "mongodb")]
+use ::mongodb::SessionCursor;
 use ::mongodb::{ClientSession, SessionCursor};
 use revolt_result::Result;
 use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
 #[derive(Debug)]
 #[allow(clippy::large_enum_variant)]
 pub enum ChunkedServerMembersGenerator {
    #[cfg(feature = "mongodb")]
    MongoDb {
-        session: ClientSession,
+        session: ::mongodb::ClientSession,
        cursor: Option<SessionCursor<Member>>,
    },
@@ -26,7 +22,7 @@ pub enum ChunkedServerMembersGenerator {
 impl ChunkedServerMembersGenerator {
    #[cfg(feature = "mongodb")]
-    pub fn new_mongo(session: ClientSession, cursor: SessionCursor<Member>) -> Self {
+    pub fn new_mongo(session: ::mongodb::ClientSession, cursor: SessionCursor<Member>) -> Self {
        ChunkedServerMembersGenerator::MongoDb {
            session,
            cursor: Some(cursor),
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::{FieldsRole, FieldsServer, PartialRole, PartialServer, Role, Server};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -2,7 +2,6 @@ use revolt_result::Result;
 use crate::UserSettings;
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -705,8 +705,6 @@ impl User {
        duration_days: Option<usize>,
        reason: Option<Vec<String>>,
    ) -> Result<()> {
        // TODO: authifier Error should implement Error
        let authifier = db.clone().to_authifier().await;
        let mut account = authifier
            .database
@@ -4,7 +4,6 @@ use revolt_result::Result;
 use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
 #[cfg(feature = "mongodb")]
 mod mongodb;
 mod reference;
@@ -212,34 +212,16 @@ impl AbstractUsers for MongoDb {
        partial: &PartialUser,
        remove: Vec<FieldsUser>,
    ) -> Result<()> {
-        if remove.contains(&FieldsUser::StatusText) && partial.status.is_some() {
+        query!(
-            // stupid-ass workaround to fix mongo conflicting the same item
+            self,
-            let _: Result<()> = query!(
+            update_one_by_id,
-                self,
+            COL,
-                update_one_by_id,
+            id,
-                COL,
+            partial,
-                id,
+            remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
-                PartialUser {
+            None
-                    ..Default::default()
+        )
-                },
+        .map(|_| ())
                remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
                None
            )
            .map(|_| ());
            query!(self, update_one_by_id, COL, id, partial, vec![], None).map(|_| ())
        } else {
            query!(
                self,
                update_one_by_id,
                COL,
                id,
                partial,
                remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
                None
            )
            .map(|_| ())
        }
    }
    /// Set relationship with another user
@@ -38,7 +38,6 @@ impl<'r> FromRequest<'r> for User {
        if let Some(user) = user {
            Outcome::Success(user.clone())
        } else {
            request.local_cache(|| Some(create_error!(InvalidSession)));
            Outcome::Error((Status::Unauthorized, authifier::Error::InvalidSession))
        }
    }
@@ -113,8 +113,6 @@ impl<'r> FromRequest<'r> for IdempotencyKey {
            let idempotency = IdempotencyKey { key };
            let mut cache = TOKEN_CACHE.lock().await;
            if cache.get(&idempotency.key).is_some() {
                request.local_cache(|| Some(create_error!(DuplicateNonce)));
                return Outcome::Error((Status::Conflict, create_error!(DuplicateNonce)));
            }
@@ -14,40 +14,41 @@ use crate::{
 };
 /// Reference to some object in the database
-pub struct Reference<'a> {
+#[derive(Serialize, Deserialize)]
 pub struct Reference {
    /// Id of object
-    pub id: &'a str,
+    pub id: String,
 }
-impl<'a> Reference<'a> {
+impl Reference {
    /// Create a Ref from an unchecked string
-    pub fn from_unchecked(id: &'a str) -> Reference<'a> {
+    pub fn from_unchecked(id: String) -> Reference {
        Reference { id }
    }
    /// Fetch ban from Ref
    pub async fn as_ban(&self, db: &Database, server: &str) -> Result<ServerBan> {
-        db.fetch_ban(server, self.id).await
+        db.fetch_ban(server, &self.id).await
    }
    /// Fetch bot from Ref
    pub async fn as_bot(&self, db: &Database) -> Result<Bot> {
-        db.fetch_bot(self.id).await
+        db.fetch_bot(&self.id).await
    }
    /// Fetch emoji from Ref
    pub async fn as_emoji(&self, db: &Database) -> Result<Emoji> {
-        db.fetch_emoji(self.id).await
+        db.fetch_emoji(&self.id).await
    }
    /// Fetch channel from Ref
    pub async fn as_channel(&self, db: &Database) -> Result<Channel> {
-        db.fetch_channel(self.id).await
+        db.fetch_channel(&self.id).await
    }
    /// Fetch invite from Ref or create invite to server if discoverable
    pub async fn as_invite(&self, db: &Database) -> Result<Invite> {
-        if ulid::Ulid::from_str(self.id).is_ok() {
+        if ulid::Ulid::from_str(&self.id).is_ok() {
            let server = self.as_server(db).await?;
            if !server.discoverable {
                return Err(create_error!(NotFound));
@@ -64,18 +65,18 @@ impl<'a> Reference<'a> {
                    .ok_or(create_error!(NotFound))?,
            })
        } else {
-            db.fetch_invite(self.id).await
+            db.fetch_invite(&self.id).await
        }
    }
    /// Fetch message from Ref
    pub async fn as_message(&self, db: &Database) -> Result<Message> {
-        db.fetch_message(self.id).await
+        db.fetch_message(&self.id).await
    }
    /// Fetch message from Ref and validate channel
    pub async fn as_message_in_channel(&self, db: &Database, channel: &str) -> Result<Message> {
-        let msg = db.fetch_message(self.id).await?;
+        let msg = db.fetch_message(&self.id).await?;
        if msg.channel != channel {
            return Err(create_error!(NotFound));
        }
@@ -85,36 +86,36 @@ impl<'a> Reference<'a> {
    /// Fetch member from Ref
    pub async fn as_member(&self, db: &Database, server: &str) -> Result<Member> {
-        db.fetch_member(server, self.id).await
+        db.fetch_member(server, &self.id).await
    }
    /// Fetch server from Ref
    pub async fn as_server(&self, db: &Database) -> Result<Server> {
-        db.fetch_server(self.id).await
+        db.fetch_server(&self.id).await
    }
    /// Fetch user from Ref
    pub async fn as_user(&self, db: &Database) -> Result<User> {
-        db.fetch_user(self.id).await
+        db.fetch_user(&self.id).await
    }
    /// Fetch webhook from Ref
    pub async fn as_webhook(&self, db: &Database) -> Result<Webhook> {
-        db.fetch_webhook(self.id).await
+        db.fetch_webhook(&self.id).await
    }
 }
 #[cfg(feature = "rocket-impl")]
-impl<'r> FromParam<'r> for Reference<'r> {
+impl<'r> FromParam<'r> for Reference {
    type Error = &'r str;
    fn from_param(param: &'r str) -> Result<Self, Self::Error> {
-        Ok(Reference::from_unchecked(param))
+        Ok(Reference::from_unchecked(param.into()))
    }
 }
 #[cfg(feature = "rocket-impl")]
-impl<'a> JsonSchema for Reference<'a> {
+impl JsonSchema for Reference {
    fn schema_name() -> String {
        "Id".to_string()
    }
@@ -14,7 +14,7 @@ imagesize = "0.13.0"
 tempfile = "3.12.0"
 base64 = "0.22.1"
-aes-gcm = { version = "0.10.3", features = ["std"] }
+aes-gcm = "0.10.3"
 typenum = "1.17.0"
 aws-config = "1.5.5"
@@ -23,7 +23,7 @@ aws-sdk-s3 = { version = "1.46.0", features = ["behavior-version-latest"] }
 revolt-config = { version = "0.8.8", path = "../config", features = [
    "report-macros",
 ] }
-revolt-result = { version = "0.8.8", path = "../result", features = ["sentry"] }
+revolt-result = { version = "0.8.8", path = "../result" }
 # image processing
 jxl-oxide = "0.8.1"
@@ -6,7 +6,7 @@ use aes_gcm::{
 };
 use image::{DynamicImage, ImageBuffer};
 use revolt_config::{config, report_internal_error, FilesS3};
-use revolt_result::{create_error, Result, ToRevoltError};
+use revolt_result::{create_error, Result};
 use aws_sdk_s3::{
    config::{Credentials, Region},
@@ -55,12 +55,13 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
    // Send a request for the file
    let mut obj =
-        client.get_object().bucket(bucket_id).key(path).send().await.to_internal_error()?;
+        report_internal_error!(client.get_object().bucket(bucket_id).key(path).send().await)?;
    // Read the file from remote
    let mut buf = vec![];
    while let Some(bytes) = obj.body.next().await {
-        buf.write_all(&bytes.to_internal_error()?).to_internal_error()?;
+        let data = report_internal_error!(bytes)?;
        report_internal_error!(buf.write_all(&data))?;
        // is there a more efficient way to do this?
        // we just want the Vec<u8>
    }
@@ -77,7 +78,7 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
    // Decrypt the file
    create_cipher(&config.files.encryption_key)
        .decrypt_in_place(nonce, b"", &mut buf)
-        .to_internal_error()?;
+        .map_err(|_| create_error!(InternalError))?;
    Ok(buf)
 }
@@ -96,17 +97,18 @@ pub async fn upload_to_s3(bucket_id: &str, path: &str, buf: &[u8]) -> Result<Str
    // Encrypt the file in place
    create_cipher(&config.files.encryption_key)
        .encrypt_in_place(&nonce, b"", &mut buf)
-        .to_internal_error()?;
+        .map_err(|_| create_error!(InternalError))?;
    // Upload the file to remote
-    client
+    report_internal_error!(
-        .put_object()
+        client
-        .bucket(bucket_id)
+            .put_object()
-        .key(path)
+            .bucket(bucket_id)
-        .body(buf.into())
+            .key(path)
-        .send()
+            .body(buf.into())
-        .await
+            .send()
-        .to_internal_error()?;
+            .await
    )?;
    Ok(BASE64_STANDARD.encode(nonce))
 }
@@ -116,13 +118,14 @@ pub async fn delete_from_s3(bucket_id: &str, path: &str) -> Result<()> {
    let config = config().await;
    let client = create_client(config.files.s3);
-    client
+    report_internal_error!(
-        .delete_object()
+        client
-        .bucket(bucket_id)
+            .delete_object()
-        .key(path)
+            .bucket(bucket_id)
-        .send()
+            .key(path)
-        .await
+            .send()
-        .to_internal_error()?;
+            .await
    )?;
    Ok(())
 }
@@ -142,7 +145,8 @@ pub fn image_size(f: &NamedTempFile) -> Option<(usize, usize)> {
 pub fn image_size_vec(v: &[u8], mime: &str) -> Option<(usize, usize)> {
    match mime {
        "image/svg+xml" => {
-            let tree = usvg::Tree::from_data(v, &Default::default()).to_internal_error().ok()?;
+            let tree =
                report_internal_error!(usvg::Tree::from_data(v, &Default::default())).ok()?;
            let size = tree.size();
            Some((size.width() as usize, size.height() as usize))
@@ -217,9 +221,9 @@ pub fn decode_image<R: Read + BufRead + Seek>(reader: &mut R, mime: &str) -> Res
        "image/svg+xml" => {
            // usvg doesn't support Read trait so copy to buffer
            let mut buf = Vec::new();
-            reader.read_to_end(&mut buf).to_internal_error()?;
+            report_internal_error!(reader.read_to_end(&mut buf))?;
-            let tree = usvg::Tree::from_data(&buf, &Default::default()).to_internal_error()?;
+            let tree = report_internal_error!(usvg::Tree::from_data(&buf, &Default::default()))?;
            let size = tree.size();
            let mut pixmap = Pixmap::new(size.width() as u32, size.height() as u32)
                .ok_or_else(|| create_error!(ImageProcessingFailed))?;
@@ -237,11 +241,10 @@ pub fn decode_image<R: Read + BufRead + Seek>(reader: &mut R, mime: &str) -> Res
            ))
        }
        // Check if we can read using image-rs crate
-        _ => image::ImageReader::new(reader)
+        _ => report_internal_error!(report_internal_error!(
-            .with_guessed_format()
+            image::ImageReader::new(reader).with_guessed_format()
-            .to_internal_error()?
+        )?
-            .decode()
+        .decode()),
            .to_internal_error()
    }
 }
@@ -132,8 +132,8 @@ auto_derived!(
        #[cfg_attr(feature = "validator", validate(length(min = 1, max = 2048)))]
        pub interactions_url: Option<String>,
        /// Fields to remove from bot object
-        #[cfg_attr(feature = "serde", serde(default))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
-        pub remove: Vec<FieldsBot>,
+        pub remove: Option<Vec<FieldsBot>>,
    }
    /// Where we are inviting a bot to
@@ -207,7 +207,7 @@ auto_derived!(
        /// Fields to remove from channel
        #[cfg_attr(feature = "serde", serde(default))]
-        pub remove: Vec<FieldsChannel>,
+        pub remove: Option<Vec<FieldsChannel>>,
    }
    /// Create new group
@@ -215,7 +215,7 @@ auto_derived!(
    #[derive(Default)]
    #[cfg_attr(feature = "validator", derive(Validate))]
    pub struct SendableEmbed {
-        #[cfg_attr(feature = "validator", validate(length(min = 1, max = 256)))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1, max = 128)))]
        pub icon_url: Option<String>,
        #[cfg_attr(feature = "validator", validate(length(min = 1, max = 256)))]
        pub url: Option<String>,
@@ -7,7 +7,6 @@ mod embeds;
 mod emojis;
 mod files;
 mod messages;
 mod onboard;
 mod policy_changes;
 mod safety_reports;
 mod server_bans;
@@ -25,7 +24,6 @@ pub use embeds::*;
 pub use emojis::*;
 pub use files::*;
 pub use messages::*;
 pub use onboard::*;
 pub use policy_changes::*;
 pub use safety_reports::*;
 pub use server_bans::*;
@@ -1,15 +0,0 @@
 auto_derived!(
    /// # New User Data
    #[derive(validator::Validate)]
    pub struct DataOnboard {
        /// New username which will be used to identify the user on the platform
        #[validate(length(min = 2, max = 32), regex = "super::RE_USERNAME")]
        pub username: String,
    }
    /// # Onboarding Status
    pub struct DataHello {
        /// Whether onboarding is required
        pub onboarding: bool,
    }
 );
@@ -1,17 +1,6 @@
 use iso8601_timestamp::Timestamp;
 auto_derived!(
    /// # Report Data
    #[derive(validator::Validate)]
    pub struct DataReportContent {
        /// Content being reported
        pub content: ReportedContent,
        /// Additional report description
        #[validate(length(min = 0, max = 1000))]
        #[serde(default)]
        pub additional_context: String,
    }
    /// User-generated platform moderation report
    pub struct Report {
        /// Unique Id
@@ -124,7 +124,7 @@ auto_derived!(
        /// Timestamp this member is timed out until
        pub timeout: Option<Timestamp>,
        /// Fields to remove from channel object
-        #[cfg_attr(feature = "serde", serde(default))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
-        pub remove: Vec<FieldsMember>,
+        pub remove: Option<Vec<FieldsMember>>,
    }
 );
@@ -1,4 +1,4 @@
-use super::{Channel, File, Member, User, RE_COLOUR};
+use super::{Channel, File, RE_COLOUR};
 use revolt_permissions::{Override, OverrideField};
 use std::collections::HashMap;
@@ -175,8 +175,6 @@ auto_derived!(
        /// Ranking position
        ///
        /// Smaller values take priority.
        ///
        /// **Removed** - no effect, use the edit server role positions route
        pub rank: Option<i64>,
    }
@@ -249,8 +247,8 @@ auto_derived!(
        pub analytics: Option<bool>,
        /// Fields to remove from server object
-        #[cfg_attr(feature = "serde", serde(default))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
-        pub remove: Vec<FieldsServer>,
+        pub remove: Option<Vec<FieldsServer>>,
    }
    /// New role information
@@ -272,8 +270,8 @@ auto_derived!(
        /// **Removed** - no effect, use the edit server role positions route
        pub rank: Option<i64>,
        /// Fields to remove from role object
-        #[cfg_attr(feature = "serde", serde(default))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
-        pub remove: Vec<FieldsRole>,
+        pub remove: Option<Vec<FieldsRole>>,
    }
    /// New role permissions
@@ -293,22 +291,4 @@ auto_derived!(
    pub struct DataEditRoleRanks {
        pub ranks: Vec<String>,
    }
    /// # Query Parameters
    #[derive(FromForm)]
    pub struct OptionsQueryMembers {
        /// String to search for
        pub query: String,
        /// Discourage use of this API
        pub experimental_api: bool,
    }
    /// # Query members by name
    pub struct MemberQueryResponse {
        /// List of members
        pub members: Vec<Member>,
        /// List of users
        pub users: Vec<User>,
    }
 );
@@ -245,8 +245,8 @@ auto_derived!(
        pub flags: Option<i32>,
        /// Fields to remove from user object
-        #[cfg_attr(feature = "serde", serde(default))]
+        #[cfg_attr(feature = "validator", validate(length(min = 1)))]
-        pub remove: Vec<FieldsUser>,
+        pub remove: Option<Vec<FieldsUser>>,
    }
    /// User flag reponse
@@ -275,17 +275,6 @@ auto_derived!(
        /// Username and discriminator combo separated by #
        pub username: String,
    }
    /// # Username Information
    #[derive(Validate)]
    pub struct DataChangeUsername {
        /// New username
        #[validate(length(min = 2, max = 32), regex = "super::RE_USERNAME")]
        pub username: String,
        /// Current account password
        #[validate(length(min = 8, max = 1024))]
        pub password: String,
    }
 );
 pub trait CheckRelationship {
@@ -15,7 +15,6 @@ utoipa = ["dep:utoipa"]
 rocket = ["dep:rocket", "dep:serde_json"]
 axum = ["dep:axum", "dep:serde_json"]
 okapi = ["dep:revolt_rocket_okapi", "dep:revolt_okapi", "schemas"]
 sentry = ["dep:sentry"]
 default = ["serde"]
@@ -35,6 +34,3 @@ revolt_okapi = { version = "0.9.1", optional = true }
 # Axum
 axum = { version = "0.7.5", optional = true }
 # Sentry
 sentry = { version = "0.31.5", optional = true }
@@ -2,9 +2,10 @@ use axum::{http::StatusCode, response::IntoResponse, Json};
 use crate::{Error, ErrorType};
-impl Error {
+/// HTTP response builder for Error enum
-    pub fn axum_status(&self) -> StatusCode {
+impl IntoResponse for Error {
-        match self.error_type {
+    fn into_response(self) -> axum::response::Response {
        let status = match self.error_type {
            ErrorType::LabelMe => StatusCode::INTERNAL_SERVER_ERROR,
            ErrorType::AlreadyOnboarded => StatusCode::FORBIDDEN,
@@ -69,15 +70,11 @@ impl Error {
            ErrorType::InvalidProperty => StatusCode::BAD_REQUEST,
            ErrorType::InvalidSession => StatusCode::UNAUTHORIZED,
            ErrorType::NotAuthenticated => StatusCode::UNAUTHORIZED,
            ErrorType::Conflict => StatusCode::CONFLICT,
            ErrorType::DuplicateNonce => StatusCode::CONFLICT,
            ErrorType::VosoUnavailable => StatusCode::BAD_REQUEST,
            ErrorType::NotFound => StatusCode::NOT_FOUND,
            ErrorType::NoEffect => StatusCode::OK,
-            ErrorType::IOError => StatusCode::BAD_REQUEST,
+            ErrorType::FailedValidation { .. } => StatusCode::BAD_REQUEST,
            ErrorType::UnprocessableEntity => StatusCode::UNPROCESSABLE_ENTITY,
            ErrorType::DeserializationError { .. } => StatusCode::UNPROCESSABLE_ENTITY,
            ErrorType::FailedValidation { .. } => StatusCode::UNPROCESSABLE_ENTITY,
            ErrorType::InvalidFlagValue => StatusCode::BAD_REQUEST,
            ErrorType::FeatureDisabled { .. } => StatusCode::BAD_REQUEST,
@@ -87,13 +84,8 @@ impl Error {
            ErrorType::FileTypeNotAllowed => StatusCode::BAD_REQUEST,
            ErrorType::ImageProcessingFailed => StatusCode::INTERNAL_SERVER_ERROR,
            ErrorType::NoEmbedData => StatusCode::BAD_REQUEST,
-        }
+        };
    }
 }
-/// HTTP response builder for Error enum
+        (status, Json(&self)).into_response()
 impl IntoResponse for Error {
    fn into_response(self) -> axum::response::Response {
        (self.axum_status(), Json(&self)).into_response()
    }
 }
@@ -1,4 +1,3 @@
 use std::panic::Location;
 use std::fmt::Display;
 #[cfg(feature = "serde")]
@@ -152,15 +151,9 @@ pub enum ErrorType {
    InvalidSession,
    InvalidFlagValue,
    NotAuthenticated,
    Conflict,
    DuplicateNonce,
    NotFound,
    NoEffect,
    IOError,
    UnprocessableEntity,
    DeserializationError {
        error: String,
    },
    FailedValidation {
        error: String,
    },
@@ -181,7 +174,7 @@ pub enum ErrorType {
    // ? Feature flag disabled in the config
    FeatureDisabled {
        feature: String,
-    }
+    },
 }
 #[macro_export]
@@ -204,62 +197,6 @@ macro_rules! create_database_error {
    };
 }
 pub trait ToRevoltError<T>: Sized {
    fn capture_error(self) -> Self;
    #[track_caller]
    fn to_internal_error(self) -> Result<T, Error>;
 }
 impl<T, E: std::error::Error> ToRevoltError<T> for Result<T, E> {
    fn capture_error(self) -> Self {
        #[allow(unused_variables)]
        self.inspect_err(|e| {
            #[cfg(feature = "sentry")]
            sentry::capture_error(e);
        })
    }
    #[track_caller]
    fn to_internal_error(self) -> Result<T, Error> {
        let loc = Location::caller();
        self
            .capture_error()
            .map_err(|_| {
                Error {
                    error_type: ErrorType::InternalError,
                    location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())
                }
            })
    }
 }
 impl<T: std::error::Error> ToRevoltError<T> for Option<T> {
    fn capture_error(self) -> Self {
        #[allow(unused_variables)]
        self.inspect(|e| {
            #[cfg(feature = "sentry")]
            sentry::capture_error(e);
        })
    }
    #[track_caller]
    fn to_internal_error(self) -> Result<T, Error> {
        let loc = Location::caller();
        self
        .capture_error()
        .ok_or_else(|| {
            Error {
                error_type: ErrorType::InternalError,
                location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column())
            }
        })
    }
 }
 #[cfg(test)]
 mod tests {
    use crate::ErrorType;
@@ -8,9 +8,10 @@ use rocket::{
 use crate::{Error, ErrorType};
-impl Error {
+/// HTTP response builder for Error enum
-    pub fn rocket_status(&self) -> Status {
+impl<'r> Responder<'r, 'static> for Error {
-        match self.error_type {
+    fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
        let status = match self.error_type {
            ErrorType::LabelMe => Status::InternalServerError,
            ErrorType::AlreadyOnboarded => Status::Forbidden,
@@ -76,15 +77,11 @@ impl Error {
            ErrorType::InvalidProperty => Status::BadRequest,
            ErrorType::InvalidSession => Status::Unauthorized,
            ErrorType::NotAuthenticated => Status::Unauthorized,
            ErrorType::Conflict => Status::Conflict,
            ErrorType::DuplicateNonce => Status::Conflict,
            ErrorType::VosoUnavailable => Status::BadRequest,
            ErrorType::NotFound => Status::NotFound,
            ErrorType::NoEffect => Status::Ok,
-            ErrorType::IOError => Status::BadRequest,
+            ErrorType::FailedValidation { .. } => Status::BadRequest,
            ErrorType::UnprocessableEntity => Status::UnprocessableEntity,
            ErrorType::DeserializationError { .. } => Status::UnprocessableEntity,
            ErrorType::FailedValidation { .. } => Status::UnprocessableEntity,
            ErrorType::FeatureDisabled { .. } => Status::BadRequest,
            ErrorType::ProxyError => Status::BadRequest,
@@ -93,13 +90,8 @@ impl Error {
            ErrorType::FileTypeNotAllowed => Status::BadRequest,
            ErrorType::ImageProcessingFailed => Status::InternalServerError,
            ErrorType::NoEmbedData => Status::BadRequest,
-        }
+        };
    }
 }
 /// HTTP response builder for Error enum
 impl<'r> Responder<'r, 'static> for Error {
    fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
        // Serialize the error data structure into JSON.
        let string = serde_json::to_string(&self).unwrap();
@@ -107,7 +99,7 @@ impl<'r> Responder<'r, 'static> for Error {
        Response::build()
            .sized_body(string.len(), Cursor::new(string))
            .header(ContentType::new("application", "json"))
-            .status(self.rocket_status())
+            .status(status)
            .ok()
    }
 }
@@ -1,32 +1,19 @@
-use std::{future::Future, time::Duration};
+use revolt_config::configure;
-
+use revolt_database::DatabaseInfo;
 use revolt_config::{configure, capture_error};
 use revolt_database::{Database, DatabaseInfo};
 use revolt_result::Result;
 use tasks::{file_deletion, prune_dangling_files};
-use tokio::{join, time::sleep};
+use tokio::try_join;
 pub mod tasks;
 pub async fn cron_task_wrapper<Fut: Future<Output = Result<()>>>(func: fn(Database) -> Fut, db: Database) {
    loop {
        if let Err(error) = func(db.clone()).await {
            log::error!("cron task failed unexpectidly: {error:?}\nRetrying after 60s");
            capture_error(&error);
        }
        sleep(Duration::from_secs(60)).await;
    }
 }
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<()> {
    configure!(crond);
    let db = DatabaseInfo::Auto.connect().await.expect("database");
-
+    try_join!(
-    join!(
+        file_deletion::task(db.clone()),
-        cron_task_wrapper(file_deletion::task, db.clone()),
+        prune_dangling_files::task(db)
-        cron_task_wrapper(prune_dangling_files::task, db.clone()),
+    )
-    );
+    .map(|_| ())
 }
@@ -8,7 +8,6 @@ license = "AGPL-3.0-or-later"
 revolt-result = { version = "0.8.8", path = "../../core/result" }
 revolt-config = { version = "0.8.8", path = "../../core/config", features = [
    "report-macros",
    "anyhow"
 ] }
 revolt-database = { version = "0.8.8", path = "../../core/database" }
 revolt-models = { version = "0.8.8", path = "../../core/models", features = [
@@ -123,26 +123,24 @@ impl AsyncConsumer for AckConsumer {
                    token: session.subscription.as_ref().unwrap().auth.clone(),
                    extras: Default::default(),
                };
                let raw_service_payload = serde_json::to_string(&service_payload);
-                match serde_json::to_string(&service_payload) {
+                if let Ok(p) = raw_service_payload {
-                    Ok(p) => {
+                    let args = BasicPublishArguments::new(
-                        let args = BasicPublishArguments::new(
+                        config.pushd.exchange.as_str(),
-                            config.pushd.exchange.as_str(),
+                        config.pushd.apn.queue.as_str(),
-                            config.pushd.apn.queue.as_str(),
+                    )
-                        )
+                    .finish();
                        .finish();
-                        log::debug!(
+                    log::debug!(
-                            "Publishing ack to apn session {}",
+                        "Publishing ack to apn session {}",
-                            session.subscription.as_ref().unwrap().auth
+                        session.subscription.as_ref().unwrap().auth
-                        );
+                    );
-                        publish_message(self, p.into(), args).await;
+                    publish_message(self, p.into(), args).await;
-                    },
+                } else {
-                    Err(e) => {
+                    log::warn!("Failed to serialize ack badge update payload!");
-                        log::warn!("Failed to serialize ack badge update payload!");
+                    revolt_config::capture_error(&raw_service_payload.unwrap_err());
                        revolt_config::capture_error(&e);
                    }
                }
            }
        }
@@ -32,6 +32,7 @@ ulid = "0.4.1"
 nanoid = "0.4.0"
 # serde
 rmp-serde = "1.0.0"
 serde_json = "1.0.57"
 serde = { version = "1.0.115", features = ["derive"] }
 validator = { version = "0.16", features = ["derive"] }
@@ -65,10 +66,12 @@ revolt_rocket_okapi = { version = "0.10.0", features = ["swagger"] }
 # rabbit
 amqprs = { version = "1.7.0" }
 lapin = { version = "3.0.0" }
 # core
 authifier = "1.0.15"
 revolt-config = { path = "../core/config" }
 revolt-broker = { path = "../core/broker" }
 revolt-database = { path = "../core/database", features = [
    "rocket-impl",
    "redis-is-patched",
@@ -79,7 +82,7 @@ revolt-models = { path = "../core/models", features = [
    "rocket",
 ] }
 revolt-presence = { path = "../core/presence" }
-revolt-result = { path = "../core/result", features = ["rocket", "okapi", "sentry"] }
+revolt-result = { path = "../core/result", features = ["rocket", "okapi"] }
 revolt-permissions = { path = "../core/permissions", features = ["schemas"] }
 [build-dependencies]
@@ -135,7 +135,6 @@ pub async fn web() -> Rocket<Build> {
        .manage(cors.clone())
        .attach(util::ratelimiter::RatelimitFairing)
        .attach(cors)
        .register("/", util::catchers::all_catchers())
        .configure(rocket::Config {
            limits: rocket::data::Limits::default().limit("string", 5.megabytes()),
            address: Ipv4Addr::new(0, 0, 0, 0).into(),
@@ -1,8 +1,9 @@
 use crate::util::json::{Json, Validate};
 use revolt_database::{Bot, Database, User};
 use revolt_models::v0;
-use revolt_result::Result;
+use revolt_result::{create_error, Result};
 use rocket::serde::json::Json;
 use rocket::State;
 use validator::Validate;
 /// # Create Bot
 ///
@@ -12,9 +13,14 @@ use rocket::State;
 pub async fn create_bot(
    db: &State<Database>,
    user: User,
-    info: Validate<Json<v0::DataCreateBot>>,
+    info: Json<v0::DataCreateBot>,
 ) -> Result<Json<v0::BotWithUserResponse>> {
-    let info = info.into_inner().into_inner();
+    let info = info.into_inner();
    info.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    let (bot, user) = Bot::create(db, info.name, &user, None).await?;
    Ok(Json(v0::BotWithUserResponse {
@@ -11,7 +11,7 @@ use rocket_empty::EmptyResponse;
 pub async fn delete_bot(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<EmptyResponse> {
    let bot = target.as_bot(db).await?;
    if bot.owner != user.id {
@@ -3,7 +3,8 @@ use revolt_models::v0::{self, DataEditBot};
 use revolt_result::{create_error, Result};
 use rocket::State;
-use crate::util::json::{Json, Validate};
+use rocket::serde::json::Json;
 use validator::Validate;
 /// # Edit Bot
 ///
@@ -13,10 +14,15 @@ use crate::util::json::{Json, Validate};
 pub async fn edit_bot(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    data: Validate<Json<DataEditBot>>,
+    data: Json<DataEditBot>,
 ) -> Result<Json<v0::BotWithUserResponse>> {
-    let data = data.into_inner().into_inner();
+    let data = data.into_inner();
    data.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    let mut bot = target.as_bot(db).await?;
    if bot.owner != user.id {
@@ -31,7 +37,7 @@ pub async fn edit_bot(
    if data.public.is_none()
        && data.analytics.is_none()
        && data.interactions_url.is_none()
-        && data.remove.is_empty()
+        && data.remove.is_none()
    {
        return Ok(Json(v0::BotWithUserResponse {
            bot: bot.into(),
@@ -58,6 +64,7 @@ pub async fn edit_bot(
        db,
        partial,
        remove
            .unwrap_or_default()
            .into_iter()
            .map(|v| v.into())
            .collect(),
@@ -93,7 +100,7 @@ mod test {
            .body(
                json!(v0::DataEditBot {
                    public: Some(true),
-                    remove: vec![FieldsBot::Token],
+                    remove: Some(vec![FieldsBot::Token]),
                    ..Default::default()
                })
                .to_string(),
@@ -1,8 +1,7 @@
 use revolt_database::{util::reference::Reference, Database, User};
 use revolt_models::v0::FetchBotResponse;
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
 use crate::util::json::Json;
 /// # Fetch Bot
 ///
@@ -12,7 +11,7 @@ use crate::util::json::Json;
 pub async fn fetch_bot(
    db: &State<Database>,
    user: User,
-    bot: Reference<'_>,
+    bot: Reference,
 ) -> Result<Json<FetchBotResponse>> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
@@ -2,7 +2,7 @@ use futures::future::join_all;
 use revolt_database::{Database, User};
 use revolt_models::v0::OwnedBotsResponse;
 use revolt_result::Result;
-use crate::util::json::Json;
+use rocket::serde::json::Json;
 use rocket::State;
 /// # Fetch Owned Bots
@@ -2,7 +2,7 @@ use revolt_database::{util::reference::Reference, Database, User};
 use revolt_models::v0::PublicBot;
 use revolt_result::{create_error, Result};
-use crate::util::json::Json;
+use rocket::serde::json::Json;
 use rocket::State;
 /// # Fetch Public Bot
@@ -13,9 +13,9 @@ use rocket::State;
 pub async fn fetch_public_bot(
    db: &State<Database>,
    user: Option<User>,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<PublicBot>> {
-    let bot = db.fetch_bot(target.id).await?;
+    let bot = db.fetch_bot(&target.id).await?;
    if !bot.public && user.is_none_or(|x| x.id != bot.owner) {
        return Err(create_error!(NotFound));
    }
@@ -8,7 +8,7 @@ use revolt_permissions::{
 use revolt_result::{create_error, Result};
 use rocket::State;
-use crate::util::json::Json;
+use rocket::serde::json::Json;
 use rocket_empty::EmptyResponse;
 /// # Invite Bot
@@ -20,7 +20,7 @@ pub async fn invite_bot(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
    dest: Json<v0::InviteBotDestination>,
 ) -> Result<EmptyResponse> {
    if user.bot.is_some() {
@@ -15,8 +15,8 @@ use rocket_empty::EmptyResponse;
 pub async fn ack(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    message: Reference<'_>,
+    message: Reference,
 ) -> Result<EmptyResponse> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
@@ -29,7 +29,7 @@ pub async fn ack(
        .throw_if_lacking_channel_permission(ChannelPermission::ViewChannel)?;
    channel
-        .ack(&user.id, message.id)
+        .ack(&user.id, &message.id)
        .await
        .map(|_| EmptyResponse)
 }
@@ -17,7 +17,7 @@ pub async fn delete(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
    options: v0::OptionsChannelDelete,
 ) -> Result<EmptyResponse> {
    let mut channel = target.as_channel(db).await?;
@@ -5,8 +5,8 @@ use revolt_database::{
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
-use crate::util::json::{Json, Validate};
+use validator::Validate;
 /// # Edit Channel
 ///
@@ -17,10 +17,15 @@ pub async fn edit(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    data: Validate<Json<v0::DataEditChannel>>,
+    data: Json<v0::DataEditChannel>,
 ) -> Result<Json<v0::Channel>> {
-    let data = data.into_inner().into_inner();
+    let data = data.into_inner();
    data.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    let mut channel = target.as_channel(db).await?;
    let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -33,7 +38,7 @@ pub async fn edit(
        && data.icon.is_none()
        && data.nsfw.is_none()
        && data.owner.is_none()
-        && data.remove.is_empty()
+        && data.remove.is_none()
    {
        return Ok(Json(channel.into()));
    }
@@ -107,21 +112,23 @@ pub async fn edit(
            nsfw,
            ..
        } => {
-            if data.remove.contains(&v0::FieldsChannel::Icon) {
+            if let Some(fields) = &data.remove {
-                if let Some(icon) = &icon {
+                if fields.contains(&v0::FieldsChannel::Icon) {
-                    db.mark_attachment_as_deleted(&icon.id).await?;
+                    if let Some(icon) = &icon {
                        db.mark_attachment_as_deleted(&icon.id).await?;
                    }
                }
            }
-            for field in &data.remove {
+                for field in fields {
-                match field {
+                    match field {
-                    v0::FieldsChannel::Description => {
+                        v0::FieldsChannel::Description => {
-                        description.take();
+                            description.take();
                        }
                        v0::FieldsChannel::Icon => {
                            icon.take();
                        }
                        _ => {}
                    }
                    v0::FieldsChannel::Icon => {
                        icon.take();
                    }
                    _ => {}
                }
            }
@@ -207,7 +214,11 @@ pub async fn edit(
                .update(
                    db,
                    partial,
-                    data.remove.into_iter().map(|f| f.into()).collect(),
+                    data.remove
                        .unwrap_or_default()
                        .into_iter()
                        .map(|f| f.into())
                        .collect(),
                )
                .await?;
        }
@@ -6,8 +6,7 @@ use revolt_database::{
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::Result;
-use rocket::State;
+use rocket::{serde::json::Json, State};
 use crate::util::json::Json;
 /// # Fetch Channel
 ///
@@ -17,7 +16,7 @@ use crate::util::json::Json;
 pub async fn fetch(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<v0::Channel>> {
    let channel = target.as_channel(db).await?;
@@ -17,8 +17,8 @@ pub async fn add_member(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    group_id: Reference<'_>,
+    group_id: Reference,
-    member_id: Reference<'_>,
+    member_id: Reference,
 ) -> Result<EmptyResponse> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
@@ -2,8 +2,9 @@ use revolt_database::{Channel, Database, RelationshipStatus, User};
 use revolt_models::v0;
 use revolt_result::{create_error, Result};
-use crate::util::json::{Json, Validate};
+use rocket::serde::json::Json;
 use rocket::State;
 use validator::Validate;
 /// # Create Group
 ///
@@ -13,13 +14,18 @@ use rocket::State;
 pub async fn create_group(
    db: &State<Database>,
    user: User,
-    data: Validate<Json<v0::DataCreateGroup>>,
+    data: Json<v0::DataCreateGroup>,
 ) -> Result<Json<v0::Channel>> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
    }
-    let data = data.into_inner().into_inner();
+    let data = data.into_inner();
    data.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    for target in &data.users {
        match user.relationship_with(target) {
@@ -14,8 +14,8 @@ pub async fn remove_member(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    member: Reference<'_>,
+    member: Reference,
 ) -> Result<EmptyResponse> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
@@ -6,8 +6,7 @@ use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
 use crate::util::json::Json;
 /// # Create Invite
 ///
@@ -19,7 +18,7 @@ use crate::util::json::Json;
 pub async fn create_invite(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<v0::Invite>> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
@@ -5,8 +5,7 @@ use revolt_database::{
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
 use crate::util::json::Json;
 /// # Fetch Group Members
 ///
@@ -18,7 +17,7 @@ use crate::util::json::Json;
 pub async fn fetch_members(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
 ) -> Result<Json<Vec<v0::User>>> {
    let channel = target.as_channel(db).await?;
    let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -6,9 +6,9 @@ use revolt_database::{
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
 use rocket_empty::EmptyResponse;
-use crate::util::json::{Json, Validate};
+use validator::Validate;
 /// # Bulk Delete Messages
 ///
@@ -22,10 +22,15 @@ use crate::util::json::{Json, Validate};
 pub async fn bulk_delete_messages(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    options: Validate<Json<v0::OptionsBulkDelete>>,
+    options: Json<v0::OptionsBulkDelete>,
 ) -> Result<EmptyResponse> {
-    let options = options.into_inner().into_inner();
+    let options = options.into_inner();
    options.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    for id in &options.ids {
        if ulid::Ulid::from_string(id)
@@ -46,7 +51,7 @@ pub async fn bulk_delete_messages(
        .await
        .throw_if_lacking_channel_permission(ChannelPermission::ManageMessages)?;
-    Message::bulk_delete(db, target.id, options.ids)
+    Message::bulk_delete(db, &target.id, options.ids)
        .await
        .map(|_| EmptyResponse)
 }
@@ -17,8 +17,8 @@ use rocket_empty::EmptyResponse;
 pub async fn clear_reactions(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    msg: Reference<'_>,
+    msg: Reference,
 ) -> Result<EmptyResponse> {
    let channel = target.as_channel(db).await?;
    let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -15,10 +15,10 @@ use rocket_empty::EmptyResponse;
 pub async fn delete(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    msg: Reference<'_>,
+    msg: Reference,
 ) -> Result<EmptyResponse> {
-    let message = msg.as_message_in_channel(db, target.id).await?;
+    let message = msg.as_message_in_channel(db, &target.id).await?;
    if message.author != user.id {
        let channel = target.as_channel(db).await?;
@@ -7,8 +7,8 @@ use revolt_database::{
 use revolt_models::v0::{self, Embed};
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
-use crate::util::json::{Json, Validate};
+use validator::Validate;
 /// # Edit Message
 ///
@@ -18,11 +18,16 @@ use crate::util::json::{Json, Validate};
 pub async fn edit(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    msg: Reference<'_>,
+    msg: Reference,
-    edit: Validate<Json<v0::DataEditMessage>>,
+    edit: Json<v0::DataEditMessage>,
 ) -> Result<Json<v0::Message>> {
-    let edit = edit.into_inner().into_inner();
+    let edit = edit.into_inner();
    edit.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    Message::validate_sum(
        &edit.content,
@@ -5,8 +5,7 @@ use revolt_database::{
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
 use crate::util::json::Json;
 /// # Fetch Message
 ///
@@ -16,8 +15,8 @@ use crate::util::json::Json;
 pub async fn fetch(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    msg: Reference<'_>,
+    msg: Reference,
 ) -> Result<Json<v0::Message>> {
    let channel = target.as_channel(db).await?;
    let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -1,7 +1,4 @@
-use revolt_database::{
+use revolt_database::{util::{permissions::DatabasePermissionQuery, reference::Reference}, Channel, Database, PartialMessage, SystemMessage, User, AMQP};
    util::{permissions::DatabasePermissionQuery, reference::Reference},
    Channel, Database, PartialMessage, SystemMessage, User, AMQP,
 };
 use revolt_models::v0::MessageAuthor;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
@@ -17,8 +14,8 @@ pub async fn message_pin(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    msg: Reference<'_>,
+    msg: Reference,
 ) -> Result<EmptyResponse> {
    let channel = target.as_channel(db).await?;
@@ -172,7 +169,7 @@ mod test {
            })
            .await;
-        let updated_message = Reference::from_unchecked(&message.id)
+        let updated_message = Reference::from_unchecked(message.id)
            .as_message(&harness.db)
            .await
            .expect("Failed to find updated message");
@@ -5,8 +5,8 @@ use revolt_database::{
 use revolt_models::v0::{self, MessageSort};
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
-use crate::util::json::{Json, Validate};
+use validator::Validate;
 /// # Fetch Messages
 ///
@@ -16,9 +16,15 @@ use crate::util::json::{Json, Validate};
 pub async fn query(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    options: Validate<v0::OptionsQueryMessages>,
+    options: v0::OptionsQueryMessages,
 ) -> Result<Json<v0::BulkMessageResponse>> {
    options.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    if let Some(MessageSort::Relevance) = options.sort {
        return Err(create_error!(InvalidOperation));
    }
@@ -37,7 +43,7 @@ pub async fn query(
        sort,
        nearby,
        include_users,
-    } = options.into_inner();
+    } = options;
    Message::fetch_with_users(
        db,
@@ -15,9 +15,9 @@ use rocket_empty::EmptyResponse;
 pub async fn react_message(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    msg: Reference<'_>,
+    msg: Reference,
-    emoji: Reference<'_>,
+    emoji: Reference,
 ) -> Result<EmptyResponse> {
    let channel = target.as_channel(db).await?;
    let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -30,7 +30,7 @@ pub async fn react_message(
    // Add the reaction
    message
-        .add_reaction(db, &user, emoji.id)
+        .add_reaction(db, &user, &emoji.id)
        .await
        .map(|_| EmptyResponse)
 }
@@ -5,8 +5,8 @@ use revolt_database::{
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result};
-use rocket::State;
+use rocket::{serde::json::Json, State};
-use crate::util::json::{Json, Validate};
+use validator::Validate;
 /// # Search for Messages
 ///
@@ -16,14 +16,19 @@ use crate::util::json::{Json, Validate};
 pub async fn search(
    db: &State<Database>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    options: Validate<Json<v0::DataMessageSearch>>,
+    options: Json<v0::DataMessageSearch>,
 ) -> Result<Json<v0::BulkMessageResponse>> {
    if user.bot.is_some() {
        return Err(create_error!(IsBot));
    }
-    let options = options.into_inner().into_inner();
+    let options = options.into_inner();
    options.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    if options.query.is_some() && options.pinned.is_some() {
        return Err(create_error!(InvalidOperation))
@@ -7,9 +7,10 @@ use revolt_database::{Interactions, Message, AMQP};
 use revolt_models::v0;
 use revolt_permissions::PermissionQuery;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
-use revolt_result::Result;
+use revolt_result::{create_error, Result};
-use crate::util::json::{Json, Validate};
+use rocket::serde::json::Json;
 use rocket::State;
 use validator::Validate;
 /// # Send Message
 ///
@@ -20,11 +21,16 @@ pub async fn message_send(
    db: &State<Database>,
    amqp: &State<AMQP>,
    user: User,
-    target: Reference<'_>,
+    target: Reference,
-    data: Validate<Json<v0::DataMessageSend>>,
+    data: Json<v0::DataMessageSend>,
    idempotency: IdempotencyKey,
 ) -> Result<Json<v0::Message>> {
-    let data = data.into_inner().into_inner();
+    let data = data.into_inner();
    data.validate().map_err(|error| {
        create_error!(FailedValidation {
            error: error.to_string()
        })
    })?;
    // Ensure we have permissions to send a message
    let channel = target.as_channel(db).await?;
@@ -196,7 +202,7 @@ mod test {
        Member::create(&harness.db, &server, &user, Some(channels.clone()))
            .await
            .expect("Failed to create member");
-        let member = Reference::from_unchecked(&user.id)
+        let member = Reference::from_unchecked(user.id.clone())
            .as_member(&harness.db, &server.id)
            .await
            .expect("Failed to get member");
@@ -236,7 +242,7 @@ mod test {
        Member::create(&harness.db, &server, &second_user, Some(channels.clone()))
            .await
            .expect("Failed to create second member");
-        let mut second_member = Reference::from_unchecked(&second_user.id)
+        let mut second_member = Reference::from_unchecked(second_user.id.clone())
            .as_member(&harness.db, &server.id)
            .await
            .expect("Failed to get second member");
Author	SHA1	Message	Date
izzy	3dc7c32723	chore: ignore WS protocol errors	2025-06-28 18:55:15 +01:00
izzy	083229c7f5	refactor: one channel per consumer	2025-06-28 18:26:51 +01:00
izzy	9d576d2430	refactor: move lapin code into revolt-broker crate	2025-06-28 17:23:44 +01:00
izzy	85989a2138	feat: switch to RabbitMQ streams from Redis Pub/Sub refactor: redo the bonfire client structure	2025-06-28 13:35:18 +01:00