From ab5bd47a39ee889de0b5ae6e7b560620853daead Mon Sep 17 00:00:00 2001 From: Tom Date: Fri, 8 May 2026 14:39:16 -0700 Subject: [PATCH] feat: Rewrite acks (#741) * feat: rewrite ack system Signed-off-by: IAmTomahawkx * feat: rewrite acks to crond + rabbit task * fix: review changes --------- Signed-off-by: IAmTomahawkx --- Cargo.lock | 638 +++++++++++++++++- Cargo.toml | 2 + compose.yml | 14 +- crates/core/config/Revolt.toml | 4 + crates/core/config/src/lib.rs | 7 + crates/core/database/Cargo.toml | 1 + crates/core/database/src/amqp/amqp.rs | 83 ++- crates/core/database/src/events/rabbit.rs | 8 + .../database/src/models/channels/model.rs | 19 +- crates/core/database/src/tasks/ack.rs | 10 +- crates/core/database/src/util/acker.rs | 77 +++ crates/core/database/src/util/mod.rs | 1 + crates/daemons/crond/Cargo.toml | 14 + crates/daemons/crond/src/main.rs | 5 +- crates/daemons/crond/src/tasks/acks.rs | 129 ++++ crates/daemons/crond/src/tasks/mod.rs | 1 + crates/delta/src/main.rs | 22 +- .../delta/src/routes/channels/channel_ack.rs | 5 +- crates/delta/src/routes/servers/server_ack.rs | 16 +- 19 files changed, 993 insertions(+), 63 deletions(-) create mode 100644 crates/core/database/src/util/acker.rs create mode 100644 crates/daemons/crond/src/tasks/acks.rs diff --git a/Cargo.lock b/Cargo.lock index 03b3406f..671d6a9d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -35,7 +35,7 @@ checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", "cipher", - "cpufeatures", + "cpufeatures 0.2.17", ] [[package]] @@ -124,6 +124,56 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "amq-protocol" +version = "10.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5b2e8843d88d935a75cbdb1c5b9ad80987da6e6472c2703914e41dc14560990" +dependencies = [ + "amq-protocol-tcp", + "amq-protocol-types", + "amq-protocol-uri", + "cookie-factory", + "nom 8.0.0", + "serde", +] + +[[package]] +name = "amq-protocol-tcp" +version = "10.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "998fb81655e11de5a336bb609042c11633678fc01f90b0772fb9a7886b6cc4c2" +dependencies = [ + "amq-protocol-uri", + "async-rs", + "cfg-if", + "tcp-stream", + "tracing", +] + +[[package]] +name = "amq-protocol-types" +version = "10.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee5b3a9e458bd2e452536995c8cf861b01c17283f55c4bbe0a1b9626b8253add" +dependencies = [ + "cookie-factory", + "nom 8.0.0", + "serde", + "serde_json", +] + +[[package]] +name = "amq-protocol-uri" +version = "10.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dca3316970d20cdcca9123f4e8feb7a2c1c8fdca572a9692fc10002db35407aa" +dependencies = [ + "amq-protocol-types", + "percent-encoding", + "url", +] + [[package]] name = "amqp_serde" version = "0.4.3" @@ -211,6 +261,45 @@ dependencies = [ "stable_deref_trait", ] +[[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.18", + "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.45", + "syn 2.0.117", + "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.45", + "syn 2.0.117", +] + [[package]] name = "async-attributes" version = "1.1.2" @@ -244,6 +333,19 @@ dependencies = [ "pin-project-lite 0.2.17", ] +[[package]] +name = "async-compat" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1ba85bc55464dcbf728b56d97e119d673f4cf9062be330a9a26f3acf504a590" +dependencies = [ + "futures-core", + "futures-io", + "once_cell", + "pin-project-lite 0.2.17", + "tokio 1.51.0", +] + [[package]] name = "async-executor" version = "1.14.0" @@ -275,6 +377,20 @@ dependencies = [ "tokio 1.51.0", ] +[[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.5.0", + "async-executor", + "async-lock 3.4.2", + "blocking", + "futures-lite", + "tokio 1.51.0", +] + [[package]] name = "async-io" version = "2.6.0" @@ -342,6 +458,22 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "async-rs" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e32bd31386d41d0c06bd79b0397ec96e544d69d9dbd6db0236c7ceefe1ad61b" +dependencies = [ + "async-compat", + "async-global-executor 3.1.0", + "async-trait", + "futures-core", + "futures-io", + "hickory-resolver 0.26.1", + "tokio 1.51.0", + "tokio-stream", +] + [[package]] name = "async-signal" version = "0.2.13" @@ -368,7 +500,7 @@ checksum = "2c8e079a4ab67ae52b7403632e4618815d6db36d2a010cfe41b02c1b1578f93b" dependencies = [ "async-attributes", "async-channel 1.9.0", - "async-global-executor", + "async-global-executor 2.4.1", "async-io", "async-lock 3.4.2", "async-process", @@ -1130,6 +1262,15 @@ dependencies = [ "ubyte", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand 2.4.0", +] + [[package]] name = "backtrace" version = "0.3.76" @@ -1289,6 +1430,15 @@ dependencies = [ "generic-array 0.14.7", ] +[[package]] +name = "block-padding" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93" +dependencies = [ + "generic-array 0.14.7", +] + [[package]] name = "block2" version = "0.6.2" @@ -1441,6 +1591,15 @@ dependencies = [ "rustversion", ] +[[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.59" @@ -1482,6 +1641,17 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chacha20" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.44" @@ -1514,6 +1684,18 @@ dependencies = [ "cc", ] +[[package]] +name = "cms" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b77c319abfd5219629c45c34c89ba945ed3c5e49fcde9d16b6c3885f118a730" +dependencies = [ + "const-oid 0.9.6", + "der 0.7.10", + "spki 0.7.3", + "x509-cert", +] + [[package]] name = "coarsetime" version = "0.1.37" @@ -1699,6 +1881,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crc" version = "3.3.0" @@ -1907,7 +2098,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "curve25519-dalek-derive", "digest", "fiat-crypto", @@ -2128,7 +2319,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "79b71cca7d95d7681a4b3b9cdf63c8dbc3730d0584c2c74e31416d64a90493f4" dependencies = [ "const-oid 0.6.2", - "der_derive", + "der_derive 0.4.1", ] [[package]] @@ -2149,10 +2340,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid 0.9.6", + "der_derive 0.7.3", + "flagset", "pem-rfc7468 0.7.0", "zeroize", ] +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom 7.1.3", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "der_derive" version = "0.4.1" @@ -2165,6 +2372,17 @@ dependencies = [ "synstructure 0.12.6", ] +[[package]] +name = "der_derive" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18" +dependencies = [ + "proc-macro2", + "quote 1.0.45", + "syn 2.0.117", +] + [[package]] name = "deranged" version = "0.5.8" @@ -2231,6 +2449,15 @@ dependencies = [ "unicode-xid 0.2.6", ] +[[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" @@ -2758,6 +2985,12 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" +[[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.9" @@ -2777,6 +3010,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "futures-core", + "futures-sink", + "spin 0.9.8", +] + [[package]] name = "fnv" version = "1.0.7" @@ -2973,6 +3217,17 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "futures-rustls" +version = "0.26.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8f2f12607f92c69b12ed746fabf9ca4f5c482cba46679c1a75b874ed7c26adb" +dependencies = [ + "futures-io", + "rustls 0.23.37", + "rustls-pki-types", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -3096,6 +3351,7 @@ dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", "wasip2", "wasip3", ] @@ -3362,6 +3618,30 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hickory-net" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2295ed2f9c31e471e1428a8f88a3f0e1f4b27c15049592138d1eebe9c35b183" +dependencies = [ + "async-trait", + "cfg-if", + "data-encoding", + "futures-channel", + "futures-io", + "futures-util", + "hickory-proto 0.26.1", + "idna 1.1.0", + "ipnet", + "jni 0.22.4", + "rand 0.10.1", + "thiserror 2.0.18", + "tinyvec", + "tokio 1.51.0", + "tracing", + "url", +] + [[package]] name = "hickory-proto" version = "0.25.2" @@ -3387,6 +3667,26 @@ dependencies = [ "url", ] +[[package]] +name = "hickory-proto" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bab31817bfb44672a252e97fe81cd0c18d1b2cf892108922f6818820df8c643" +dependencies = [ + "data-encoding", + "idna 1.1.0", + "ipnet", + "jni 0.22.4", + "once_cell", + "prefix-trie", + "rand 0.10.1", + "ring", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "url", +] + [[package]] name = "hickory-resolver" version = "0.25.2" @@ -3395,7 +3695,7 @@ checksum = "dc62a9a99b0bfb44d2ab95a7208ac952d31060efc16241c87eaf36406fecf87a" dependencies = [ "cfg-if", "futures-util", - "hickory-proto", + "hickory-proto 0.25.2", "ipconfig", "moka", "once_cell", @@ -3408,6 +3708,32 @@ dependencies = [ "tracing", ] +[[package]] +name = "hickory-resolver" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d58d28879ceecde6607729660c2667a081ccdc082e082675042793960f178c" +dependencies = [ + "cfg-if", + "futures-util", + "hickory-net", + "hickory-proto 0.26.1", + "ipconfig", + "ipnet", + "jni 0.22.4", + "moka", + "ndk-context", + "once_cell", + "parking_lot", + "rand 0.10.1", + "resolv-conf", + "smallvec", + "system-configuration 0.7.0", + "thiserror 2.0.18", + "tokio 1.51.0", + "tracing", +] + [[package]] name = "hkdf" version = "0.12.4" @@ -3977,6 +4303,7 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ + "block-padding", "generic-array 0.14.7", ] @@ -4018,6 +4345,9 @@ name = "ipnet" version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +dependencies = [ + "serde", +] [[package]] name = "iri-string" @@ -4129,6 +4459,36 @@ dependencies = [ "windows-sys 0.45.0", ] +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys 0.4.1", + "log", + "simd_cesu8", + "thiserror 2.0.18", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote 1.0.45", + "rustc_version", + "simd_cesu8", + "syn 2.0.117", +] + [[package]] name = "jni-sys" version = "0.3.1" @@ -4460,6 +4820,24 @@ dependencies = [ "log", ] +[[package]] +name = "lapin" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "478790661081f7434e111a31953acc1cf23654cf2a2d815d0e12cef9a2aed720" +dependencies = [ + "amq-protocol", + "async-rs", + "async-trait", + "atomic-waker", + "backon", + "cfg-if", + "flume", + "futures-core", + "futures-io", + "tracing", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -5008,8 +5386,8 @@ dependencies = [ "futures-io", "futures-util", "hex", - "hickory-proto", - "hickory-resolver", + "hickory-proto 0.25.2", + "hickory-resolver 0.25.2", "hmac", "macro_magic", "md-5", @@ -5118,6 +5496,12 @@ dependencies = [ "tempfile", ] +[[package]] +name = "ndk-context" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" + [[package]] name = "new_debug_unreachable" version = "1.0.6" @@ -5471,6 +5855,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -5575,6 +5968,29 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" +[[package]] +name = "p12-keystore" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ffb9bf5222606eb712d3bb30e01bc9420545b00859970897e70c682353a034f2" +dependencies = [ + "base64 0.22.1", + "cbc", + "cms", + "der 0.7.10", + "des", + "hex", + "hmac", + "pkcs12", + "pkcs5", + "rand 0.10.1", + "rc2", + "sha1", + "sha2", + "thiserror 2.0.18", + "x509-parser", +] + [[package]] name = "p256" version = "0.11.1" @@ -5701,6 +6117,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2" dependencies = [ "digest", + "hmac", ] [[package]] @@ -6092,6 +6509,36 @@ dependencies = [ "spki 0.7.3", ] +[[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", + "scrypt", + "sha2", + "spki 0.7.3", +] + [[package]] name = "pkcs8" version = "0.9.0" @@ -6165,7 +6612,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "opaque-debug", "universal-hash", ] @@ -6206,6 +6653,17 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "925383efa346730478fb4838dbe9137d2a47675ad789c546d150a6e1dd4ab31c" +[[package]] +name = "prefix-trie" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90f561214012d3fc240a1f9c817cc4d57f5310910d066069c1b093f766bb5966" +dependencies = [ + "either", + "ipnet", + "num-traits", +] + [[package]] name = "pretty_env_logger" version = "0.4.0" @@ -6609,6 +7067,17 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rand" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" +dependencies = [ + "chacha20", + "getrandom 0.4.2", + "rand_core 0.10.1", +] + [[package]] name = "rand_chacha" version = "0.3.1" @@ -6647,6 +7116,12 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "rav1e" version = "0.8.1" @@ -6717,6 +7192,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rc2" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62c64daa8e9438b84aaae55010a93f396f8e60e3911590fcba770d04643fc1dd" +dependencies = [ + "cipher", +] + [[package]] name = "redis" version = "0.23.3" @@ -6929,7 +7413,7 @@ dependencies = [ "quinn", "rustls 0.23.37", "rustls-pki-types", - "rustls-platform-verifier", + "rustls-platform-verifier 0.6.2", "serde", "serde_json", "serde_urlencoded", @@ -7069,11 +7553,19 @@ dependencies = [ name = "revolt-crond" version = "0.12.1" dependencies = [ + "futures-lite", + "iso8601-timestamp", + "lapin", "log", + "redis-kiss", "revolt-config", "revolt-database", "revolt-files", + "revolt-permissions", "revolt-result", + "revolt_optional_struct", + "serde", + "serde_json", "tokio 1.51.0", ] @@ -7108,6 +7600,7 @@ dependencies = [ "rand 0.8.5", "redis-kiss", "regex", + "revolt-coalesced", "revolt-config", "revolt-models", "revolt-parser", @@ -7793,6 +8286,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "rustix" version = "0.38.44" @@ -7861,6 +8363,21 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-connector" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f26bcb6901a3319d57589047c0da93a0f3228f13abf8dd949deef024749cb5e2" +dependencies = [ + "futures-io", + "futures-rustls", + "log", + "rustls 0.23.37", + "rustls-pki-types", + "rustls-platform-verifier 0.7.0", + "rustls-webpki 0.103.10", +] + [[package]] name = "rustls-native-certs" version = "0.6.3" @@ -7921,7 +8438,28 @@ checksum = "1d99feebc72bae7ab76ba994bb5e121b8d83d910ca40b36e0921f53becc41784" dependencies = [ "core-foundation 0.10.1", "core-foundation-sys", - "jni", + "jni 0.21.1", + "log", + "once_cell", + "rustls 0.23.37", + "rustls-native-certs 0.8.3", + "rustls-platform-verifier-android", + "rustls-webpki 0.103.10", + "security-framework 3.7.0", + "security-framework-sys", + "webpki-root-certs", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls-platform-verifier" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" +dependencies = [ + "core-foundation 0.10.1", + "core-foundation-sys", + "jni 0.22.4", "log", "once_cell", "rustls 0.23.37", @@ -8003,6 +8541,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "salsa20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" +dependencies = [ + "cipher", +] + [[package]] name = "same-file" version = "1.0.6" @@ -8075,6 +8622,17 @@ dependencies = [ "tendril", ] +[[package]] +name = "scrypt" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f" +dependencies = [ + "pbkdf2", + "salsa20", + "sha2", +] + [[package]] name = "sct" version = "0.7.1" @@ -8493,7 +9051,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f5058ada175748e33390e40e872bd0fe59a19f265d0158daa551c5a88a76009c" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -8504,7 +9062,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -8521,7 +9079,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -8576,6 +9134,16 @@ version = "0.3.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" +[[package]] +name = "simd_cesu8" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94f90157bb87cddf702797c5dadfa0be7d266cdf49e22da2fcaa32eff75b2c33" +dependencies = [ + "rustc_version", + "simdutf8", +] + [[package]] name = "simd_helpers" version = "0.1.0" @@ -8679,6 +9247,9 @@ name = "spin" version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] [[package]] name = "spin" @@ -8970,6 +9541,19 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" +[[package]] +name = "tcp-stream" +version = "0.34.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd7219422d3348cddeaf9073772997c452085c33e22d0b08cbd19652e1b16da5" +dependencies = [ + "async-rs", + "cfg-if", + "futures-io", + "p12-keystore", + "rustls-connector", +] + [[package]] name = "tempfile" version = "3.27.0" @@ -10733,6 +11317,34 @@ dependencies = [ "tap", ] +[[package]] +name = "x509-cert" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94" +dependencies = [ + "const-oid 0.9.6", + "der 0.7.10", + "spki 0.7.3", +] + +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom 7.1.3", + "oid-registry", + "rusticata-macros", + "thiserror 2.0.18", + "time", +] + [[package]] name = "xmlparser" version = "0.13.6" diff --git a/Cargo.toml b/Cargo.toml index f10bfb26..6ce826ef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -159,6 +159,7 @@ authifier = "1.0.16" # RabbitMQ amqprs = "1.7.0" +lapin = "4.7.1" # Voice livekit-api = "0.4.4" @@ -184,6 +185,7 @@ url = "2.2.2" impl_ops = "0.1.1" lazy_static = "1.5.0" mime = "0.3.17" +futures-lite = "2.6.1" # Build Dependencies vergen = "7.5.0" diff --git a/compose.yml b/compose.yml index a01c965e..bb4cb8fe 100644 --- a/compose.yml +++ b/compose.yml @@ -8,10 +8,20 @@ services: # MongoDB database: image: mongo + command: mongod --replSet rs0 ports: - "27017:27017" volumes: - ./.data/db:/data/db + extra_hosts: + - "host.docker.internal:host-gateway" + healthcheck: + test: echo "try { rs.status() } catch (err) { rs.initiate({_id:'rs0',members:[{_id:0,host:'work-laptop.van-acoustic.ts.net:27017'}]}) }" | mongosh --port 27017 --quiet + interval: 5s + timeout: 30s + start_period: 0s + start_interval: 1s + retries: 30 ulimits: nofile: soft: 65536 @@ -19,8 +29,8 @@ services: # MinIO minio: - image: minio/minio - command: server /data + image: firstfinger/minio:latest + #command: server /data environment: MINIO_ROOT_USER: minioautumn MINIO_ROOT_PASSWORD: minioautumn diff --git a/crates/core/config/Revolt.toml b/crates/core/config/Revolt.toml index 2440bb31..ad2d6992 100644 --- a/crates/core/config/Revolt.toml +++ b/crates/core/config/Revolt.toml @@ -30,6 +30,10 @@ host = "rabbit" port = 5672 username = "rabbituser" password = "rabbitpass" +default_exchange = "revolt" + +[rabbit.queues] +acks = "internal.ack" [api] diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 3dc2771a..1ba041ca 100644 --- a/crates/core/config/src/lib.rs +++ b/crates/core/config/src/lib.rs @@ -122,12 +122,19 @@ pub struct Database { pub redis_pubsub: Option, } +#[derive(Deserialize, Debug, Clone)] +pub struct RabbitQueues { + pub acks: String, +} + #[derive(Deserialize, Debug, Clone)] pub struct Rabbit { pub host: String, pub port: u16, pub username: String, pub password: String, + pub default_exchange: String, + pub queues: RabbitQueues, } #[derive(Deserialize, Debug, Clone)] diff --git a/crates/core/database/Cargo.toml b/crates/core/database/Cargo.toml index ba5b7d1b..be3119c5 100644 --- a/crates/core/database/Cargo.toml +++ b/crates/core/database/Cargo.toml @@ -38,6 +38,7 @@ revolt-models = { workspace = true, features = ["validator"] } revolt-presence = { workspace = true } revolt-permissions = { workspace = true, features = ["serde", "bson"] } revolt-parser = { workspace = true } +revolt-coalesced = { workspace = true } # Utility log = { workspace = true } diff --git a/crates/core/database/src/amqp/amqp.rs b/crates/core/database/src/amqp/amqp.rs index b29105ed..5358bc34 100644 --- a/crates/core/database/src/amqp/amqp.rs +++ b/crates/core/database/src/amqp/amqp.rs @@ -2,7 +2,10 @@ use std::collections::HashSet; use crate::events::rabbit::*; use crate::User; -use amqprs::channel::{BasicPublishArguments, ExchangeDeclareArguments}; +use amqprs::channel::{ + BasicPublishArguments, ExchangeDeclareArguments, ExchangeType, QueueBindArguments, + QueueDeclareArguments, +}; use amqprs::connection::OpenConnectionArguments; use amqprs::{channel::Channel, connection::Connection, error::Error as AMQPError}; use amqprs::{BasicProperties, FieldTable}; @@ -55,6 +58,43 @@ impl AMQP { AMQP::new(connection, channel) } + pub async fn configure_channels(&self) -> revolt_result::Result<()> { + let config = revolt_config::config().await; + + self.channel + .exchange_declare( + ExchangeDeclareArguments::new( + &config.rabbit.default_exchange, + &ExchangeType::Topic.to_string(), + ) + .durable(true) + .finish(), + ) + .await + .expect("Failed to declare exchange"); + + // Configure acks channel & routing + self.channel + .queue_declare( + QueueDeclareArguments::new(&config.rabbit.queues.acks) + .durable(true) + .no_wait(true) + .finish(), + ) + .await + .expect("Failed to bind queue"); + + self.channel + .queue_bind(QueueBindArguments::new( + &config.rabbit.queues.acks, + &config.rabbit.default_exchange, + &config.rabbit.queues.acks, + )) + .await + .expect("Failed to bind channel"); + Ok(()) + } + pub async fn friend_request_accepted( &self, accepted_request_user: &User, @@ -232,7 +272,9 @@ impl AMQP { .await } - pub async fn ack_message( + /// # Sends an ack to pushd to update badges on iPhones. + /// Not to be confused with the process_ack function, which handles sending all acks to crond for processing. + pub async fn ack_notification_message( &self, user_id: String, channel_id: String, @@ -316,4 +358,41 @@ impl AMQP { ) .await } + + /// # Send an ack to crond for processing + pub async fn process_ack( + &self, + user_id: &str, + channel_id: Option<&str>, + server_id: Option<&str>, + ) -> Result<(), AMQPError> { + let config = revolt_config::config().await; + + let payload = AckEventPayload { + user_id: user_id.to_string(), + channel_id: channel_id.map(|value| value.to_string()), + server_id: server_id.map(|value| value.to_string()), + }; + let payload = to_string(&payload).unwrap(); + + info!( + "Sending ack processor event on exchange {}, channel {}: {}", + config.rabbit.default_exchange, config.rabbit.queues.acks, payload + ); + + self.channel + .basic_publish( + BasicProperties::default() + .with_content_type("application/json") + .with_persistence(true) + //.with_headers(headers) + .finish(), + payload.into(), + BasicPublishArguments::new( + &config.rabbit.default_exchange, + &config.rabbit.queues.acks, + ), + ) + .await + } } diff --git a/crates/core/database/src/events/rabbit.rs b/crates/core/database/src/events/rabbit.rs index 6f5c9ab3..1673f4a4 100644 --- a/crates/core/database/src/events/rabbit.rs +++ b/crates/core/database/src/events/rabbit.rs @@ -78,3 +78,11 @@ pub struct AckPayload { pub channel_id: String, pub message_id: String, } + +/// This is not the same as the AckPayload above, as the state for this event is stored in redis to allow for state updates while the event is queued. +#[derive(Serialize, Deserialize, Debug)] +pub struct AckEventPayload { + pub user_id: String, + pub channel_id: Option, + pub server_id: Option, +} diff --git a/crates/core/database/src/models/channels/model.rs b/crates/core/database/src/models/channels/model.rs index 025b7934..74b56e11 100644 --- a/crates/core/database/src/models/channels/model.rs +++ b/crates/core/database/src/models/channels/model.rs @@ -1,6 +1,7 @@ #![allow(deprecated)] use std::{borrow::Cow, collections::HashMap}; +use redis_kiss::get_connection; use revolt_config::config; use revolt_models::v0::{self, MessageAuthor}; use revolt_permissions::OverrideField; @@ -212,7 +213,7 @@ impl Channel { role_permissions: HashMap::new(), nsfw: data.nsfw.unwrap_or(false), voice: data.voice.map(|voice| voice.into()), - slowmode: None + slowmode: None, }, v0::LegacyServerChannelType::Voice => Channel::TextChannel { id: id.clone(), @@ -225,7 +226,7 @@ impl Channel { role_permissions: HashMap::new(), nsfw: data.nsfw.unwrap_or(false), voice: Some(data.voice.unwrap_or_default().into()), - slowmode: None + slowmode: None, }, }; @@ -643,7 +644,7 @@ impl Channel { } /// Acknowledge a message - pub async fn ack(&self, user: &str, message: &str) -> Result<()> { + pub async fn ack(&self, user: &str, message: &str, amqp: &AMQP) -> Result<()> { EventV1::ChannelAck { id: self.id().to_string(), user: user.to_string(), @@ -652,17 +653,7 @@ impl Channel { .private(user.to_string()) .await; - #[cfg(feature = "tasks")] - crate::tasks::ack::queue_ack( - self.id().to_string(), - user.to_string(), - crate::tasks::ack::AckEvent::AckMessage { - id: message.to_string(), - }, - ) - .await; - - Ok(()) + crate::util::acker::ack_channel(user, self.id(), message, amqp).await } /// Remove user from a group diff --git a/crates/core/database/src/tasks/ack.rs b/crates/core/database/src/tasks/ack.rs index c26de03b..24e3805e 100644 --- a/crates/core/database/src/tasks/ack.rs +++ b/crates/core/database/src/tasks/ack.rs @@ -105,7 +105,11 @@ pub async fn handle_ack_event( if mentions_acked > 0 { if let Err(err) = amqp - .ack_message(user.to_string(), channel.to_string(), id.to_owned()) + .ack_notification_message( + user.to_string(), + channel.to_string(), + id.to_owned(), + ) .await { revolt_config::capture_error(&err); @@ -192,9 +196,7 @@ pub async fn handle_ack_event( .expect("Failed to fetch channel from db"); if let TextChannel { server, .. } = channel { - if let Err(err) = - amqp.mass_mention_message_sent(server, mass_mentions).await - { + if let Err(err) = amqp.mass_mention_message_sent(server, mass_mentions).await { revolt_config::capture_error(&err); } } else { diff --git a/crates/core/database/src/util/acker.rs b/crates/core/database/src/util/acker.rs new file mode 100644 index 00000000..8cc104c5 --- /dev/null +++ b/crates/core/database/src/util/acker.rs @@ -0,0 +1,77 @@ +use redis_kiss::{get_connection, AsyncCommands}; +use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; +use revolt_result::{Result, ToRevoltError}; + +use crate::{events::client::EventV1, Channel, Database, Server, User, AMQP}; + +pub async fn ack_channel(user: &str, channel: &str, message: &str, amqp: &AMQP) -> Result<()> { + let mut redis = get_connection() + .await + .map_err(|_| create_error!(InternalError))?; + + let old: Option = redis + .getset(format!("acker:{user}+{channel}"), message) + .await + .to_internal_error()?; + + if old.is_none() || old.unwrap() == message { + amqp.process_ack(user, Some(channel), None) + .await + .to_internal_error()?; + } + + Ok(()) +} + +pub async fn ack_server(user: &User, server: &Server, db: &Database, amqp: &AMQP) -> Result<()> { + let mut redis = get_connection() + .await + .map_err(|_| create_error!(InternalError))?; + + let channels = db.fetch_channels(&server.channels).await?; + let query = crate::util::permissions::DatabasePermissionQuery::new(db, user).server(server); + + for channel in channels { + let channel_id = channel.id(); + let mut q = query.clone().channel(&channel); + + if calculate_channel_permissions(&mut q) + .await + .has_channel_permission(ChannelPermission::ViewChannel) + { + let channel_last_msg = match &channel { + Channel::TextChannel { + last_message_id, .. + } => last_message_id, + _ => unreachable!(), + } + .clone(); + + if let Some(channel_last_msg) = channel_last_msg { + let old: Option = redis + .getset( + format!("acker:{}+{}", user.id, channel_id), + &channel_last_msg, + ) + .await + .to_internal_error()?; + + if old.is_none() || old.unwrap() == channel_last_msg { + amqp.process_ack(&user.id, Some(channel_id), Some(&server.id)) + .await + .to_internal_error()?; + + EventV1::ChannelAck { + id: channel_id.to_string(), + user: user.id.clone(), + message_id: channel_last_msg, + } + .private(user.id.clone()) + .await; + } + } + } + } + + Ok(()) +} diff --git a/crates/core/database/src/util/mod.rs b/crates/core/database/src/util/mod.rs index 1a03a176..ae2817ac 100644 --- a/crates/core/database/src/util/mod.rs +++ b/crates/core/database/src/util/mod.rs @@ -1,3 +1,4 @@ +pub mod acker; pub mod bridge; pub mod bulk_permissions; mod funcs; diff --git a/crates/daemons/crond/Cargo.toml b/crates/daemons/crond/Cargo.toml index fe33d2b7..6bd90052 100644 --- a/crates/daemons/crond/Cargo.toml +++ b/crates/daemons/crond/Cargo.toml @@ -16,8 +16,22 @@ log = { workspace = true } # Async tokio = { workspace = true } +# Redis +redis-kiss = { workspace = true } + +# RabbitMQ +lapin = { workspace = true } +futures-lite = { workspace = true } + +# Processing +serde_json = { workspace = true } +revolt_optional_struct = { workspace = true } +serde = { workspace = true } +iso8601-timestamp = { workspace = true, features = ["serde", "bson"] } + # Core revolt-database = { workspace = true } revolt-result = { workspace = true } revolt-config = { workspace = true } revolt-files = { workspace = true } +revolt-permissions = { workspace = true } diff --git a/crates/daemons/crond/src/main.rs b/crates/daemons/crond/src/main.rs index b232a10a..c45780d5 100644 --- a/crates/daemons/crond/src/main.rs +++ b/crates/daemons/crond/src/main.rs @@ -1,7 +1,7 @@ use revolt_config::configure; use revolt_database::DatabaseInfo; use revolt_result::Result; -use tasks::{file_deletion, prune_dangling_files, prune_members}; +use tasks::{acks, file_deletion, prune_dangling_files, prune_members}; use tokio::try_join; pub mod tasks; @@ -14,7 +14,8 @@ async fn main() -> Result<()> { try_join!( file_deletion::task(db.clone()), prune_dangling_files::task(db.clone()), - prune_members::task(db.clone()) + prune_members::task(db.clone()), + acks::task(db.clone()) ) .map(|_| ()) } diff --git a/crates/daemons/crond/src/tasks/acks.rs b/crates/daemons/crond/src/tasks/acks.rs new file mode 100644 index 00000000..68b13144 --- /dev/null +++ b/crates/daemons/crond/src/tasks/acks.rs @@ -0,0 +1,129 @@ +use futures_lite::stream::StreamExt; +use lapin::{ + options::*, + types::FieldTable, + uri::{AMQPAuthority, AMQPQueryString, AMQPUri, AMQPUserInfo}, + ConnectionBuilder, ConnectionProperties, +}; +use log::info; +use redis_kiss::{get_connection, AsyncCommands, Conn as RedisConnection}; +use revolt_config::config; +use revolt_database::{events::rabbit::AckEventPayload, Database}; +use revolt_result::{Result, ToRevoltError}; +use serde_json; + +pub async fn task(db: Database) -> Result<()> { + let config = config().await; + + let mut redis = get_connection() + .await + .expect("Failed to get redis connection"); + + let uri = AMQPUri { + scheme: lapin::uri::AMQPScheme::AMQP, + authority: AMQPAuthority { + userinfo: AMQPUserInfo { + username: config.rabbit.username, + password: config.rabbit.password, + }, + host: config.rabbit.host, + port: config.rabbit.port, + }, + vhost: "/".to_string(), + query: AMQPQueryString::default(), + }; + + let connection = ConnectionBuilder::new() + .expect("Builder") + .with_uri(uri) + .with_properties(ConnectionProperties::default()) + .connect() + .await + .expect("Failed to connect to rabbitmq"); + + let reader_channel = connection + .create_channel() + .await + .expect("Failed to create channel"); + + let mut consumer = reader_channel + .basic_consume( + config.rabbit.queues.acks.into(), + "crond-ack-consumer".into(), + BasicConsumeOptions::default(), + FieldTable::default(), + ) + .await + .expect("Failed to create consumer"); + + while let Some(delivery) = consumer.next().await { + if let Ok(delivery) = delivery { + let payload: std::result::Result = + serde_json::from_slice(&delivery.data); + if let Ok(payload) = payload { + info!("{:?}", payload); + if let Err(e) = process_channel_ack( + &db, + payload.user_id, + payload.channel_id.unwrap(), + &mut redis, + ) + .await + { + revolt_config::capture_error(&e); + _ = delivery.reject(BasicRejectOptions { requeue: false }).await; + } else { + _ = delivery.ack(BasicAckOptions { multiple: false }).await; + } + } else { + revolt_config::capture_message( + format!("Failed to decode ack data: {:?}", delivery.data).as_str(), + revolt_config::Level::Error, + ); + } + } + } + Ok(()) +} + +#[allow(clippy::disallowed_methods)] +async fn process_channel_ack( + db: &Database, + user: String, + channel: String, + redis: &mut RedisConnection, +) -> Result<()> { + let message_id: Option = redis + .get_del(format!("acker:{user}+{channel}")) + .await + .to_internal_error()?; + + if let Some(message_id) = message_id { + // This will be uncommented eventually, but we need to sort out the transition to lapin first. For now we'll simply disable the badge update logic. + // We also drop a db request as a bonus. + + //let unread = db.fetch_unread(&user, &channel).await?; + let _updated = db.acknowledge_message(&channel, &user, &message_id).await?; + info!("Set new state for ack: {}:{}:{}", channel, user, message_id); + + // if let (Some(before), Some(after)) = (unread, updated) { + // let before_mentions = before.mentions.unwrap_or_default().len(); + // let after_mentions = after.mentions.unwrap_or_default().len(); + + // let mentions_acked = before_mentions - after_mentions; + + // if mentions_acked > 0 { + // if let Err(err) = amqp + // .ack_message(user.to_string(), channel.to_string(), payload.message_id) + // .await + // { + // revolt_config::capture_error(&err); + // } + // }; + // } + + Ok(()) + } else { + Err(message_id.to_internal_error().expect_err("no err")) + } +} diff --git a/crates/daemons/crond/src/tasks/mod.rs b/crates/daemons/crond/src/tasks/mod.rs index 060f55d8..a7dd9040 100644 --- a/crates/daemons/crond/src/tasks/mod.rs +++ b/crates/daemons/crond/src/tasks/mod.rs @@ -1,3 +1,4 @@ +pub mod acks; pub mod file_deletion; pub mod prune_dangling_files; pub mod prune_members; diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index 882609cc..e3522de1 100644 --- a/crates/delta/src/main.rs +++ b/crates/delta/src/main.rs @@ -24,8 +24,8 @@ use amqprs::{ }; use async_std::channel::unbounded; use authifier::AuthifierEvent; -use rocket::data::ToByteUnit; use revolt_database::voice::VoiceClient; +use rocket::data::ToByteUnit; pub async fn web() -> Rocket { // Get settings @@ -93,22 +93,6 @@ pub async fn web() -> Rocket { ) .into(); - let swagger_0_8 = revolt_rocket_okapi::swagger_ui::make_swagger_ui( - &revolt_rocket_okapi::swagger_ui::SwaggerUIConfig { - url: "/0.8/openapi.json".to_owned(), - ..Default::default() - }, - ) - .into(); - - let swagger_0_8 = revolt_rocket_okapi::swagger_ui::make_swagger_ui( - &revolt_rocket_okapi::swagger_ui::SwaggerUIConfig { - url: "/0.8/openapi.json".to_owned(), - ..Default::default() - }, - ) - .into(); - // Voice handler let voice_client = VoiceClient::new(config.api.livekit.nodes.clone()); // Configure Rabbit @@ -136,6 +120,9 @@ pub async fn web() -> Rocket { .expect("Failed to declare exchange"); let amqp = AMQP::new(connection, channel); + amqp.configure_channels() + .await + .expect("Failed to configure channels"); // Launch background task workers revolt_database::tasks::start_workers(db.clone(), amqp.clone()); @@ -153,7 +140,6 @@ pub async fn web() -> Rocket { .mount("/", rocket_cors::catch_all_options_routes()) .mount("/", ratelimiter::routes()) .mount("/swagger/", swagger) - .mount("/0.8/swagger/", swagger_0_8) .manage(authifier) .manage(db) .manage(amqp) diff --git a/crates/delta/src/routes/channels/channel_ack.rs b/crates/delta/src/routes/channels/channel_ack.rs index 2ae0d3b0..e20da642 100644 --- a/crates/delta/src/routes/channels/channel_ack.rs +++ b/crates/delta/src/routes/channels/channel_ack.rs @@ -1,6 +1,6 @@ use revolt_database::{ util::{permissions::DatabasePermissionQuery, reference::Reference}, - Database, User, + Database, User, AMQP, }; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; @@ -14,6 +14,7 @@ use rocket_empty::EmptyResponse; #[put("//ack/")] pub async fn ack( db: &State, + amqp: &State, user: User, target: Reference<'_>, message: Reference<'_>, @@ -29,7 +30,7 @@ pub async fn ack( .throw_if_lacking_channel_permission(ChannelPermission::ViewChannel)?; channel - .ack(&user.id, message.id) + .ack(&user.id, message.id, amqp) .await .map(|_| EmptyResponse) } diff --git a/crates/delta/src/routes/servers/server_ack.rs b/crates/delta/src/routes/servers/server_ack.rs index 789617e1..81ae683e 100644 --- a/crates/delta/src/routes/servers/server_ack.rs +++ b/crates/delta/src/routes/servers/server_ack.rs @@ -1,6 +1,6 @@ use revolt_database::{ - util::{permissions::DatabasePermissionQuery, reference::Reference}, - Database, User, + util::{acker, permissions::DatabasePermissionQuery, reference::Reference}, + Database, User, AMQP, }; use revolt_permissions::PermissionQuery; use revolt_result::{create_error, Result}; @@ -12,7 +12,12 @@ use rocket_empty::EmptyResponse; /// Mark all channels in a server as read. #[openapi(tag = "Server Information")] #[put("//ack")] -pub async fn ack(db: &State, user: User, target: Reference<'_>) -> Result { +pub async fn ack( + db: &State, + amqp: &State, + user: User, + target: Reference<'_>, +) -> Result { if user.bot.is_some() { return Err(create_error!(IsBot)); } @@ -23,7 +28,6 @@ pub async fn ack(db: &State, user: User, target: Reference<'_>) -> Res return Err(create_error!(NotFound)); } - db.acknowledge_channels(&user.id, &server.channels) - .await - .map(|_| EmptyResponse) + acker::ack_server(&user, &server, db, amqp).await?; + Ok(EmptyResponse) }