From 5372296dc003e58babc99d756e34d1d4413d3bdf Mon Sep 17 00:00:00 2001 From: Paul Makles Date: Sun, 27 Aug 2023 11:17:03 +0100 Subject: [PATCH] feat(core): implement message tasks includes: - acknowledgements - last_message_id - embeds - web push --- Cargo.lock | 826 +++++++++++++++++- crates/core/database/Cargo.toml | 14 +- crates/core/database/src/drivers/mod.rs | 11 + crates/core/database/src/lib.rs | 1 + .../database/src/models/messages/model.rs | 120 ++- .../src/models/server_members/model.rs | 2 +- crates/core/database/src/tasks/ack.rs | 122 +++ .../database/src/tasks/last_message_id.rs | 87 ++ crates/core/database/src/tasks/mod.rs | 58 ++ .../core/database/src/tasks/process_embeds.rs | 170 ++++ crates/core/database/src/tasks/web_push.rs | 122 +++ crates/core/database/src/util/bridge/v0.rs | 10 + crates/core/models/Cargo.toml | 1 + crates/core/models/src/v0/messages.rs | 124 ++- crates/delta/src/main.rs | 3 +- 15 files changed, 1608 insertions(+), 63 deletions(-) create mode 100644 crates/core/database/src/tasks/ack.rs create mode 100644 crates/core/database/src/tasks/last_message_id.rs create mode 100644 crates/core/database/src/tasks/mod.rs create mode 100644 crates/core/database/src/tasks/process_embeds.rs create mode 100644 crates/core/database/src/tasks/web_push.rs diff --git a/Cargo.lock b/Cargo.lock index 506c62d4..788d0cec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -189,12 +189,11 @@ dependencies = [ [[package]] name = "async-lock" -version = "2.6.0" +version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8101efe8695a6c17e02911402145357e718ac92d3ff88ae8419e84b1707b685" +checksum = "287272293e9d8c41773cec55e365490fe034813a2f172f502d6ddcf75b2f582b" dependencies = [ "event-listener", - "futures-lite", ] [[package]] @@ -416,6 +415,12 @@ dependencies = [ "rustc-demangle", ] +[[package]] +name = "base16ct" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "349a06037c7bf932dd7e7d1f653678b2038b9ad46a74102f1fc7bd7872678cce" + [[package]] name = "base32" version = "0.4.0" @@ -440,9 +445,15 @@ checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd" [[package]] name = "base64" -version = "0.21.2" +version = "0.21.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "604178f6c5c21f02dc555784810edfb88d34ac2c73b2eae109655649ee73ce3d" +checksum = "414dcefbc63d77c526a76b3afcf6fbb9b5e2791c19c3aa2297733208750c6e53" + +[[package]] +name = "base64ct" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b" [[package]] name = "binascii" @@ -459,6 +470,12 @@ dependencies = [ "serde", ] +[[package]] +name = "binstring" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e0d60973d9320722cb1206f412740e162a33b8547ea8d6be75d7cff237c7a85" + [[package]] name = "bitfield" version = "0.13.2" @@ -581,6 +598,48 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1db59621ec70f09c5e9b597b220c7a2b43611f4710dc03ceb8748637775692c" +[[package]] +name = "cached" +version = "0.44.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b195e4fbc4b6862bbd065b991a34750399c119797efff72492f28a5864de8700" +dependencies = [ + "async-trait", + "cached_proc_macro", + "cached_proc_macro_types", + "futures", + "hashbrown 0.13.2", + "instant", + "once_cell", + "thiserror", + "tokio 1.18.2", +] + +[[package]] +name = "cached_proc_macro" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b48814962d2fd604c50d2b9433c2a41a0ab567779ee2c02f7fba6eca1221f082" +dependencies = [ + "cached_proc_macro_types", + "darling 0.14.4", + "proc-macro2", + "quote 1.0.26", + "syn 1.0.107", +] + +[[package]] +name = "cached_proc_macro_types" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a4f925191b4367301851c6d99b09890311d74b0d43f274c0b34c86d308a3663" + +[[package]] +name = "castaway" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2698f953def977c68f935bb0dfa959375ad4638570e969e2f1e9f433cbf1af6" + [[package]] name = "cc" version = "1.0.73" @@ -635,6 +694,18 @@ dependencies = [ "bitflags", ] +[[package]] +name = "coarsetime" +version = "0.1.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a90d114103adbc625300f346d4d09dfb4ab1c4a8df6868435dd903392ecf4354" +dependencies = [ + "libc", + "once_cell", + "wasi 0.11.0+wasi-snapshot-preview1", + "wasm-bindgen", +] + [[package]] name = "combine" version = "4.6.4" @@ -658,6 +729,37 @@ dependencies = [ "cache-padded", ] +[[package]] +name = "config" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d379af7f68bfc21714c6c7dea883544201741d2ce8274bb12fa54f89507f52a7" +dependencies = [ + "async-trait", + "json5", + "lazy_static", + "nom", + "pathdiff", + "ron", + "rust-ini", + "serde", + "serde_json", + "toml", + "yaml-rust", +] + +[[package]] +name = "const-oid" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d6f2aa4d0537bcc1c74df8755072bd31c1ef1a3a1b85a68e8404a8c353b7b8b" + +[[package]] +name = "const-oid" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28c122c3980598d243d63d9a704629a2d748d101f278052ff068be5a4423ab6f" + [[package]] name = "constant_time_eq" version = "0.1.5" @@ -727,6 +829,18 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "crypto-bigint" +version = "0.4.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef2b4b23cddf68b89b8f8069890e8c270d54e2d5fe1b143820234805e4cb17ef" +dependencies = [ + "generic-array 0.14.5", + "rand_core 0.6.3", + "subtle", + "zeroize", +] + [[package]] name = "crypto-common" version = "0.1.3" @@ -737,6 +851,12 @@ dependencies = [ "typenum", ] +[[package]] +name = "ct-codecs" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3b7eb4404b8195a9abb6356f4ac07d8ba267045c8d6d220ac4dc992e6cc75df" + [[package]] name = "ctor" version = "0.1.22" @@ -756,14 +876,55 @@ dependencies = [ "cipher", ] +[[package]] +name = "curl" +version = "0.4.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "509bd11746c7ac09ebd19f0b17782eae80aadee26237658a6b4808afb5c11a22" +dependencies = [ + "curl-sys", + "libc", + "openssl-probe", + "openssl-sys", + "schannel", + "socket2 0.4.4", + "winapi 0.3.9", +] + +[[package]] +name = "curl-sys" +version = "0.4.65+curl-8.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "961ba061c9ef2fe34bbd12b807152d96f0badd2bebe7b90ce6c8c8b7572a0986" +dependencies = [ + "cc", + "libc", + "libnghttp2-sys", + "libz-sys", + "openssl-sys", + "pkg-config", + "vcpkg", + "winapi 0.3.9", +] + [[package]] name = "darling" version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a01d95850c592940db9b8194bc39f4bc0e89dee5c4265e4b1807c34a9aba453c" dependencies = [ - "darling_core", - "darling_macro", + "darling_core 0.13.4", + "darling_macro 0.13.4", +] + +[[package]] +name = "darling" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b750cb3417fd1b327431a470f388520309479ab0bf5e323505daf0290cd3850" +dependencies = [ + "darling_core 0.14.4", + "darling_macro 0.14.4", ] [[package]] @@ -780,13 +941,38 @@ dependencies = [ "syn 1.0.107", ] +[[package]] +name = "darling_core" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "109c1ca6e6b7f82cc233a97004ea8ed7ca123a9af07a8230878fcfda9b158bf0" +dependencies = [ + "fnv", + "ident_case", + "proc-macro2", + "quote 1.0.26", + "strsim", + "syn 1.0.107", +] + [[package]] name = "darling_macro" version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c972679f83bdf9c42bd905396b6c3588a843a17f0f16dfcfa3e2c5d57441835" dependencies = [ - "darling_core", + "darling_core 0.13.4", + "quote 1.0.26", + "syn 1.0.107", +] + +[[package]] +name = "darling_macro" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4aab4dbc9f7611d8b55048a3a16d2d010c2c8334e46304b40ac1cc14bf3b48e" +dependencies = [ + "darling_core 0.14.4", "quote 1.0.26", "syn 1.0.107", ] @@ -811,9 +997,9 @@ checksum = "3ee2393c4a91429dffb4bedf19f4d6abf27d8a732c8ce4980305d782e5426d57" [[package]] name = "deadqueue" -version = "0.2.2" +version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "455d501abc72bae6432dbe500059887cab4700ba83303045db9ce2e824177d1c" +checksum = "16a2561fd313df162315935989dceb8c99db4ee1933358270a57a3cfb8c957f3" dependencies = [ "crossbeam-queue", "tokio 1.18.2", @@ -835,6 +1021,39 @@ version = "1.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "808127a7de612079ec37bfc1abc48ed77a6015a971a8bd7d4178d79147cbc839" +[[package]] +name = "der" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79b71cca7d95d7681a4b3b9cdf63c8dbc3730d0584c2c74e31416d64a90493f4" +dependencies = [ + "const-oid 0.6.2", + "der_derive", +] + +[[package]] +name = "der" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1a467a65c5e759bce6e65eaf91cc29f466cdc57cb65777bd646872a8a1fd4de" +dependencies = [ + "const-oid 0.9.5", + "pem-rfc7468", + "zeroize", +] + +[[package]] +name = "der_derive" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8aed3b3c608dc56cf36c45fe979d04eda51242e6703d8d0bb03426ef7c41db6a" +dependencies = [ + "proc-macro2", + "quote 1.0.26", + "syn 1.0.107", + "synstructure", +] + [[package]] name = "derivative" version = "2.2.0" @@ -890,15 +1109,22 @@ dependencies = [ [[package]] name = "digest" -version = "0.10.3" +version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2fb860ca6fafa5552fb6d0e816a69c8e49f0908bf524e30a90d97c85892d506" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.2", + "const-oid 0.9.5", "crypto-common", "subtle", ] +[[package]] +name = "dlv-list" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0688c2a7f92e427f44895cd63841bff7b29f8d7a1648b9e7e07a4a365b2e1257" + [[package]] name = "dotenv" version = "0.15.0" @@ -911,12 +1137,74 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "21e50f3adc76d6a43f5ed73b698a87d0760ca74617f60f7c3b879003536fdd28" +[[package]] +name = "ecdsa" +version = "0.14.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413301934810f597c1d19ca71c8710e99a3f1ba28a0d2ebc01551a2daeea3c5c" +dependencies = [ + "der 0.6.1", + "elliptic-curve", + "rfc6979", + "signature", +] + +[[package]] +name = "ece" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8dd5463ffecc0677adcd786c4481f73b215714d4757edf2eb37a573c03d00459" +dependencies = [ + "base64 0.13.0", + "byteorder", + "hex", + "hkdf", + "lazy_static", + "once_cell", + "openssl", + "serde", + "sha2", + "thiserror", +] + +[[package]] +name = "ed25519-compact" +version = "2.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a3d382e8464107391c8706b4c14b087808ecb909f6c15c34114bc42e53a9e4c" +dependencies = [ + "ct-codecs", + "getrandom", +] + [[package]] name = "either" version = "1.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457" +[[package]] +name = "elliptic-curve" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7bb888ab5300a19b8e5bceef25ac745ad065f3c9f7efc6de1b91958110891d3" +dependencies = [ + "base16ct", + "crypto-bigint", + "der 0.6.1", + "digest 0.10.7", + "ff", + "generic-array 0.14.5", + "group", + "hkdf", + "pem-rfc7468", + "pkcs8", + "rand_core 0.6.3", + "sec1", + "subtle", + "zeroize", +] + [[package]] name = "email-encoding" version = "0.1.0" @@ -1010,6 +1298,30 @@ dependencies = [ "instant", ] +[[package]] +name = "fcm" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d8d0da8a6bd63bdec888b6d7a87c5698230005c1800823d28ddd5adb6f2550f" +dependencies = [ + "chrono", + "erased-serde", + "log", + "reqwest", + "serde", + "serde_json", +] + +[[package]] +name = "ff" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d013fc25338cc558c5c2cfbad646908fb23591e2404481826742b651c9af7160" +dependencies = [ + "rand_core 0.6.3", + "subtle", +] + [[package]] name = "figment" version = "0.10.6" @@ -1152,6 +1464,17 @@ dependencies = [ "waker-fn", ] +[[package]] +name = "futures-locks" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "45ec6fe3675af967e67c5536c0b9d44e34e6c52f86bedc4ea49c5317b8e94d06" +dependencies = [ + "futures-channel", + "futures-task", + "tokio 1.18.2", +] + [[package]] name = "futures-macro" version = "0.3.21" @@ -1301,6 +1624,17 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "group" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5dfbfb3a6cfbd390d5c9564ab283a0349b9b9fcd46a706c1eb10e0db70bfbac7" +dependencies = [ + "ff", + "rand_core 0.6.3", + "subtle", +] + [[package]] name = "h2" version = "0.2.7" @@ -1368,6 +1702,15 @@ name = "hashbrown" version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db0d4cf898abf0081f964436dc980e96670a0f36863e4b83aaacdb65c9d7ccc3" +dependencies = [ + "ahash 0.7.6", +] + +[[package]] +name = "hashbrown" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43a3c133739dddd0d2990f9a4bdf8eb4b21ef50e4851ca85ab661199821d510e" [[package]] name = "hashbrown" @@ -1415,7 +1758,31 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" dependencies = [ - "digest 0.10.3", + "digest 0.10.7", +] + +[[package]] +name = "hmac-sha1-compact" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dff9d405ec732fa3fcde87264e54a32a84956a377b3e3107de96e59b798c84a7" + +[[package]] +name = "hmac-sha256" +version = "1.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3688e69b38018fec1557254f64c8dc2cc8ec502890182f395dbb0aa997aa5735" +dependencies = [ + "digest 0.10.7", +] + +[[package]] +name = "hmac-sha512" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e4ce1f4656bae589a3fab938f9f09bf58645b7ed01a2c5f8a3c238e01a4ef78a" +dependencies = [ + "digest 0.10.7", ] [[package]] @@ -1644,6 +2011,35 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879d54834c8c76457ef4293a689b2a8c59b076067ad77b15efafbb05f92a592b" +[[package]] +name = "isahc" +version = "1.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "334e04b4d781f436dc315cb1e7515bd96826426345d498149e4bde36b67f8ee9" +dependencies = [ + "async-channel", + "castaway", + "crossbeam-utils", + "curl", + "curl-sys", + "encoding_rs", + "event-listener", + "futures-lite", + "http", + "log", + "mime", + "once_cell", + "polling", + "serde", + "serde_json", + "slab", + "sluice", + "tracing", + "tracing-futures", + "url", + "waker-fn", +] + [[package]] name = "iso8601-timestamp" version = "0.1.10" @@ -1698,6 +2094,55 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "json5" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96b0db21af676c1ce64250b5f40f3ce2cf27e4e47cb91ed91eb6fe9350b430c1" +dependencies = [ + "pest", + "pest_derive", + "serde", +] + +[[package]] +name = "jwt-simple" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "529a00f2d42d7dc349c994e65917c81bf53225831a65361f6c0454124c550f63" +dependencies = [ + "anyhow", + "binstring", + "coarsetime", + "ct-codecs", + "ed25519-compact", + "hmac-sha1-compact", + "hmac-sha256", + "hmac-sha512", + "k256", + "p256", + "p384", + "rand 0.8.5", + "rsa", + "serde", + "serde_json", + "spki", + "thiserror", + "zeroize", +] + +[[package]] +name = "k256" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72c1e0b51e7ec0a97369623508396067a486bd0cbed95a2659a4b863d28cfc8b" +dependencies = [ + "cfg-if 1.0.0", + "ecdsa", + "elliptic-curve", + "sha2", +] + [[package]] name = "kernel32-sys" version = "0.2.2" @@ -1722,6 +2167,9 @@ name = "lazy_static" version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" +dependencies = [ + "spin 0.5.2", +] [[package]] name = "lettre" @@ -1762,6 +2210,22 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "libm" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7012b1bbb0719e1097c47611d3898568c546d597c2e74d66f6087edd5233ff4" + +[[package]] +name = "libnghttp2-sys" +version = "0.1.8+1.55.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fae956c192dadcdb5dace96db71fa0b827333cce7c7b38dc71446f024d8a340" +dependencies = [ + "cc", + "libc", +] + [[package]] name = "libz-sys" version = "1.1.8" @@ -1893,7 +2357,7 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "658646b21e0b72f7866c7038ab086d3d5e1cd6271f060fd37defb241949d0582" dependencies = [ - "digest 0.10.3", + "digest 0.10.7", ] [[package]] @@ -2141,6 +2605,23 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "num-bigint-dig" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc84195820f291c7697304f3cbdadd1cb7199c0efc917ff5eafd71225c136151" +dependencies = [ + "byteorder", + "lazy_static", + "libm", + "num-integer", + "num-iter", + "num-traits", + "rand 0.8.5", + "smallvec", + "zeroize", +] + [[package]] name = "num-integer" version = "0.1.45" @@ -2151,6 +2632,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-iter" +version = "0.1.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d03e6c028c5dc5cac6e2dec0efda81fc887605bb3d884578bb6d6bf7514e252" +dependencies = [ + "autocfg 1.1.0", + "num-integer", + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.15" @@ -2158,6 +2650,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "578ede34cf02f8924ab9447f50c28075b4d3e5b269972345e7e0372b38c6cdcd" dependencies = [ "autocfg 1.1.0", + "libm", ] [[package]] @@ -2223,9 +2716,9 @@ dependencies = [ [[package]] name = "once_cell" -version = "1.17.1" +version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3" +checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d" [[package]] name = "opaque-debug" @@ -2284,6 +2777,16 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "ordered-multimap" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccd746e37177e1711c20dd619a1620f34f5c8b569c53590a72dedd5344d8924a" +dependencies = [ + "dlv-list", + "hashbrown 0.12.1", +] + [[package]] name = "os_info" version = "3.7.0" @@ -2295,6 +2798,28 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "p256" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51f44edd08f51e2ade572f141051021c5af22677e42b7dd28a88155151c33594" +dependencies = [ + "ecdsa", + "elliptic-curve", + "sha2", +] + +[[package]] +name = "p384" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfc8c5bf642dde52bb9e87c0ecd8ca5a76faac2eeed98dedb7c717997e1080aa" +dependencies = [ + "ecdsa", + "elliptic-curve", + "sha2", +] + [[package]] name = "parking" version = "2.0.0" @@ -2330,13 +2855,19 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c520e05135d6e763148b6426a837e239041653ba7becd2e538c076c738025fc" +[[package]] +name = "pathdiff" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8835116a5c179084a830efb3adc117ab007512b535bc1a21c991d3b32a6b44dd" + [[package]] name = "pbkdf2" version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271779f35b581956db91a3e55737327a03aa051e90b1c47aeb189508533adfd7" dependencies = [ - "digest 0.10.3", + "digest 0.10.7", ] [[package]] @@ -2362,6 +2893,35 @@ dependencies = [ "syn 1.0.107", ] +[[package]] +name = "pem" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd56cbd21fea48d0c440b41cd69c589faacade08c992d9a54e471b79d0fd13eb" +dependencies = [ + "base64 0.13.0", + "once_cell", + "regex", +] + +[[package]] +name = "pem" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8835c273a76a90455d7344889b0964598e3316e2a79ede8e36f16bdcf2228b8" +dependencies = [ + "base64 0.13.0", +] + +[[package]] +name = "pem-rfc7468" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d159833a9105500e0398934e205e0773f0b27529557134ecfc51c27646adac" +dependencies = [ + "base64ct", +] + [[package]] name = "percent-encoding" version = "2.1.0" @@ -2449,6 +3009,28 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pkcs1" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eff33bdbdfc54cc98a2eca766ebdec3e1b8fb7387523d5c9c9a2891da856f719" +dependencies = [ + "der 0.6.1", + "pkcs8", + "spki", + "zeroize", +] + +[[package]] +name = "pkcs8" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9eca2c590a5f85da82668fa685c09ce2888b9430e83299debf1f34b65fd4a4ba" +dependencies = [ + "der 0.6.1", + "spki", +] + [[package]] name = "pkg-config" version = "0.3.25" @@ -2928,19 +3510,37 @@ dependencies = [ "serde_json", ] +[[package]] +name = "revolt-config" +version = "0.6.7" +dependencies = [ + "async-std", + "cached", + "config", + "futures-locks", + "once_cell", + "serde", +] + [[package]] name = "revolt-database" version = "0.6.7" dependencies = [ + "async-lock", "async-recursion", "async-std", "async-trait", "authifier", + "base64 0.21.3", "bson", + "deadqueue", "decancer", + "fcm", "futures", "indexmap", + "isahc", "iso8601-timestamp 0.2.11", + "linkify 0.8.1", "log", "lru 0.11.0", "mongodb", @@ -2949,6 +3549,7 @@ dependencies = [ "rand 0.8.5", "redis-kiss", "regex", + "revolt-config", "revolt-models", "revolt-permissions", "revolt-presence", @@ -2962,6 +3563,8 @@ dependencies = [ "serde_json", "ulid 1.0.0", "unicode-segmentation", + "url-escape", + "web-push 0.10.0", ] [[package]] @@ -3014,6 +3617,7 @@ dependencies = [ "iso8601-timestamp 0.2.11", "once_cell", "regex", + "revolt-config", "revolt-permissions", "revolt_optional_struct", "schemars", @@ -3097,7 +3701,7 @@ dependencies = [ "serde_json", "ulid 0.5.0", "validator 0.16.0", - "web-push", + "web-push 0.7.2", ] [[package]] @@ -3156,13 +3760,24 @@ version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cc6620569d8ac8f0a1690fcca13f488503807a60e96ebf729749b59aca1dbef9" dependencies = [ - "darling", + "darling 0.13.4", "proc-macro2", "quote 1.0.26", "rocket_http", "syn 1.0.107", ] +[[package]] +name = "rfc6979" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7743f17af12fa0b03b803ba12cd6a8d9483a587e89c69445e3909655c0b9fabb" +dependencies = [ + "crypto-bigint", + "hmac", + "zeroize", +] + [[package]] name = "ring" version = "0.16.20" @@ -3335,6 +3950,38 @@ dependencies = [ "rocket", ] +[[package]] +name = "ron" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88073939a61e5b7680558e6be56b419e208420c2adb92be54921fa6b72283f1a" +dependencies = [ + "base64 0.13.0", + "bitflags", + "serde", +] + +[[package]] +name = "rsa" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "094052d5470cbcef561cb848a7209968c9f12dfa6d668f4bca048ac5de51099c" +dependencies = [ + "byteorder", + "digest 0.10.7", + "num-bigint-dig", + "num-integer", + "num-iter", + "num-traits", + "pkcs1", + "pkcs8", + "rand_core 0.6.3", + "signature", + "smallvec", + "subtle", + "zeroize", +] + [[package]] name = "rust-argon2" version = "1.0.0" @@ -3347,6 +3994,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rust-ini" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6d5f2436026b4f6e79dc829837d467cc7e9a55ee40e750d716713540715a2df" +dependencies = [ + "cfg-if 1.0.0", + "ordered-multimap", +] + [[package]] name = "rustc-demangle" version = "0.1.21" @@ -3477,6 +4134,31 @@ dependencies = [ "untrusted 0.7.1", ] +[[package]] +name = "sec1" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3be24c1842290c45df0a7bf069e0c268a747ad05a192f2fd7dcfdbc1cba40928" +dependencies = [ + "base16ct", + "der 0.6.1", + "generic-array 0.14.5", + "pkcs8", + "subtle", + "zeroize", +] + +[[package]] +name = "sec1_decode" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6326ddc956378a0739200b2c30892dccaf198992dfd7323274690b9e188af23" +dependencies = [ + "der 0.4.5", + "pem 0.8.3", + "thiserror", +] + [[package]] name = "security-framework" version = "2.6.1" @@ -3709,7 +4391,7 @@ version = "1.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e182d6ec6f05393cc0e5ed1bf81ad6db3a8feedf8ee515ecdd369809bcce8082" dependencies = [ - "darling", + "darling 0.13.4", "proc-macro2", "quote 1.0.26", "syn 1.0.107", @@ -3735,7 +4417,7 @@ checksum = "028f48d513f9678cda28f6e4064755b3fbb2af6acd672f2c209b62323f7aea0f" dependencies = [ "cfg-if 1.0.0", "cpufeatures", - "digest 0.10.3", + "digest 0.10.7", ] [[package]] @@ -3752,7 +4434,7 @@ checksum = "55deaec60f81eefe3cce0dc50bda92d6d8e88f2a27df7c5033b42afeb1ed2676" dependencies = [ "cfg-if 1.0.0", "cpufeatures", - "digest 0.10.3", + "digest 0.10.7", ] [[package]] @@ -3783,6 +4465,16 @@ dependencies = [ "libc", ] +[[package]] +name = "signature" +version = "1.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74233d3b3b2f6d4b006dc19dee745e73e2a6bfb6f93607cd3b02bd5b00797d7c" +dependencies = [ + "digest 0.10.7", + "rand_core 0.6.3", +] + [[package]] name = "slab" version = "0.4.6" @@ -3790,10 +4482,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eb703cfe953bccee95685111adeedb76fabe4e97549a58d16f03ea7b9367bb32" [[package]] -name = "smallvec" -version = "1.8.0" +name = "sluice" +version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2dd574626839106c320a323308629dcb1acfc96e32a8cba364ddc61ac23ee83" +checksum = "6d7400c0eff44aa2fcb5e31a5f24ba9716ed90138769e4977a2ba6014ae63eb5" +dependencies = [ + "async-channel", + "futures-core", + "futures-io", +] + +[[package]] +name = "smallvec" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62bb4feee49fdd9f707ef802e22365a35de4b7b299de4763d44bfea899442ff9" [[package]] name = "socket2" @@ -3828,6 +4531,16 @@ version = "0.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c530c2b0d0bf8b69304b39fe2001993e267461948b890cd037d8ad4293fa1a0d" +[[package]] +name = "spki" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67cf02bbac7a337dc36e4f5a693db6c21e7863f45070f7064577eb4367a3212b" +dependencies = [ + "base64ct", + "der 0.6.1", +] + [[package]] name = "stable-pattern" version = "0.1.0" @@ -3910,6 +4623,18 @@ dependencies = [ "unicode-xid 0.0.4", ] +[[package]] +name = "synstructure" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f36bdaa60a83aca3921b5259d5400cbf5e90fc51931376a9bd4a0eb79aa7210f" +dependencies = [ + "proc-macro2", + "quote 1.0.26", + "syn 1.0.107", + "unicode-xid 0.2.3", +] + [[package]] name = "sysinfo" version = "0.27.6" @@ -4064,6 +4789,7 @@ dependencies = [ "mio 0.8.3", "num_cpus", "once_cell", + "parking_lot", "pin-project-lite 0.2.9", "signal-hook-registry", "socket2 0.4.4", @@ -4182,7 +4908,7 @@ version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5cc496875d9c8fe9a0ce19e3ee8e8808c60376831a439543f0aac71c9dd129fa" dependencies = [ - "digest 0.10.3", + "digest 0.10.7", "hmac", "sha-1 0.10.0", "sha2", @@ -4504,7 +5230,7 @@ version = "2.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b11c96ac7ee530603dcdf68ed1557050f374ce55a5a07193ebf8cbc9f8927e9" dependencies = [ - "base64 0.21.2", + "base64 0.21.3", "log", "native-tls", "once_cell", @@ -4524,6 +5250,15 @@ dependencies = [ "serde", ] +[[package]] +name = "url-escape" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44e0ce4d1246d075ca5abec4b41d33e87a6054d08e2366b63205665e950db218" +dependencies = [ + "percent-encoding", +] + [[package]] name = "utf-8" version = "0.7.6" @@ -4772,6 +5507,28 @@ dependencies = [ "untrusted 0.6.2", ] +[[package]] +name = "web-push" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30bdf799b4be6f04bfee94807548718de568834c9ea020e73b59af3dac66d72c" +dependencies = [ + "async-trait", + "base64 0.13.0", + "chrono", + "ece", + "futures-lite", + "http", + "isahc", + "jwt-simple", + "log", + "pem 1.1.1", + "sec1_decode", + "serde", + "serde_derive", + "serde_json", +] + [[package]] name = "web-sys" version = "0.3.57" @@ -4930,8 +5687,23 @@ dependencies = [ "winapi-build", ] +[[package]] +name = "yaml-rust" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56c1936c4cc7a1c9ab21a1ebb602eb942ba868cbd44a99cb7cdc5892335e1c85" +dependencies = [ + "linked-hash-map", +] + [[package]] name = "yansi" version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09041cd90cf85f7f8b2df60c646f853b7f535ce68f85244eb6731cf89fa498ec" + +[[package]] +name = "zeroize" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a0956f1ba7c7909bfb66c2e9e4124ab6f6482560f6628b5aaeba39207c9aad9" diff --git a/crates/core/database/Cargo.toml b/crates/core/database/Cargo.toml index ce5044a6..5a1043c6 100644 --- a/crates/core/database/Cargo.toml +++ b/crates/core/database/Cargo.toml @@ -13,15 +13,17 @@ description = "Revolt Backend: Database Implementation" mongodb = ["dep:mongodb", "bson"] # ... Other +tasks = ["isahc", "linkify", "url-escape"] async-std-runtime = ["async-std"] rocket-impl = ["rocket", "schemars", "revolt_okapi", "revolt_rocket_okapi"] redis-is-patched = ["revolt-presence/redis-is-patched"] # Default Features -default = ["mongodb", "async-std-runtime"] +default = ["mongodb", "async-std-runtime", "tasks"] [dependencies] # Core +revolt-config = { version = "0.6.7", path = "../config" } revolt-result = { version = "0.6.7", path = "../result" } revolt-models = { version = "0.6.7", path = "../models" } revolt-presence = { version = "0.6.7", path = "../presence" } @@ -36,9 +38,14 @@ lru = "0.11.0" rand = "0.8.5" ulid = "1.0.0" nanoid = "0.4.0" +base64 = "0.21.3" once_cell = "1.17" indexmap = "1.9.1" decancer = "1.6.2" +deadqueue = "0.2.4" +linkify = { optional = true, version = "0.8.1" } +url-escape = { optional = true, version = "0.1.1" } +isahc = { optional = true, version = "1.7", features = ["json"] } # Serialisation serde_json = "1" @@ -59,6 +66,7 @@ regex = "1" # Async Language Features futures = "0.3.19" +async-lock = "2.8.0" async-trait = "0.1.51" async-recursion = "1.0.4" @@ -73,5 +81,9 @@ rocket = { version = "0.5.0-rc.2", default-features = false, features = [ revolt_okapi = { version = "0.9.1", optional = true } revolt_rocket_okapi = { version = "0.9.1", optional = true } +# Notifications +fcm = "0.9.2" +web-push = "0.10.0" + # Authifier authifier = { version = "1.0" } diff --git a/crates/core/database/src/drivers/mod.rs b/crates/core/database/src/drivers/mod.rs index 79cc03f1..01dc4455 100644 --- a/crates/core/database/src/drivers/mod.rs +++ b/crates/core/database/src/drivers/mod.rs @@ -65,3 +65,14 @@ impl DatabaseInfo { }) } } + +impl From for authifier::Database { + fn from(value: Database) -> Self { + match value { + Database::Reference(_) => Default::default(), + Database::MongoDb(MongoDb(client, _)) => authifier::Database::MongoDb( + authifier::database::MongoDb(client.database("revolt")), + ), + } + } +} diff --git a/crates/core/database/src/lib.rs b/crates/core/database/src/lib.rs index 03f0df41..afc591e3 100644 --- a/crates/core/database/src/lib.rs +++ b/crates/core/database/src/lib.rs @@ -81,6 +81,7 @@ pub mod util; pub use models::*; pub mod events; +pub mod tasks; /// Utility function to check if a boolean value is false pub fn if_false(t: &bool) -> bool { diff --git a/crates/core/database/src/models/messages/model.rs b/crates/core/database/src/models/messages/model.rs index b2dfd345..c99a675e 100644 --- a/crates/core/database/src/models/messages/model.rs +++ b/crates/core/database/src/models/messages/model.rs @@ -1,10 +1,15 @@ use indexmap::{IndexMap, IndexSet}; use iso8601_timestamp::Timestamp; -use revolt_models::v0::{Embed, MessageSort, MessageWebhook}; +use revolt_models::v0::{Embed, MessageAuthor, MessageSort, MessageWebhook, PushNotification}; use revolt_result::Result; +use serde_json::json; use ulid::Ulid; -use crate::{events::client::EventV1, Database, File}; +use crate::{ + events::client::EventV1, + tasks::{self, ack::AckEvent}, + Channel, Database, File, +}; auto_derived_partial!( /// Message @@ -170,7 +175,12 @@ auto_derived!( #[allow(clippy::disallowed_methods)] impl Message { /// Send a message without any notifications - pub async fn send_without_notifications(&mut self, db: &Database) -> Result<()> { + pub async fn send_without_notifications( + &mut self, + db: &Database, + is_dm: bool, + generate_embeds: bool, + ) -> Result<()> { db.insert_message(self).await?; // Fan out events @@ -178,21 +188,90 @@ impl Message { .p(self.channel.to_string()) .await; - // TODO: update last_message_id - // TODO: add mentions for affected users - // TODO: generate embeds + // Update last_message_id + tasks::last_message_id::queue(self.channel.to_string(), self.id.to_string(), is_dm).await; + + // Add mentions for affected users + if let Some(mentions) = &self.mentions { + for user in mentions { + tasks::ack::queue( + self.channel.to_string(), + user.to_string(), + AckEvent::AddMention { + ids: vec![self.id.to_string()], + }, + ) + .await; + } + } + + // Generate embeds + if generate_embeds { + if let Some(content) = &self.content { + tasks::process_embeds::queue( + self.channel.to_string(), + self.id.to_string(), + content.clone(), + ) + .await; + } + } Ok(()) } /// Send a message - pub async fn send(&mut self, db: &Database) -> Result<()> { - self.send_without_notifications(db).await?; + pub async fn send( + &mut self, + db: &Database, + author: MessageAuthor<'_>, + channel: &Channel, + generate_embeds: bool, + ) -> Result<()> { + self.send_without_notifications( + db, + matches!(channel, Channel::DirectMessage { .. }), + generate_embeds, + ) + .await?; - // TODO: web push / FCM + // Push out Web Push notifications + crate::tasks::web_push::queue( + { + match channel { + Channel::DirectMessage { recipients, .. } + | Channel::Group { recipients, .. } => recipients.clone(), + Channel::TextChannel { .. } => self.mentions.clone().unwrap_or_default(), + _ => vec![], + } + }, + json!(PushNotification::from(self.clone().into(), Some(author), &channel.id()).await) + .to_string(), + ) + .await; todo!() } + + /// Append content to message + pub async fn append( + db: &Database, + id: String, + channel: String, + append: AppendMessage, + ) -> Result<()> { + db.append_message(&id, &append).await?; + + EventV1::MessageAppend { + id, + channel: channel.to_string(), + append: append.into(), + } + .p(channel) + .await; + + Ok(()) + } } impl SystemMessage { @@ -209,29 +288,6 @@ impl SystemMessage { } impl Interactions { - /// Validate interactions info is correct - /* pub async fn validate( - &self, - db: &Database, - permissions: &mut PermissionCalculator<'_>, - ) -> Result<()> { - if let Some(reactions) = &self.reactions { - permissions.throw_permission(db, Permission::React).await?; - - if reactions.len() > 20 { - return Err(Error::InvalidOperation); - } - - for reaction in reactions { - if !Emoji::can_use(db, reaction).await? { - return Err(Error::InvalidOperation); - } - } - } - - Ok(()) - }*/ - /// Check if we can use a given emoji to react pub fn can_use(&self, emoji: &str) -> bool { if self.restrict_reactions { diff --git a/crates/core/database/src/models/server_members/model.rs b/crates/core/database/src/models/server_members/model.rs index b3f2a274..7722dbb8 100644 --- a/crates/core/database/src/models/server_members/model.rs +++ b/crates/core/database/src/models/server_members/model.rs @@ -131,7 +131,7 @@ impl Member { id: user.id.clone(), } .into_message(id.to_string()) - .send_without_notifications(db) + .send_without_notifications(db, false, false) .await .ok(); } diff --git a/crates/core/database/src/tasks/ack.rs b/crates/core/database/src/tasks/ack.rs new file mode 100644 index 00000000..c055fb4a --- /dev/null +++ b/crates/core/database/src/tasks/ack.rs @@ -0,0 +1,122 @@ +// Queue Type: Debounced +use crate::Database; + +use deadqueue::limited::Queue; +use once_cell::sync::Lazy; +use std::{collections::HashMap, time::Duration}; + +use super::DelayedTask; + +/// Enumeration of possible events +#[derive(Debug, Eq, PartialEq)] +pub enum AckEvent { + /// Add mentions for a user in a channel + AddMention { + /// Message IDs + ids: Vec, + }, + + /// Acknowledge message in a channel for a user + AckMessage { + /// Message ID + id: String, + }, +} + +/// Task information +struct Data { + /// Channel to ack + channel: String, + /// User to ack for + user: String, + /// Event + event: AckEvent, +} + +#[derive(Debug)] +struct Task { + event: AckEvent, +} + +static Q: Lazy> = Lazy::new(|| Queue::new(10_000)); + +/// Queue a new task for a worker +pub async fn queue(channel: String, user: String, event: AckEvent) { + Q.try_push(Data { + channel, + user, + event, + }) + .ok(); + + info!("Queue is using {} slots from {}.", Q.len(), Q.capacity()); +} + +/// Start a new worker +pub async fn worker(db: Database) { + let mut tasks = HashMap::<(String, String), DelayedTask>::new(); + let mut keys = vec![]; + + loop { + // Find due tasks. + for (key, task) in &tasks { + if task.should_run() { + keys.push(key.clone()); + } + } + + // Commit any due tasks to the database. + for key in &keys { + if let Some(task) = tasks.remove(key) { + let Task { event } = task.data; + let (user, channel) = key; + + if let Err(err) = match &event { + #[allow(clippy::disallowed_methods)] // event is sent by higher level function + AckEvent::AckMessage { id } => db.acknowledge_message(channel, user, id).await, + AckEvent::AddMention { ids } => { + db.add_mention_to_unread(channel, user, ids).await + } + } { + error!("{err:?} for {event:?}. ({user}, {channel})"); + } else { + info!("User {user} ack in {channel} with {event:?}"); + } + } + } + + // Clear keys + keys.clear(); + + // Queue incoming tasks. + while let Some(Data { + channel, + user, + mut event, + }) = Q.try_pop() + { + let key = (user, channel); + if let Some(task) = tasks.get_mut(&key) { + task.delay(); + + match &mut event { + AckEvent::AddMention { ids } => { + if let AckEvent::AddMention { ids: existing } = &mut task.data.event { + existing.append(ids); + } else { + task.data.event = event; + } + } + AckEvent::AckMessage { .. } => { + task.data.event = event; + } + } + } else { + tasks.insert(key, DelayedTask::new(Task { event })); + } + } + + // Sleep for an arbitrary amount of time. + async_std::task::sleep(Duration::from_secs(1)).await; + } +} diff --git a/crates/core/database/src/tasks/last_message_id.rs b/crates/core/database/src/tasks/last_message_id.rs new file mode 100644 index 00000000..94c14c17 --- /dev/null +++ b/crates/core/database/src/tasks/last_message_id.rs @@ -0,0 +1,87 @@ +// Queue Type: Debounced +use deadqueue::limited::Queue; +use once_cell::sync::Lazy; +use std::{collections::HashMap, time::Duration}; + +use crate::{Database, PartialChannel}; + +use super::DelayedTask; + +/// Task information +struct Data { + /// Channel to update + channel: String, + /// Latest message ID + id: String, + /// Whether the channel is a DM + is_dm: bool, +} + +/// Task information +#[derive(Debug)] +struct Task { + /// Latest message ID + id: String, + /// Whether the channel is a DM + is_dm: bool, +} + +static Q: Lazy> = Lazy::new(|| Queue::new(10_000)); + +/// Queue a new task for a worker +pub async fn queue(channel: String, id: String, is_dm: bool) { + Q.try_push(Data { channel, id, is_dm }).ok(); + info!("Queue is using {} slots from {}.", Q.len(), Q.capacity()); +} + +/// Start a new worker +pub async fn worker(db: Database) { + let mut tasks = HashMap::>::new(); + let mut keys = vec![]; + + loop { + // Find due tasks. + for (key, task) in &tasks { + if task.should_run() { + keys.push(key.clone()); + } + } + + // Commit any due tasks to the database. + for key in &keys { + if let Some(task) = tasks.remove(key) { + let Task { id, is_dm, .. } = task.data; + + let mut channel = PartialChannel { + last_message_id: Some(id.to_string()), + ..Default::default() + }; + + if is_dm { + channel.active = Some(true); + } + + match db.update_channel(key, &channel, vec![]).await { + Ok(_) => info!("Updated last_message_id for {key} to {id}."), + Err(err) => error!("Failed to update last_message_id with {err:?}!"), + } + } + } + + // Clear keys + keys.clear(); + + // Queue incoming tasks. + while let Some(Data { channel, id, is_dm }) = Q.try_pop() { + if let Some(task) = tasks.get_mut(&channel) { + task.data.id = id; + task.delay(); + } else { + tasks.insert(channel, DelayedTask::new(Task { id, is_dm })); + } + } + + // Sleep for an arbitrary amount of time. + async_std::task::sleep(Duration::from_secs(1)).await; + } +} diff --git a/crates/core/database/src/tasks/mod.rs b/crates/core/database/src/tasks/mod.rs new file mode 100644 index 00000000..93f15a5a --- /dev/null +++ b/crates/core/database/src/tasks/mod.rs @@ -0,0 +1,58 @@ +//! Semi-important background task management + +use crate::Database; + +use async_std::task; +use std::time::Instant; + +const WORKER_COUNT: usize = 5; + +pub mod ack; +pub mod last_message_id; +pub mod process_embeds; +pub mod web_push; + +/// Spawn background workers +pub async fn start_workers(db: Database) { + for _ in 0..WORKER_COUNT { + task::spawn(ack::worker(db.clone())); + task::spawn(last_message_id::worker(db.clone())); + task::spawn(process_embeds::worker(db.clone())); + task::spawn(web_push::worker(db.clone().into())); + } +} + +/// Task with additional information on when it should run +pub struct DelayedTask { + pub data: T, + last_updated: Instant, + first_seen: Instant, +} + +/// Commit to database every 30 seconds if the task is particularly active. +static EXPIRE_CONSTANT: u64 = 30; + +/// Otherwise, commit to database after 5 seconds. +static SAVE_CONSTANT: u64 = 5; + +impl DelayedTask { + /// Create a new delayed task + pub fn new(data: T) -> Self { + DelayedTask { + data, + last_updated: Instant::now(), + first_seen: Instant::now(), + } + } + + /// Push a task further back in time + pub fn delay(&mut self) { + self.last_updated = Instant::now() + } + + /// Check if a task should run yet + pub fn should_run(&self) -> bool { + self.first_seen.elapsed().as_secs() > EXPIRE_CONSTANT + || self.last_updated.elapsed().as_secs() > SAVE_CONSTANT + } +} diff --git a/crates/core/database/src/tasks/process_embeds.rs b/crates/core/database/src/tasks/process_embeds.rs new file mode 100644 index 00000000..e03ba914 --- /dev/null +++ b/crates/core/database/src/tasks/process_embeds.rs @@ -0,0 +1,170 @@ +use crate::{models::Message, AppendMessage, Database}; + +use futures::future::join_all; +use linkify::{LinkFinder, LinkKind}; +use regex::Regex; +use revolt_config::config; +use revolt_result::Result; + +use async_lock::Semaphore; +use async_std::task::spawn; +use deadqueue::limited::Queue; +use once_cell::sync::Lazy; +use revolt_models::v0::Embed; +use std::{collections::HashSet, sync::Arc}; + +use isahc::prelude::*; + +/// Task information +#[derive(Debug)] +struct EmbedTask { + /// Channel we're processing the event in + channel: String, + /// ID of the message we're processing + id: String, + /// Content of the message + content: String, +} + +static Q: Lazy> = Lazy::new(|| Queue::new(10_000)); + +/// Queue a new task for a worker +pub async fn queue(channel: String, id: String, content: String) { + Q.try_push(EmbedTask { + channel, + id, + content, + }) + .ok(); + + info!("Queue is using {} slots from {}.", Q.len(), Q.capacity()); +} + +/// Start a new worker +pub async fn worker(db: Database) { + let semaphore = Arc::new(Semaphore::new( + config().await.api.workers.max_concurrent_connections, + )); + + loop { + let task = Q.pop().await; + let db = db.clone(); + let semaphore = semaphore.clone(); + + spawn(async move { + let config = config().await; + let embeds = generate( + task.content, + &config.hosts.january, + config.features.limits.default.message_embeds, + semaphore, + ) + .await; + + if let Ok(embeds) = embeds { + if let Err(err) = Message::append( + &db, + task.id, + task.channel, + AppendMessage { + embeds: Some(embeds), + }, + ) + .await + { + error!("Encountered an error appending to message: {:?}", err); + } + } + }); + } +} + +static RE_CODE: Lazy = Lazy::new(|| Regex::new("```(?:.|\n)+?```|`(?:.|\n)+?`").unwrap()); +static RE_IGNORED: Lazy = Lazy::new(|| Regex::new("()").unwrap()); + +pub async fn generate( + content: String, + host: &str, + max_embeds: usize, + semaphore: Arc, +) -> Result> { + // Ignore code blocks. + let content = RE_CODE.replace_all(&content, ""); + + // Ignore all content between angle brackets starting with http. + let content = RE_IGNORED.replace_all(&content, ""); + + let content = content + // Ignore quoted lines. + .split('\n') + .map(|v| { + if let Some(c) = v.chars().next() { + if c == '>' { + return ""; + } + } + + v + }) + .collect::>() + .join("\n"); + + let mut finder = LinkFinder::new(); + finder.kinds(&[LinkKind::Url]); + + // Process all links, stripping anchors and + // only taking up to `max_embeds` of links. + let links: Vec = finder + .links(&content) + .map(|x| { + x.as_str() + .chars() + .take_while(|&ch| ch != '#') + .collect::() + }) + .collect::>() + .into_iter() + .take(max_embeds) + .collect(); + + // If no links, fail out. + if links.is_empty() { + return Err(create_error!(LabelMe)); + } + + // ! FIXME: batch request to january + let mut tasks = Vec::new(); + + for link in links { + let semaphore = semaphore.clone(); + let host = host.to_string(); + tasks.push(spawn(async move { + let guard = semaphore.acquire().await; + + if let Ok(mut response) = isahc::get_async(format!( + "{host}/embed?url={}", + url_escape::encode_component(&link) + )) + .await + { + drop(guard); + response.json::().await.ok() + } else { + None + } + })); + } + + let embeds = join_all(tasks) + .await + .into_iter() + .flatten() + .collect::>(); + + // Prevent database update when no embeds are found. + if !embeds.is_empty() { + Ok(embeds) + } else { + Err(create_error!(LabelMe)) + } +} diff --git a/crates/core/database/src/tasks/web_push.rs b/crates/core/database/src/tasks/web_push.rs new file mode 100644 index 00000000..1d833ad8 --- /dev/null +++ b/crates/core/database/src/tasks/web_push.rs @@ -0,0 +1,122 @@ +use std::collections::HashSet; + +use authifier::Database; +use base64::{ + alphabet, + engine::{self}, + Engine as _, +}; +use deadqueue::limited::Queue; +use once_cell::sync::Lazy; +use revolt_presence::filter_online; +use web_push::{ + ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder, + WebPushClient, WebPushMessageBuilder, +}; + +/// Task information +#[derive(Debug)] +struct PushTask { + /// User IDs of the targets that are to receive this notification + recipients: Vec, + /// Raw JSON payload to send to clients + payload: String, +} + +static Q: Lazy> = Lazy::new(|| Queue::new(10_000)); + +/// Queue a new task for a worker +pub async fn queue(recipients: Vec, payload: String) { + if recipients.is_empty() { + return; + } + + let online_ids = filter_online(&recipients).await; + let recipients = (&recipients.into_iter().collect::>() - &online_ids) + .into_iter() + .collect::>(); + + Q.try_push(PushTask { + recipients, + payload, + }) + .ok(); + + info!("Queue is using {} slots from {}.", Q.len(), Q.capacity()); +} + +/// Start a new worker +pub async fn worker(db: Database) { + let client = IsahcWebPushClient::new().unwrap(); + let key = engine::GeneralPurpose::new(&alphabet::URL_SAFE, engine::general_purpose::NO_PAD) + .decode(std::env::var("REVOLT_VAPID_PRIVATE_KEY").unwrap()) + .expect("valid `VAPID_PRIVATE_KEY`"); + + loop { + let task = Q.pop().await; + + if let Ok(sessions) = db.find_sessions_with_subscription(&task.recipients).await { + for session in sessions { + if let Some(sub) = session.subscription { + if sub.endpoint == "fcm" { + // Use Firebase Cloud Messaging + info!("Skipping FCM client, unimplemented..."); + } else { + // Use Web Push Standard + let subscription = SubscriptionInfo { + endpoint: sub.endpoint, + keys: SubscriptionKeys { + auth: sub.auth, + p256dh: sub.p256dh, + }, + }; + + match VapidSignatureBuilder::from_pem( + std::io::Cursor::new(&key), + &subscription, + ) { + Ok(sig_builder) => match sig_builder.build() { + Ok(signature) => { + let mut builder = WebPushMessageBuilder::new(&subscription); + builder.set_vapid_signature(signature); + builder.set_payload( + ContentEncoding::AesGcm, + task.payload.as_bytes(), + ); + + match builder.build() { + Ok(msg) => match client.send(msg).await { + Ok(_) => { + info!( + "Sent Web Push notification to {:?}.", + session.id + ) + } + Err(err) => { + error!("Hit error sending Web Push! {:?}", err) + } + }, + Err(err) => { + error!( + "Failed to build message for {}! {:?}", + session.user_id, err + ) + } + } + } + Err(err) => error!( + "Failed to build signature for {}! {:?}", + session.user_id, err + ), + }, + Err(err) => error!( + "Failed to create signature builder for {}! {:?}", + session.user_id, err + ), + } + } + } + } + } + } +} diff --git a/crates/core/database/src/util/bridge/v0.rs b/crates/core/database/src/util/bridge/v0.rs index f2111c7e..f03c9f08 100644 --- a/crates/core/database/src/util/bridge/v0.rs +++ b/crates/core/database/src/util/bridge/v0.rs @@ -391,6 +391,16 @@ impl From for Interactions { } } +impl From for AppendMessage { + fn from(value: crate::AppendMessage) -> Self { + AppendMessage { + embeds: value + .embeds + .map(|embeds| embeds.into_iter().map(|embed| embed.into()).collect()), + } + } +} + impl From for Masquerade { fn from(value: crate::Masquerade) -> Self { Masquerade { diff --git a/crates/core/models/Cargo.toml b/crates/core/models/Cargo.toml index fe293ab1..ea327d38 100644 --- a/crates/core/models/Cargo.toml +++ b/crates/core/models/Cargo.toml @@ -18,6 +18,7 @@ default = ["serde", "partials"] [dependencies] # Core +revolt-config = { version = "0.6.7", path = "../config" } revolt-permissions = { version = "0.6.7", path = "../permissions" } # Utility diff --git a/crates/core/models/src/v0/messages.rs b/crates/core/models/src/v0/messages.rs index d67d44e6..ceb8715f 100644 --- a/crates/core/models/src/v0/messages.rs +++ b/crates/core/models/src/v0/messages.rs @@ -1,7 +1,11 @@ +use std::time::SystemTime; + +use revolt_config::config; + use indexmap::{IndexMap, IndexSet}; use iso8601_timestamp::Timestamp; -use super::{Embed, File, MessageWebhook}; +use super::{Embed, File, MessageWebhook, User, Webhook}; auto_derived_partial!( /// Message @@ -129,11 +133,129 @@ auto_derived!( /// Sort by the oldest messages first Oldest, } + + /// Push Notification + pub struct PushNotification { + /// Known author name + pub author: String, + /// URL to author avatar + pub icon: String, + /// URL to first matching attachment + #[serde(skip_serializing_if = "Option::is_none")] + pub image: Option, + /// Message content or system message information + pub body: String, + /// Unique tag, usually the channel ID + pub tag: String, + /// Timestamp at which this notification was created + pub timestamp: u64, + /// URL to open when clicking notification + pub url: String, + } ); +/// Message Author Abstraction +pub enum MessageAuthor<'a> { + User(&'a User), + Webhook(&'a Webhook), +} + impl Interactions { /// Check if default initialisation of fields pub fn is_default(&self) -> bool { !self.restrict_reactions && self.reactions.is_none() } } + +impl<'a> MessageAuthor<'a> { + pub fn id(&self) -> &str { + match self { + MessageAuthor::User(user) => &user.id, + MessageAuthor::Webhook(webhook) => &webhook.id, + } + } + + pub fn avatar(&self) -> Option<&str> { + match self { + MessageAuthor::User(user) => user.avatar.as_ref().map(|file| file.id.as_str()), + MessageAuthor::Webhook(webhook) => webhook.avatar.as_ref().map(|file| file.id.as_str()), + } + } + + pub fn username(&self) -> &str { + match self { + MessageAuthor::User(user) => &user.username, + MessageAuthor::Webhook(webhook) => &webhook.name, + } + } +} + +impl From for String { + fn from(s: SystemMessage) -> String { + match s { + SystemMessage::Text { content } => content, + SystemMessage::UserAdded { .. } => "User added to the channel.".to_string(), + SystemMessage::UserRemove { .. } => "User removed from the channel.".to_string(), + SystemMessage::UserJoined { .. } => "User joined the channel.".to_string(), + SystemMessage::UserLeft { .. } => "User left the channel.".to_string(), + SystemMessage::UserKicked { .. } => "User kicked from the channel.".to_string(), + SystemMessage::UserBanned { .. } => "User banned from the channel.".to_string(), + SystemMessage::ChannelRenamed { .. } => "Channel renamed.".to_string(), + SystemMessage::ChannelDescriptionChanged { .. } => { + "Channel description changed.".to_string() + } + SystemMessage::ChannelIconChanged { .. } => "Channel icon changed.".to_string(), + SystemMessage::ChannelOwnershipChanged { .. } => { + "Channel ownership changed.".to_string() + } + } + } +} + +impl PushNotification { + /// Create a new notification from a given message, author and channel ID + pub async fn from(msg: Message, author: Option>, channel_id: &str) -> Self { + let config = config().await; + + let icon = if let Some(author) = &author { + if let Some(avatar) = author.avatar() { + format!("{}/avatars/{}", config.hosts.autumn, avatar) + } else { + format!("{}/users/{}/default_avatar", config.hosts.api, author.id()) + } + } else { + format!("{}/assets/logo.png", config.hosts.app) + }; + + let image = msg.attachments.and_then(|attachments| { + attachments + .first() + .map(|v| format!("{}/attachments/{}", config.hosts.autumn, v.id)) + }); + + let body = if let Some(sys) = msg.system { + sys.into() + } else if let Some(text) = msg.content { + text + } else { + "Empty Message".to_string() + }; + + let timestamp = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .expect("Time went backwards") + .as_secs(); + + Self { + author: author + .map(|x| x.username().to_string()) + .unwrap_or_else(|| "Revolt".to_string()), + icon, + image, + body, + tag: channel_id.to_string(), + timestamp, + url: format!("{}/channel/{}/{}", config.hosts.app, channel_id, msg.id), + } + } +} diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index d8a2b3b3..38406a0d 100644 --- a/crates/delta/src/main.rs +++ b/crates/delta/src/main.rs @@ -39,7 +39,7 @@ async fn rocket() -> _ { // Setup Authifier let authifier = Authifier { - database: legacy_db.clone().into(), + database: db.clone().into(), config: revolt_quark::util::authifier::config(), event_channel: Some(sender), }; @@ -61,6 +61,7 @@ async fn rocket() -> _ { }); // Launch background task workers + async_std::task::spawn(revolt_database::tasks::start_workers(db.clone())); async_std::task::spawn(revolt_quark::tasks::start_workers(legacy_db.clone())); // Configure CORS