From c721005b20f6f496ecdb23ad3981df7d174b00bc Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Thu, 27 Nov 2025 10:09:48 +0200 Subject: [PATCH 1/7] fix: redesign and compact Signed-off-by: Leonid Kaganov --- .gitignore | 1 + Cargo.lock | 368 ++-------------------------------------- Cargo.toml | 49 +++--- README.md | 8 +- src/config.rs | 18 +- src/config/default.toml | 2 - src/db.rs | 291 +++++++++++++++++++++---------- src/handlers_http.rs | 51 +++--- src/handlers_ws.rs | 87 +++++++--- src/hub_service.rs | 45 ++--- src/main.rs | 180 +++++++++++++------- src/memory.rs | 24 +-- src/redis.rs | 83 +++------ src/workspace_owner.rs | 8 +- 14 files changed, 547 insertions(+), 668 deletions(-) diff --git a/.gitignore b/.gitignore index 543c428394..8ffa5d5078 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ /scripts/TEST_LLEOTOKEN.html Justfile commit.sh +bloat.sh /src/GO.sh /src/GOT.sh GO.sh diff --git a/Cargo.lock b/Cargo.lock index b6064ca942..0ae060db12 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,31 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "actix" -version = "0.13.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de7fa236829ba0841304542f7614c42b80fca007455315c45c785ccfa873a85b" -dependencies = [ - "actix-macros", - "actix-rt", - "actix_derive", - "bitflags 2.9.4", - "bytes", - "crossbeam-channel", - "futures-core", - "futures-sink", - "futures-task", - "futures-util", - "log", - "once_cell", - "parking_lot 0.12.4", - "pin-project-lite", - "smallvec", - "tokio", - "tokio-util", -] - [[package]] name = "actix-codec" version = "0.5.2" @@ -70,7 +45,7 @@ dependencies = [ "actix-service", "actix-tls", "actix-utils", - "base64 0.22.1", + "base64", "bitflags 2.9.4", "brotli", "bytes", @@ -234,24 +209,6 @@ dependencies = [ "url", ] -[[package]] -name = "actix-web-actors" -version = "4.3.1+deprecated" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f98c5300b38fd004fe7d2a964f9a90813fdbe8a81fed500587e78b1b71c6f980" -dependencies = [ - "actix", - "actix-codec", - "actix-http", - "actix-web", - "bytes", - "bytestring", - "futures-core", - "pin-project-lite", - "tokio", - "tokio-util", -] - [[package]] name = "actix-web-codegen" version = "4.3.0" @@ -278,17 +235,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "actix_derive" -version = "0.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6ac1e58cded18cb28ddc17143c4dea5345b3ad575e14f32f66e4054a56eb271" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "addr2line" version = "0.25.1" @@ -363,12 +309,6 @@ version = "1.0.100" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" -[[package]] -name = "arraydeque" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236" - [[package]] name = "async-trait" version = "0.1.89" @@ -423,12 +363,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "base64" -version = "0.21.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" - [[package]] name = "base64" version = "0.22.1" @@ -461,9 +395,6 @@ name = "bitflags" version = "2.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2261d10cca569e4643e526d8dc2e62e433cc8aba21ab764233731f8d369bf394" -dependencies = [ - "serde", -] [[package]] name = "block-buffer" @@ -476,9 +407,9 @@ dependencies = [ [[package]] name = "borrow-or-share" -version = "0.2.2" +version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3eeab4423108c5d7c744f4d234de88d18d636100093ae04caf4825134b9c3a32" +checksum = "dc0b364ead1874514c8c2855ab558056ebfeb775653e7ae45ff72f28f8f3166c" [[package]] name = "brotli" @@ -503,9 +434,9 @@ dependencies = [ [[package]] name = "bstr" -version = "1.12.0" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "234113d19d0d7d613b40e86fb654acf958910802bcceab913a4f9e7cda03b1a4" +checksum = "63044e1ae8e69f3b5a92c736ca6269b8d12fa7efe39bf34ddb06d102cf0e2cab" dependencies = [ "memchr", "serde", @@ -612,47 +543,11 @@ version = "0.15.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "180e549344080374f9b32ed41bf3b6b57885ff6a289367b3dbc10eea8acc1918" dependencies = [ - "async-trait", - "convert_case", - "json5", "pathdiff", - "ron", - "rust-ini", - "serde-untagged", "serde_core", "serde_json", "toml", "winnow", - "yaml-rust2", -] - -[[package]] -name = "const-random" -version = "0.1.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" -dependencies = [ - "const-random-macro", -] - -[[package]] -name = "const-random-macro" -version = "0.1.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" -dependencies = [ - "getrandom 0.2.16", - "once_cell", - "tiny-keccak", -] - -[[package]] -name = "convert_case" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec182b0ca2f35d8fc196cf3404988fd8b8c739a4d270ff118a398feb0cbec1ca" -dependencies = [ - "unicode-segmentation", ] [[package]] @@ -700,27 +595,12 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "crossbeam-channel" -version = "0.5.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "crossbeam-utils" version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" -[[package]] -name = "crunchy" -version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" - [[package]] name = "crypto-common" version = "0.1.6" @@ -904,15 +784,6 @@ dependencies = [ "syn", ] -[[package]] -name = "dlv-list" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "442039f5147480ba31067cb00ada1adae6892028e40e45fc5de7b7df6dcc1b5f" -dependencies = [ - "const-random", -] - [[package]] name = "dyn-clone" version = "1.0.20" @@ -943,17 +814,6 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" -[[package]] -name = "erased-serde" -version = "0.4.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "259d404d09818dec19332e31d94558aeb442fea04c817006456c24b5460bbd4b" -dependencies = [ - "serde", - "serde_core", - "typeid", -] - [[package]] name = "fancy-regex" version = "0.14.0" @@ -1163,9 +1023,9 @@ checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" [[package]] name = "globset" -version = "0.4.16" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "54a1028dfc5f5df5da8a56a73e6c153c9a9708ec57232470703592a3f18e49f5" +checksum = "52dfc19153a48bde0cbd630453615c8151bce3a5adfac7a0aebfbf0a1e1f57e3" dependencies = [ "aho-corasick", "bstr", @@ -1244,15 +1104,6 @@ version = "0.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5419bdc4f6a9207fbeba6d11b604d481addf78ecd10c11ad51e76c2f6482748d" -[[package]] -name = "hashlink" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" -dependencies = [ - "hashbrown 0.15.5", -] - [[package]] name = "heck" version = "0.5.0" @@ -1324,12 +1175,10 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.2.2" +version = "0.3.2" dependencies = [ - "actix", "actix-cors", "actix-web", - "actix-web-actors", "actix-ws", "anyhow", "config", @@ -1343,9 +1192,7 @@ dependencies = [ "serde", "serde_json", "serde_with", - "size", "strum", - "strum_macros", "tokio", "tokio-stream", "tokio-tungstenite", @@ -1436,7 +1283,7 @@ version = "0.1.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c6995591a8f1380fcb4ba966a252a4b29188d51d2b89e3a252f5305be65aea8" dependencies = [ - "base64 0.22.1", + "base64", "bytes", "futures-channel", "futures-core", @@ -1685,17 +1532,6 @@ 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 = "jsonschema" version = "0.30.0" @@ -1703,7 +1539,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1b46a0365a611fbf1d2143104dcf910aada96fafd295bab16c60b802bf6fa1d" dependencies = [ "ahash", - "base64 0.22.1", + "base64", "bytecount", "email_address", "fancy-regex", @@ -1728,7 +1564,7 @@ version = "9.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde" dependencies = [ - "base64 0.22.1", + "base64", "js-sys", "pem", "ring 0.17.14", @@ -1854,15 +1690,6 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21" -[[package]] -name = "nu-ansi-term" -version = "0.50.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4a28e057d01f97e61255210fcff094d74ed0466038633e95017f5beb68e4399" -dependencies = [ - "windows-sys 0.52.0", -] - [[package]] name = "num" version = "0.4.3" @@ -1969,16 +1796,6 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" -[[package]] -name = "ordered-multimap" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49203cdcae0030493bad186b28da2fa25645fa276a51b6fec8010d281e02ef79" -dependencies = [ - "dlv-list", - "hashbrown 0.14.5", -] - [[package]] name = "outref" version = "0.5.2" @@ -2045,7 +1862,7 @@ version = "3.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38af38e8470ac9dee3ce1bae1af9c1671fffc44ddfd8bd1d0a3445bf349a8ef3" dependencies = [ - "base64 0.22.1", + "base64", "serde", ] @@ -2055,50 +1872,6 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" -[[package]] -name = "pest" -version = "2.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21e0a3a33733faeaf8651dfee72dd0f388f0c8e5ad496a3478fa5a922f49cfa8" -dependencies = [ - "memchr", - "thiserror 2.0.17", - "ucd-trie", -] - -[[package]] -name = "pest_derive" -version = "2.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc58706f770acb1dbd0973e6530a3cff4746fb721207feb3a8a6064cd0b6c663" -dependencies = [ - "pest", - "pest_generator", -] - -[[package]] -name = "pest_generator" -version = "2.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d4f36811dfe07f7b8573462465d5cb8965fffc2e71ae377a33aecf14c2c9a2f" -dependencies = [ - "pest", - "pest_meta", - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "pest_meta" -version = "2.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42919b05089acbd0a5dcd5405fb304d17d1053847b81163d09c4ad18ce8e8420" -dependencies = [ - "pest", - "sha2", -] - [[package]] name = "phf" version = "0.12.1" @@ -2469,7 +2242,7 @@ version = "0.12.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d429f34c8092b2d42c7c93cec323bb4adeb7c67698f70839adec842ec10c7ceb" dependencies = [ - "base64 0.22.1", + "base64", "bytes", "futures-core", "http 1.3.1", @@ -2608,28 +2381,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "ron" -version = "0.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b91f7eff05f748767f183df4320a63d6936e9c6107d97c9e6bdd9784f4289c94" -dependencies = [ - "base64 0.21.7", - "bitflags 2.9.4", - "serde", - "serde_derive", -] - -[[package]] -name = "rust-ini" -version = "0.21.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "796e8d2b6696392a43bea58116b667fb4c29727dc5abd27d6acf338bb4f688c7" -dependencies = [ - "cfg-if", - "ordered-multimap", -] - [[package]] name = "rustc-demangle" version = "0.1.26" @@ -2866,18 +2617,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "serde-untagged" -version = "0.1.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9faf48a4a2d2693be24c6289dbe26552776eb7737074e6722891fadbe6c5058" -dependencies = [ - "erased-serde", - "serde", - "serde_core", - "typeid", -] - [[package]] name = "serde_core" version = "1.0.228" @@ -2938,7 +2677,7 @@ version = "3.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c522100790450cf78eeac1507263d0a350d4d5b30df0c8e1fe051a10c22b376e" dependencies = [ - "base64 0.22.1", + "base64", "chrono", "hex", "indexmap 1.9.3", @@ -2994,17 +2733,6 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" -[[package]] -name = "sha2" -version = "0.10.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" -dependencies = [ - "cfg-if", - "cpufeatures", - "digest", -] - [[package]] name = "sharded-slab" version = "0.1.7" @@ -3047,15 +2775,6 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56199f7ddabf13fe5074ce809e7d3f42b42ae711800501b5b16ea82ad029c39d" -[[package]] -name = "size" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b6709c7b6754dca1311b3c73e79fcce40dd414c782c66d88e8823030093b02b" -dependencies = [ - "serde", -] - [[package]] name = "slab" version = "0.4.11" @@ -3253,15 +2972,6 @@ dependencies = [ "time-core", ] -[[package]] -name = "tiny-keccak" -version = "2.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237" -dependencies = [ - "crunchy", -] - [[package]] name = "tinystr" version = "0.8.1" @@ -3522,18 +3232,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678" dependencies = [ "once_cell", - "valuable", -] - -[[package]] -name = "tracing-log" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" -dependencies = [ - "log", - "once_cell", - "tracing-core", ] [[package]] @@ -3542,12 +3240,9 @@ version = "0.3.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5" dependencies = [ - "nu-ansi-term", "sharded-slab", - "smallvec", "thread_local", "tracing-core", - "tracing-log", ] [[package]] @@ -3594,36 +3289,18 @@ dependencies = [ "utf-8", ] -[[package]] -name = "typeid" -version = "1.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" - [[package]] name = "typenum" version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" -[[package]] -name = "ucd-trie" -version = "0.1.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" - [[package]] name = "unicode-ident" version = "1.0.19" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f63a545481291138910575129486daeaf8ac54aee4387fe7906919f7830c7d9d" -[[package]] -name = "unicode-segmentation" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6ccf251212114b54433ec949fd6a7841275f9ada20dddd2f29e9ceea4501493" - [[package]] name = "unicode-xid" version = "0.2.6" @@ -3696,12 +3373,6 @@ dependencies = [ "vsimd", ] -[[package]] -name = "valuable" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" - [[package]] name = "version_check" version = "0.9.5" @@ -4163,17 +3834,6 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ea2f10b9bb0928dfb1b42b65e1f9e36f7f54dbdf08457afefb38afcdec4fa2bb" -[[package]] -name = "yaml-rust2" -version = "0.10.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2462ea039c445496d8793d052e13787f2b90e750b833afee748e601c17621ed9" -dependencies = [ - "arraydeque", - "encoding_rs", - "hashlink", -] - [[package]] name = "yoke" version = "0.8.0" diff --git a/Cargo.toml b/Cargo.toml index 6352ab12c2..b43d69568e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,38 +1,36 @@ [package] name = "hulypulse" -version = "0.2.2" +version = "0.3.2" edition = "2024" [dependencies] -tokio = { version = "1", features = ["full", "macros", "rt-multi-thread"] } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "signal"] } tracing = "0.1.41" -tracing-subscriber = "0.3.19" +tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] } +config = { version = "0.15", default-features = false, features = ["json", "toml"] } anyhow = "1.0.97" -config = "0.15.4" serde = { version = "1.0.219", features = ["derive"] } -actix = "0.13.5" -actix-web = "4.10.2" -actix-cors = "0.7.1" -actix-web-actors = "4.2.0" -redis = { version = "=0.32.5", features = ["aio", "tokio-comp", "sentinel"] } -md5 = "0.8.0" serde_with = "3" -url = "2" -size = { version = "0.5.0", features = ["serde"] } -uuid = { version = "1.7", features = ["v4", "serde"] } -hex = "0.4.3" serde_json = "1.0" -hulyrs = { git = "https://github.com/hcengineering/hulyrs.git", features = [ - "actix", -] } -secrecy = "0.10.3" +actix-web = { version = "4.10", default-features = false, features = ["macros"] } +actix-cors = "0.7.1" +actix-ws = "0.3.0" +md5 = "0.8.0" +url = "2" +hex = "0.4.3" tokio-stream = "0.1" strum = { version = "0.27.2", features = ["derive"] } -strum_macros = "0.27.2" -regorus = "0.5.0" -actix-ws = "0.3.0" futures-util = "0.3" +# auth: +regorus = { version = "0.5.0", optional = true } +uuid = { version = "1.7", features = ["v4", "serde"], optional = true } +hulyrs = { git = "https://github.com/hcengineering/hulyrs.git", features = [ "actix" ], optional = true } +secrecy = { version = "0.10.3", optional = true } + +#redis +redis = { version = "=0.32.5", features = ["aio", "tokio-comp", "sentinel"], optional = true } + [[bin]] name = "hulypulse" path = "src/main.rs" @@ -42,7 +40,8 @@ tokio-tungstenite = { version = "0.21", default-features = false, features = [ "rustls-tls-native-roots", "connect", ] } -tokio = { version = "1", features = ["macros", "rt-multi-thread"] } -futures-util = "0.3" -serde_json = "1" -uuid = "1" + +[features] +default = ["db-redis","auth"] +auth = ["regorus", "uuid", "hulyrs", "secrecy"] +db-redis = ["redis"] \ No newline at end of file diff --git a/README.md b/README.md index 3b4a447a9f..90940eb91b 100644 --- a/README.md +++ b/README.md @@ -191,9 +191,15 @@ Size of data is limited to some reasonable size ## Special options in config/default.toml - ```memory_mode = true``` Use native memory storage instead Redis - - ```no_authorization = true``` Don't check authorization - ```max_size = 100``` Max value size in bytes +## Special cargo build options + - "db-redis" (default) - use Redis (Memory instead) + - "auth" (default) - use huly-authorization + Disable both: + cargo build --no-default-features + Enable one: + cargo build --no-default-features --features "auth" ## Running diff --git a/src/config.rs b/src/config.rs index ab204576f0..b99e3a9e95 100644 --- a/src/config.rs +++ b/src/config.rs @@ -15,10 +15,16 @@ use std::{path::Path, sync::LazyLock}; +#[cfg(feature = "auth")] use secrecy::SecretString; + use serde::Deserialize; +#[cfg(feature = "db-redis")] +use serde_with::StringWithSeparator; +#[cfg(feature = "db-redis")] use serde_with::formats::CommaSeparator; -use serde_with::{StringWithSeparator, serde_as}; +use serde_with::serde_as; +#[cfg(feature = "db-redis")] use url::Url; use config::FileFormat; @@ -43,23 +49,27 @@ pub struct Config { pub bind_port: u16, pub bind_host: String, + #[cfg(feature = "auth")] pub token_secret: SecretString, + #[cfg(feature = "db-redis")] #[serde_as(as = "StringWithSeparator::")] pub redis_urls: Vec, + #[cfg(feature = "db-redis")] pub redis_password: String, + #[cfg(feature = "db-redis")] pub redis_mode: RedisMode, + #[cfg(feature = "db-redis")] pub redis_service: String, pub max_ttl: usize, pub max_size: Option, - pub backend: BackendType, - pub no_authorization: bool, - + // pub backend: BackendType, pub heartbeat_timeout: u64, pub ping_timeout: u64, + #[cfg(feature = "auth")] pub policy_file: Option, } diff --git a/src/config/default.toml b/src/config/default.toml index 5e8d1228ec..dd35595557 100644 --- a/src/config/default.toml +++ b/src/config/default.toml @@ -9,8 +9,6 @@ redis_mode = "direct" redis_service = "mymaster" max_ttl = 3600 -backend = "redis" -no_authorization = false heartbeat_timeout = 90 ping_timeout = 30 diff --git a/src/db.rs b/src/db.rs index aadac7c1ab..72d07962aa 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,72 +1,179 @@ +#[cfg(not(feature = "db-redis"))] use std::sync::Arc; -use crate::{ - hub_service::{HubState, RedisEvent, RedisEventAction, broadcast_event}, - memory::{MemoryBackend, memory_delete, memory_info, memory_list, memory_read, memory_save}, - redis::{ - RedisArray, SaveMode, Ttl, redis_delete, redis_info, redis_list, redis_read, redis_save, - }, +#[cfg(not(feature = "db-redis"))] +use crate::hub_service::{HubState, RedisEvent, RedisEventAction, broadcast_event}; + +#[cfg(not(feature = "db-redis"))] +use crate::memory::{ + MemoryBackend, memory_delete, memory_info, memory_list, memory_read, memory_save, }; + +#[cfg(feature = "db-redis")] +use crate::redis::{redis_delete, redis_info, redis_list, redis_read, redis_save}; + +#[cfg(feature = "db-redis")] use ::redis::aio::MultiplexedConnection; + +#[cfg(feature = "db-redis")] +pub type DbError = redis::RedisError; + +#[cfg(not(feature = "db-redis"))] +#[derive(Debug)] +pub struct DbError(pub String); + +pub type DbResult = Result; + +#[cfg(not(feature = "db-redis"))] +impl std::fmt::Display for DbError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +#[cfg(not(feature = "db-redis"))] +impl std::error::Error for DbError {} + +#[cfg(not(feature = "db-redis"))] use tokio::sync::RwLock; +use serde::Serialize; + +#[derive(Debug, Serialize)] +pub struct DbArray { + pub key: String, + pub data: String, + pub ttl: u64, // sec to expire TTL + pub etag: String, // md5 hash (data) +} + +#[derive(serde::Serialize)] +pub enum Ttl { + Sec(usize), // EX + At(u64), // EXAT (timestamp in seconds) +} + +#[derive(Debug)] +pub enum SaveMode { + Upsert, // default: set or overwrite + Insert, // only if not exists (NX) + Update, // only if exists (XX) + Equal(String), // only if md5 matches provided +} + +/// return Error +// pub fn error(code: u16, msg: impl Into) -> DbResult { +// let msg = msg.into(); +// let full = format!("{}: {}", code, msg); +// Err(redis::RedisError::from(( +// redis::ErrorKind::ExtensionError, +// "", +// full, +// ))) +// } + +pub fn error(code: u16, msg: impl Into) -> DbResult { + let msg = format!("{}: {}", code, msg.into()); + + #[cfg(feature = "db-redis")] + { + return Err(redis::RedisError::from(( + redis::ErrorKind::ExtensionError, + "", + msg, + ))); + } + + #[cfg(not(feature = "db-redis"))] + { + return Err(DbError(msg)); + } +} + +/// Check for redis-deprecated symbols +pub fn deprecated_symbol(s: &str) -> bool { + s.chars().any(|c| { + matches!( + c, + '*' | '?' | '[' | ']' | '\\' | '\0'..='\x1F' | '\x7F' | '"' | '\'' + ) + }) +} + +pub fn deprecated_symbol_error(s: &str) -> DbResult<()> { + if deprecated_symbol(s) { + error(412, "Deprecated symbol in key") + } else { + Ok(()) + } +} + #[derive(Clone)] pub struct Db { - inner: DbInner, + #[cfg(feature = "db-redis")] + db: MultiplexedConnection, + #[cfg(not(feature = "db-redis"))] + db: MemoryBackend, + #[cfg(not(feature = "db-redis"))] hub: Arc>, } -#[derive(Clone)] -enum DbInner { - Memory(MemoryBackend), - Redis(MultiplexedConnection), -} - impl Db { - pub fn new_memory(m: MemoryBackend, hub: Arc>) -> Self { + // pub fn new_db(hub: Arc>) -> Self { + // Self { + // inner: DbInner::Memory(m), + // hub, + // } + // } + // pub fn new_redis(c: MultiplexedConnection, hub: Arc>) -> Self { + // Self { + // inner: DbInner::Redis(c), + // hub, + // } + // } + + pub fn new_db( + #[cfg(not(feature = "db-redis"))] db: MemoryBackend, + #[cfg(feature = "db-redis")] db: MultiplexedConnection, + #[cfg(not(feature = "db-redis"))] hub: Arc>, + ) -> Self { Self { - inner: DbInner::Memory(m), - hub, - } - } - pub fn new_redis(c: MultiplexedConnection, hub: Arc>) -> Self { - Self { - inner: DbInner::Redis(c), + db, + #[cfg(not(feature = "db-redis"))] hub, } } - pub async fn info(&self) -> redis::RedisResult { - // String { - // let res = - match &self.inner { - DbInner::Memory(m) => memory_info(m).await, - DbInner::Redis(conn) => { - let mut c = conn.clone(); - redis_info(&mut c).await - } - } - // }; - // res.unwrap_or_else(|_| "error".to_string()) - } + pub async fn info(&self) -> DbResult { + #[cfg(not(feature = "db-redis"))] + return memory_info(&self.db).await; - pub async fn list(&self, key: &str) -> redis::RedisResult> { - match &self.inner { - DbInner::Memory(m) => memory_list(m, key).await, - DbInner::Redis(conn) => { - let mut c = conn.clone(); - redis_list(&mut c, key).await - } + #[cfg(feature = "db-redis")] + { + let mut c = self.db.clone(); + redis_info(&mut c).await } } - pub async fn read(&self, key: &str) -> redis::RedisResult> { - match &self.inner { - DbInner::Memory(m) => memory_read(m, key).await, - DbInner::Redis(conn) => { - let mut c = conn.clone(); - redis_read(&mut c, key).await - } + pub async fn list(&self, key: &str) -> DbResult> { + #[cfg(not(feature = "db-redis"))] + return memory_list(&self.db, key).await; + + #[cfg(feature = "db-redis")] + { + let mut c = self.db.clone(); + redis_list(&mut c, key).await + } + } + + pub async fn read(&self, key: &str) -> DbResult> { + #[cfg(not(feature = "db-redis"))] + return memory_read(&self.db, key).await; + + #[cfg(feature = "db-redis")] + { + let mut c = self.db.clone(); + redis_read(&mut c, key).await } } @@ -76,53 +183,55 @@ impl Db { value: V, ttl: Option, mode: Option, - ) -> redis::RedisResult<()> { - match &self.inner { - DbInner::Memory(m) => { - memory_save(m, key, value.as_ref(), ttl, mode).await?; - // Send events - let value_str = std::str::from_utf8(value.as_ref()) - .ok() - .map(|s| s.to_string()); - broadcast_event( - &self.hub, - RedisEvent { - message: RedisEventAction::Set, - key: key.to_string(), - }, - value_str, - ) - .await; - Ok(()) - } - DbInner::Redis(conn) => { - let mut c = conn.clone(); - redis_save(&mut c, key, value.as_ref(), ttl, mode).await - } + ) -> DbResult<()> { + #[cfg(not(feature = "db-redis"))] + { + memory_save(&self.db, key, value.as_ref(), ttl, mode).await?; + // Send events + let value_str = std::str::from_utf8(value.as_ref()) + .ok() + .map(|s| s.to_string()); + broadcast_event( + &self.hub, + RedisEvent { + message: RedisEventAction::Set, + key: key.to_string(), + }, + value_str, + ) + .await; + return Ok(()); + } + + #[cfg(feature = "db-redis")] + { + let mut c = self.db.clone(); + redis_save(&mut c, key, value.as_ref(), ttl, mode).await } } - pub async fn delete(&self, key: &str, mode: Option) -> redis::RedisResult { - match &self.inner { - DbInner::Memory(m) => { - let deleted = memory_delete(m, key, mode).await?; - if deleted { - broadcast_event( - &self.hub, - RedisEvent { - message: RedisEventAction::Del, - key: key.to_string(), - }, - None, - ) - .await; - } - Ok(deleted) - } - DbInner::Redis(conn) => { - let mut c = conn.clone(); - redis_delete(&mut c, key, mode).await + pub async fn delete(&self, key: &str, mode: Option) -> DbResult { + #[cfg(not(feature = "db-redis"))] + { + let deleted = memory_delete(&self.db, key, mode).await?; + if deleted { + broadcast_event( + &self.hub, + RedisEvent { + message: RedisEventAction::Del, + key: key.to_string(), + }, + None, + ) + .await; } + return Ok(deleted); + } + + #[cfg(feature = "db-redis")] + { + let mut c = self.db.clone(); + redis_delete(&mut c, key, mode).await } } } diff --git a/src/handlers_http.rs b/src/handlers_http.rs index 4de72250e4..5efabece51 100644 --- a/src/handlers_http.rs +++ b/src/handlers_http.rs @@ -18,18 +18,19 @@ use std::str::FromStr; use tracing::*; use actix_web::{ - Error, HttpRequest, HttpResponse, + Error, HttpResponse, error::ParseError, http::header::{self, HeaderName, HeaderValue, IfMatch, IfNoneMatch, TryIntoHeaderValue}, web, }; -use crate::{ - config::CONFIG, - db::Db, - redis::{SaveMode, Ttl}, - workspace_owner::test_rego_http, -}; +#[cfg(feature = "auth")] +use actix_web::HttpRequest; + +use crate::db::{Db, SaveMode, Ttl}; + +#[cfg(feature = "auth")] +use crate::workspace_owner::test_rego_http; pub fn map_redis_error(err: impl std::fmt::Display) -> Error { let msg = err.to_string(); @@ -60,7 +61,7 @@ pub struct TtlExpiresAtHeader(Option); /// list pub async fn list( - req: HttpRequest, + #[cfg(feature = "auth")] req: HttpRequest, path: web::Path, db: web::Data, ) -> Result { @@ -68,8 +69,11 @@ pub async fn list( let key = format!("{}/{}", ¶ms.workspace, ¶ms.key); trace!(key, "list request"); - if !CONFIG.no_authorization && !test_rego_http(req, "List", &key) { - return Err(actix_web::error::ErrorForbidden("forbidden")); + #[cfg(feature = "auth")] + { + if !test_rego_http(req, "List", &key) { + return Err(actix_web::error::ErrorForbidden("forbidden")); + } } let entries = db.list(&key).await.map_err(map_redis_error)?; @@ -78,7 +82,7 @@ pub async fn list( /// get pub async fn get( - req: HttpRequest, + #[cfg(feature = "auth")] req: HttpRequest, path: web::Path, db: web::Data, ) -> Result { @@ -86,8 +90,11 @@ pub async fn get( let key = format!("{}/{}", ¶ms.workspace, ¶ms.key); trace!(key, "get request"); - if !CONFIG.no_authorization && !test_rego_http(req, "Get", &key) { - return Err(actix_web::error::ErrorForbidden("forbidden")); + #[cfg(feature = "auth")] + { + if !test_rego_http(req, "Get", &key) { + return Err(actix_web::error::ErrorForbidden("forbidden")); + } } let entry_opt = db.read(&key).await.map_err(map_redis_error)?; @@ -102,7 +109,7 @@ pub async fn get( /// put pub async fn put( - req: HttpRequest, + #[cfg(feature = "auth")] req: HttpRequest, path: web::Path, body: web::Bytes, db: web::Data, @@ -119,8 +126,11 @@ pub async fn put( let key = format!("{}/{}", ¶ms.workspace, ¶ms.key); trace!(key, "put request"); - if !CONFIG.no_authorization && !test_rego_http(req, "Put", &key) { - return Err(actix_web::error::ErrorForbidden("forbidden")); + #[cfg(feature = "auth")] + { + if !test_rego_http(req, "Put", &key) { + return Err(actix_web::error::ErrorForbidden("forbidden")); + } } // TTL logic @@ -162,7 +172,7 @@ pub async fn put( /// delete pub async fn delete( - req: HttpRequest, + #[cfg(feature = "auth")] req: HttpRequest, path: web::Path, db: web::Data, if_match: web::Header, @@ -171,8 +181,11 @@ pub async fn delete( let key = format!("{}/{}", ¶ms.workspace, ¶ms.key); trace!(key, "delete request"); - if !CONFIG.no_authorization && !test_rego_http(req, "Delete", &key) { - return Err(actix_web::error::ErrorForbidden("forbidden")); + #[cfg(feature = "auth")] + { + if !test_rego_http(req, "Delete", &key) { + return Err(actix_web::error::ErrorForbidden("forbidden")); + } } // MODE logic diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 5aa291c93c..acf73539cc 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -16,27 +16,36 @@ use actix_ws; use futures_util::StreamExt; // use tracing::info; -use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web}; +use actix_web::{Error, HttpRequest, HttpResponse, web}; + +#[cfg(feature = "auth")] +use actix_web::HttpMessage; + use serde::Deserialize; use serde_json::{Value, json}; use std::sync::Arc; use tokio::sync::RwLock; use crate::{ - config::CONFIG, - db::Db, + BACKEND, + db::{Db, SaveMode, Ttl}, hub_service::{HubState, SessionId, new_session_id}, - redis::{SaveMode, Ttl}, - workspace_owner::check_workspace_core, - workspace_owner::test_rego_claims, }; -use strum_macros::AsRefStr; +#[cfg(feature = "auth")] +use crate::workspace_owner::check_workspace_core; + +#[cfg(feature = "auth")] +use crate::workspace_owner::test_rego_claims; + +// use strum_macros::AsRefStr; +use strum::AsRefStr; #[derive(Deserialize, Debug, AsRefStr)] #[serde(rename_all = "lowercase", tag = "type")] pub enum WsCommand { Put { + #[serde(default = "default_corr")] correlation: String, key: String, data: String, @@ -59,6 +68,7 @@ pub enum WsCommand { }, Delete { + #[serde(default = "default_corr")] correlation: String, key: String, @@ -68,36 +78,48 @@ pub enum WsCommand { }, Get { + #[serde(default = "default_corr")] correlation: String, key: String, }, List { + #[serde(default = "default_corr")] correlation: String, key: String, }, Sub { + #[serde(default = "default_corr")] correlation: String, key: String, }, Unsub { + #[serde(default = "default_corr")] correlation: String, key: String, }, Sublist { + #[serde(default = "default_corr")] correlation: String, }, Info { + #[serde(default = "default_corr")] correlation: String, }, } +fn default_corr() -> String { + "1".to_string() +} + +#[cfg(feature = "auth")] use hulyrs::services::jwt::Claims; +#[cfg(feature = "auth")] async fn wrong_workspace( claims: &Option, key: &str, @@ -132,7 +154,7 @@ async fn handle_command( cmd: WsCommand, db: &Db, hub_state: &Arc>, - claims: Option, + #[cfg(feature = "auth")] claims: Option, session_id: SessionId, ) { match cmd { @@ -140,7 +162,15 @@ async fn handle_command( WsCommand::Info { correlation } => { tracing::info!("INFO"); match db.info().await { - Ok(info) => result(info, &correlation, ws).await, + Ok(info) => { + let info = json!({ + "memory_info": info, + "backend": BACKEND, + "websockets": hub_state.read().await.count(), + "version": env!("CARGO_PKG_VERSION"), + }); + result(info, &correlation, ws).await + } Err(e) => result_err(e.to_string(), &correlation, ws).await, } } @@ -156,6 +186,8 @@ async fn handle_command( correlation, } => { tracing::info!("PUT {} = {}", &key, &data); + + #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { return; } @@ -198,6 +230,8 @@ async fn handle_command( if_match, } => { tracing::info!("DELETE {}", &key); // correlation:{:?} , &correlation + + #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { return; } @@ -224,6 +258,8 @@ async fn handle_command( WsCommand::Get { key, correlation } => { tracing::info!("GET {}", &key); + + #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { return; } @@ -240,9 +276,12 @@ async fn handle_command( WsCommand::List { key, correlation } => { tracing::info!("LIST {:?}", &key); + + #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { return; } + match db.list(&key).await { Ok(data) => { let values: Vec = data.into_iter().map(|item| json!(item)).collect(); @@ -254,9 +293,12 @@ async fn handle_command( WsCommand::Sub { key, correlation } => { tracing::info!("SUB {}", &key); + + #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { return; } + hub_state.write().await.subscribe(session_id, key); result("OK", &correlation, ws).await; } @@ -267,9 +309,11 @@ async fn handle_command( hub_state.write().await.unsubscribe_all(session_id); result("OK", &correlation, ws).await; } else { + #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { return; } + hub_state.write().await.unsubscribe(session_id, key); result("OK", &correlation, ws).await; } @@ -291,16 +335,13 @@ pub async fn handler( db: web::Data, hub_state: web::Data>>, ) -> Result { - let claims = if !CONFIG.no_authorization { - Some( - req.extensions() - .get::() - .expect("Missing claims") - .to_owned(), - ) - } else { - None - }; + #[cfg(feature = "auth")] + let claims = Some( + req.extensions() + .get::() + .expect("Missing claims") + .to_owned(), + ); let (response, mut session, mut msg_stream) = actix_ws::handle(&req, payload)?; @@ -311,7 +352,9 @@ pub async fn handler( actix_web::rt::spawn(async move { while let Some(Ok(msg)) = msg_stream.next().await { - tracing::debug!("WebSocket message: {:?}", msg); + if !matches!(msg, actix_ws::Message::Pong(_)) { + tracing::debug!("WebSocket message: {:?}", msg); + } // renew heartbeat to unixtime (all messages is activity, including "ping") hub_state.write().await.renew_heartbeat(session_id); @@ -336,7 +379,8 @@ pub async fn handler( actix_ws::Message::Text(text) => match serde_json::from_str::(&text) { Ok(cmd) => { - if !CONFIG.no_authorization { + #[cfg(feature = "auth")] + { let key = match &cmd { WsCommand::Put { key, .. } | WsCommand::Delete { key, .. } @@ -360,6 +404,7 @@ pub async fn handler( cmd, &db, &hub_state, + #[cfg(feature = "auth")] claims.clone(), session_id, ) diff --git a/src/hub_service.rs b/src/hub_service.rs index 38082a8566..4ddc8b1d38 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -14,7 +14,10 @@ // use crate::config::CONFIG; -use redis::aio::MultiplexedConnection; + +// #[cfg(feature = "db-redis")] +// use redis::aio::MultiplexedConnection; + use serde::Serialize; use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -50,8 +53,10 @@ pub fn new_session_id() -> SessionId { pub enum RedisEventAction { Set, Del, + #[cfg(feature = "db-redis")] Unlink, Expired, + #[cfg(feature = "db-redis")] Other(String), } @@ -169,26 +174,26 @@ pub async fn broadcast_event( } } -pub async fn push_event( - hub_state: &Arc>, - redis: &mut MultiplexedConnection, - ev: RedisEvent, -) { - // Value only for Set - let mut value: Option = None; - if matches!(ev.message, RedisEventAction::Set) { - match ::redis::cmd("GET") - .arg(&ev.key) - .query_async::>(redis) - .await - { - Ok(v) => value = v, - Err(e) => tracing::warn!("redis GET {} failed: {}", &ev.key, e), - } - } +// pub async fn push_event( +// hub_state: &Arc>, +// redis: &mut MultiplexedConnection, +// ev: RedisEvent, +// ) { +// // Value only for Set +// let mut value: Option = None; +// if matches!(ev.message, RedisEventAction::Set) { +// match ::redis::cmd("GET") +// .arg(&ev.key) +// .query_async::>(redis) +// .await +// { +// Ok(v) => value = v, +// Err(e) => tracing::warn!("redis GET {} failed: {}", &ev.key, e), +// } +// } - broadcast_event(hub_state, ev, value).await; -} +// broadcast_event(hub_state, ev, value).await; +// } pub fn check_heartbeat(hub_state: Arc>) { tokio::spawn(async move { diff --git a/src/main.rs b/src/main.rs index 0fc2ae93f1..47e261ecfb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,35 +15,63 @@ use actix_cors::Cors; use actix_web::{ - App, Error, HttpMessage, HttpResponse, HttpServer, - body::MessageBody, - dev::{ServiceRequest, ServiceResponse}, - middleware::{self, Next}, - web::{self, Path, Query}, + App, HttpResponse, HttpServer, + middleware::{self}, + web::{self}, }; +#[cfg(feature = "auth")] +use actix_web::{ + Error, HttpMessage, + body::MessageBody, + dev::{ServiceRequest, ServiceResponse}, + middleware::Next, + web::{Path, Query}, +}; + +#[cfg(feature = "auth")] use hulyrs::services::jwt::{Claims, actix::ServiceRequestExt}; + +#[cfg(feature = "auth")] use secrecy::ExposeSecret; + use serde_json::json; + +#[cfg(feature = "auth")] use tracing::*; + +#[cfg(feature = "auth")] use uuid::Uuid; mod config; mod handlers_http; mod handlers_ws; + +#[cfg(feature = "db-redis")] mod redis; + +#[cfg(feature = "auth")] mod workspace_owner; mod hub_service; use hub_service::HubState; +#[cfg(feature = "db-redis")] +pub const BACKEND: &str = "REDIS"; +#[cfg(not(feature = "db-redis"))] +pub const BACKEND: &str = "MEMORY"; + use config::CONFIG; mod db; -mod memory; +use crate::db::Db; +#[cfg(not(feature = "db-redis"))] +mod memory; +#[cfg(not(feature = "db-redis"))] use crate::memory::MemoryBackend; -use crate::{db::Db, hub_service::check_heartbeat}; + +use crate::hub_service::check_heartbeat; fn initialize_tracing(level: tracing::Level) { use tracing_subscriber::{filter::targets::Targets, prelude::*}; @@ -59,6 +87,7 @@ fn initialize_tracing(level: tracing::Level) { .init(); } +#[cfg(feature = "auth")] async fn extract_claims( mut request: ServiceRequest, next: Next, @@ -68,40 +97,35 @@ async fn extract_claims( token: Option, } - if !CONFIG.no_authorization { - let query = request.extract::>().await?.into_inner(); + let query = request.extract::>().await?.into_inner(); - let claims = if let Some(token) = query.token { - Claims::from_token(token, CONFIG.token_secret.expose_secret()).unwrap() - } else { - request.extract_claims(&CONFIG.token_secret)? - }; - request.extensions_mut().insert(claims); - } + let claims = if let Some(token) = query.token { + Claims::from_token(token, CONFIG.token_secret.expose_secret()).unwrap() + } else { + request.extract_claims(&CONFIG.token_secret)? + }; + request.extensions_mut().insert(claims); next.call(request).await } +#[cfg(feature = "auth")] async fn check_workspace( mut request: ServiceRequest, next: Next, ) -> Result, Error> { - if !CONFIG.no_authorization { - let workspace = Uuid::parse_str(&request.extract::>().await?); - let claims = request.extensions().get::().cloned().unwrap(); + let workspace = Uuid::parse_str(&request.extract::>().await?); + let claims = request.extensions().get::().cloned().unwrap(); - if claims.is_system() || Ok(claims.workspace.clone()) == workspace.clone().map(Some) { - next.call(request).await - } else { - warn!( - expected = ?claims.workspace, - actual = ?workspace, - "Unauthorized request, workspace mismatch" - ); - Err(actix_web::error::ErrorUnauthorized("Unauthorized").into()) - } - } else { + if claims.is_system() || Ok(claims.workspace.clone()) == workspace.clone().map(Some) { next.call(request).await + } else { + warn!( + expected = ?claims.workspace, + actual = ?workspace, + "Unauthorized request, workspace mismatch" + ); + Err(actix_web::error::ErrorUnauthorized("Unauthorized").into()) } } @@ -117,29 +141,48 @@ async fn main() -> anyhow::Result<()> { // starting heartbeat checker check_heartbeat(hub_state.clone()); - let db_backend = match CONFIG.backend { - config::BackendType::Memory => { - let memory = MemoryBackend::new(); - memory.spawn_ticker(hub_state.clone()); - tracing::info!("Memory mode enabled"); - Db::new_memory(memory, hub_state.clone()) + let db_backend = { + #[cfg(feature = "db-redis")] + { + let redis_client = redis::client().await?; + let db_connection = redis_client + .get_multiplexed_async_connection() + .await + .map_err(|e| { + tracing::error!( + "REDIS not found: {:?}", + &CONFIG + .redis_urls + .iter() + .map(|u| u.as_str()) + .collect::>() + .join(", ") + ); + e + })?; + tokio::spawn(crate::redis::receiver(redis_client, hub_state.clone())); + Db::new_db(db_connection) } - config::BackendType::Redis => { - let redis_client = redis::client().await?; - let redis_connection = redis_client.get_multiplexed_async_connection().await?; - tokio::spawn(crate::redis::receiver(redis_client, hub_state.clone())); - tracing::info!("Redis mode enabled"); - Db::new_redis(redis_connection, hub_state.clone()) + #[cfg(not(feature = "db-redis"))] + { + let db_connection = MemoryBackend::new(); + db_connection.spawn_ticker(hub_state.clone()); + Db::new_db(db_connection, hub_state.clone()) } }; + tracing::info!("DB mode: {}", BACKEND); + let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port); let url = format!("http://{}:{}", &CONFIG.bind_host, &CONFIG.bind_port); tracing::info!("Server running at {}", &url); - tracing::info!("HTTP API: {}/api", &url); - tracing::info!("WebSocket API: {}/ws", &url); + tracing::info!("API: {}/api", &url); + tracing::info!( + "WS: {}/ws", + format!("ws://{}:{}", &CONFIG.bind_host, &CONFIG.bind_port) + ); tracing::info!("Status: {}/status", &url); use std::sync::Arc; @@ -153,26 +196,49 @@ async fn main() -> anyhow::Result<()> { .supports_credentials() .max_age(3600); - App::new() - .app_data(web::Data::new(db_backend.clone())) - .app_data(web::Data::new(hub_state.clone())) - .wrap(middleware::Logger::default()) - .wrap(cors) - .service( + let api_scope = { + #[cfg(feature = "auth")] + { web::scope("/api/{workspace}") .wrap(middleware::from_fn(check_workspace)) .wrap(middleware::from_fn(extract_claims)) .route("/{key:.+/}", web::get().to(handlers_http::list)) .route("/{key:.+}", web::get().to(handlers_http::get)) .route("/{key:.+}", web::put().to(handlers_http::put)) - .route("/{key:.+}", web::delete().to(handlers_http::delete)), - ) + .route("/{key:.+}", web::delete().to(handlers_http::delete)) + } + + #[cfg(not(feature = "auth"))] + { + web::scope("/api/{workspace}") + .route("/{key:.+/}", web::get().to(handlers_http::list)) + .route("/{key:.+}", web::get().to(handlers_http::get)) + .route("/{key:.+}", web::put().to(handlers_http::put)) + .route("/{key:.+}", web::delete().to(handlers_http::delete)) + } + }; + + let ws_route = { + let r = web::get().to(handlers_ws::handler); + + #[cfg(feature = "auth")] + let r = r.wrap(middleware::from_fn(extract_claims)); + + r + }; + + App::new() + .app_data(web::Data::new(db_backend.clone())) + .app_data(web::Data::new(hub_state.clone())) + .wrap(middleware::Logger::default()) + .wrap(cors) + .service(api_scope) .route( "/ws", - web::get() - .to(handlers_ws::handler) - .wrap(middleware::from_fn(extract_claims)), - ) // WebSocket + ws_route, // web::get() + // .to(handlers_ws::handler) + // .wrap(middleware::from_fn(extract_claims)), + ) .route( "/status", web::get().to({ @@ -186,7 +252,7 @@ async fn main() -> anyhow::Result<()> { let count = hub_state.read().await.count(); Ok::<_, actix_web::Error>(HttpResponse::Ok().json(json!({ "memory_info": info, - "backend": CONFIG.backend.to_string().to_lowercase(), + "backend": BACKEND, "websockets": count, "status": "OK", "version": env!("CARGO_PKG_VERSION"), diff --git a/src/memory.rs b/src/memory.rs index 4774f36980..5003f2aa57 100644 --- a/src/memory.rs +++ b/src/memory.rs @@ -15,8 +15,8 @@ use crate::{ config::CONFIG, + db::{DbArray, DbResult, SaveMode, Ttl, deprecated_symbol_error, error}, hub_service::{HubState, RedisEvent, RedisEventAction, broadcast_event}, - redis::{RedisArray, SaveMode, Ttl, deprecated_symbol_error, error}, }; use std::{ collections::HashMap, @@ -95,10 +95,7 @@ impl MemoryBackend { } /// memory_list(&backend, "prefix/") → Vec -pub async fn memory_list( - backend: &MemoryBackend, - key_prefix: &str, -) -> redis::RedisResult> { +pub async fn memory_list(backend: &MemoryBackend, key_prefix: &str) -> DbResult> { deprecated_symbol_error(key_prefix)?; if !key_prefix.ends_with('/') { return error(412, "Key must end with slash"); @@ -125,7 +122,7 @@ pub async fn memory_list( let expires = v.ttl.wrapping_sub(current_tick); - results.push(RedisArray { + results.push(DbArray { key: k.clone(), data: v.data.clone(), ttl: expires as u64, @@ -137,7 +134,7 @@ pub async fn memory_list( } /// memory_info(&backend) -pub async fn memory_info(backend: &MemoryBackend) -> redis::RedisResult { +pub async fn memory_info(backend: &MemoryBackend) -> DbResult { let map = backend.inner.read().await; let keys = map.len(); let memory: usize = map.values().map(|v| v.data.len()).sum(); @@ -145,10 +142,7 @@ pub async fn memory_info(backend: &MemoryBackend) -> redis::RedisResult } /// memory_read(&backend, "key") -pub async fn memory_read( - backend: &MemoryBackend, - key: &str, -) -> redis::RedisResult> { +pub async fn memory_read(backend: &MemoryBackend, key: &str) -> DbResult> { deprecated_symbol_error(key)?; if key.ends_with('/') { return error(412, "Key must not end with a slash"); @@ -163,7 +157,7 @@ pub async fn memory_read( let current_tick = *backend.tick.read().await; let expires = entry.ttl.wrapping_sub(current_tick); - Ok(Some(RedisArray { + Ok(Some(DbArray { key: key.to_string(), data: data.clone(), ttl: expires as u64, @@ -174,7 +168,7 @@ pub async fn memory_read( } /// TTL in sec -fn compute_ttl_u8(ttl: Option) -> redis::RedisResult { +fn compute_ttl_u8(ttl: Option) -> DbResult { let sec_usize = match ttl { Some(Ttl::Sec(secs)) => secs, Some(Ttl::At(timestamp)) => { @@ -209,7 +203,7 @@ pub async fn memory_save>( bytes_value: V, ttl: Option, mode: Option, -) -> redis::RedisResult<()> { +) -> DbResult<()> { // u8 - String let value = match std::str::from_utf8(bytes_value.as_ref()) { Ok(s) => s.to_string(), @@ -300,7 +294,7 @@ pub async fn memory_delete( backend: &MemoryBackend, key: &str, mode: Option, -) -> redis::RedisResult { +) -> DbResult { deprecated_symbol_error(key)?; if key.ends_with('/') { return error(412, "Key must not end with a slash"); diff --git a/src/redis.rs b/src/redis.rs index cb230e247e..78c5775edd 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -25,66 +25,39 @@ use tracing::*; use crate::{ config::{CONFIG, RedisMode}, - hub_service::{HubState, RedisEvent, RedisEventAction, push_event}, + db::{DbArray, DbResult, SaveMode, Ttl, deprecated_symbol_error, error}, + hub_service::{HubState, RedisEvent, RedisEventAction}, }; -#[derive(serde::Serialize)] -pub enum Ttl { - Sec(usize), // EX - At(u64), // EXAT (timestamp in seconds) -} - -#[derive(Debug)] -pub enum SaveMode { - Upsert, // default: set or overwrite - Insert, // only if not exists (NX) - Update, // only if exists (XX) - Equal(String), // only if md5 matches provided -} - use redis::{ Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, RedisResult, ToRedisArgs, aio::MultiplexedConnection, }; -use serde::Serialize; +// use serde::Serialize; + +use crate::hub_service::broadcast_event; static MAX_LOOP_COUNT: usize = 1000; // to avoid infinite loops -#[derive(Debug, Serialize)] -pub struct RedisArray { - pub key: String, - pub data: String, - pub ttl: u64, // sec to expire TTL - pub etag: String, // md5 hash (data) -} - -/// return Error -pub fn error(code: u16, msg: impl Into) -> redis::RedisResult { - let msg = msg.into(); - let full = format!("{}: {}", code, msg); - Err(redis::RedisError::from(( - redis::ErrorKind::ExtensionError, - "", - full, - ))) -} - -/// Check for redis-deprecated symbols -pub fn deprecated_symbol(s: &str) -> bool { - s.chars().any(|c| { - matches!( - c, - '*' | '?' | '[' | ']' | '\\' | '\0'..='\x1F' | '\x7F' | '"' | '\'' - ) - }) -} - -pub fn deprecated_symbol_error(s: &str) -> redis::RedisResult<()> { - if deprecated_symbol(s) { - error(412, "Deprecated symbol in key") - } else { - Ok(()) +pub async fn push_event( + hub_state: &Arc>, + redis: &mut MultiplexedConnection, + ev: RedisEvent, +) { + // Value only for Set + let mut value: Option = None; + if matches!(ev.message, RedisEventAction::Set) { + match ::redis::cmd("GET") + .arg(&ev.key) + .query_async::>(redis) + .await + { + Ok(v) => value = v, + Err(e) => tracing::warn!("redis GET {} failed: {}", &ev.key, e), + } } + + broadcast_event(hub_state, ev, value).await; } /// redis_info(&connection) @@ -121,7 +94,7 @@ pub async fn redis_info(conn: &mut MultiplexedConnection) -> redis::RedisResult< pub async fn redis_list( conn: &mut MultiplexedConnection, key: &str, -) -> redis::RedisResult> { +) -> redis::RedisResult> { deprecated_symbol_error(key)?; if !key.ends_with('/') { return error(412, "Key must end with slash"); @@ -154,7 +127,7 @@ pub async fn redis_list( // Get TTL let ttl: i64 = redis::cmd("TTL").arg(&k).query_async(conn).await?; if ttl >= 0 { - results.push(RedisArray { + results.push(DbArray { key: k, data: value.clone(), ttl: ttl as u64, @@ -176,7 +149,7 @@ pub async fn redis_list( pub async fn redis_read( conn: &mut MultiplexedConnection, key: &str, -) -> redis::RedisResult> { +) -> redis::RedisResult> { deprecated_symbol_error(key)?; if key.ends_with('/') { @@ -197,7 +170,7 @@ pub async fn redis_read( _ => {} // ttl >= 0, ок } - Ok(Some(RedisArray { + Ok(Some(DbArray { key: key.to_string(), data: data.clone(), ttl: ttl as u64, @@ -212,7 +185,7 @@ pub async fn redis_save( value: T, ttl: Option, mode: Option, -) -> RedisResult<()> { +) -> DbResult<()> { deprecated_symbol_error(&key)?; if key.ends_with('/') { diff --git a/src/workspace_owner.rs b/src/workspace_owner.rs index d477abfef6..794dbef1a2 100644 --- a/src/workspace_owner.rs +++ b/src/workspace_owner.rs @@ -19,17 +19,17 @@ use serde_json::json; use std::{fs, path::Path, sync::LazyLock}; use uuid::Uuid; -use crate::{config::CONFIG, redis::deprecated_symbol}; +use crate::{config::CONFIG, db::deprecated_symbol}; // common checker + pub fn check_workspace_core(claims_opt: Option, key: &str) -> Result<(), &'static str> { if deprecated_symbol(key) { return Err("Invalid key: deprecated symbols"); } - if CONFIG.no_authorization { - return Ok(()); - } + #[cfg(not(feature = "auth"))] + return Ok(()); let claims = claims_opt.ok_or("Missing authorization")?; From 5abd5139a20faa608c26538d3653a0b3c018ef27 Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Thu, 27 Nov 2025 10:12:26 +0200 Subject: [PATCH 2/7] fix: comments Signed-off-by: Leonid Kaganov --- .gitignore | 2 ++ Cargo.lock | 2 +- Cargo.toml | 2 +- src/db.rs | 13 ------------- src/hub_service.rs | 21 --------------------- src/pulse-status.sh | 1 - 6 files changed, 4 insertions(+), 37 deletions(-) delete mode 100755 src/pulse-status.sh diff --git a/.gitignore b/.gitignore index 8ffa5d5078..e980812ead 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,7 @@ commit.sh bloat.sh /src/GO.sh /src/GOT.sh +pulse-status.sh GO.sh TEST.sh TEST_WS.sh @@ -18,3 +19,4 @@ DOCKER.sh /lleo /client /scripts + diff --git a/Cargo.lock b/Cargo.lock index 0ae060db12..d7b7478d27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.3.2" +version = "0.3.3" dependencies = [ "actix-cors", "actix-web", diff --git a/Cargo.toml b/Cargo.toml index b43d69568e..ae88a323ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.3.2" +version = "0.3.3" edition = "2024" [dependencies] diff --git a/src/db.rs b/src/db.rs index 72d07962aa..b33836d7b2 100644 --- a/src/db.rs +++ b/src/db.rs @@ -119,19 +119,6 @@ pub struct Db { } impl Db { - // pub fn new_db(hub: Arc>) -> Self { - // Self { - // inner: DbInner::Memory(m), - // hub, - // } - // } - // pub fn new_redis(c: MultiplexedConnection, hub: Arc>) -> Self { - // Self { - // inner: DbInner::Redis(c), - // hub, - // } - // } - pub fn new_db( #[cfg(not(feature = "db-redis"))] db: MemoryBackend, #[cfg(feature = "db-redis")] db: MultiplexedConnection, diff --git a/src/hub_service.rs b/src/hub_service.rs index 4ddc8b1d38..7440f44215 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -174,27 +174,6 @@ pub async fn broadcast_event( } } -// pub async fn push_event( -// hub_state: &Arc>, -// redis: &mut MultiplexedConnection, -// ev: RedisEvent, -// ) { -// // Value only for Set -// let mut value: Option = None; -// if matches!(ev.message, RedisEventAction::Set) { -// match ::redis::cmd("GET") -// .arg(&ev.key) -// .query_async::>(redis) -// .await -// { -// Ok(v) => value = v, -// Err(e) => tracing::warn!("redis GET {} failed: {}", &ev.key, e), -// } -// } - -// broadcast_event(hub_state, ev, value).await; -// } - pub fn check_heartbeat(hub_state: Arc>) { tokio::spawn(async move { let mut ticker = tokio::time::interval(std::time::Duration::from_secs(2)); diff --git a/src/pulse-status.sh b/src/pulse-status.sh deleted file mode 100755 index cadc091b2d..0000000000 --- a/src/pulse-status.sh +++ /dev/null @@ -1 +0,0 @@ -curl -i https://pulse.hc.engineering/status \ No newline at end of file From 7624cd768c5b4eef3fcdd205bb894ad2acfc1131 Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Thu, 27 Nov 2025 21:50:28 +0200 Subject: [PATCH 3/7] fix: correctly dropping connections in timeout Signed-off-by: Leonid Kaganov --- Cargo.lock | 3 +- Cargo.toml | 3 +- src/config.rs | 2 + src/config/default.toml | 3 + src/handlers_ws.rs | 181 ++++++++++++++++++++-------------------- src/hub_service.rs | 87 ++++++++++++++----- src/main.rs | 45 ++++------ 7 files changed, 180 insertions(+), 144 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d7b7478d27..28d6e8a2da 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1175,13 +1175,14 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.3.3" +version = "0.3.6" dependencies = [ "actix-cors", "actix-web", "actix-ws", "anyhow", "config", + "futures", "futures-util", "hex", "hulyrs", diff --git a/Cargo.toml b/Cargo.toml index ae88a323ce..558e61e276 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.3.3" +version = "0.3.6" edition = "2024" [dependencies] @@ -21,6 +21,7 @@ hex = "0.4.3" tokio-stream = "0.1" strum = { version = "0.27.2", features = ["derive"] } futures-util = "0.3" +futures = "0.3" # auth: regorus = { version = "0.5.0", optional = true } diff --git a/src/config.rs b/src/config.rs index b99e3a9e95..50f555c117 100644 --- a/src/config.rs +++ b/src/config.rs @@ -71,6 +71,8 @@ pub struct Config { #[cfg(feature = "auth")] pub policy_file: Option, + + pub loglevel: String, } pub static CONFIG: LazyLock = LazyLock::new(|| { diff --git a/src/config/default.toml b/src/config/default.toml index dd35595557..46779e7197 100644 --- a/src/config/default.toml +++ b/src/config/default.toml @@ -13,7 +13,10 @@ max_ttl = 3600 heartbeat_timeout = 90 ping_timeout = 30 +loglevel = "INFO" + # optional settings # max_size = 100 # permit_file = "/home/user/hulipulse/permit.rego" + diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index acf73539cc..05bc8015fc 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -15,7 +15,9 @@ use actix_ws; use futures_util::StreamExt; -// use tracing::info; + +use futures::future::{AbortHandle, Abortable}; + use actix_web::{Error, HttpRequest, HttpResponse, web}; #[cfg(feature = "auth")] @@ -27,7 +29,6 @@ use std::sync::Arc; use tokio::sync::RwLock; use crate::{ - BACKEND, db::{Db, SaveMode, Ttl}, hub_service::{HubState, SessionId, new_session_id}, }; @@ -38,7 +39,6 @@ use crate::workspace_owner::check_workspace_core; #[cfg(feature = "auth")] use crate::workspace_owner::test_rego_claims; -// use strum_macros::AsRefStr; use strum::AsRefStr; #[derive(Deserialize, Debug, AsRefStr)] @@ -160,19 +160,9 @@ async fn handle_command( match cmd { // INFO WsCommand::Info { correlation } => { - tracing::info!("INFO"); - match db.info().await { - Ok(info) => { - let info = json!({ - "memory_info": info, - "backend": BACKEND, - "websockets": hub_state.read().await.count(), - "version": env!("CARGO_PKG_VERSION"), - }); - result(info, &correlation, ws).await - } - Err(e) => result_err(e.to_string(), &correlation, ws).await, - } + tracing::debug!("INFO"); + let info = hub_state.read().await.info_json(db).await; + result(info, &correlation, ws).await } // PUT @@ -185,7 +175,7 @@ async fn handle_command( if_none_match, correlation, } => { - tracing::info!("PUT {} = {}", &key, &data); + tracing::debug!("PUT {} = {}", &key, &data); #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { @@ -229,7 +219,7 @@ async fn handle_command( correlation, if_match, } => { - tracing::info!("DELETE {}", &key); // correlation:{:?} , &correlation + tracing::debug!("DELETE {}", &key); // correlation:{:?} , &correlation #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { @@ -257,7 +247,7 @@ async fn handle_command( } WsCommand::Get { key, correlation } => { - tracing::info!("GET {}", &key); + tracing::debug!("GET {}", &key); #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { @@ -275,7 +265,7 @@ async fn handle_command( } WsCommand::List { key, correlation } => { - tracing::info!("LIST {:?}", &key); + tracing::debug!("LIST {:?}", &key); #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { @@ -292,7 +282,7 @@ async fn handle_command( } WsCommand::Sub { key, correlation } => { - tracing::info!("SUB {}", &key); + tracing::debug!("SUB {}", &key); #[cfg(feature = "auth")] if wrong_workspace(&claims, &key, &correlation, ws).await { @@ -304,7 +294,7 @@ async fn handle_command( } WsCommand::Unsub { key, correlation } => { - tracing::info!("UNSUB {}", &key); + tracing::debug!("UNSUB {}", &key); if key == "*" { hub_state.write().await.unsubscribe_all(session_id); result("OK", &correlation, ws).await; @@ -320,7 +310,7 @@ async fn handle_command( } WsCommand::Sublist { correlation } => { - tracing::info!("SUBLIST"); + tracing::debug!("SUBLIST"); // w/o Check workspace! let keys = hub_state.read().await.subscribe_list(session_id); result(keys, &correlation, ws).await; @@ -347,91 +337,100 @@ pub async fn handler( let session_id = new_session_id(); - hub_state.write().await.connect(session_id, session.clone()); - tracing::info!("WebSocket connected: {}", session_id); + let (abort_handle, abort_reg) = AbortHandle::new_pair(); - actix_web::rt::spawn(async move { - while let Some(Ok(msg)) = msg_stream.next().await { - if !matches!(msg, actix_ws::Message::Pong(_)) { - tracing::debug!("WebSocket message: {:?}", msg); - } + hub_state + .write() + .await + .connect(session_id, session.clone(), abort_handle); + tracing::debug!("WebSocket connected: {}", session_id); - // renew heartbeat to unixtime (all messages is activity, including "ping") - hub_state.write().await.renew_heartbeat(session_id); - - match msg { - actix_ws::Message::Ping(bytes) => { - session.pong(&bytes).await.ok(); - continue; + actix_web::rt::spawn(Abortable::new( + async move { + while let Some(Ok(msg)) = msg_stream.next().await { + if !matches!(msg, actix_ws::Message::Pong(_)) { + tracing::debug!("WebSocket message: {:?}", msg); } - actix_ws::Message::Pong(_) => { - continue; - } + // renew heartbeat to unixtime (all messages is activity, including "ping") + hub_state.write().await.renew_heartbeat(session_id); - actix_ws::Message::Text(text) if text == "ping" => { - let _ = session.text("pong").await; - continue; - } - actix_ws::Message::Text(text) if text == "pong" => { - continue; - } + match msg { + actix_ws::Message::Ping(bytes) => { + session.pong(&bytes).await.ok(); + continue; + } - actix_ws::Message::Text(text) => match serde_json::from_str::(&text) { - Ok(cmd) => { - #[cfg(feature = "auth")] - { - let key = match &cmd { - WsCommand::Put { key, .. } - | WsCommand::Delete { key, .. } - | WsCommand::Get { key, .. } - | WsCommand::List { key, .. } - | WsCommand::Sub { key, .. } - | WsCommand::Unsub { key, .. } => key.as_str(), - _ => "", - }; + actix_ws::Message::Pong(_) => { + continue; + } - if let Some(ref claim) = claims { - if !test_rego_claims(claim, cmd.as_ref(), key) { - let _ = session.text("Unauthorized: Rego policy").await; - break; + actix_ws::Message::Text(text) if text == "ping" => { + let _ = session.text("pong").await; + continue; + } + actix_ws::Message::Text(text) if text == "pong" => { + continue; + } + + actix_ws::Message::Text(text) => match serde_json::from_str::(&text) + { + Ok(cmd) => { + #[cfg(feature = "auth")] + { + let key = match &cmd { + WsCommand::Put { key, .. } + | WsCommand::Delete { key, .. } + | WsCommand::Get { key, .. } + | WsCommand::List { key, .. } + | WsCommand::Sub { key, .. } + | WsCommand::Unsub { key, .. } => key.as_str(), + _ => "", + }; + + if let Some(ref claim) = claims { + if !test_rego_claims(claim, cmd.as_ref(), key) { + let _ = session.text("Unauthorized: Rego policy").await; + break; + } } } + + handle_command( + &mut session, + cmd, + &db, + &hub_state, + #[cfg(feature = "auth")] + claims.clone(), + session_id, + ) + .await; } - handle_command( - &mut session, - cmd, - &db, - &hub_state, - #[cfg(feature = "auth")] - claims.clone(), - session_id, - ) - .await; + Err(err) => { + let _ = session.text(format!("Invalid JSON: {}", err)).await; + } + }, + + actix_ws::Message::Close(reason) => { + if let Err(e) = session.close(reason).await { + tracing::warn!("WS close error: {:?}", e); + } + break; } - Err(err) => { - let _ = session.text(format!("Invalid JSON: {}", err)).await; + _ => { + tracing::warn!("Unhandled WS message: {:?}", msg); } - }, - - actix_ws::Message::Close(reason) => { - if let Err(e) = session.close(reason).await { - tracing::warn!("WS close error: {:?}", e); - } - break; - } - - _ => { - tracing::info!("Unhandled WS message: {:?}", msg); } } - } - hub_state.write().await.disconnect(session_id); - tracing::info!("WebSocket disconnected: {}", session_id); - }); + hub_state.write().await.disconnect(session_id); + tracing::debug!("WebSocket disconnected by client: {}", session_id); + }, + abort_reg, + )); Ok(response) } diff --git a/src/hub_service.rs b/src/hub_service.rs index 7440f44215..e716634881 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -15,15 +15,16 @@ use crate::config::CONFIG; -// #[cfg(feature = "db-redis")] -// use redis::aio::MultiplexedConnection; - use serde::Serialize; use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::sync::RwLock; +use serde_json::{Value, json}; + +use crate::{BACKEND, db::Db}; + fn subscription_matches(sub_key: &str, key: &str) -> bool { if sub_key == key { return true; @@ -73,8 +74,11 @@ pub struct HubState { subs: HashMap>, heartbeats: HashMap, serverping: HashMap, + abort_handles: HashMap, } +use futures::future::AbortHandle; + impl HubState { pub fn renew_heartbeat(&mut self, session_id: SessionId) { if self.sessions.contains_key(&session_id) { @@ -84,22 +88,34 @@ impl HubState { } } - pub fn connect(&mut self, session_id: SessionId, session: actix_ws::Session) { + pub fn connect( + &mut self, + session_id: SessionId, + session: actix_ws::Session, + abort_handle: AbortHandle, + ) { self.sessions.insert(session_id, session); self.heartbeats .insert(session_id, std::time::Instant::now()); self.serverping .insert(session_id, std::time::Instant::now()); + self.abort_handles.insert(session_id, abort_handle); } pub fn disconnect(&mut self, session_id: SessionId) { self.sessions.remove(&session_id); self.heartbeats.remove(&session_id); self.serverping.remove(&session_id); + self.abort_handles.remove(&session_id); self.subs.retain(|_, ids| { ids.remove(&session_id); !ids.is_empty() }); + tracing::debug!( + "hub.disconnected {}, all: {}", + session_id, + self.sessions.len() + ); } pub fn subscribe(&mut self, session_id: SessionId, key: String) { @@ -134,8 +150,20 @@ impl HubState { .collect() } - pub fn count(&self) -> usize { - self.sessions.len() + pub async fn info_json(&self, db: &Db) -> Value { + let info = db.info().await.unwrap_or_else(|_| "error".to_string()); + json!({ + "memory_info": info, + "backend": BACKEND, + "websockets": self.sessions.len(), + "subscriptions": self.subs.len(), + "heartbeats": self.heartbeats.len(), + "serverping": self.serverping.len(), + "loops": self.abort_handles.len(), + "loglevel": &CONFIG.loglevel, + "status": "OK", + "version": env!("CARGO_PKG_VERSION"), + }) } pub fn recipients_for_key(&self, key: &str) -> Vec { @@ -168,7 +196,6 @@ pub async fn broadcast_event( // Send let payload = ServerMessage { event: ev, value }; for mut rcpt in recipients { - // let _ = rcpt.do_send(payload.clone()); let json = serde_json::to_string(&payload).unwrap(); let _ = rcpt.text(json).await; } @@ -186,19 +213,22 @@ pub fn check_heartbeat(hub_state: Arc>) { let hub = hub_state.read().await; - let expired: Vec = hub + let ids_expired: Vec = hub .heartbeats .iter() - .filter_map(|(&sid, &last_beat)| { - if last_beat < timelimit { - hub.sessions.get(&sid).cloned() - } else { - None - } - }) + .filter_map( + |(&sid, &last)| { + if last < timelimit { Some(sid) } else { None } + }, + ) .collect(); - let to_ping: Vec = hub + let expired_sessions: Vec = ids_expired + .iter() + .filter_map(|sid| hub.sessions.get(sid).cloned()) + .collect(); + + let ids_to_ping: Vec = hub .serverping .iter() .filter_map(|(&sid, &last_ping)| { @@ -212,21 +242,32 @@ pub fn check_heartbeat(hub_state: Arc>) { drop(hub); - if !expired.is_empty() { - for addr in &expired { - // addr.do_send(crate::handlers_ws::ForceDisconnect); - let _ = addr.clone().close(None).await; - } + for session in &expired_sessions { + let _ = session.clone().close(None).await; } - if !to_ping.is_empty() { + if !ids_to_ping.is_empty() || !ids_expired.is_empty() { let mut hub = hub_state.write().await; - for sid in &to_ping { + + for sid in &ids_expired { + if let Some(abort_handle) = hub.abort_handles.get(sid) { + abort_handle.abort(); + } + tracing::debug!("WebSocket disconnected by timeout: {}", sid); + hub.disconnect(*sid); + } + + for sid in &ids_to_ping { + if ids_expired.contains(sid) { + continue; + } + if let Some(session) = hub.sessions.get_mut(sid) { let _ = session.ping(&[]).await; } hub.serverping.insert(*sid, now); } + drop(hub); } } diff --git a/src/main.rs b/src/main.rs index 47e261ecfb..6e6e20c4a6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -35,8 +35,6 @@ use hulyrs::services::jwt::{Claims, actix::ServiceRequestExt}; #[cfg(feature = "auth")] use secrecy::ExposeSecret; -use serde_json::json; - #[cfg(feature = "auth")] use tracing::*; @@ -56,11 +54,6 @@ mod workspace_owner; mod hub_service; use hub_service::HubState; -#[cfg(feature = "db-redis")] -pub const BACKEND: &str = "REDIS"; -#[cfg(not(feature = "db-redis"))] -pub const BACKEND: &str = "MEMORY"; - use config::CONFIG; mod db; @@ -73,17 +66,21 @@ use crate::memory::MemoryBackend; use crate::hub_service::check_heartbeat; +#[cfg(feature = "db-redis")] +pub const BACKEND: &str = "REDIS"; +#[cfg(not(feature = "db-redis"))] +pub const BACKEND: &str = "MEMORY"; + fn initialize_tracing(level: tracing::Level) { use tracing_subscriber::{filter::targets::Targets, prelude::*}; let filter = Targets::default() .with_target(env!("CARGO_BIN_NAME"), level) .with_target("actix", tracing::Level::WARN); - let format = tracing_subscriber::fmt::layer().compact(); tracing_subscriber::registry() .with(filter) - .with(format) + .with(tracing_subscriber::fmt::layer().compact()) .init(); } @@ -131,7 +128,14 @@ async fn check_workspace( #[actix_web::main] async fn main() -> anyhow::Result<()> { - initialize_tracing(tracing::Level::TRACE); + initialize_tracing(match CONFIG.loglevel.as_str() { + "TRACE" => tracing::Level::TRACE, // full + "DEBUG" => tracing::Level::DEBUG, // for developer + "INFO" => tracing::Level::INFO, // normal + "WARN" => tracing::Level::WARN, // something went wrong + "ERROR" => tracing::Level::ERROR, // serious error + _ => tracing::Level::ERROR, + }); tracing::info!("{}/{}", env!("CARGO_BIN_NAME"), env!("CARGO_PKG_VERSION")); @@ -233,30 +237,15 @@ async fn main() -> anyhow::Result<()> { .wrap(middleware::Logger::default()) .wrap(cors) .service(api_scope) - .route( - "/ws", - ws_route, // web::get() - // .to(handlers_ws::handler) - // .wrap(middleware::from_fn(extract_claims)), - ) + .route("/ws", ws_route) .route( "/status", web::get().to({ move |hub_state: web::Data>>, db_backend: web::Data| { let hub_state = hub_state.clone(); async move { - let info = db_backend - .info() - .await - .unwrap_or_else(|_| "error".to_string()); - let count = hub_state.read().await.count(); - Ok::<_, actix_web::Error>(HttpResponse::Ok().json(json!({ - "memory_info": info, - "backend": BACKEND, - "websockets": count, - "status": "OK", - "version": env!("CARGO_PKG_VERSION"), - }))) + let info = hub_state.read().await.info_json(&db_backend).await; + Ok::<_, actix_web::Error>(HttpResponse::Ok().json(info)) } } }), From 5a076fb3859fac7c10cee4f951320f49a85a92bf Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Thu, 27 Nov 2025 21:56:43 +0200 Subject: [PATCH 4/7] feature: loglevel in config Signed-off-by: Leonid Kaganov --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/main.rs | 28 +++++++++++++++------------- 3 files changed, 17 insertions(+), 15 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 28d6e8a2da..5ecd2e434c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.3.6" +version = "0.3.7" dependencies = [ "actix-cors", "actix-web", diff --git a/Cargo.toml b/Cargo.toml index 558e61e276..f80f449ad2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.3.6" +version = "0.3.7" edition = "2024" [dependencies] diff --git a/src/main.rs b/src/main.rs index 6e6e20c4a6..66f4b2bcc8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -71,15 +71,24 @@ pub const BACKEND: &str = "REDIS"; #[cfg(not(feature = "db-redis"))] pub const BACKEND: &str = "MEMORY"; -fn initialize_tracing(level: tracing::Level) { +fn initialize_tracing() { use tracing_subscriber::{filter::targets::Targets, prelude::*}; - let filter = Targets::default() - .with_target(env!("CARGO_BIN_NAME"), level) - .with_target("actix", tracing::Level::WARN); + let level = match CONFIG.loglevel.as_str() { + "TRACE" => tracing::Level::TRACE, // full + "DEBUG" => tracing::Level::DEBUG, // for developer + "INFO" => tracing::Level::INFO, // normal + "WARN" => tracing::Level::WARN, // something went wrong + "ERROR" => tracing::Level::ERROR, // serious error + _ => tracing::Level::ERROR, + }; tracing_subscriber::registry() - .with(filter) + .with( + Targets::new() + .with_target(env!("CARGO_BIN_NAME"), level) + .with_target("actix", tracing::Level::WARN), + ) .with(tracing_subscriber::fmt::layer().compact()) .init(); } @@ -128,14 +137,7 @@ async fn check_workspace( #[actix_web::main] async fn main() -> anyhow::Result<()> { - initialize_tracing(match CONFIG.loglevel.as_str() { - "TRACE" => tracing::Level::TRACE, // full - "DEBUG" => tracing::Level::DEBUG, // for developer - "INFO" => tracing::Level::INFO, // normal - "WARN" => tracing::Level::WARN, // something went wrong - "ERROR" => tracing::Level::ERROR, // serious error - _ => tracing::Level::ERROR, - }); + initialize_tracing(); tracing::info!("{}/{}", env!("CARGO_BIN_NAME"), env!("CARGO_PKG_VERSION")); From 139368235aa0d021321171ffc20c45083f24a029 Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Thu, 27 Nov 2025 22:02:58 +0200 Subject: [PATCH 5/7] fix: loglevel in config Signed-off-by: Leonid Kaganov --- src/main.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index 66f4b2bcc8..596c751fcf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -80,7 +80,7 @@ fn initialize_tracing() { "INFO" => tracing::Level::INFO, // normal "WARN" => tracing::Level::WARN, // something went wrong "ERROR" => tracing::Level::ERROR, // serious error - _ => tracing::Level::ERROR, + _ => tracing::Level::TRACE, }; tracing_subscriber::registry() @@ -184,6 +184,7 @@ async fn main() -> anyhow::Result<()> { let url = format!("http://{}:{}", &CONFIG.bind_host, &CONFIG.bind_port); tracing::info!("Server running at {}", &url); + tracing::info!("Log level: {}", &CONFIG.loglevel); tracing::info!("API: {}/api", &url); tracing::info!( "WS: {}/ws", From 1d007f0e023c2a583295d5baebc6f4bdb8e478ac Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Sat, 29 Nov 2025 01:40:17 +0200 Subject: [PATCH 6/7] features lopt: direct personal messages between websockets by username Signed-off-by: Leonid Kaganov --- Cargo.lock | 2 +- Cargo.toml | 5 ++-- src/handlers_ws.rs | 69 +++++++++++++++++++++++++++++++++++++++++++--- src/hub_service.rs | 34 +++++++++++++++++++++++ src/main.rs | 1 + src/redis.rs | 2 +- 6 files changed, 105 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5ecd2e434c..1668acc159 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.3.7" +version = "0.4.0" dependencies = [ "actix-cors", "actix-web", diff --git a/Cargo.toml b/Cargo.toml index f80f449ad2..0cdcc1233a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.3.7" +version = "0.4.0" edition = "2024" [dependencies] @@ -43,6 +43,7 @@ tokio-tungstenite = { version = "0.21", default-features = false, features = [ ] } [features] -default = ["db-redis","auth"] +default = ["db-redis","auth"] # lopt auth = ["regorus", "uuid", "hulyrs", "secrecy"] +lopt = [] db-redis = ["redis"] \ No newline at end of file diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 05bc8015fc..b836f7254d 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -33,6 +33,9 @@ use crate::{ hub_service::{HubState, SessionId, new_session_id}, }; +#[cfg(feature = "lopt")] +use crate::hub_service::send_to_name; + #[cfg(feature = "auth")] use crate::workspace_owner::check_workspace_core; @@ -44,6 +47,20 @@ use strum::AsRefStr; #[derive(Deserialize, Debug, AsRefStr)] #[serde(rename_all = "lowercase", tag = "type")] pub enum WsCommand { + #[cfg(feature = "lopt")] + Personal { + to: String, + correlation: String, + data: String, + }, + + #[cfg(feature = "lopt")] + Answer { + to: String, + correlation: String, + data: String, + }, + Put { #[serde(default = "default_corr")] correlation: String, @@ -156,8 +173,39 @@ async fn handle_command( hub_state: &Arc>, #[cfg(feature = "auth")] claims: Option, session_id: SessionId, + #[cfg(feature = "lopt")] client_name: &str, ) { match cmd { + #[cfg(feature = "lopt")] + WsCommand::Personal { + to, + correlation, + data, + } => { + use crate::hub_service::send_to_name; + + tracing::debug!("PERSONAL from {} to {}", &client_name, &to); + let payload = + json!({ "personal": client_name, "correlation": correlation, "data": data }); + if !send_to_name(&hub_state, &to, payload).await { + tracing::debug!("PERSONAL send from [{}] to [{}] failed", &client_name, &to); + result_err("failed", &correlation, ws).await; + } + } + + #[cfg(feature = "lopt")] + WsCommand::Answer { + to, + correlation, + data, + } => { + tracing::debug!("ANSWER from {} to {}", &client_name, &to); + let payload = json!({ "correlation": correlation, "data": data }); + if !send_to_name(&hub_state, &to, payload).await { + tracing::debug!("PERSONAL send_to failed: no such session {}", to); + } + } + // INFO WsCommand::Info { correlation } => { tracing::debug!("INFO"); @@ -333,16 +381,26 @@ pub async fn handler( .to_owned(), ); + #[cfg(feature = "lopt")] + let client_name = req + .match_info() + .get("client_name") + .unwrap_or("") + .to_string(); + let (response, mut session, mut msg_stream) = actix_ws::handle(&req, payload)?; let session_id = new_session_id(); let (abort_handle, abort_reg) = AbortHandle::new_pair(); - hub_state - .write() - .await - .connect(session_id, session.clone(), abort_handle); + hub_state.write().await.connect( + session_id, + session.clone(), + abort_handle, + #[cfg(feature = "lopt")] + client_name.clone(), + ); tracing::debug!("WebSocket connected: {}", session_id); actix_web::rt::spawn(Abortable::new( @@ -385,6 +443,7 @@ pub async fn handler( | WsCommand::List { key, .. } | WsCommand::Sub { key, .. } | WsCommand::Unsub { key, .. } => key.as_str(), + // | WsCommand::Personal { key, .. } => key.as_str(), _ => "", }; @@ -404,6 +463,8 @@ pub async fn handler( #[cfg(feature = "auth")] claims.clone(), session_id, + #[cfg(feature = "lopt")] + &client_name, ) .await; } diff --git a/src/hub_service.rs b/src/hub_service.rs index e716634881..137bdf5013 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -75,6 +75,11 @@ pub struct HubState { heartbeats: HashMap, serverping: HashMap, abort_handles: HashMap, + // client_ids: HashMap, + #[cfg(feature = "lopt")] + name_by_session: HashMap, + #[cfg(feature = "lopt")] + session_by_name: HashMap, } use futures::future::AbortHandle; @@ -93,6 +98,7 @@ impl HubState { session_id: SessionId, session: actix_ws::Session, abort_handle: AbortHandle, + #[cfg(feature = "lopt")] client_name: String, ) { self.sessions.insert(session_id, session); self.heartbeats @@ -100,6 +106,11 @@ impl HubState { self.serverping .insert(session_id, std::time::Instant::now()); self.abort_handles.insert(session_id, abort_handle); + + #[cfg(feature = "lopt")] + self.name_by_session.insert(session_id, client_name.clone()); + #[cfg(feature = "lopt")] + self.session_by_name.insert(client_name, session_id); } pub fn disconnect(&mut self, session_id: SessionId) { @@ -111,6 +122,12 @@ impl HubState { ids.remove(&session_id); !ids.is_empty() }); + + #[cfg(feature = "lopt")] + if let Some(client_id) = self.name_by_session.remove(&session_id) { + self.session_by_name.remove(&client_id); + } + tracing::debug!( "hub.disconnected {}, all: {}", session_id, @@ -201,6 +218,23 @@ pub async fn broadcast_event( } } +#[cfg(feature = "lopt")] +pub async fn send_to_name(hub_state: &Arc>, to: &str, payload: Value) -> bool { + let hub = hub_state.read().await; + + let to_sid = if let Some(&sid) = hub.session_by_name.get(to) { + sid + } else { + return false; + }; + + let Some(mut session) = hub.sessions.get(&to_sid).cloned() else { + return false; + }; + + session.text(payload.to_string()).await.is_ok() +} + pub fn check_heartbeat(hub_state: Arc>) { tokio::spawn(async move { let mut ticker = tokio::time::interval(std::time::Duration::from_secs(2)); diff --git a/src/main.rs b/src/main.rs index 596c751fcf..7221b2c9f9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -240,6 +240,7 @@ async fn main() -> anyhow::Result<()> { .wrap(middleware::Logger::default()) .wrap(cors) .service(api_scope) + .route("/ws/{client_name}", web::get().to(handlers_ws::handler)) .route("/ws", ws_route) .route( "/status", diff --git a/src/redis.rs b/src/redis.rs index 78c5775edd..37ea4bcdbd 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -433,7 +433,7 @@ pub async fn receiver( while let Some(message) = messages.next().await { match RedisEvent::try_from(message) { Ok(ev) => { - push_event(&hub_state, &mut redis, ev).await; + push_event(&hub_state, &mut redis, ev); // .await; } Err(e) => { warn!("invalid redis message: {e}"); From 75ebcdd61b32fc2f2e57894c5996259e71d40646 Mon Sep 17 00:00:00 2001 From: Artem Savchenko Date: Fri, 20 Feb 2026 14:10:00 +0700 Subject: [PATCH 7/7] Support in-memory mode for hulypulse Signed-off-by: Artem Savchenko --- Cargo.lock | 2 +- Cargo.toml | 7 +- README.md | 20 ++- src/config.rs | 15 +-- src/config/default.toml | 2 +- src/db.rs | 286 ++++++++++++++++++++++------------------ src/handlers_http.rs | 24 ++-- src/hub_service.rs | 6 +- src/main.rs | 41 +++--- src/redis.rs | 18 +-- tests/rest_api.rs | 7 +- 11 files changed, 233 insertions(+), 195 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1668acc159..52644883b8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.4.0" +version = "0.4.1" dependencies = [ "actix-cors", "actix-web", diff --git a/Cargo.toml b/Cargo.toml index 0cdcc1233a..50f1dafe32 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.4.0" +version = "0.4.1" edition = "2024" [dependencies] @@ -30,7 +30,7 @@ hulyrs = { git = "https://github.com/hcengineering/hulyrs.git", features = [ "ac secrecy = { version = "0.10.3", optional = true } #redis -redis = { version = "=0.32.5", features = ["aio", "tokio-comp", "sentinel"], optional = true } +redis = { version = "=0.32.5", features = ["aio", "tokio-comp", "sentinel"] } [[bin]] name = "hulypulse" @@ -43,7 +43,6 @@ tokio-tungstenite = { version = "0.21", default-features = false, features = [ ] } [features] -default = ["db-redis","auth"] # lopt +default = ["auth"] # lopt auth = ["regorus", "uuid", "hulyrs", "secrecy"] lopt = [] -db-redis = ["redis"] \ No newline at end of file diff --git a/README.md b/README.md index 90940eb91b..8e457bd91c 100644 --- a/README.md +++ b/README.md @@ -190,15 +190,14 @@ Size of data is limited to some reasonable size - `{"message":"Del","key":"00000000-0000-0000-0000-000000000001/foo/bar"}` ## Special options in config/default.toml - - ```memory_mode = true``` Use native memory storage instead Redis + - ```backend = "memory"``` Use native memory storage instead Redis - ```max_size = 100``` Max value size in bytes ## Special cargo build options - - "db-redis" (default) - use Redis (Memory instead) - "auth" (default) - use huly-authorization - Disable both: + Disable auth: cargo build --no-default-features - Enable one: + Enable auth: cargo build --no-default-features --features "auth" ## Running @@ -207,7 +206,17 @@ Pre-build docker images is available at: hardcoreeng/service_hulypulse:{tag}. You can use the following command to run the image locally: ```bash -docker run -p 8095:8095 -it --rm hardcoreeng/service_hulypulse:{tag}" +docker run -p 8095:8095 -it --rm hardcoreeng/service_hulypulse:{tag} +``` + +Run from source using Redis: +```bash +HULY_REDIS_URLS=redis://huly.local:6379 cargo run +``` + +Run from source in in-memory mode: +```bash +HULY_BACKEND=memory cargo run ``` If you want to run the service as a part of local huly development environment use the following command: @@ -228,6 +237,7 @@ The following environment variables are used to configure hulypulse: - ```HULY_BIND_HOST```: host to bind the server to (default: 0.0.0.0) - ```HULY_BIND_PORT```: port to bind the server to (default: 8094) - ```HULY_TOKEN_SECRET```: secret used to sign JWT tokens (default: secret) + - ```HULY_BACKEND```: storage backend "redis" or "memory" (default: "redis") - ```HULY_REDIS_URLS```: redis connection string (default: redis://huly.local:6379) - ```HULY_REDIS_PASSWORD```: redis password (default: "<invalid>") - ```HULY_REDIS_MODE```: redis mode "direct" or "sentinel" (default: "direct") diff --git a/src/config.rs b/src/config.rs index 50f555c117..e5e7f07440 100644 --- a/src/config.rs +++ b/src/config.rs @@ -19,12 +19,9 @@ use std::{path::Path, sync::LazyLock}; use secrecy::SecretString; use serde::Deserialize; -#[cfg(feature = "db-redis")] use serde_with::StringWithSeparator; -#[cfg(feature = "db-redis")] use serde_with::formats::CommaSeparator; use serde_with::serde_as; -#[cfg(feature = "db-redis")] use url::Url; use config::FileFormat; @@ -43,6 +40,10 @@ pub enum BackendType { Redis, } +fn default_backend() -> BackendType { + BackendType::Redis +} + #[serde_as] #[derive(Deserialize, Debug)] pub struct Config { @@ -52,20 +53,18 @@ pub struct Config { #[cfg(feature = "auth")] pub token_secret: SecretString, - #[cfg(feature = "db-redis")] + #[serde(default = "default_backend")] + pub backend: BackendType, + #[serde_as(as = "StringWithSeparator::")] pub redis_urls: Vec, - #[cfg(feature = "db-redis")] pub redis_password: String, - #[cfg(feature = "db-redis")] pub redis_mode: RedisMode, - #[cfg(feature = "db-redis")] pub redis_service: String, pub max_ttl: usize, pub max_size: Option, - // pub backend: BackendType, pub heartbeat_timeout: u64, pub ping_timeout: u64, diff --git a/src/config/default.toml b/src/config/default.toml index 46779e7197..844d97f781 100644 --- a/src/config/default.toml +++ b/src/config/default.toml @@ -3,6 +3,7 @@ bind_host = "0.0.0.0" token_secret = "secret" +backend = "redis" redis_urls = "redis://huly.local:6379" redis_password = "" redis_mode = "direct" @@ -19,4 +20,3 @@ loglevel = "INFO" # max_size = 100 # permit_file = "/home/user/hulipulse/permit.rego" - diff --git a/src/db.rs b/src/db.rs index b33836d7b2..006728d6fd 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,43 +1,38 @@ -#[cfg(not(feature = "db-redis"))] use std::sync::Arc; -#[cfg(not(feature = "db-redis"))] use crate::hub_service::{HubState, RedisEvent, RedisEventAction, broadcast_event}; - -#[cfg(not(feature = "db-redis"))] use crate::memory::{ MemoryBackend, memory_delete, memory_info, memory_list, memory_read, memory_save, }; - -#[cfg(feature = "db-redis")] use crate::redis::{redis_delete, redis_info, redis_list, redis_read, redis_save}; +use redis::aio::MultiplexedConnection; +use serde::Serialize; +use tokio::sync::RwLock; -#[cfg(feature = "db-redis")] -use ::redis::aio::MultiplexedConnection; - -#[cfg(feature = "db-redis")] -pub type DbError = redis::RedisError; - -#[cfg(not(feature = "db-redis"))] #[derive(Debug)] -pub struct DbError(pub String); +pub enum DbError { + Redis(redis::RedisError), + Message(String), +} pub type DbResult = Result; -#[cfg(not(feature = "db-redis"))] impl std::fmt::Display for DbError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) + match self { + Self::Redis(err) => write!(f, "{err}"), + Self::Message(msg) => write!(f, "{msg}"), + } } } -#[cfg(not(feature = "db-redis"))] impl std::error::Error for DbError {} -#[cfg(not(feature = "db-redis"))] -use tokio::sync::RwLock; - -use serde::Serialize; +impl From for DbError { + fn from(value: redis::RedisError) -> Self { + Self::Redis(value) + } +} #[derive(Debug, Serialize)] pub struct DbArray { @@ -61,33 +56,8 @@ pub enum SaveMode { Equal(String), // only if md5 matches provided } -/// return Error -// pub fn error(code: u16, msg: impl Into) -> DbResult { -// let msg = msg.into(); -// let full = format!("{}: {}", code, msg); -// Err(redis::RedisError::from(( -// redis::ErrorKind::ExtensionError, -// "", -// full, -// ))) -// } - pub fn error(code: u16, msg: impl Into) -> DbResult { - let msg = format!("{}: {}", code, msg.into()); - - #[cfg(feature = "db-redis")] - { - return Err(redis::RedisError::from(( - redis::ErrorKind::ExtensionError, - "", - msg, - ))); - } - - #[cfg(not(feature = "db-redis"))] - { - return Err(DbError(msg)); - } + Err(DbError::Message(format!("{}: {}", code, msg.into()))) } /// Check for redis-deprecated symbols @@ -108,59 +78,67 @@ pub fn deprecated_symbol_error(s: &str) -> DbResult<()> { } } +#[derive(Clone)] +enum DbBackend { + Redis(MultiplexedConnection), + Memory { + db: MemoryBackend, + hub: Arc>, + }, +} + #[derive(Clone)] pub struct Db { - #[cfg(feature = "db-redis")] - db: MultiplexedConnection, - #[cfg(not(feature = "db-redis"))] - db: MemoryBackend, - #[cfg(not(feature = "db-redis"))] - hub: Arc>, + backend: DbBackend, } impl Db { - pub fn new_db( - #[cfg(not(feature = "db-redis"))] db: MemoryBackend, - #[cfg(feature = "db-redis")] db: MultiplexedConnection, - #[cfg(not(feature = "db-redis"))] hub: Arc>, - ) -> Self { + pub fn new_redis(db: MultiplexedConnection) -> Self { Self { - db, - #[cfg(not(feature = "db-redis"))] - hub, + backend: DbBackend::Redis(db), + } + } + + pub fn new_memory(db: MemoryBackend, hub: Arc>) -> Self { + Self { + backend: DbBackend::Memory { db, hub }, + } + } + + pub fn mode(&self) -> &'static str { + match &self.backend { + DbBackend::Redis(_) => "redis", + DbBackend::Memory { .. } => "memory", } } pub async fn info(&self) -> DbResult { - #[cfg(not(feature = "db-redis"))] - return memory_info(&self.db).await; - - #[cfg(feature = "db-redis")] - { - let mut c = self.db.clone(); - redis_info(&mut c).await + match &self.backend { + DbBackend::Memory { db, .. } => memory_info(db).await, + DbBackend::Redis(conn) => { + let mut c = conn.clone(); + redis_info(&mut c).await + } } } pub async fn list(&self, key: &str) -> DbResult> { - #[cfg(not(feature = "db-redis"))] - return memory_list(&self.db, key).await; - - #[cfg(feature = "db-redis")] - { - let mut c = self.db.clone(); - redis_list(&mut c, key).await + match &self.backend { + DbBackend::Memory { db, .. } => memory_list(db, key).await, + DbBackend::Redis(conn) => { + let mut c = conn.clone(); + redis_list(&mut c, key).await + } } } pub async fn read(&self, key: &str) -> DbResult> { - #[cfg(not(feature = "db-redis"))] - return memory_read(&self.db, key).await; - - #[cfg(feature = "db-redis")] - { - let mut c = self.db.clone(); - redis_read(&mut c, key).await + match &self.backend { + DbBackend::Memory { db, .. } => memory_read(db, key).await, + DbBackend::Redis(conn) => { + let mut c = conn.clone(); + redis_read(&mut c, key).await + } } } @@ -171,54 +149,112 @@ impl Db { ttl: Option, mode: Option, ) -> DbResult<()> { - #[cfg(not(feature = "db-redis"))] - { - memory_save(&self.db, key, value.as_ref(), ttl, mode).await?; - // Send events - let value_str = std::str::from_utf8(value.as_ref()) - .ok() - .map(|s| s.to_string()); - broadcast_event( - &self.hub, - RedisEvent { - message: RedisEventAction::Set, - key: key.to_string(), - }, - value_str, - ) - .await; - return Ok(()); - } - - #[cfg(feature = "db-redis")] - { - let mut c = self.db.clone(); - redis_save(&mut c, key, value.as_ref(), ttl, mode).await + match &self.backend { + DbBackend::Memory { db, hub } => { + memory_save(db, key, value.as_ref(), ttl, mode).await?; + let value_str = std::str::from_utf8(value.as_ref()) + .ok() + .map(|s| s.to_string()); + broadcast_event( + hub, + RedisEvent { + message: RedisEventAction::Set, + key: key.to_string(), + }, + value_str, + ) + .await; + Ok(()) + } + DbBackend::Redis(conn) => { + let mut c = conn.clone(); + redis_save(&mut c, key, value.as_ref(), ttl, mode).await + } } } pub async fn delete(&self, key: &str, mode: Option) -> DbResult { - #[cfg(not(feature = "db-redis"))] - { - let deleted = memory_delete(&self.db, key, mode).await?; - if deleted { - broadcast_event( - &self.hub, - RedisEvent { - message: RedisEventAction::Del, - key: key.to_string(), - }, - None, - ) - .await; + match &self.backend { + DbBackend::Memory { db, hub } => { + let deleted = memory_delete(db, key, mode).await?; + if deleted { + broadcast_event( + hub, + RedisEvent { + message: RedisEventAction::Del, + key: key.to_string(), + }, + None, + ) + .await; + } + Ok(deleted) + } + DbBackend::Redis(conn) => { + let mut c = conn.clone(); + redis_delete(&mut c, key, mode).await } - return Ok(deleted); - } - - #[cfg(feature = "db-redis")] - { - let mut c = self.db.clone(); - redis_delete(&mut c, key, mode).await } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::hub_service::HubState; + use crate::memory::MemoryBackend; + use std::sync::Arc; + use tokio::sync::RwLock; + + fn memory_db() -> Db { + let hub = Arc::new(RwLock::new(HubState::default())); + let backend = MemoryBackend::new(); + Db::new_memory(backend, hub) + } + + #[tokio::test] + async fn memory_db_mode_and_crud_work() { + let db = memory_db(); + assert_eq!(db.mode(), "memory"); + + db.save("workspace/tests/key1", b"hello", Some(Ttl::Sec(60)), None) + .await + .expect("save should succeed"); + + let item = db + .read("workspace/tests/key1") + .await + .expect("read should succeed") + .expect("key should exist"); + assert_eq!(item.data, "hello"); + + let list = db + .list("workspace/tests/") + .await + .expect("list should succeed"); + assert_eq!(list.len(), 1); + assert_eq!(list[0].key, "workspace/tests/key1"); + + let deleted = db + .delete("workspace/tests/key1", None) + .await + .expect("delete should succeed"); + assert!(deleted); + assert!( + db.read("workspace/tests/key1") + .await + .expect("read should succeed") + .is_none() + ); + } + + #[tokio::test] + async fn memory_db_status_reports_memory_backend() { + let hub = Arc::new(RwLock::new(HubState::default())); + let db = Db::new_memory(MemoryBackend::new(), hub.clone()); + + let info = hub.read().await.info_json(&db).await; + assert_eq!(info["backend"], "memory"); + assert_eq!(info["status"], "OK"); + } +} diff --git a/src/handlers_http.rs b/src/handlers_http.rs index 5efabece51..ee6a8a9368 100644 --- a/src/handlers_http.rs +++ b/src/handlers_http.rs @@ -35,17 +35,19 @@ use crate::workspace_owner::test_rego_http; pub fn map_redis_error(err: impl std::fmt::Display) -> Error { let msg = err.to_string(); - if let Some(detail) = msg.split(" - ExtensionError: ").nth(1) { - if let Some((code, text)) = detail.split_once(": ") { - let text = format!("{} {}", code, text); - return match code { - "400" => actix_web::error::ErrorBadRequest(text), - "404" => actix_web::error::ErrorNotFound(text), - "412" => actix_web::error::ErrorPreconditionFailed(text), - "500" => actix_web::error::ErrorInternalServerError(text), - _ => actix_web::error::ErrorInternalServerError("unexpected error"), - }; - } + let detail = msg + .split(" - ExtensionError: ") + .nth(1) + .unwrap_or(msg.as_str()); + if let Some((code, text)) = detail.split_once(": ") { + let text = format!("{} {}", code, text); + return match code { + "400" => actix_web::error::ErrorBadRequest(text), + "404" => actix_web::error::ErrorNotFound(text), + "412" => actix_web::error::ErrorPreconditionFailed(text), + "500" => actix_web::error::ErrorInternalServerError(text), + _ => actix_web::error::ErrorInternalServerError("unexpected error"), + }; } actix_web::error::ErrorInternalServerError("internal error") } diff --git a/src/hub_service.rs b/src/hub_service.rs index 137bdf5013..0327a883ab 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -23,7 +23,7 @@ use tokio::sync::RwLock; use serde_json::{Value, json}; -use crate::{BACKEND, db::Db}; +use crate::db::Db; fn subscription_matches(sub_key: &str, key: &str) -> bool { if sub_key == key { @@ -54,10 +54,8 @@ pub fn new_session_id() -> SessionId { pub enum RedisEventAction { Set, Del, - #[cfg(feature = "db-redis")] Unlink, Expired, - #[cfg(feature = "db-redis")] Other(String), } @@ -171,7 +169,7 @@ impl HubState { let info = db.info().await.unwrap_or_else(|_| "error".to_string()); json!({ "memory_info": info, - "backend": BACKEND, + "backend": db.mode(), "websockets": self.sessions.len(), "subscriptions": self.subs.len(), "heartbeats": self.heartbeats.len(), diff --git a/src/main.rs b/src/main.rs index 7221b2c9f9..083d9365a9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -19,6 +19,8 @@ use actix_web::{ middleware::{self}, web::{self}, }; +use std::sync::Arc; +use tokio::sync::RwLock; #[cfg(feature = "auth")] use actix_web::{ @@ -45,7 +47,7 @@ mod config; mod handlers_http; mod handlers_ws; -#[cfg(feature = "db-redis")] +mod memory; mod redis; #[cfg(feature = "auth")] @@ -57,20 +59,12 @@ use hub_service::HubState; use config::CONFIG; mod db; +use crate::config::BackendType; use crate::db::Db; - -#[cfg(not(feature = "db-redis"))] -mod memory; -#[cfg(not(feature = "db-redis"))] use crate::memory::MemoryBackend; use crate::hub_service::check_heartbeat; -#[cfg(feature = "db-redis")] -pub const BACKEND: &str = "REDIS"; -#[cfg(not(feature = "db-redis"))] -pub const BACKEND: &str = "MEMORY"; - fn initialize_tracing() { use tracing_subscriber::{filter::targets::Targets, prelude::*}; @@ -147,9 +141,8 @@ async fn main() -> anyhow::Result<()> { // starting heartbeat checker check_heartbeat(hub_state.clone()); - let db_backend = { - #[cfg(feature = "db-redis")] - { + let db_backend = match &CONFIG.backend { + BackendType::Redis => { let redis_client = redis::client().await?; let db_connection = redis_client .get_multiplexed_async_connection() @@ -166,19 +159,24 @@ async fn main() -> anyhow::Result<()> { ); e })?; - tokio::spawn(crate::redis::receiver(redis_client, hub_state.clone())); - Db::new_db(db_connection) + tokio::spawn({ + let hub_state = hub_state.clone(); + async move { + if let Err(err) = crate::redis::receiver(redis_client, hub_state).await { + tracing::error!("Redis receiver stopped: {err}"); + } + } + }); + Db::new_redis(db_connection) } - - #[cfg(not(feature = "db-redis"))] - { + BackendType::Memory => { let db_connection = MemoryBackend::new(); db_connection.spawn_ticker(hub_state.clone()); - Db::new_db(db_connection, hub_state.clone()) + Db::new_memory(db_connection, hub_state.clone()) } }; - tracing::info!("DB mode: {}", BACKEND); + tracing::info!("DB mode: {}", db_backend.mode()); let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port); @@ -192,9 +190,6 @@ async fn main() -> anyhow::Result<()> { ); tracing::info!("Status: {}/status", &url); - use std::sync::Arc; - use tokio::sync::RwLock; - let server = HttpServer::new(move || { let cors = Cors::default() .allow_any_origin() diff --git a/src/redis.rs b/src/redis.rs index 37ea4bcdbd..4cb83fcac8 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -30,7 +30,7 @@ use crate::{ }; use redis::{ - Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, RedisResult, ToRedisArgs, + Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, ToRedisArgs, aio::MultiplexedConnection, }; // use serde::Serialize; @@ -61,7 +61,7 @@ pub async fn push_event( } /// redis_info(&connection) -pub async fn redis_info(conn: &mut MultiplexedConnection) -> redis::RedisResult { +pub async fn redis_info(conn: &mut MultiplexedConnection) -> DbResult { let info: String = redis::cmd("INFO").query_async(conn).await?; let mut redis_keys: Option = None; @@ -91,10 +91,7 @@ pub async fn redis_info(conn: &mut MultiplexedConnection) -> redis::RedisResult< } /// redis_list(&connection,prefix) -pub async fn redis_list( - conn: &mut MultiplexedConnection, - key: &str, -) -> redis::RedisResult> { +pub async fn redis_list(conn: &mut MultiplexedConnection, key: &str) -> DbResult> { deprecated_symbol_error(key)?; if !key.ends_with('/') { return error(412, "Key must end with slash"); @@ -146,10 +143,7 @@ pub async fn redis_list( } /// redis_read(&connection,key) -pub async fn redis_read( - conn: &mut MultiplexedConnection, - key: &str, -) -> redis::RedisResult> { +pub async fn redis_read(conn: &mut MultiplexedConnection, key: &str) -> DbResult> { deprecated_symbol_error(key)?; if key.ends_with('/') { @@ -306,7 +300,7 @@ pub async fn redis_delete( conn: &mut MultiplexedConnection, key: &str, mode: Option, -) -> RedisResult { +) -> DbResult { deprecated_symbol_error(key)?; if key.ends_with('/') { @@ -433,7 +427,7 @@ pub async fn receiver( while let Some(message) = messages.next().await { match RedisEvent::try_from(message) { Ok(ev) => { - push_event(&hub_state, &mut redis, ev); // .await; + push_event(&hub_state, &mut redis, ev).await; } Err(e) => { warn!("invalid redis message: {e}"); diff --git a/tests/rest_api.rs b/tests/rest_api.rs index 6a9f86fa70..a072d8deac 100644 --- a/tests/rest_api.rs +++ b/tests/rest_api.rs @@ -31,7 +31,12 @@ async fn status(base: &str, client: &reqwest::Client) -> () { let text = resp.text().await.unwrap(); let json: Value = serde_json::from_str(&text).unwrap(); - assert_eq!(json["backend"], "memory"); + let backend = json["backend"].as_str().unwrap_or_default(); + if let Ok(expected_backend) = env::var("TEST_BACKEND") { + assert_eq!(backend, expected_backend); + } else { + assert!(backend == "memory" || backend == "redis"); + } assert_eq!(json["status"], "OK"); assert!(json.get("memory_info").is_some()); assert!(json.get("websockets").is_some());