Compare commits

..
Author SHA1 Message Date
Zomatree 1a60aecdb3 chore: switch to method chaining over macro 2025-08-15 02:07:26 +01:00
Zomatree 789dedd9f1 chore: cleanup errors 2025-08-15 02:07:26 +01:00
Zomatree d7cf809424 fix: make sentry dep optional 2025-08-15 02:07:25 +01:00
Zomatree 249a4818fc feat: error docs 2025-08-15 02:07:25 +01:00
Zomatree 46e127ccd2 chore: send rest of the errors to the catchers 2025-08-15 02:07:25 +01:00
Zomatree cf4fe859bf feat: move all json and validate to new utils 2025-08-15 02:07:25 +01:00
Zomatree 3d6f39a0eb feat: better error handling 2025-08-15 02:00:12 +01:00
Zomatree ed22b3a5ce chore: switch Reference to using an actual reference 2025-08-12 03:44:29 +01:00
Zomatree 65fbd36624 fix: allow for disabling default features 2025-08-12 02:35:09 +01:00
izzy 050ba16d4a fix: correct shebang for try-tag-and-release 2025-08-07 12:58:52 +02:00
izzy 65bc6c8fc6 chore: add just tool to nix shell 2025-08-07 12:58:15 +02:00
izzy 6ad3da5f35 fix: permit empty remove array in edit requests 2025-08-07 12:57:33 +02:00
izzy 947eb15771 fix: newly created roles should be ranked the lowest 2025-08-07 12:29:42 +02:00
IAmTomahawkx f4ee35fb09 fix: don't update the same field while trying to remove it
fixes #392
2025-08-06 14:18:40 -07:00
IAmTomahawkx 6048587d34 fix: rename openapi operation ids
fixes #406
2025-08-06 12:28:58 -07:00
Builderb 80cf8fc4e8 Increase limit of embed icon url due to new cdn changes causing longer url (#426)
Signed-off-by: Builderb <builderbgamer@gmail.com>
2025-08-03 11:54:22 -07:00
higgs01 4f54227495 chore: use mc alias set instead of removed mc config (#423) 2025-07-18 09:28:31 +01:00
183 changed files with 1692 additions and 2155 deletions
Generated
+15 -461
View File
@@ -97,54 +97,6 @@ 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"
@@ -229,45 +181,6 @@ 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"
@@ -332,31 +245,6 @@ 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"
@@ -415,18 +303,6 @@ 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"
@@ -464,7 +340,7 @@ checksum = "730294c1c08c2e0f85759590518f6333f0d5a0a766a27d519c1b244c3dfd8a24"
dependencies = [
"async-attributes",
"async-channel 1.9.0",
"async-global-executor 2.4.1",
"async-global-executor",
"async-io",
"async-lock 3.4.0",
"async-process",
@@ -634,7 +510,7 @@ dependencies = [
"anyhow",
"arrayvec",
"log",
"nom 7.1.3",
"nom",
"num-rational",
"v_frame",
]
@@ -1380,15 +1256,6 @@ 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"
@@ -1519,15 +1386,6 @@ 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"
@@ -1545,7 +1403,7 @@ version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
dependencies = [
"nom 7.1.3",
"nom",
]
[[package]]
@@ -1629,18 +1487,6 @@ 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"
@@ -1690,7 +1536,7 @@ dependencies = [
"async-trait",
"json5",
"lazy_static",
"nom 7.1.3",
"nom",
"pathdiff",
"ron",
"rust-ini",
@@ -2168,7 +2014,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "79b71cca7d95d7681a4b3b9cdf63c8dbc3730d0584c2c74e31416d64a90493f4"
dependencies = [
"const-oid 0.6.2",
"der_derive 0.4.1",
"der_derive",
]
[[package]]
@@ -2189,26 +2035,10 @@ 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"
@@ -2221,17 +2051,6 @@ 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"
@@ -2277,15 +2096,6 @@ 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"
@@ -2574,7 +2384,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cea14ef9355e3beab063703aa9dab15afd25f0667c341310c1e5274bb1d0da18"
dependencies = [
"libc",
"windows-sys 0.59.0",
"windows-sys 0.52.0",
]
[[package]]
@@ -2604,15 +2414,6 @@ 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"
@@ -2721,12 +2522,6 @@ 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"
@@ -2746,17 +2541,6 @@ 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"
@@ -3963,7 +3747,6 @@ version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
dependencies = [
"block-padding",
"generic-array 0.14.7",
]
@@ -4023,7 +3806,7 @@ checksum = "e04d7f318608d35d4b61ddd75cbdaee86b023ebe2bd5a66ee0915f0bf93095a9"
dependencies = [
"hermit-abi 0.5.1",
"libc",
"windows-sys 0.59.0",
"windows-sys 0.52.0",
]
[[package]]
@@ -4328,25 +4111,6 @@ 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"
@@ -4384,7 +4148,7 @@ dependencies = [
"idna 0.3.0",
"mime",
"native-tls",
"nom 7.1.3",
"nom",
"once_cell",
"quoted_printable",
"socket2 0.4.10",
@@ -4891,7 +4655,7 @@ dependencies = [
"md-5",
"mongodb-internal-macros",
"once_cell",
"pbkdf2 0.11.0",
"pbkdf2",
"percent-encoding",
"rand 0.8.5",
"rustc_version_runtime",
@@ -4994,15 +4758,6 @@ 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"
@@ -5183,15 +4938,6 @@ 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"
@@ -5281,28 +5027,6 @@ 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"
@@ -5388,16 +5112,6 @@ 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"
@@ -5668,36 +5382,6 @@ 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"
@@ -6252,15 +5936,6 @@ 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"
@@ -6270,18 +5945,6 @@ 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"
@@ -6330,7 +5993,7 @@ dependencies = [
"cookie-factory",
"crc16",
"log",
"nom 7.1.3",
"nom",
]
[[package]]
@@ -6565,15 +6228,12 @@ 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",
@@ -6587,19 +6247,6 @@ 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"
@@ -6644,12 +6291,10 @@ dependencies = [
"bson",
"deadqueue",
"decancer",
"fcm_v1",
"futures",
"indexmap 1.9.3",
"isahc",
"iso8601-timestamp",
"lapin",
"linkify 0.8.1",
"log",
"lru 0.11.1",
@@ -6659,18 +6304,15 @@ 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",
@@ -6679,7 +6321,6 @@ dependencies = [
"unicode-segmentation",
"url-escape",
"validator 0.16.1",
"web-push",
]
[[package]]
@@ -6697,7 +6338,6 @@ dependencies = [
"futures",
"impl_ops",
"iso8601-timestamp",
"lapin",
"lettre",
"linkify 0.6.0",
"log",
@@ -6709,7 +6349,6 @@ dependencies = [
"redis-kiss",
"regex",
"reqwest 0.11.27",
"revolt-broker",
"revolt-config",
"revolt-database",
"revolt-models",
@@ -6717,7 +6356,6 @@ dependencies = [
"revolt-presence",
"revolt-result",
"revolt_rocket_okapi",
"rmp-serde",
"rocket",
"rocket_authifier",
"rocket_cors",
@@ -6873,6 +6511,7 @@ dependencies = [
"revolt_rocket_okapi",
"rocket",
"schemars",
"sentry",
"serde",
"serde_json",
"utoipa",
@@ -7251,15 +6890,6 @@ 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"
@@ -7270,7 +6900,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys 0.4.15",
"windows-sys 0.59.0",
"windows-sys 0.52.0",
]
[[package]]
@@ -7283,7 +6913,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys 0.9.4",
"windows-sys 0.59.0",
"windows-sys 0.52.0",
]
[[package]]
@@ -7326,19 +6956,6 @@ 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"
@@ -7453,15 +7070,6 @@ 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"
@@ -7524,17 +7132,6 @@ 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"
@@ -8126,9 +7723,6 @@ name = "spin"
version = "0.9.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
dependencies = [
"lock_api",
]
[[package]]
name = "spki"
@@ -8433,18 +8027,6 @@ 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"
@@ -8455,7 +8037,7 @@ dependencies = [
"getrandom 0.3.3",
"once_cell",
"rustix 1.0.7",
"windows-sys 0.59.0",
"windows-sys 0.52.0",
]
[[package]]
@@ -9652,7 +9234,7 @@ version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
dependencies = [
"windows-sys 0.59.0",
"windows-sys 0.48.0",
]
[[package]]
@@ -10056,34 +9638,6 @@ 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"
+4
View File
@@ -26,3 +26,7 @@ disallowed-methods = [
# Prefer to use Object::delete(&self)
"revolt_database::models::bots::ops::AbstractBots::delete_bot",
]
disallowed-types = [
"rocket::serde::json::Json",
]
+1 -1
View File
@@ -34,7 +34,7 @@ services:
- minio
entrypoint: >
/bin/sh -c "while ! /usr/bin/mc ready minio; do
/usr/bin/mc config host add minio http://minio:9000 minioautumn minioautumn;
/usr/bin/mc alias set minio http://minio:9000 minioautumn minioautumn;
echo 'Waiting minio...' && sleep 1;
done; /usr/bin/mc mb minio/revolt-uploads; exit 0;"
+1 -6
View File
@@ -9,7 +9,6 @@ edition = "2021"
[dependencies]
# util
log = "*"
rand = "*"
sentry = "0.31.5"
lru = "0.7.6"
ulid = "0.5.0"
@@ -38,8 +37,7 @@ async-std = { version = "1.8.0", features = [
# core
authifier = { version = "1.0.15" }
revolt-result = { path = "../core/result" }
revolt-broker = { path = "../core/broker" }
revolt-result = { path = "../core/result", features = ["sentry"] }
revolt-models = { path = "../core/models" }
revolt-config = { path = "../core/config" }
revolt-database = { path = "../core/database" }
@@ -48,6 +46,3 @@ revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
# redis
fred = { version = "8.0.1", features = ["subscriber-client"] }
# rabbit
lapin = { version = "3.0.0" }
-129
View File
@@ -1,129 +0,0 @@
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;
}
}
-3
View File
@@ -1,3 +0,0 @@
pub mod core;
pub mod subscriber;
pub mod worker;
-112
View File
@@ -1,112 +0,0 @@
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;
}
-124
View File
@@ -1,124 +0,0 @@
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();
}
}
_ => {}
}
}
}
}
}
+11 -5
View File
@@ -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};
use revolt_result::{create_error, Result, ToRevoltError};
use serde::{Deserialize, Serialize};
/// Enumeration of supported protocol formats
@@ -38,16 +38,22 @@ impl ProtocolConfiguration {
match self.format {
ProtocolFormat::Json => {
if let Message::Text(text) = msg {
serde_json::from_str(text).map_err(|_| create_error!(InternalError))
// Log the error in-case we make a breaking change to the payload
serde_json::from_str(text)
.capture_error()
.map_err(|_| create_error!(UnprocessableEntity))
} else {
Err(create_error!(InternalError))
Err(create_error!(UnprocessableEntity))
}
}
ProtocolFormat::Msgpack => {
if let Message::Binary(buf) = msg {
rmp_serde::from_slice(buf).map_err(|_| create_error!(InternalError))
rmp_serde::from_slice(buf)
.capture_error()
.map_err(|_| create_error!(UnprocessableEntity))
} else {
Err(create_error!(InternalError))
Err(create_error!(UnprocessableEntity))
}
}
}
-3
View File
@@ -66,7 +66,6 @@ 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,
@@ -88,7 +87,6 @@ impl State {
..Default::default()
};
let user_id = user.id.clone();
cache.users.insert(user.id.clone(), user);
State {
@@ -98,7 +96,6 @@ impl State {
Duration::from_secs(900),
5,
))),
user_id,
session_id,
private_topic,
state: SubscriptionStateChange::Reset,
+2 -3
View File
@@ -6,11 +6,10 @@ use revolt_presence::clear_region;
#[macro_use]
extern crate log;
pub mod client;
pub mod config;
pub mod events;
mod config;
mod database;
mod events;
mod websocket;
#[async_std::main]
+478 -8
View File
@@ -1,11 +1,39 @@
use std::net::SocketAddr;
use std::{collections::HashSet, net::SocketAddr, sync::Arc};
use async_std::net::TcpStream;
use futures::channel::oneshot;
use revolt_database::Database;
use async_tungstenite::WebSocketStream;
use authifier::AuthifierEvent;
use fred::{
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 crate::client::core::client_core;
use crate::config::WebsocketHandshakeCallback;
use async_std::{
net::TcpStream,
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.
@@ -25,7 +53,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(config) = receiver.await else {
let Ok(mut config) = receiver.await else {
return;
};
@@ -35,5 +63,447 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
config.get_protocol_format()
);
client_core(db, ws, config).await;
// 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!("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();
}
}
_ => {}
}
}
}
}
}
-25
View File
@@ -1,25 +0,0 @@
[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" }
@@ -1,180 +0,0 @@
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
}
}
@@ -1,7 +0,0 @@
mod consumer;
mod pool;
mod publish;
pub use consumer::Consumer;
pub use pool::{create_channel, get_connection};
pub use publish::publish_event;
@@ -1,99 +0,0 @@
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
}
@@ -1,37 +0,0 @@
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
}
-18
View File
@@ -1,18 +0,0 @@
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()
}
+3 -2
View File
@@ -11,8 +11,9 @@ description = "Revolt Backend: Configuration"
[features]
anyhow = ["dep:sentry-anyhow"]
report-macros = ["revolt-result"]
sentry = ["dep:sentry"]
test = ["async-std"]
default = ["test", "anyhow"]
default = ["test", "sentry"]
[dependencies]
# Utility
@@ -32,7 +33,7 @@ log = "0.4.14"
pretty_env_logger = "0.4.0"
# Sentry
sentry = "0.31.5"
sentry = { version = "0.31.5", optional = true }
sentry-anyhow = { version = "0.38.1", optional = true }
# Core
-14
View File
@@ -28,20 +28,6 @@ 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]
+7 -15
View File
@@ -6,10 +6,12 @@ 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(feature = "report-macros")]
#[cfg(all(feature = "report-macros", feature = "sentry"))]
#[macro_export]
macro_rules! report_error {
( $expr: expr, $error: ident $( $tt:tt )? ) => {
@@ -24,7 +26,7 @@ macro_rules! report_error {
};
}
#[cfg(feature = "report-macros")]
#[cfg(all(feature = "report-macros", feature = "sentry"))]
#[macro_export]
macro_rules! capture_internal_error {
( $expr: expr ) => {
@@ -35,7 +37,7 @@ macro_rules! capture_internal_error {
};
}
#[cfg(feature = "report-macros")]
#[cfg(all(feature = "report-macros", feature = "sentry"))]
#[macro_export]
macro_rules! report_internal_error {
( $expr: expr ) => {
@@ -108,24 +110,12 @@ 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)]
@@ -431,6 +421,7 @@ 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");
@@ -456,6 +447,7 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry:
}
}
#[cfg(feature = "sentry")]
#[macro_export]
macro_rules! configure {
($application: ident) => {
+4 -12
View File
@@ -10,12 +10,12 @@ description = "Revolt Backend: Database Implementation"
[features]
# Databases
mongodb = ["dep:mongodb", "bson"]
mongodb = ["dep:mongodb", "bson", "authifier/database-mongodb"]
# ... Other
tasks = ["isahc", "linkify", "url-escape"]
async-std-runtime = ["async-std"]
rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi"]
async-std-runtime = ["async-std", "authifier/async-std-runtime"]
rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi", "authifier/rocket_impl"]
axum-impl = ["axum"]
redis-is-patched = ["revolt-presence/redis-is-patched"]
@@ -37,7 +37,6 @@ 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"
@@ -57,7 +56,6 @@ 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"] }
@@ -93,14 +91,8 @@ 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", features = ["rocket_impl"] }
authifier = { version = "1.0.15" }
# RabbitMQ
amqprs = { version = "1.7.0" }
lapin = { version = "3.0.0" }
+31 -12
View File
@@ -1,3 +1,4 @@
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -13,6 +14,7 @@ use authifier::Authifier;
use rand::Rng;
use revolt_config::config;
#[cfg(feature = "mongodb")]
pub use self::mongodb::*;
pub use self::reference::*;
@@ -25,8 +27,10 @@ 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),
}
@@ -36,6 +40,7 @@ pub enum Database {
/// Mock database
Reference(ReferenceDb),
/// MongoDB database
#[cfg(feature = "mongodb")]
MongoDb(MongoDb),
}
@@ -45,7 +50,7 @@ impl DatabaseInfo {
pub async fn connect(self) -> Result<Database, String> {
let config = config().await;
Ok(match self {
match self {
DatabaseInfo::Auto => {
if std::env::var("TEST_DB").is_ok() {
DatabaseInfo::Test(format!(
@@ -53,16 +58,20 @@ impl DatabaseInfo {
rand::thread_rng().gen_range(1_000_000..10_000_000)
))
.connect()
.await?
.await
} else if !config.database.mongodb.is_empty() {
DatabaseInfo::MongoDb {
#[cfg(feature = "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) => {
@@ -70,30 +79,36 @@ 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" => {
DatabaseInfo::MongoDb {
#[cfg(feature = "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 => Database::Reference(Default::default()),
DatabaseInfo::Reference => Ok(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())?;
Database::MongoDb(MongoDb(client, database_name))
Ok(Database::MongoDb(MongoDb(client, database_name)))
}
#[cfg(feature = "mongodb")]
DatabaseInfo::MongoDbFromClient(client, database_name) => {
Database::MongoDb(MongoDb(client, database_name))
Ok(Database::MongoDb(MongoDb(client, database_name)))
}
})
}
}
}
@@ -219,12 +234,16 @@ 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,7 +10,6 @@ use serde::de::DeserializeOwned;
use serde::Serialize;
database_derived!(
#[cfg(feature = "mongodb")]
/// MongoDB implementation
pub struct MongoDb(pub ::mongodb::Client, pub String);
);
+4 -7
View File
@@ -1,5 +1,4 @@
use authifier::AuthifierEvent;
use revolt_broker::event_stream;
use revolt_result::Error;
use serde::{Deserialize, Serialize};
@@ -254,16 +253,14 @@ 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)]
result.unwrap();
redis_kiss::publish(channel, self).await.unwrap();
}
/// Publish user event
+4
View File
@@ -25,6 +25,9 @@ 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 {
@@ -103,6 +106,7 @@ pub mod util;
pub use models::*;
pub mod events;
#[cfg(feature = "tasks")]
pub mod tasks;
mod amqp;
@@ -1,3 +1,4 @@
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::{Bot, FieldsBot, PartialBot};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::Invite;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::ChannelUnread;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::{FieldsWebhook, PartialWebhook, Webhook};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -8,10 +8,13 @@ use serde::{Deserialize, Serialize};
use ulid::Ulid;
use crate::{
events::client::EventV1, tasks::ack::AckEvent, Database, File, IntoDocumentPath, PartialServer,
events::client::EventV1, Database, File, PartialServer,
Server, SystemMessage, User, AMQP,
};
#[cfg(feature = "mongodb")]
use crate::IntoDocumentPath;
auto_derived!(
#[serde(tag = "channel_type")]
pub enum Channel {
@@ -646,10 +649,11 @@ impl Channel {
.private(user.to_string())
.await;
#[cfg(feature = "tasks")]
crate::tasks::ack::queue_ack(
self.id().to_string(),
user.to_string(),
AckEvent::AckMessage {
crate::tasks::ack::AckEvent::AckMessage {
id: message.to_string(),
},
)
@@ -766,6 +770,7 @@ impl Channel {
}
}
#[cfg(feature = "mongodb")]
impl IntoDocumentPath for FieldsChannel {
fn as_path(&self) -> Option<&'static str> {
Some(match self {
@@ -1,5 +1,7 @@
use crate::{revolt_result::Result, Channel, FieldsChannel, PartialChannel};
use revolt_permissions::OverrideField;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::Emoji;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::FileHash;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -4,6 +4,7 @@ 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": report_internal_error!(to_document(&used_for))?,
"used_for": to_document(&used_for).to_internal_error()?,
"uploader_id": uploader_id
}
},
@@ -10,11 +10,9 @@ 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,
@@ -22,6 +20,9 @@ use crate::{
Channel, Database, Emoji, File, User, AMQP,
};
#[cfg(feature = "tasks")]
use crate::tasks::{self, ack::AckEvent};
auto_derived_partial!(
/// Message
pub struct Message {
@@ -487,31 +488,28 @@ 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;
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_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await?;
user_mentions.retain(|m| valid_mentions.contains(m)); // quick pass, validate mentions are in the server
let valid_mentions = HashSet::<&String, RandomState>::from_iter(
valid_members.iter().map(|m| &m.id.user),
);
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;
user_mentions.retain(|m| valid_mentions.contains(m)); // quick pass, validate mentions are in the server
user_mentions
.retain(|m| *member_channel_view_perms.get(m).unwrap_or(&false));
}
} else {
revolt_config::capture_error(&valid_members.unwrap_err());
return Err(create_error!(InternalError));
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;
user_mentions
.retain(|m| *member_channel_view_perms.get(m).unwrap_or(&false));
}
}
Channel::SavedMessages { .. } => {
user_mentions.clear();
@@ -616,9 +614,11 @@ 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,6 +637,7 @@ impl Message {
}
// Generate embeds
#[cfg(feature = "tasks")]
if generate_embeds {
if let Some(content) = &self.content {
tasks::process_embeds::queue(
@@ -673,10 +674,12 @@ 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 {
@@ -710,12 +713,6 @@ 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,6 +2,7 @@ use revolt_result::Result;
use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
+7 -1
View File
@@ -38,7 +38,10 @@ pub use servers::*;
pub use user_settings::*;
pub use users::*;
use crate::{Database, MongoDb, ReferenceDb};
use crate::{Database, ReferenceDb};
#[cfg(feature = "mongodb")]
use crate::MongoDb;
pub trait AbstractDatabase:
Sync
@@ -66,6 +69,8 @@ pub trait AbstractDatabase:
}
impl AbstractDatabase for ReferenceDb {}
#[cfg(feature = "mongodb")]
impl AbstractDatabase for MongoDb {}
impl std::ops::Deref for Database {
@@ -74,6 +79,7 @@ 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,6 +2,7 @@ use revolt_result::Result;
use crate::PolicyChange;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -1,6 +1,8 @@
use std::time::Duration;
use crate::{revolt_result::Result, RatelimitEvent, RatelimitEventType};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::Report;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::Snapshot;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::{MemberCompositeKey, ServerBan};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -1,16 +1,20 @@
use ::mongodb::SessionCursor;
#[cfg(feature = "mongodb")]
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: ::mongodb::ClientSession,
session: ClientSession,
cursor: Option<SessionCursor<Member>>,
},
@@ -22,7 +26,7 @@ pub enum ChunkedServerMembersGenerator {
impl ChunkedServerMembersGenerator {
#[cfg(feature = "mongodb")]
pub fn new_mongo(session: ::mongodb::ClientSession, cursor: SessionCursor<Member>) -> Self {
pub fn new_mongo(session: ClientSession, cursor: SessionCursor<Member>) -> Self {
ChunkedServerMembersGenerator::MongoDb {
session,
cursor: Some(cursor),
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::{FieldsRole, FieldsServer, PartialRole, PartialServer, Role, Server};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -2,6 +2,7 @@ use revolt_result::Result;
use crate::UserSettings;
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -705,6 +705,8 @@ 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,6 +4,7 @@ use revolt_result::Result;
use crate::{FieldsUser, PartialUser, RelationshipStatus, User};
#[cfg(feature = "mongodb")]
mod mongodb;
mod reference;
@@ -212,16 +212,34 @@ impl AbstractUsers for MongoDb {
partial: &PartialUser,
remove: Vec<FieldsUser>,
) -> Result<()> {
query!(
self,
update_one_by_id,
COL,
id,
partial,
remove.iter().map(|x| x as &dyn IntoDocumentPath).collect(),
None
)
.map(|_| ())
if remove.contains(&FieldsUser::StatusText) && partial.status.is_some() {
// stupid-ass workaround to fix mongo conflicting the same item
let _: Result<()> = query!(
self,
update_one_by_id,
COL,
id,
PartialUser {
..Default::default()
},
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,6 +38,7 @@ 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,6 +113,8 @@ 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)));
}
+19 -20
View File
@@ -14,41 +14,40 @@ use crate::{
};
/// Reference to some object in the database
#[derive(Serialize, Deserialize)]
pub struct Reference {
pub struct Reference<'a> {
/// Id of object
pub id: String,
pub id: &'a str,
}
impl Reference {
impl<'a> Reference<'a> {
/// Create a Ref from an unchecked string
pub fn from_unchecked(id: String) -> Reference {
pub fn from_unchecked(id: &'a str) -> Reference<'a> {
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));
@@ -65,18 +64,18 @@ impl Reference {
.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));
}
@@ -86,36 +85,36 @@ impl Reference {
/// 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 {
impl<'r> FromParam<'r> for Reference<'r> {
type Error = &'r str;
fn from_param(param: &'r str) -> Result<Self, Self::Error> {
Ok(Reference::from_unchecked(param.into()))
Ok(Reference::from_unchecked(param))
}
}
#[cfg(feature = "rocket-impl")]
impl JsonSchema for Reference {
impl<'a> JsonSchema for Reference<'a> {
fn schema_name() -> String {
"Id".to_string()
}
+2 -2
View File
@@ -14,7 +14,7 @@ imagesize = "0.13.0"
tempfile = "3.12.0"
base64 = "0.22.1"
aes-gcm = "0.10.3"
aes-gcm = { version = "0.10.3", features = ["std"] }
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" }
revolt-result = { version = "0.8.8", path = "../result", features = ["sentry"] }
# image processing
jxl-oxide = "0.8.1"
+28 -31
View File
@@ -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};
use revolt_result::{create_error, Result, ToRevoltError};
use aws_sdk_s3::{
config::{Credentials, Region},
@@ -55,13 +55,12 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
// Send a request for the file
let mut obj =
report_internal_error!(client.get_object().bucket(bucket_id).key(path).send().await)?;
client.get_object().bucket(bucket_id).key(path).send().await.to_internal_error()?;
// Read the file from remote
let mut buf = vec![];
while let Some(bytes) = obj.body.next().await {
let data = report_internal_error!(bytes)?;
report_internal_error!(buf.write_all(&data))?;
buf.write_all(&bytes.to_internal_error()?).to_internal_error()?;
// is there a more efficient way to do this?
// we just want the Vec<u8>
}
@@ -78,7 +77,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)
.map_err(|_| create_error!(InternalError))?;
.to_internal_error()?;
Ok(buf)
}
@@ -97,18 +96,17 @@ 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)
.map_err(|_| create_error!(InternalError))?;
.to_internal_error()?;
// Upload the file to remote
report_internal_error!(
client
.put_object()
.bucket(bucket_id)
.key(path)
.body(buf.into())
.send()
.await
)?;
client
.put_object()
.bucket(bucket_id)
.key(path)
.body(buf.into())
.send()
.await
.to_internal_error()?;
Ok(BASE64_STANDARD.encode(nonce))
}
@@ -118,14 +116,13 @@ pub async fn delete_from_s3(bucket_id: &str, path: &str) -> Result<()> {
let config = config().await;
let client = create_client(config.files.s3);
report_internal_error!(
client
.delete_object()
.bucket(bucket_id)
.key(path)
.send()
.await
)?;
client
.delete_object()
.bucket(bucket_id)
.key(path)
.send()
.await
.to_internal_error()?;
Ok(())
}
@@ -145,8 +142,7 @@ 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 =
report_internal_error!(usvg::Tree::from_data(v, &Default::default())).ok()?;
let tree = usvg::Tree::from_data(v, &Default::default()).to_internal_error().ok()?;
let size = tree.size();
Some((size.width() as usize, size.height() as usize))
@@ -221,9 +217,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();
report_internal_error!(reader.read_to_end(&mut buf))?;
reader.read_to_end(&mut buf).to_internal_error()?;
let tree = report_internal_error!(usvg::Tree::from_data(&buf, &Default::default()))?;
let tree = usvg::Tree::from_data(&buf, &Default::default()).to_internal_error()?;
let size = tree.size();
let mut pixmap = Pixmap::new(size.width() as u32, size.height() as u32)
.ok_or_else(|| create_error!(ImageProcessingFailed))?;
@@ -241,10 +237,11 @@ pub fn decode_image<R: Read + BufRead + Seek>(reader: &mut R, mime: &str) -> Res
))
}
// Check if we can read using image-rs crate
_ => report_internal_error!(report_internal_error!(
image::ImageReader::new(reader).with_guessed_format()
)?
.decode()),
_ => image::ImageReader::new(reader)
.with_guessed_format()
.to_internal_error()?
.decode()
.to_internal_error()
}
}
+2 -2
View File
@@ -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 = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsBot>>,
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsBot>,
}
/// Where we are inviting a bot to
+1 -1
View File
@@ -207,7 +207,7 @@ auto_derived!(
/// Fields to remove from channel
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Option<Vec<FieldsChannel>>,
pub remove: Vec<FieldsChannel>,
}
/// Create new group
+1 -1
View File
@@ -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 = 128)))]
#[cfg_attr(feature = "validator", validate(length(min = 1, max = 256)))]
pub icon_url: Option<String>,
#[cfg_attr(feature = "validator", validate(length(min = 1, max = 256)))]
pub url: Option<String>,
+2
View File
@@ -7,6 +7,7 @@ mod embeds;
mod emojis;
mod files;
mod messages;
mod onboard;
mod policy_changes;
mod safety_reports;
mod server_bans;
@@ -24,6 +25,7 @@ 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::*;
+15
View File
@@ -0,0 +1,15 @@
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,6 +1,17 @@
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
+2 -2
View File
@@ -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 = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsMember>>,
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsMember>,
}
);
+25 -5
View File
@@ -1,4 +1,4 @@
use super::{Channel, File, RE_COLOUR};
use super::{Channel, File, Member, User, RE_COLOUR};
use revolt_permissions::{Override, OverrideField};
use std::collections::HashMap;
@@ -175,6 +175,8 @@ auto_derived!(
/// Ranking position
///
/// Smaller values take priority.
///
/// **Removed** - no effect, use the edit server role positions route
pub rank: Option<i64>,
}
@@ -247,8 +249,8 @@ auto_derived!(
pub analytics: Option<bool>,
/// Fields to remove from server object
#[cfg_attr(feature = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsServer>>,
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsServer>,
}
/// New role information
@@ -270,8 +272,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 = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsRole>>,
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsRole>,
}
/// New role permissions
@@ -291,4 +293,22 @@ 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>,
}
);
+13 -2
View File
@@ -245,8 +245,8 @@ auto_derived!(
pub flags: Option<i32>,
/// Fields to remove from user object
#[cfg_attr(feature = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsUser>>,
#[cfg_attr(feature = "serde", serde(default))]
pub remove: Vec<FieldsUser>,
}
/// User flag reponse
@@ -275,6 +275,17 @@ 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 {
+4
View File
@@ -15,6 +15,7 @@ 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"]
@@ -34,3 +35,6 @@ revolt_okapi = { version = "0.9.1", optional = true }
# Axum
axum = { version = "0.7.5", optional = true }
# Sentry
sentry = { version = "0.31.5", optional = true }
+15 -7
View File
@@ -2,10 +2,9 @@ use axum::{http::StatusCode, response::IntoResponse, Json};
use crate::{Error, ErrorType};
/// HTTP response builder for Error enum
impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response {
let status = match self.error_type {
impl Error {
pub fn axum_status(&self) -> StatusCode {
match self.error_type {
ErrorType::LabelMe => StatusCode::INTERNAL_SERVER_ERROR,
ErrorType::AlreadyOnboarded => StatusCode::FORBIDDEN,
@@ -70,11 +69,15 @@ impl IntoResponse for 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::FailedValidation { .. } => StatusCode::BAD_REQUEST,
ErrorType::IOError => 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,
@@ -84,8 +87,13 @@ impl IntoResponse for Error {
ErrorType::FileTypeNotAllowed => StatusCode::BAD_REQUEST,
ErrorType::ImageProcessingFailed => StatusCode::INTERNAL_SERVER_ERROR,
ErrorType::NoEmbedData => StatusCode::BAD_REQUEST,
};
}
}
}
(status, Json(&self)).into_response()
/// HTTP response builder for Error enum
impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response {
(self.axum_status(), Json(&self)).into_response()
}
}
+64 -1
View File
@@ -1,3 +1,4 @@
use std::panic::Location;
use std::fmt::Display;
#[cfg(feature = "serde")]
@@ -151,9 +152,15 @@ pub enum ErrorType {
InvalidSession,
InvalidFlagValue,
NotAuthenticated,
Conflict,
DuplicateNonce,
NotFound,
NoEffect,
IOError,
UnprocessableEntity,
DeserializationError {
error: String,
},
FailedValidation {
error: String,
},
@@ -174,7 +181,7 @@ pub enum ErrorType {
// ? Feature flag disabled in the config
FeatureDisabled {
feature: String,
},
}
}
#[macro_export]
@@ -197,6 +204,62 @@ 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;
+15 -7
View File
@@ -8,10 +8,9 @@ use rocket::{
use crate::{Error, ErrorType};
/// HTTP response builder for Error enum
impl<'r> Responder<'r, 'static> for Error {
fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
let status = match self.error_type {
impl Error {
pub fn rocket_status(&self) -> Status {
match self.error_type {
ErrorType::LabelMe => Status::InternalServerError,
ErrorType::AlreadyOnboarded => Status::Forbidden,
@@ -77,11 +76,15 @@ impl<'r> Responder<'r, 'static> for 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::FailedValidation { .. } => Status::BadRequest,
ErrorType::IOError => Status::BadRequest,
ErrorType::UnprocessableEntity => Status::UnprocessableEntity,
ErrorType::DeserializationError { .. } => Status::UnprocessableEntity,
ErrorType::FailedValidation { .. } => Status::UnprocessableEntity,
ErrorType::FeatureDisabled { .. } => Status::BadRequest,
ErrorType::ProxyError => Status::BadRequest,
@@ -90,8 +93,13 @@ impl<'r> Responder<'r, 'static> for 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();
@@ -99,7 +107,7 @@ impl<'r> Responder<'r, 'static> for Error {
Response::build()
.sized_body(string.len(), Cursor::new(string))
.header(ContentType::new("application", "json"))
.status(status)
.status(self.rocket_status())
.ok()
}
}
+22 -9
View File
@@ -1,19 +1,32 @@
use revolt_config::configure;
use revolt_database::DatabaseInfo;
use std::{future::Future, time::Duration};
use revolt_config::{configure, capture_error};
use revolt_database::{Database, DatabaseInfo};
use revolt_result::Result;
use tasks::{file_deletion, prune_dangling_files};
use tokio::try_join;
use tokio::{join, time::sleep};
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() -> Result<()> {
async fn main() {
configure!(crond);
let db = DatabaseInfo::Auto.connect().await.expect("database");
try_join!(
file_deletion::task(db.clone()),
prune_dangling_files::task(db)
)
.map(|_| ())
join!(
cron_task_wrapper(file_deletion::task, db.clone()),
cron_task_wrapper(prune_dangling_files::task, db.clone()),
);
}
+1
View File
@@ -8,6 +8,7 @@ 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,24 +123,26 @@ 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);
if let Ok(p) = raw_service_payload {
let args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
match serde_json::to_string(&service_payload) {
Ok(p) => {
let args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
log::debug!(
"Publishing ack to apn session {}",
session.subscription.as_ref().unwrap().auth
);
log::debug!(
"Publishing ack to apn session {}",
session.subscription.as_ref().unwrap().auth
);
publish_message(self, p.into(), args).await;
} else {
log::warn!("Failed to serialize ack badge update payload!");
revolt_config::capture_error(&raw_service_payload.unwrap_err());
publish_message(self, p.into(), args).await;
},
Err(e) => {
log::warn!("Failed to serialize ack badge update payload!");
revolt_config::capture_error(&e);
}
}
}
}
+1 -4
View File
@@ -32,7 +32,6 @@ 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"] }
@@ -66,12 +65,10 @@ 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",
@@ -82,7 +79,7 @@ revolt-models = { path = "../core/models", features = [
"rocket",
] }
revolt-presence = { path = "../core/presence" }
revolt-result = { path = "../core/result", features = ["rocket", "okapi"] }
revolt-result = { path = "../core/result", features = ["rocket", "okapi", "sentry"] }
revolt-permissions = { path = "../core/permissions", features = ["schemas"] }
[build-dependencies]
+1
View File
@@ -135,6 +135,7 @@ 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(),
+4 -10
View File
@@ -1,9 +1,8 @@
use crate::util::json::{Json, Validate};
use revolt_database::{Bot, Database, User};
use revolt_models::v0;
use revolt_result::{create_error, Result};
use rocket::serde::json::Json;
use revolt_result::Result;
use rocket::State;
use validator::Validate;
/// # Create Bot
///
@@ -13,14 +12,9 @@ use validator::Validate;
pub async fn create_bot(
db: &State<Database>,
user: User,
info: Json<v0::DataCreateBot>,
info: Validate<Json<v0::DataCreateBot>>,
) -> Result<Json<v0::BotWithUserResponse>> {
let info = info.into_inner();
info.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let info = info.into_inner().into_inner();
let (bot, user) = Bot::create(db, info.name, &user, None).await?;
Ok(Json(v0::BotWithUserResponse {
+1 -1
View File
@@ -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 {
+6 -13
View File
@@ -3,8 +3,7 @@ use revolt_models::v0::{self, DataEditBot};
use revolt_result::{create_error, Result};
use rocket::State;
use rocket::serde::json::Json;
use validator::Validate;
use crate::util::json::{Json, Validate};
/// # Edit Bot
///
@@ -14,15 +13,10 @@ use validator::Validate;
pub async fn edit_bot(
db: &State<Database>,
user: User,
target: Reference,
data: Json<DataEditBot>,
target: Reference<'_>,
data: Validate<Json<DataEditBot>>,
) -> Result<Json<v0::BotWithUserResponse>> {
let data = data.into_inner();
data.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let data = data.into_inner().into_inner();
let mut bot = target.as_bot(db).await?;
if bot.owner != user.id {
@@ -37,7 +31,7 @@ pub async fn edit_bot(
if data.public.is_none()
&& data.analytics.is_none()
&& data.interactions_url.is_none()
&& data.remove.is_none()
&& data.remove.is_empty()
{
return Ok(Json(v0::BotWithUserResponse {
bot: bot.into(),
@@ -64,7 +58,6 @@ pub async fn edit_bot(
db,
partial,
remove
.unwrap_or_default()
.into_iter()
.map(|v| v.into())
.collect(),
@@ -100,7 +93,7 @@ mod test {
.body(
json!(v0::DataEditBot {
public: Some(true),
remove: Some(vec![FieldsBot::Token]),
remove: vec![FieldsBot::Token],
..Default::default()
})
.to_string(),
+3 -2
View File
@@ -1,7 +1,8 @@
use revolt_database::{util::reference::Reference, Database, User};
use revolt_models::v0::FetchBotResponse;
use revolt_result::{create_error, Result};
use rocket::{serde::json::Json, State};
use rocket::State;
use crate::util::json::Json;
/// # Fetch Bot
///
@@ -11,7 +12,7 @@ use rocket::{serde::json::Json, State};
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));
+1 -1
View File
@@ -2,7 +2,7 @@ use futures::future::join_all;
use revolt_database::{Database, User};
use revolt_models::v0::OwnedBotsResponse;
use revolt_result::Result;
use rocket::serde::json::Json;
use crate::util::json::Json;
use rocket::State;
/// # Fetch Owned Bots
+3 -3
View File
@@ -2,7 +2,7 @@ use revolt_database::{util::reference::Reference, Database, User};
use revolt_models::v0::PublicBot;
use revolt_result::{create_error, Result};
use rocket::serde::json::Json;
use crate::util::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));
}
+2 -2
View File
@@ -8,7 +8,7 @@ use revolt_permissions::{
use revolt_result::{create_error, Result};
use rocket::State;
use rocket::serde::json::Json;
use crate::util::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,
message: Reference,
target: 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::{serde::json::Json, State};
use validator::Validate;
use rocket::State;
use crate::util::json::{Json, Validate};
/// # Edit Channel
///
@@ -17,15 +17,10 @@ pub async fn edit(
db: &State<Database>,
amqp: &State<AMQP>,
user: User,
target: Reference,
data: Json<v0::DataEditChannel>,
target: Reference<'_>,
data: Validate<Json<v0::DataEditChannel>>,
) -> Result<Json<v0::Channel>> {
let data = data.into_inner();
data.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let data = data.into_inner().into_inner();
let mut channel = target.as_channel(db).await?;
let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -38,7 +33,7 @@ pub async fn edit(
&& data.icon.is_none()
&& data.nsfw.is_none()
&& data.owner.is_none()
&& data.remove.is_none()
&& data.remove.is_empty()
{
return Ok(Json(channel.into()));
}
@@ -112,23 +107,21 @@ pub async fn edit(
nsfw,
..
} => {
if let Some(fields) = &data.remove {
if fields.contains(&v0::FieldsChannel::Icon) {
if let Some(icon) = &icon {
db.mark_attachment_as_deleted(&icon.id).await?;
}
if data.remove.contains(&v0::FieldsChannel::Icon) {
if let Some(icon) = &icon {
db.mark_attachment_as_deleted(&icon.id).await?;
}
}
for field in fields {
match field {
v0::FieldsChannel::Description => {
description.take();
}
v0::FieldsChannel::Icon => {
icon.take();
}
_ => {}
for field in &data.remove {
match field {
v0::FieldsChannel::Description => {
description.take();
}
v0::FieldsChannel::Icon => {
icon.take();
}
_ => {}
}
}
@@ -214,11 +207,7 @@ pub async fn edit(
.update(
db,
partial,
data.remove
.unwrap_or_default()
.into_iter()
.map(|f| f.into())
.collect(),
data.remove.into_iter().map(|f| f.into()).collect(),
)
.await?;
}
@@ -6,7 +6,8 @@ use revolt_database::{
use revolt_models::v0;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::Result;
use rocket::{serde::json::Json, State};
use rocket::State;
use crate::util::json::Json;
/// # Fetch Channel
///
@@ -16,7 +17,7 @@ use rocket::{serde::json::Json, State};
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,
member_id: Reference,
group_id: Reference<'_>,
member_id: Reference<'_>,
) -> Result<EmptyResponse> {
if user.bot.is_some() {
return Err(create_error!(IsBot));
@@ -2,9 +2,8 @@ use revolt_database::{Channel, Database, RelationshipStatus, User};
use revolt_models::v0;
use revolt_result::{create_error, Result};
use rocket::serde::json::Json;
use crate::util::json::{Json, Validate};
use rocket::State;
use validator::Validate;
/// # Create Group
///
@@ -14,18 +13,13 @@ use validator::Validate;
pub async fn create_group(
db: &State<Database>,
user: User,
data: Json<v0::DataCreateGroup>,
data: Validate<Json<v0::DataCreateGroup>>,
) -> Result<Json<v0::Channel>> {
if user.bot.is_some() {
return Err(create_error!(IsBot));
}
let data = data.into_inner();
data.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let data = data.into_inner().into_inner();
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,
member: Reference,
target: Reference<'_>,
member: Reference<'_>,
) -> Result<EmptyResponse> {
if user.bot.is_some() {
return Err(create_error!(IsBot));
@@ -6,7 +6,8 @@ use revolt_models::v0;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
use rocket::{serde::json::Json, State};
use rocket::State;
use crate::util::json::Json;
/// # Create Invite
///
@@ -18,7 +19,7 @@ use rocket::{serde::json::Json, State};
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,7 +5,8 @@ use revolt_database::{
use revolt_models::v0;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
use rocket::{serde::json::Json, State};
use rocket::State;
use crate::util::json::Json;
/// # Fetch Group Members
///
@@ -17,7 +18,7 @@ use rocket::{serde::json::Json, State};
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::{serde::json::Json, State};
use rocket::State;
use rocket_empty::EmptyResponse;
use validator::Validate;
use crate::util::json::{Json, Validate};
/// # Bulk Delete Messages
///
@@ -22,15 +22,10 @@ use validator::Validate;
pub async fn bulk_delete_messages(
db: &State<Database>,
user: User,
target: Reference,
options: Json<v0::OptionsBulkDelete>,
target: Reference<'_>,
options: Validate<Json<v0::OptionsBulkDelete>>,
) -> Result<EmptyResponse> {
let options = options.into_inner();
options.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let options = options.into_inner().into_inner();
for id in &options.ids {
if ulid::Ulid::from_string(id)
@@ -51,7 +46,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,
msg: Reference,
target: 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,
msg: Reference,
target: 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::{serde::json::Json, State};
use validator::Validate;
use rocket::State;
use crate::util::json::{Json, Validate};
/// # Edit Message
///
@@ -18,16 +18,11 @@ use validator::Validate;
pub async fn edit(
db: &State<Database>,
user: User,
target: Reference,
msg: Reference,
edit: Json<v0::DataEditMessage>,
target: Reference<'_>,
msg: Reference<'_>,
edit: Validate<Json<v0::DataEditMessage>>,
) -> Result<Json<v0::Message>> {
let edit = edit.into_inner();
edit.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let edit = edit.into_inner().into_inner();
Message::validate_sum(
&edit.content,
@@ -5,7 +5,8 @@ use revolt_database::{
use revolt_models::v0;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
use rocket::{serde::json::Json, State};
use rocket::State;
use crate::util::json::Json;
/// # Fetch Message
///
@@ -15,8 +16,8 @@ use rocket::{serde::json::Json, State};
pub async fn fetch(
db: &State<Database>,
user: User,
target: Reference,
msg: Reference,
target: Reference<'_>,
msg: Reference<'_>,
) -> Result<Json<v0::Message>> {
let channel = target.as_channel(db).await?;
let mut query = DatabasePermissionQuery::new(db, &user).channel(&channel);
@@ -1,4 +1,7 @@
use revolt_database::{util::{permissions::DatabasePermissionQuery, reference::Reference}, Channel, Database, PartialMessage, SystemMessage, User, AMQP};
use revolt_database::{
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};
@@ -14,8 +17,8 @@ pub async fn message_pin(
db: &State<Database>,
amqp: &State<AMQP>,
user: User,
target: Reference,
msg: Reference,
target: Reference<'_>,
msg: Reference<'_>,
) -> Result<EmptyResponse> {
let channel = target.as_channel(db).await?;
@@ -169,7 +172,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::{serde::json::Json, State};
use validator::Validate;
use rocket::State;
use crate::util::json::{Json, Validate};
/// # Fetch Messages
///
@@ -16,15 +16,9 @@ use validator::Validate;
pub async fn query(
db: &State<Database>,
user: User,
target: Reference,
options: v0::OptionsQueryMessages,
target: Reference<'_>,
options: Validate<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));
}
@@ -43,7 +37,7 @@ pub async fn query(
sort,
nearby,
include_users,
} = options;
} = options.into_inner();
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,
msg: Reference,
emoji: Reference,
target: Reference<'_>,
msg: 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::{serde::json::Json, State};
use validator::Validate;
use rocket::State;
use crate::util::json::{Json, Validate};
/// # Search for Messages
///
@@ -16,19 +16,14 @@ use validator::Validate;
pub async fn search(
db: &State<Database>,
user: User,
target: Reference,
options: Json<v0::DataMessageSearch>,
target: Reference<'_>,
options: Validate<Json<v0::DataMessageSearch>>,
) -> Result<Json<v0::BulkMessageResponse>> {
if user.bot.is_some() {
return Err(create_error!(IsBot));
}
let options = options.into_inner();
options.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let options = options.into_inner().into_inner();
if options.query.is_some() && options.pinned.is_some() {
return Err(create_error!(InvalidOperation))
@@ -7,10 +7,9 @@ use revolt_database::{Interactions, Message, AMQP};
use revolt_models::v0;
use revolt_permissions::PermissionQuery;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
use rocket::serde::json::Json;
use revolt_result::Result;
use crate::util::json::{Json, Validate};
use rocket::State;
use validator::Validate;
/// # Send Message
///
@@ -21,16 +20,11 @@ pub async fn message_send(
db: &State<Database>,
amqp: &State<AMQP>,
user: User,
target: Reference,
data: Json<v0::DataMessageSend>,
target: Reference<'_>,
data: Validate<Json<v0::DataMessageSend>>,
idempotency: IdempotencyKey,
) -> Result<Json<v0::Message>> {
let data = data.into_inner();
data.validate().map_err(|error| {
create_error!(FailedValidation {
error: error.to_string()
})
})?;
let data = data.into_inner().into_inner();
// Ensure we have permissions to send a message
let channel = target.as_channel(db).await?;
@@ -202,7 +196,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.clone())
let member = Reference::from_unchecked(&user.id)
.as_member(&harness.db, &server.id)
.await
.expect("Failed to get member");
@@ -242,7 +236,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.clone())
let mut second_member = Reference::from_unchecked(&second_user.id)
.as_member(&harness.db, &server.id)
.await
.expect("Failed to get second member");

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