From 5279324beb05a2befb1dc5f38d396458a3cb8d7d Mon Sep 17 00:00:00 2001 From: Raphael Date: Sun, 9 Aug 2026 00:21:16 +0200 Subject: [PATCH 1/2] feat: websocket action --- actions/websocket-action/Cargo.lock | 1716 +++++++++++++++++ actions/websocket-action/Cargo.toml | 25 + .../websocket-action/src/auth/credentials.rs | 133 ++ actions/websocket-action/src/auth/jwt.rs | 131 ++ actions/websocket-action/src/auth/mod.rs | 80 + actions/websocket-action/src/auth/settings.rs | 32 + actions/websocket-action/src/auth/types.rs | 88 + actions/websocket-action/src/data_types.rs | 103 + actions/websocket-action/src/events.rs | 132 ++ actions/websocket-action/src/flow_setting.rs | 27 + actions/websocket-action/src/functions.rs | 93 + actions/websocket-action/src/main.rs | 101 + actions/websocket-action/src/outbound.rs | 42 + actions/websocket-action/src/route.rs | 288 +++ actions/websocket-action/src/server.rs | 317 +++ 15 files changed, 3308 insertions(+) create mode 100644 actions/websocket-action/Cargo.lock create mode 100644 actions/websocket-action/Cargo.toml create mode 100644 actions/websocket-action/src/auth/credentials.rs create mode 100644 actions/websocket-action/src/auth/jwt.rs create mode 100644 actions/websocket-action/src/auth/mod.rs create mode 100644 actions/websocket-action/src/auth/settings.rs create mode 100644 actions/websocket-action/src/auth/types.rs create mode 100644 actions/websocket-action/src/data_types.rs create mode 100644 actions/websocket-action/src/events.rs create mode 100644 actions/websocket-action/src/flow_setting.rs create mode 100644 actions/websocket-action/src/functions.rs create mode 100644 actions/websocket-action/src/main.rs create mode 100644 actions/websocket-action/src/outbound.rs create mode 100644 actions/websocket-action/src/route.rs create mode 100644 actions/websocket-action/src/server.rs diff --git a/actions/websocket-action/Cargo.lock b/actions/websocket-action/Cargo.lock new file mode 100644 index 0000000..563d13d --- /dev/null +++ b/actions/websocket-action/Cargo.lock @@ -0,0 +1,1716 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "sync_wrapper", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + +[[package]] +name = "bitflags" +version = "2.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" + +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cc" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "num-traits", +] + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + +[[package]] +name = "defmt" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1" +dependencies = [ + "bitflags 1.3.2", + "defmt-macros", +] + +[[package]] +name = "defmt-macros" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8" +dependencies = [ + "defmt-parser", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "defmt-parser" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" +dependencies = [ + "thiserror", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", +] + +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + +[[package]] +name = "either" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" + +[[package]] +name = "env_filter" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "900d271a03799a1ee8d1ca9b19893b48ca674a9284fefcfb85f05e74ed314217" +dependencies = [ + "log", + "regex", +] + +[[package]] +name = "env_logger" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de671bd27a75a797dc9ae289ba1e77276e75e2026408aab65185384e2d5cd3f6" +dependencies = [ + "anstream", + "anstyle", + "env_filter", + "jiff", + "log", +] + +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + +[[package]] +name = "find-msvc-tools" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" + +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" + +[[package]] +name = "futures-sink" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" + +[[package]] +name = "futures-task" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" + +[[package]] +name = "futures-util" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" +dependencies = [ + "futures-core", + "futures-sink", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "rand_core", +] + +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "hercules" +version = "0.0.0" +source = "git+https://github.com/code0-tech/hercules.git?branch=%2322-primitive-rust-sdk#c6ec1336b39e246cc739ab698291e06fd1f08978" +dependencies = [ + "async-trait", + "hercules-macros", + "inventory", + "log", + "schemars", + "serde", + "serde_json", + "thiserror", + "tokio", + "tokio-stream", + "tonic", + "tucana", + "uuid", +] + +[[package]] +name = "hercules-macros" +version = "0.0.0" +source = "git+https://github.com/code0-tech/hercules.git?branch=%2322-primitive-rust-sdk#c6ec1336b39e246cc739ab698291e06fd1f08978" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hybrid-array" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +dependencies = [ + "typenum", +] + +[[package]] +name = "hyper" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d22053281f852e11534f5198498373cbb59295120a20771d90f7ed1897490a72" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-timeout" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" +dependencies = [ + "hyper", + "hyper-util", + "pin-project-lite", + "tokio", + "tower-service", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "hyper", + "libc", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown 0.17.1", +] + +[[package]] +name = "inventory" +version = "0.3.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4f0c30c76f2f4ccee3fe55a2435f691ca00c0e4bd87abe4f4a851b1d4dac39b" +dependencies = [ + "rustversion", +] + +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "jiff" +version = "0.2.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "668b7183bd07af9a4885f5c35b0cc5c83c4607a913c16b7e17291832910d2dcc" +dependencies = [ + "defmt", + "jiff-core", + "jiff-static", + "log", + "portable-atomic", + "portable-atomic-util", + "serde_core", +] + +[[package]] +name = "jiff-core" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7feca88439efe53da3754500c1851dedf3cb36c524dd5cf8225cc0794de95d09" +dependencies = [ + "defmt", +] + +[[package]] +name = "jiff-static" +version = "0.2.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a69dcb3a21cfb32ce1cd056169337ca284af0766dd766e7878819b251a49204" +dependencies = [ + "jiff-core", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "js-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "multimap" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + +[[package]] +name = "pbjson" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8edd1efdd8ab23ba9cb9ace3d9987a72663d5d7c9f74fa00b51d6213645cf6c" +dependencies = [ + "base64", + "serde", +] + +[[package]] +name = "pbjson-build" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ed4d5c6ae95e08ac768883c8401cf0e8deb4e6e1d6a4e1fd3d2ec4f0ec63200" +dependencies = [ + "heck", + "itertools", + "prost", + "prost-types", +] + +[[package]] +name = "pbjson-types" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a14e2757d877c0f607a82ce1b8560e224370f159d66c5d52eb55ea187ef0350e" +dependencies = [ + "bytes", + "chrono", + "pbjson", + "pbjson-build", + "prost", + "prost-build", + "serde", +] + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "petgraph" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455" +dependencies = [ + "fixedbitset", + "hashbrown 0.15.5", + "indexmap", +] + +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "portable-atomic" +version = "1.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d20d5497ef88037a52ff98267d066e7f11fcc5e99bbfbd58a42336193aacec3" + +[[package]] +name = "portable-atomic-util" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" +dependencies = [ + "portable-atomic", +] + +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn 2.0.119", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-build" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" +dependencies = [ + "heck", + "itertools", + "log", + "multimap", + "petgraph", + "prettyplease", + "prost", + "prost-types", + "pulldown-cmark", + "pulldown-cmark-to-cmark", + "regex", + "syn 2.0.119", + "tempfile", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "prost-types" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a" +dependencies = [ + "prost", +] + +[[package]] +name = "pulldown-cmark" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f068eba8e7071c5f9511831b44f32c740d5adf574e990f946ddb53db2f314e" +dependencies = [ + "bitflags 2.13.1", + "memchr", + "unicase", +] + +[[package]] +name = "pulldown-cmark-to-cmark" +version = "22.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50793def1b900256624a709439404384204a5dc3a6ec580281bfaac35e882e90" +dependencies = [ + "pulldown-cmark", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "ref-cast" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "216e8f773d7923bcba9ceb86a86c93cabb3903a11872fc3f138c49630e50b96d" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9283685feec7d69af75fb0e858d5e7378f33fe4fc699383b2916ab9273e03c" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "regex" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags 2.13.1", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "schemars" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a" +dependencies = [ + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98c67716b46af2f0b8cf752abc930f6f9aecfbf671ecfb531db8a31dbe4e2ba" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.3", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "indexmap", + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.4.3", + "once_cell", + "rustix", + "windows-sys 0.61.2", +] + +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", + "tokio-util", +] + +[[package]] +name = "tokio-tungstenite" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17a073bfed563fa236697a068031408a93cd9522e08abf9933ead3e73411bd71" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tonic" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" +dependencies = [ + "async-trait", + "axum", + "base64", + "bytes", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "socket2", + "sync_wrapper", + "tokio", + "tokio-stream", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-build" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c68f61875ac5293cf72e6c8cf0158086428c82c37229e98c840878f1706b0322" +dependencies = [ + "prettyplease", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tonic-prost" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0" +dependencies = [ + "bytes", + "prost", + "tonic", +] + +[[package]] +name = "tonic-prost-build" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "654e5643eff75d7f8c99197ce1440ed19a3474eada74c12bbac488b2cafdae27" +dependencies = [ + "prettyplease", + "proc-macro2", + "prost-build", + "prost-types", + "quote", + "syn 2.0.119", + "tempfile", + "tonic-build", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "indexmap", + "pin-project-lite", + "slab", + "sync_wrapper", + "tokio", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + +[[package]] +name = "tucana" +version = "0.0.80" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0cfe93386a4f88e5ea97e6ce64b72b975a8bb5464062cb66aad12e46544b959" +dependencies = [ + "pbjson", + "pbjson-build", + "pbjson-types", + "prost", + "prost-build", + "prost-types", + "serde", + "serde_json", + "tonic", + "tonic-build", + "tonic-prost", + "tonic-prost-build", +] + +[[package]] +name = "tungstenite" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e48ac77174b19c110a50ab2128b24215ac9cb40e0e12e093fb602d175c569d22" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand", + "sha1", + "thiserror", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicase" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "uuid" +version = "1.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasm-bindgen" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 2.0.119", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "websocket-action" +version = "0.0.0" +dependencies = [ + "base64", + "env_logger", + "form_urlencoded", + "futures-util", + "hercules", + "http", + "log", + "percent-encoding", + "regex", + "ring", + "schemars", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tokio-tungstenite", + "tucana", + "uuid", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/actions/websocket-action/Cargo.toml b/actions/websocket-action/Cargo.toml new file mode 100644 index 0000000..693dc22 --- /dev/null +++ b/actions/websocket-action/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "websocket-action" +version = "0.0.0" +edition = "2024" +publish = false + +[dependencies] +hercules = { git = "https://github.com/code0-tech/hercules.git", branch = "#22-primitive-rust-sdk" } +tucana = { version = "0.0.80", default-features = false } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time"] } +tokio-stream = "0.1" +tokio-tungstenite = "0.30" +futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] } +http = "1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +schemars = "1" +log = "0.4" +env_logger = "0.11" +regex = "1" +percent-encoding = "2.3" +form_urlencoded = "1.2" +base64 = "0.22" +ring = "0.17" +uuid = { version = "1", features = ["v4"] } diff --git a/actions/websocket-action/src/auth/credentials.rs b/actions/websocket-action/src/auth/credentials.rs new file mode 100644 index 0000000..d5b7a02 --- /dev/null +++ b/actions/websocket-action/src/auth/credentials.rs @@ -0,0 +1,133 @@ +use base64::Engine; +use tucana::shared::{Struct, Value, value::Kind}; + +use super::jwt::validate_hs256_jwt; +use super::types::AuthenticationType; + +pub(super) fn matches_authorization( + auth_type: AuthenticationType, + auth_value: &Value, + authorization: &str, +) -> bool { + match auth_type { + AuthenticationType::BearerJwt => { + let Some(secret) = value_as_string(auth_value) else { + return false; + }; + validate_hs256_jwt(authorization, secret) + } + AuthenticationType::BearerStatic => { + let Some(expected_token) = value_as_string(auth_value) else { + return false; + }; + authorization.trim() == format!("Bearer {}", expected_token.trim()) + } + AuthenticationType::Basic => { + let Some(credentials) = basic_credentials(auth_value) else { + return false; + }; + let expected_encoded = + base64::engine::general_purpose::STANDARD.encode(credentials.as_bytes()); + authorization.trim() == format!("Basic {}", expected_encoded) + } + } +} + +fn basic_credentials(value: &Value) -> Option { + if let Some(credentials) = value_as_string(value) { + return Some(credentials.trim().to_string()); + } + + let Some(Kind::StructValue(Struct { fields })) = value.kind.as_ref() else { + return None; + }; + + let username = fields + .get("username") + .or_else(|| fields.get("user")) + .and_then(value_as_string)?; + let password = fields + .get("password") + .or_else(|| fields.get("pass")) + .and_then(value_as_string)?; + + Some(format!("{username}:{password}")) +} + +fn value_as_string(value: &Value) -> Option<&str> { + match value.kind.as_ref() { + Some(Kind::StringValue(value)) => Some(value.as_str()), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use tucana::shared::{Struct, Value, value::Kind}; + + use super::matches_authorization; + use crate::auth::jwt::tests::create_hs256_jwt; + use crate::auth::types::AuthenticationType; + + #[test] + fn bearer_static_matches_expected_token() { + let value = string_value("secret"); + assert!(matches_authorization( + AuthenticationType::BearerStatic, + &value, + "Bearer secret" + )); + assert!(!matches_authorization( + AuthenticationType::BearerStatic, + &value, + "Bearer other" + )); + } + + #[test] + fn bearer_jwt_verifies_hs256_token() { + let secret = string_value("jwt-secret"); + let token = create_hs256_jwt("jwt-secret", r#"{"sub":"123"}"#); + assert!(matches_authorization( + AuthenticationType::BearerJwt, + &secret, + &format!("Bearer {token}") + )); + assert!(!matches_authorization( + AuthenticationType::BearerJwt, + &secret, + "Bearer header.payload.bad-signature" + )); + } + + #[test] + fn basic_matches_encoded_username_password_object_pair() { + let value = basic_value("user", "pass"); + assert!(matches_authorization( + AuthenticationType::Basic, + &value, + "Basic dXNlcjpwYXNz" + )); + assert!(!matches_authorization( + AuthenticationType::Basic, + &value, + "Basic dXNlcjpvdGhlcg==" + )); + } + + fn string_value(value: &str) -> Value { + Value { + kind: Some(Kind::StringValue(value.to_string())), + } + } + + fn basic_value(username: &str, password: &str) -> Value { + let mut fields = HashMap::new(); + fields.insert("username".to_string(), string_value(username)); + fields.insert("password".to_string(), string_value(password)); + Value { + kind: Some(Kind::StructValue(Struct { fields })), + } + } +} diff --git a/actions/websocket-action/src/auth/jwt.rs b/actions/websocket-action/src/auth/jwt.rs new file mode 100644 index 0000000..3d057ad --- /dev/null +++ b/actions/websocket-action/src/auth/jwt.rs @@ -0,0 +1,131 @@ +use base64::Engine; +use ring::hmac; +use std::time::{SystemTime, UNIX_EPOCH}; + +pub(super) fn validate_hs256_jwt(authorization: &str, secret: &str) -> bool { + let Some(token) = authorization.trim().strip_prefix("Bearer ") else { + return false; + }; + + validate_token(token, secret) +} + +fn validate_token(token: &str, secret: &str) -> bool { + let mut segments = token.split('.'); + let Some(header_segment) = segments.next() else { + return false; + }; + let Some(payload_segment) = segments.next() else { + return false; + }; + let Some(signature_segment) = segments.next() else { + return false; + }; + + if segments.next().is_some() { + return false; + } + + let Some(header) = decode_json_segment(header_segment) else { + return false; + }; + + // The flow stores only one shared secret, so JWT auth is intentionally + // limited to HS256. Supporting RS/ES algorithms would require a public + // key or JWKS setting instead of a secret string. + if header.get("alg").and_then(|alg| alg.as_str()) != Some("HS256") { + return false; + } + + if !verify_signature(header_segment, payload_segment, signature_segment, secret) { + return false; + } + + let Some(payload) = decode_json_segment(payload_segment) else { + return false; + }; + + token_is_not_expired(&payload) +} + +fn verify_signature( + header_segment: &str, + payload_segment: &str, + signature_segment: &str, + secret: &str, +) -> bool { + let Some(signature) = decode_base64_url(signature_segment) else { + return false; + }; + + let signing_input = format!("{header_segment}.{payload_segment}"); + let key = hmac::Key::new(hmac::HMAC_SHA256, secret.as_bytes()); + + hmac::verify(&key, signing_input.as_bytes(), &signature).is_ok() +} + +fn token_is_not_expired(payload: &serde_json::Value) -> bool { + let Some(exp) = payload.get("exp").and_then(|exp| exp.as_i64()) else { + return true; + }; + + let Ok(now) = SystemTime::now().duration_since(UNIX_EPOCH) else { + return false; + }; + + exp > now.as_secs() as i64 +} + +fn decode_json_segment(segment: &str) -> Option { + let bytes = decode_base64_url(segment)?; + serde_json::from_slice::(&bytes).ok() +} + +fn decode_base64_url(value: &str) -> Option> { + base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(value) + .or_else(|_| base64::engine::general_purpose::URL_SAFE.decode(value)) + .ok() +} + +#[cfg(test)] +pub(crate) mod tests { + use base64::Engine; + use ring::hmac; + + use super::validate_hs256_jwt; + + #[test] + fn verifies_hs256_token() { + let token = create_hs256_jwt("jwt-secret", r#"{"sub":"123"}"#); + assert!(validate_hs256_jwt(&format!("Bearer {token}"), "jwt-secret")); + } + + #[test] + fn rejects_expired_token() { + let token = create_hs256_jwt("jwt-secret", r#"{"exp":1}"#); + assert!(!validate_hs256_jwt( + &format!("Bearer {token}"), + "jwt-secret" + )); + } + + #[test] + fn rejects_wrong_secret() { + let token = create_hs256_jwt("jwt-secret", r#"{"sub":"123"}"#); + assert!(!validate_hs256_jwt(&format!("Bearer {token}"), "wrong")); + } + + pub(crate) fn create_hs256_jwt(secret: &str, payload: &str) -> String { + let header = r#"{"alg":"HS256","typ":"JWT"}"#; + let header_segment = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(header); + let payload_segment = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload); + let signing_input = format!("{header_segment}.{payload_segment}"); + let key = hmac::Key::new(hmac::HMAC_SHA256, secret.as_bytes()); + let signature = hmac::sign(&key, signing_input.as_bytes()); + let signature_segment = + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(signature.as_ref()); + + format!("{signing_input}.{signature_segment}") + } +} diff --git a/actions/websocket-action/src/auth/mod.rs b/actions/websocket-action/src/auth/mod.rs new file mode 100644 index 0000000..8cb9458 --- /dev/null +++ b/actions/websocket-action/src/auth/mod.rs @@ -0,0 +1,80 @@ +mod credentials; +mod jwt; +mod settings; +mod types; + +use std::collections::HashMap; + +use http::{HeaderMap, HeaderValue, header::AUTHORIZATION}; +use tucana::aquila::ActionFlow; + +use self::credentials::matches_authorization; +use self::settings::{FlowAuthConfig, flow_auth_config}; +pub use self::types::AuthenticationError; +use crate::flow_setting; + +/// Validates a flow's `ws_auth`/`ws_auth_value` settings against the +/// incoming handshake request. Unlike rest-action (an `Authorization` +/// header on every request), a browser `WebSocket` client can't set custom +/// headers at all, so as a fallback a bearer-style token is also accepted +/// via an `access_token` query parameter — `Basic` auth has no such +/// fallback since it needs two values. +pub fn validate_flow_auth( + flow: &ActionFlow, + headers: &HeaderMap, + query_params: &HashMap, +) -> Result<(), AuthenticationError> { + let auth_type = match flow_auth_config(flow) { + FlowAuthConfig::Unauthenticated => return Ok(()), + FlowAuthConfig::Invalid => { + log::warn!( + "auth reject: flow_id={} reason=invalid_ws_auth", + flow.flow_id + ); + return Err(AuthenticationError::InvalidAuthorization); + } + FlowAuthConfig::Authenticated(auth_type) => auth_type, + }; + + let Some(auth_value) = flow_setting::value(flow, "ws_auth_value") else { + log::warn!( + "auth reject: flow_id={} reason=missing_or_invalid_ws_auth_value", + flow.flow_id + ); + return Err(AuthenticationError::invalid_for(auth_type)); + }; + + let Some(authorization) = authorization_value(headers, query_params) else { + log::debug!( + "auth reject: flow_id={} reason=missing_authorization", + flow.flow_id + ); + return Err(AuthenticationError::missing_for(auth_type)); + }; + + if matches_authorization(auth_type, auth_value, &authorization) { + Ok(()) + } else { + log::debug!( + "auth reject: flow_id={} reason=authorization_mismatch", + flow.flow_id + ); + Err(AuthenticationError::invalid_for(auth_type)) + } +} + +/// The effective `Authorization`-style header value: the real header if +/// present, otherwise a synthetic `Bearer ` built from the +/// `access_token` query parameter (see the module doc). +fn authorization_value( + headers: &HeaderMap, + query_params: &HashMap, +) -> Option { + if let Some(value) = headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) { + return Some(value.to_string()); + } + + query_params + .get("access_token") + .map(|token| format!("Bearer {token}")) +} diff --git a/actions/websocket-action/src/auth/settings.rs b/actions/websocket-action/src/auth/settings.rs new file mode 100644 index 0000000..a9d8234 --- /dev/null +++ b/actions/websocket-action/src/auth/settings.rs @@ -0,0 +1,32 @@ +use tucana::aquila::ActionFlow; + +use super::types::{AuthenticationType, is_unauthenticated_value}; +use crate::flow_setting; + +pub(super) enum FlowAuthConfig { + Unauthenticated, + Authenticated(AuthenticationType), + Invalid, +} + +pub(super) fn flow_auth_config(flow: &ActionFlow) -> FlowAuthConfig { + let Some(raw_auth_type) = flow_setting::as_string(flow, "ws_auth") else { + return FlowAuthConfig::Unauthenticated; + }; + + if is_unauthenticated_value(raw_auth_type) { + return FlowAuthConfig::Unauthenticated; + } + + match AuthenticationType::parse(raw_auth_type) { + Some(auth_type) => FlowAuthConfig::Authenticated(auth_type), + None => { + log::warn!( + "auth config invalid: flow_id={} ws_auth={:?}", + flow.flow_id, + raw_auth_type + ); + FlowAuthConfig::Invalid + } + } +} diff --git a/actions/websocket-action/src/auth/types.rs b/actions/websocket-action/src/auth/types.rs new file mode 100644 index 0000000..8fd9b29 --- /dev/null +++ b/actions/websocket-action/src/auth/types.rs @@ -0,0 +1,88 @@ +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub(crate) enum AuthenticationType { + BearerJwt, + BearerStatic, + Basic, +} + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub enum AuthenticationError { + MissingAuthorization(AuthenticationType), + InvalidAuthorizationFor(AuthenticationType), + InvalidAuthorization, +} + +impl AuthenticationError { + pub(super) fn missing_for(auth_type: AuthenticationType) -> Self { + Self::MissingAuthorization(auth_type) + } + + pub(super) fn invalid_for(auth_type: AuthenticationType) -> Self { + Self::InvalidAuthorizationFor(auth_type) + } + + pub fn message(self) -> &'static str { + match self { + Self::MissingAuthorization(_) => "Missing authorization", + Self::InvalidAuthorizationFor(_) | Self::InvalidAuthorization => { + "Invalid authorization" + } + } + } +} + +impl AuthenticationType { + pub(super) fn parse(value: &str) -> Option { + let normalized = normalize_auth_type(value); + + match normalized.as_str() { + "bearerjwt" | "jwt" => Some(Self::BearerJwt), + "bearerstatic" | "bearer" | "staticbearer" => Some(Self::BearerStatic), + "basicaccessauth" | "basic" | "basicauth" => Some(Self::Basic), + _ => None, + } + } +} + +pub(super) fn is_unauthenticated_value(value: &str) -> bool { + matches!( + normalize_auth_type(value).as_str(), + "" | "none" | "noauth" | "unauthenticated" + ) +} + +fn normalize_auth_type(value: &str) -> String { + value + .trim() + .replace(['_', '-', ' '], "") + .to_ascii_lowercase() +} + +#[cfg(test)] +mod tests { + use super::{AuthenticationType, is_unauthenticated_value}; + + #[test] + fn ws_auth_type_values_parse() { + assert_eq!( + AuthenticationType::parse("Bearer JWT"), + Some(AuthenticationType::BearerJwt) + ); + assert_eq!( + AuthenticationType::parse("Bearer static"), + Some(AuthenticationType::BearerStatic) + ); + assert_eq!( + AuthenticationType::parse("Basic"), + Some(AuthenticationType::Basic) + ); + } + + #[test] + fn unauthenticated_values_are_explicit() { + assert!(is_unauthenticated_value("unauthenticated")); + assert!(is_unauthenticated_value("no-auth")); + assert!(is_unauthenticated_value("none")); + assert!(is_unauthenticated_value("")); + } +} diff --git a/actions/websocket-action/src/data_types.rs b/actions/websocket-action/src/data_types.rs new file mode 100644 index 0000000..0265ac3 --- /dev/null +++ b/actions/websocket-action/src/data_types.rs @@ -0,0 +1,103 @@ +//! Data types specific to this adapter. + +use hercules::JsonSchema; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +#[hercules::data_type( + identifier = "WEBSOCKET_AUTH_TYPE", + name(en_US = "WebSocket credential variant"), + display_message(en_US = "WebSocket credential variant"), + alias(en_US = "websocket;ws;auth;credential;type;variant;bearer;basic;jwt"), + // A closed string union with `undefined`/`null` isn't representable via + // schemars/JSON Schema on a Rust type, so the structural type is + // hand-written to match code0-definition's reference exactly (mirrors + // rest-action's REST_AUTH_TYPE). + type_override = "'Bearer JWT' | 'Bearer static' | 'Basic' | undefined | null" +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(transparent)] +pub struct WebsocketAuthType(pub String); + +#[hercules::data_type( + identifier = "WEBSOCKET_AUTH_VALUE", + name(en_US = "WebSocket credential value"), + display_message(en_US = "WebSocket credential value"), + alias(en_US = "websocket;ws;auth;credential;value;bearer;basic;username;password;token"), + generic_keys("T"), + type_override = "T extends 'Basic' ? { username: string, password: string } : T extends undefined ? undefined : T extends null ? null : string" +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(transparent)] +pub struct WebsocketAuthValue(pub serde_json::Value); + +#[hercules::data_type( + identifier = "WEBSOCKET_CONNECT_INPUT", + name(en_US = "WebSocket Connect Input"), + display_message(en_US = "WebSocket Connect Input"), + alias(en_US = "websocket;ws;connect;input") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketConnectInput { + /// A locally-unique id for this connection. Pass it to `send` (via the + /// flow execution it triggers) to push a frame back out on it — see + /// `functions.rs`. + pub connection_id: String, + pub path: String, + pub headers: HashMap, + pub query_params: HashMap, + pub path_params: HashMap, +} + +#[hercules::data_type( + identifier = "WEBSOCKET_MESSAGE_INPUT", + name(en_US = "WebSocket Message Input"), + display_message(en_US = "WebSocket Message Input"), + alias(en_US = "websocket;ws;message;input"), + generic_keys("T"), + type_override = "{ connection_id: TEXT, path: TEXT, is_binary: BOOLEAN, payload: T, path_params: OBJECT<{}> }", + linked_data_type_identifiers("TEXT", "BOOLEAN", "OBJECT") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketMessageInput { + pub connection_id: String, + pub path: String, + /// `true` if the inbound frame was a binary frame. Binary payloads are + /// carried as a base64-encoded string; text payloads are carried as + /// parsed JSON when they parse as JSON, or as a plain string otherwise. + pub is_binary: bool, + pub payload: serde_json::Value, + pub path_params: HashMap, +} + +#[hercules::data_type( + identifier = "WEBSOCKET_DISCONNECT_INPUT", + name(en_US = "WebSocket Disconnect Input"), + display_message(en_US = "WebSocket Disconnect Input"), + alias(en_US = "websocket;ws;disconnect;input") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketDisconnectInput { + pub connection_id: String, + pub path: String, + /// The WebSocket close code, if the peer (or this action) sent a close + /// frame. Absent for an ungraceful disconnect (e.g. TCP reset). + pub code: Option, + pub reason: Option, + /// `true` if the connection closed via a clean close-frame handshake, + /// `false` if it dropped due to an error/reset. + pub was_clean: bool, +} + +#[hercules::data_type( + identifier = "WEBSOCKET_ERROR_INPUT", + name(en_US = "WebSocket Error Input"), + display_message(en_US = "WebSocket Error Input"), + alias(en_US = "websocket;ws;error;input") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketErrorInput { + pub connection_id: String, + pub path: String, + pub message: String, +} diff --git a/actions/websocket-action/src/events.rs b/actions/websocket-action/src/events.rs new file mode 100644 index 0000000..869b300 --- /dev/null +++ b/actions/websocket-action/src/events.rs @@ -0,0 +1,132 @@ +//! The four `WebSocket` flow triggers. +//! +//! Registering these `RuntimeEvent`s is what lets users attach a flow to a +//! WebSocket endpoint in the platform UI; Aquila then pushes each such flow +//! down to this action as an `ActionFlow` with its own event's settings +//! filled in. `server.rs` matches incoming connections/messages against +//! those settings. +//! +//! Unlike `rest-action` (a single `REST` trigger), a WebSocket connection has +//! several distinct moments a flow can react to, so there are four separate +//! events here instead of one. `ActionFlow` (as pushed down by Aquila) does +//! not currently carry which runtime event a flow was defined against, so +//! each event below uses its own uniquely-named path setting +//! (`ws_connect_path` / `ws_message_path` / `ws_disconnect_path` / +//! `ws_error_path`) rather than sharing a single `ws_path` identifier — that +//! is what lets `route.rs` tell, from `flow.settings` alone, which of the +//! four events a given flow belongs to. See a note on this in the +//! `websocket-action` README/PR description. + +#[hercules::runtime_event( + identifier = "WEBSOCKET_CONNECT", + signature = "(ws_connect_path: HTTP_URL, ws_auth: A, ws_auth_value: WEBSOCKET_AUTH_VALUE): WEBSOCKET_CONNECT_INPUT", + name(en_US = "WebSocket Connected"), + description( + en_US = "Fires once when a client completes the WebSocket handshake on the configured path, giving the flow a chance to run connection setup logic (e.g. sending a welcome message via 'send')." + ), + display_message(en_US = "WebSocket client connected on ${ws_connect_path}"), + alias(en_US = "websocket;ws;socket;connect;open"), + display_icon = "tabler:plug-connected", + linked_data_type_identifiers("WEBSOCKET_CONNECT_INPUT"), + editable +)] +#[setting( + identifier = "ws_connect_path", + unique = "project", + name(en_US = "Path"), + description(en_US = "The WebSocket endpoint path this flow listens on (e.g. /chat)."), + linked_data_type_identifiers("HTTP_URL") +)] +#[setting( + identifier = "ws_auth", + name(en_US = "Authentication type"), + description( + en_US = "Specifies the authentication mechanism required for the incoming WebSocket handshake (e.g. Bearer JWT, Bearer static, Basic)." + ), + linked_data_type_identifiers("WEBSOCKET_AUTH_TYPE"), + optional +)] +#[setting( + identifier = "ws_auth_value", + name(en_US = "Authentication value"), + description( + en_US = "Provides the credential value matching the selected authentication type (e.g. token string or username/password pair)." + ), + linked_data_type_identifiers("WEBSOCKET_AUTH_VALUE"), + optional +)] +pub struct WsConnectRuntimeEvent; + +#[hercules::runtime_event( + identifier = "WEBSOCKET_MESSAGE", + signature = "(ws_message_path: HTTP_URL, input_schema: TYPE): WEBSOCKET_MESSAGE_INPUT", + name(en_US = "WebSocket Message Received"), + description( + en_US = "Fires for every inbound WebSocket frame (text or binary) received on a connection to the configured path. Runs fire-and-forget, one flow execution per message; there is no built-in request/response correlation, use the 'send' function to push a reply back out on the same connection." + ), + display_message(en_US = "WebSocket message received on ${ws_message_path}"), + alias(en_US = "websocket;ws;socket;message;frame"), + display_icon = "tabler:message-2", + linked_data_type_identifiers("WEBSOCKET_MESSAGE_INPUT"), + editable +)] +#[setting( + identifier = "ws_message_path", + unique = "project", + name(en_US = "Path"), + description(en_US = "The WebSocket endpoint path this flow listens on (e.g. /chat)."), + linked_data_type_identifiers("HTTP_URL") +)] +#[setting( + identifier = "input_schema", + name(en_US = "Input schema"), + description( + en_US = "Input schema which defines the expected structure of an incoming message's payload (used only when the message is text and parses as JSON)." + ), + optional +)] +pub struct WsMessageRuntimeEvent; + +#[hercules::runtime_event( + identifier = "WEBSOCKET_DISCONNECT", + signature = "(ws_disconnect_path: HTTP_URL): WEBSOCKET_DISCONNECT_INPUT", + name(en_US = "WebSocket Disconnected"), + description( + en_US = "Fires once when a connection to the configured path closes, whether cleanly (a close frame) or due to an error (the connection dropping)." + ), + display_message(en_US = "WebSocket client disconnected on ${ws_disconnect_path}"), + alias(en_US = "websocket;ws;socket;disconnect;close"), + display_icon = "tabler:plug-connected-x", + linked_data_type_identifiers("WEBSOCKET_DISCONNECT_INPUT"), + editable +)] +#[setting( + identifier = "ws_disconnect_path", + unique = "project", + name(en_US = "Path"), + description(en_US = "The WebSocket endpoint path this flow listens on (e.g. /chat)."), + linked_data_type_identifiers("HTTP_URL") +)] +pub struct WsDisconnectRuntimeEvent; + +#[hercules::runtime_event( + identifier = "WEBSOCKET_ERROR", + signature = "(ws_error_path: HTTP_URL): WEBSOCKET_ERROR_INPUT", + name(en_US = "WebSocket Error"), + description( + en_US = "Fires when a protocol-level error occurs on a connection to the configured path (e.g. a malformed frame). This does not necessarily mean the connection was closed; a 'WebSocket Disconnected' event follows separately once it actually closes." + ), + display_message(en_US = "WebSocket error on ${ws_error_path}"), + alias(en_US = "websocket;ws;socket;error"), + display_icon = "tabler:alert-triangle", + linked_data_type_identifiers("WEBSOCKET_ERROR_INPUT"), + editable +)] +#[setting( + identifier = "ws_error_path", + unique = "project", + name(en_US = "Path"), + description(en_US = "The WebSocket endpoint path this flow listens on (e.g. /chat)."), + linked_data_type_identifiers("HTTP_URL") +)] +pub struct WsErrorRuntimeEvent; diff --git a/actions/websocket-action/src/flow_setting.rs b/actions/websocket-action/src/flow_setting.rs new file mode 100644 index 0000000..49a6e61 --- /dev/null +++ b/actions/websocket-action/src/flow_setting.rs @@ -0,0 +1,27 @@ +//! Reads a typed value out of an `ActionFlow`'s settings by identifier. +//! Ported from rest-action's `flow_setting.rs`. + +use tucana::aquila::ActionFlow; +use tucana::shared::{Struct, Value, value::Kind}; + +pub fn value<'a>(flow: &'a ActionFlow, flow_setting_id: &str) -> Option<&'a Value> { + flow.settings + .iter() + .find(|setting| setting.flow_setting_id == flow_setting_id) + .and_then(|setting| setting.value.as_ref()) +} + +pub fn as_string<'a>(flow: &'a ActionFlow, flow_setting_id: &str) -> Option<&'a str> { + match value(flow, flow_setting_id)?.kind.as_ref()? { + Kind::StringValue(value) => Some(value.as_str()), + _ => None, + } +} + +#[allow(dead_code)] +pub fn as_struct<'a>(flow: &'a ActionFlow, flow_setting_id: &str) -> Option<&'a Struct> { + match value(flow, flow_setting_id)?.kind.as_ref()? { + Kind::StructValue(value) => Some(value), + _ => None, + } +} diff --git a/actions/websocket-action/src/functions.rs b/actions/websocket-action/src/functions.rs new file mode 100644 index 0000000..1abcfa8 --- /dev/null +++ b/actions/websocket-action/src/functions.rs @@ -0,0 +1,93 @@ +//! `websocket::control::send`: the function a flow calls to push a frame +//! back out on the connection its trigger fired for. Dispatched to this +//! action like any other remote function; `Send::run` just enqueues onto +//! whichever `server.rs` connection is still registered for +//! `FunctionContext::execution_id`. Analogous to rest-action's +//! `functions.rs::Respond`, but fire-and-forget (no reply to correlate) and +//! reusable any number of times for as long as the triggering flow execution +//! is still running. + +use hercules::{Arguments, FunctionContext, PlainValue, Result, RuntimeFunctionHandler, async_trait}; +use tokio_tungstenite::tungstenite::Message; + +use crate::outbound::{self, OutboundConnections}; + +#[hercules::runtime_function( + identifier = "websocket::control::send", + signature = "(payload: TEXT, is_binary?: BOOLEAN): void", + name(en_US = "Send"), + description( + en_US = "Sends a WebSocket frame back out on the connection that triggered this flow execution (a 'WebSocket Connected' or 'WebSocket Message Received' event). Text is sent as a text frame by default; set is_binary to send the payload (base64-decoded) as a binary frame instead." + ), + display_message(en_US = "Sends WebSocket message ${payload}"), + alias(en_US = "send;control;websocket;ws"), + display_icon = "tabler:cube-send", + linked_data_type_identifiers("TEXT", "BOOLEAN"), + manual +)] +#[parameter( + runtime_name = "payload", + name(en_US = "Payload"), + description( + en_US = "The message to send. For a text frame this is sent as-is; for a binary frame (is_binary = true) this is expected to be base64-encoded." + ) +)] +#[parameter( + runtime_name = "is_binary", + name(en_US = "Send as binary"), + description(en_US = "If true, the payload is base64-decoded and sent as a binary frame instead of a text frame."), + optional +)] +pub struct Send { + connections: OutboundConnections, +} + +impl Send { + pub fn new(connections: OutboundConnections) -> Self { + Self { connections } + } +} + +#[async_trait] +impl RuntimeFunctionHandler for Send { + async fn run(&self, context: &FunctionContext, args: &Arguments) -> Result { + let payload: String = args.get("payload")?; + let is_binary: bool = args.get("is_binary").unwrap_or(false); + + let message = if is_binary { + use base64::Engine; + let bytes = base64::engine::general_purpose::STANDARD + .decode(payload.as_bytes()) + .map_err(|err| { + hercules::HerculesError::runtime( + "INVALID_BASE64_PAYLOAD", + Some(err.to_string()), + ) + })?; + Message::Binary(bytes.into()) + } else { + Message::Text(payload.into()) + }; + + match outbound::get(&self.connections, &context.execution_id) { + Some(sender) => { + if sender.send(message).is_err() { + log::info!( + "send called for execution {} but the connection already closed", + context.execution_id + ); + } + } + // Not every execution of a flow that calls `send` came from an + // inbound connection event still open (a manual/test execution, + // or a connection that already closed). That's expected, not a + // failure — same reasoning as rest-action's `Respond`. + None => log::info!( + "send called for execution {} with no registered connection", + context.execution_id + ), + } + + Ok(PlainValue::Null) + } +} diff --git a/actions/websocket-action/src/main.rs b/actions/websocket-action/src/main.rs new file mode 100644 index 0000000..37e290d --- /dev/null +++ b/actions/websocket-action/src/main.rs @@ -0,0 +1,101 @@ +mod auth; +mod data_types; +mod events; +mod flow_setting; +mod functions; +mod outbound; +mod route; +mod server; + +use hercules::{Action, HerculesEvent, ScalingOption, Translation}; +use tokio_stream::StreamExt; + +use functions::Send; +use outbound::new_outbound_connections; + +fn env(key: &str, default: &str) -> String { + std::env::var(key).unwrap_or_else(|_| default.to_string()) +} + +/// `WebsocketAuthType`/`WebsocketAuthValue`/`Websocket*Input` and the four +/// `Ws*RuntimeEvent`s never appear below. Attaching `#[hercules::data_type]` +/// / `#[hercules::runtime_event]` registered each of them automatically as +/// part of `Action::new` (see `hercules::registration`). `Send` is the one +/// exception: it needs the shared `outbound` connection map injected at +/// construction time, so it's registered by hand below instead (see its +/// `manual` attribute). +fn build_action(outbound: outbound::OutboundConnections) -> Action { + let mut action = Action::new( + env("HERCULES_ACTION_ID", "websocket-action"), + env("HERCULES_SDK_VERSION", "0.0.0"), + ) + .aquila_url(env("HERCULES_AQUILA_URL", "127.0.0.1:8081")) + // Every instance needs to be reachable to accept WebSocket connections, + // so every instance gets every flow rather than splitting them up + // (mirrors rest-action). + .scaling(ScalingOption::Disabled) + .author("CodeZero") + .icon("tabler:plug-connected") + .documentation("A WebSocket server that lets flows react to incoming connections, messages, disconnects, and errors, and push messages back out.") + .name([Translation::new("en-US", "WebSocket")]); + + action.register_runtime_function(Send::new(outbound)); + action +} + +#[tokio::main] +async fn main() -> hercules::Result<()> { + env_logger::Builder::from_env( + env_logger::Env::default().default_filter_or("websocket_action=info,hercules=info"), + ) + .init(); + + let outbound = new_outbound_connections(); + + let mut args = std::env::args().skip(1); + if args.next().as_deref() == Some("export") { + let dir = args + .next() + .unwrap_or_else(|| "./out/draco-websocket".to_string()); + build_action(outbound) + .export(&dir) + .unwrap_or_else(|err| panic!("failed to export to {dir:?}: {err}")); + println!("wrote module definitions to {dir}"); + return Ok(()); + } + + let action = build_action(outbound.clone()); + let mut events = action.subscribe(); + + let aquila_url = env("HERCULES_AQUILA_URL", "127.0.0.1:8081"); + log::info!("connecting to Aquila at {aquila_url}"); + + let connected = action + .connect(env("HERCULES_AUTH_TOKEN", "value"), None) + .await + .unwrap_or_else(|err| panic!("failed to connect to Aquila: {err}")); + + let host = env("WEBSOCKET_SERVER_HOST", "0.0.0.0"); + let port = env("WEBSOCKET_SERVER_PORT", "8080"); + let addr = format!("{host}:{port}") + .parse() + .unwrap_or_else(|err| panic!("invalid WEBSOCKET_SERVER_HOST/WEBSOCKET_SERVER_PORT: {err}")); + + tokio::spawn(server::serve(addr, connected, outbound)); + + while let Some(event) = events.next().await { + match event { + HerculesEvent::Connected => log::info!("connected to Aquila"), + HerculesEvent::Error(error) => panic!("Aquila stream error: {error}"), + HerculesEvent::FlowUpserted(flow) => { + log::info!("flow {} was created/updated", flow.flow_id); + } + HerculesEvent::FlowDeleted(flow_id) => { + log::info!("flow {flow_id} was deleted"); + } + _ => {} + } + } + + Ok(()) +} diff --git a/actions/websocket-action/src/outbound.rs b/actions/websocket-action/src/outbound.rs new file mode 100644 index 0000000..72de07e --- /dev/null +++ b/actions/websocket-action/src/outbound.rs @@ -0,0 +1,42 @@ +//! Correlates a `send` function call (identified by +//! `FunctionContext::execution_id`) back to the WebSocket connection whose +//! flow execution it belongs to. Mirrors rest-action's `pending.rs`, but +//! simplified: there's no request/response correlation to do here (a +//! WebSocket message isn't a request), so instead of a one-shot channel that +//! gets taken exactly once, this just hands out clones of a per-connection +//! `mpsc` sender that `send` can push onto any number of times for as long +//! as the triggering flow execution is registered. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; + +pub type OutboundSender = mpsc::UnboundedSender; +pub type OutboundConnections = Arc>>; + +pub fn new_outbound_connections() -> OutboundConnections { + Arc::new(Mutex::new(HashMap::new())) +} + +/// Registers `sender` as the outbound channel for `execution_id`, to be +/// looked up by [`get`] for the lifetime of that flow execution. Call +/// [`remove`] once the execution finishes so the map doesn't grow unbounded. +pub fn register(connections: &OutboundConnections, execution_id: String, sender: OutboundSender) { + lock(connections).insert(execution_id, sender); +} + +/// Looks up (without removing) the outbound sender for `execution_id`, if +/// its flow execution is still registered. +pub fn get(connections: &OutboundConnections, execution_id: &str) -> Option { + lock(connections).get(execution_id).cloned() +} + +pub fn remove(connections: &OutboundConnections, execution_id: &str) { + lock(connections).remove(execution_id); +} + +fn lock(connections: &OutboundConnections) -> std::sync::MutexGuard<'_, HashMap> { + connections.lock().unwrap_or_else(|err| err.into_inner()) +} diff --git a/actions/websocket-action/src/route.rs b/actions/websocket-action/src/route.rs new file mode 100644 index 0000000..44e7969 --- /dev/null +++ b/actions/websocket-action/src/route.rs @@ -0,0 +1,288 @@ +//! Matches an incoming WebSocket connection/message against this action's +//! flows. Path pattern compilation is ported from rest-action's +//! `route.rs` (same `:param`/`*`/legacy-regex syntax); what's different here +//! is that there are four distinct trigger kinds sharing the same path +//! syntax, so there are four `find_matching_*_flows` entry points below, one +//! per `flow_setting_id` used to both scope a flow to a path *and* tell +//! which of the four `WEBSOCKET_*` events it was defined against (see the +//! module doc on `events.rs`). + +use std::collections::HashMap; +use tucana::aquila::ActionFlow; + +use crate::flow_setting; + +pub const CONNECT_PATH_SETTING: &str = "ws_connect_path"; +pub const MESSAGE_PATH_SETTING: &str = "ws_message_path"; +pub const DISCONNECT_PATH_SETTING: &str = "ws_disconnect_path"; +pub const ERROR_PATH_SETTING: &str = "ws_error_path"; + +pub fn find_matching_connect_flows(flows: &[ActionFlow], path: &str) -> Vec { + find_matching_flows(flows, CONNECT_PATH_SETTING, path) +} + +pub fn find_matching_message_flows(flows: &[ActionFlow], path: &str) -> Vec { + find_matching_flows(flows, MESSAGE_PATH_SETTING, path) +} + +pub fn find_matching_disconnect_flows(flows: &[ActionFlow], path: &str) -> Vec { + find_matching_flows(flows, DISCONNECT_PATH_SETTING, path) +} + +pub fn find_matching_error_flows(flows: &[ActionFlow], path: &str) -> Vec { + find_matching_flows(flows, ERROR_PATH_SETTING, path) +} + +/// `true` if any flow is configured (on any of the four events) to accept +/// connections on `path` — used to decide whether to accept the WebSocket +/// handshake at all before firing any flow. +pub fn any_flow_matches_path(flows: &[ActionFlow], path: &str) -> bool { + !find_matching_connect_flows(flows, path).is_empty() + || !find_matching_message_flows(flows, path).is_empty() + || !find_matching_disconnect_flows(flows, path).is_empty() + || !find_matching_error_flows(flows, path).is_empty() +} + +/// Reads back the named `:param` captures for `path` against whichever of +/// `flow`'s path settings (`path_setting_id`) matched it — used to populate +/// a data type's `path_params` field, mirroring rest-action's +/// `extract_path_params`. +pub fn extract_path_params( + flow: &ActionFlow, + path_setting_id: &str, + path: &str, +) -> HashMap { + let Some(flow_ws_path) = flow_setting::as_string(flow, path_setting_id) else { + return HashMap::new(); + }; + + extract_named_route_captures(&flow_route_pattern(flow, flow_ws_path), path) +} + +fn extract_named_route_captures(pattern: &str, route: &str) -> HashMap { + let Ok(anchored_pattern) = compile_route_pattern(pattern) else { + return HashMap::new(); + }; + let Ok(regex) = regex::Regex::new(&anchored_pattern) else { + return HashMap::new(); + }; + + let Some(captures) = regex.captures(route) else { + return HashMap::new(); + }; + + regex + .capture_names() + .flatten() + .filter_map(|name| { + captures.name(name).map(|value| { + ( + name.to_string(), + percent_encoding::percent_decode_str(value.as_str()) + .decode_utf8_lossy() + .into_owned(), + ) + }) + }) + .collect() +} + +fn find_matching_flows( + flows: &[ActionFlow], + path_setting_id: &str, + path: &str, +) -> Vec { + flows + .iter() + .filter(|flow| matches_request(flow, path_setting_id, path)) + .cloned() + .collect() +} + +fn matches_request(flow: &ActionFlow, path_setting_id: &str, path: &str) -> bool { + // A flow only carries the settings for the one event it was defined + // against, so the presence of `path_setting_id` on this flow is what + // scopes matching to the right event kind (see events.rs's module doc). + let Some(flow_ws_path) = flow_setting::as_string(flow, path_setting_id) else { + return false; + }; + + let pattern = flow_route_pattern(flow, flow_ws_path); + let is_match = matches_route_pattern(&pattern, path); + log::debug!( + "route check: flow_id={} setting={} pattern={:?} path={:?} matched={}", + flow.flow_id, + path_setting_id, + pattern, + path, + is_match + ); + is_match +} + +fn flow_route_pattern(flow: &ActionFlow, flow_ws_path: &str) -> String { + format!("/{}{}", flow.project_slug, flow_ws_path) +} + +fn matches_route_pattern(pattern: &str, route: &str) -> bool { + let Ok(anchored_pattern) = compile_route_pattern(pattern) else { + return false; + }; + let Ok(regex) = regex::Regex::new(&anchored_pattern) else { + return false; + }; + regex.is_match(route) +} + +fn compile_route_pattern(pattern: &str) -> Result { + if is_url_pattern_style(pattern) { + return compile_url_pattern_path(pattern).map(|pattern| format!("^{}$", pattern)); + } + Ok(format!("^{}$", pattern)) +} + +fn is_url_pattern_style(pattern: &str) -> bool { + let bytes = pattern.as_bytes(); + bytes.iter().enumerate().any(|(index, byte)| { + (*byte == b':' + && bytes + .get(index + 1) + .is_some_and(|next| next.is_ascii_alphabetic() || *next == b'_')) + || (*byte == b'*' && bytes.get(index.wrapping_sub(1)) != Some(&b'.')) + }) +} + +fn compile_url_pattern_path(pattern: &str) -> Result { + let mut compiled = String::new(); + let chars: Vec = pattern.chars().collect(); + let mut index = 0; + + while index < chars.len() { + match chars[index] { + ':' if is_param_start(chars.get(index + 1).copied()) => { + let (name, next_index) = read_param_name(&chars, index + 1); + index = next_index; + + let (capture_pattern, next_index) = if chars.get(index) == Some(&'(') { + read_balanced_group(&chars, index)? + } else { + (String::from("[^/]+"), index) + }; + + compiled.push_str(&format!("(?P<{name}>{capture_pattern})")); + index = next_index; + } + '*' => { + compiled.push_str(".*"); + index += 1; + } + value => { + compiled.push_str(®ex::escape(&value.to_string())); + index += 1; + } + } + } + + Ok(compiled) +} + +fn is_param_start(value: Option) -> bool { + value.is_some_and(|value| value.is_ascii_alphabetic() || value == '_') +} + +fn read_param_name(chars: &[char], start: usize) -> (String, usize) { + let mut index = start; + let mut name = String::new(); + + while let Some(value) = chars.get(index) { + if value.is_ascii_alphanumeric() || *value == '_' { + name.push(*value); + index += 1; + } else { + break; + } + } + + (name, index) +} + +fn read_balanced_group(chars: &[char], start: usize) -> Result<(String, usize), String> { + let mut depth = 0; + let mut index = start; + let mut pattern = String::new(); + + while let Some(value) = chars.get(index) { + match value { + '(' => { + depth += 1; + if depth > 1 { + pattern.push(*value); + } + } + ')' => { + depth -= 1; + if depth == 0 { + return Ok((pattern, index + 1)); + } + pattern.push(*value); + } + _ => pattern.push(*value), + } + index += 1; + } + + Err(String::from("unclosed parameter regex group")) +} + +#[cfg(test)] +mod tests { + use super::*; + use tucana::aquila::ActionFlow; + use tucana::shared::{FlowSetting, Value, value::Kind}; + + fn flow_with_setting(id: &str, setting_id: &str, path: &str) -> ActionFlow { + ActionFlow { + flow_id: id.parse().unwrap_or(1), + project_slug: "acme".to_string(), + settings: vec![FlowSetting { + flow_setting_id: setting_id.to_string(), + value: Some(Value { + kind: Some(Kind::StringValue(path.to_string())), + }), + ..Default::default() + }], + ..Default::default() + } + } + + #[test] + fn matches_only_the_requested_event_kind() { + let flows = vec![ + flow_with_setting("1", CONNECT_PATH_SETTING, "/chat"), + flow_with_setting("2", MESSAGE_PATH_SETTING, "/chat"), + ]; + + assert_eq!( + find_matching_connect_flows(&flows, "/acme/chat").len(), + 1 + ); + assert_eq!( + find_matching_message_flows(&flows, "/acme/chat").len(), + 1 + ); + assert_eq!(find_matching_disconnect_flows(&flows, "/acme/chat").len(), 0); + } + + #[test] + fn dynamic_route_params_match_path_segments() { + let flows = vec![flow_with_setting("1", MESSAGE_PATH_SETTING, "/rooms/:id")]; + assert_eq!( + find_matching_message_flows(&flows, "/acme/rooms/42").len(), + 1 + ); + assert_eq!( + find_matching_message_flows(&flows, "/acme/rooms/42/extra").len(), + 0 + ); + } +} diff --git a/actions/websocket-action/src/server.rs b/actions/websocket-action/src/server.rs new file mode 100644 index 0000000..aeeba9f --- /dev/null +++ b/actions/websocket-action/src/server.rs @@ -0,0 +1,317 @@ +//! The inbound WebSocket server: accepts TCP connections, performs the +//! WebSocket upgrade handshake (routing/auth against this action's +//! `WEBSOCKET_CONNECT`/`WEBSOCKET_MESSAGE` flows, matched via +//! `Connected::flows()` the same way rest-action's `server.rs` matches HTTP +//! requests), then for each open connection: fires `WEBSOCKET_CONNECT` once, +//! `WEBSOCKET_MESSAGE` per inbound frame, and finally `WEBSOCKET_DISCONNECT` +//! (with `WEBSOCKET_ERROR` firing first on a protocol error). Every fired +//! flow execution is fire-and-forget (no response is awaited), matching +//! cron-action's `execute_flow` style rather than rest-action's +//! request/response correlation — a WebSocket frame isn't a request. A flow +//! that wants to talk back calls the `send` runtime function +//! (`functions.rs`), which looks up this connection's outbound channel in +//! `outbound.rs` by the triggering execution's id. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Mutex; + +use futures_util::{SinkExt, StreamExt}; +use hercules::{Connected, PlainValue}; +use http::{HeaderMap, HeaderValue, Response, StatusCode}; +use serde_json::json; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request}; +use tokio_tungstenite::tungstenite::protocol::CloseFrame; +use tokio_tungstenite::tungstenite::Message; + +use crate::outbound::{self, OutboundConnections, OutboundSender}; +use crate::{auth, route}; + +pub async fn serve( + addr: SocketAddr, + connected: Connected, + outbound: OutboundConnections, +) -> std::io::Result<()> { + let listener = TcpListener::bind(addr).await?; + log::info!("listening for websocket connections on {addr}"); + + loop { + let (stream, peer_addr) = listener.accept().await?; + let connected = connected.clone(); + let outbound = outbound.clone(); + + tokio::spawn(async move { + if let Err(err) = handle_connection(stream, connected, outbound).await { + log::debug!("connection from {peer_addr} closed with error: {err}"); + } + }); + } +} + +/// What the (synchronous) handshake callback captured about the accepted +/// request, for use once we're back in async code. +#[derive(Default, Clone)] +struct HandshakeInfo { + path: String, + headers: HeaderMap, + query_params: HashMap, +} + +async fn handle_connection( + stream: TcpStream, + connected: Connected, + outbound: OutboundConnections, +) -> Result<(), String> { + let captured: Mutex> = Mutex::new(None); + let connected_for_handshake = connected.clone(); + + let callback = |req: &Request, response: Response<()>| { + let path = req.uri().path().to_string(); + let query_params = parse_query(req.uri().query()); + let headers = req.headers().clone(); + + let flows = connected_for_handshake.flows(); + if !route::any_flow_matches_path(&flows, &path) { + log::info!("websocket handshake on {path}: no flow matched"); + return Err(reject(StatusCode::NOT_FOUND, "No flow found for path")); + } + + // Fail closed: if any WEBSOCKET_CONNECT flow bound to this path + // requires auth, the whole handshake needs valid credentials — there + // is no way to accept the TCP connection but selectively withhold it + // from just the flows that wanted auth. + for flow in route::find_matching_connect_flows(&flows, &path) { + if let Err(err) = auth::validate_flow_auth(&flow, &headers, &query_params) { + log::warn!( + "websocket handshake on {path}: flow {} rejected: {}", + flow.flow_id, + err.message() + ); + return Err(reject(StatusCode::UNAUTHORIZED, err.message())); + } + } + + *captured.lock().unwrap_or_else(|err| err.into_inner()) = Some(HandshakeInfo { + path, + headers, + query_params, + }); + + Ok(response) + }; + + let ws_stream = tokio_tungstenite::accept_hdr_async(stream, callback) + .await + .map_err(|err| format!("handshake failed: {err}"))?; + + let HandshakeInfo { + path, + headers, + query_params, + } = captured + .into_inner() + .unwrap_or_else(|err| err.into_inner()) + .ok_or_else(|| "handshake accepted without capturing request info".to_string())?; + + let connection_id = uuid::Uuid::new_v4().to_string(); + log::info!("{connection_id}: websocket connected on {path}"); + + let (write, mut read) = ws_stream.split(); + let (tx, mut rx) = mpsc::unbounded_channel::(); + + tokio::spawn(async move { + let mut write = write; + while let Some(message) = rx.recv().await { + if let Err(err) = write.send(message).await { + log::debug!("failed to write websocket frame: {err}"); + break; + } + } + }); + + for flow in route::find_matching_connect_flows(&connected.flows(), &path) { + let path_params = route::extract_path_params(&flow, route::CONNECT_PATH_SETTING, &path); + let payload = json!({ + "connection_id": connection_id, + "path": path, + "headers": header_map_to_json(&headers), + "query_params": query_params, + "path_params": path_params, + }); + fire(&connected, &outbound, &tx, flow.flow_id, payload); + } + + let (was_clean, code, reason) = loop { + match read.next().await { + Some(Ok(Message::Text(text))) => { + let payload = parse_text_payload(text.as_str()); + dispatch_message(&connected, &outbound, &tx, &connection_id, &path, false, payload); + } + Some(Ok(Message::Binary(bytes))) => { + use base64::Engine; + let encoded = base64::engine::general_purpose::STANDARD.encode(&bytes); + dispatch_message( + &connected, + &outbound, + &tx, + &connection_id, + &path, + true, + serde_json::Value::String(encoded), + ); + } + Some(Ok(Message::Ping(payload))) => { + let _ = tx.send(Message::Pong(payload)); + } + Some(Ok(Message::Pong(_))) => {} + Some(Ok(Message::Close(frame))) => { + break close_details(frame, true); + } + Some(Ok(Message::Frame(_))) => {} + Some(Err(err)) => { + log::warn!("{connection_id}: websocket protocol error: {err}"); + dispatch_error(&connected, &outbound, &tx, &connection_id, &path, err.to_string()); + break (false, None, Some(err.to_string())); + } + None => { + break (false, None, None); + } + } + }; + + log::info!("{connection_id}: websocket disconnected on {path} (clean={was_clean})"); + for flow in route::find_matching_disconnect_flows(&connected.flows(), &path) { + let path_params = route::extract_path_params(&flow, route::DISCONNECT_PATH_SETTING, &path); + let payload = json!({ + "connection_id": connection_id, + "path": path, + "code": code, + "reason": reason, + "was_clean": was_clean, + "path_params": path_params, + }); + fire(&connected, &outbound, &tx, flow.flow_id, payload); + } + + Ok(()) +} + +fn dispatch_message( + connected: &Connected, + outbound: &OutboundConnections, + tx: &OutboundSender, + connection_id: &str, + path: &str, + is_binary: bool, + payload: serde_json::Value, +) { + for flow in route::find_matching_message_flows(&connected.flows(), path) { + let path_params = route::extract_path_params(&flow, route::MESSAGE_PATH_SETTING, path); + let flow_payload = json!({ + "connection_id": connection_id, + "path": path, + "is_binary": is_binary, + "payload": payload, + "path_params": path_params, + }); + fire(connected, outbound, tx, flow.flow_id, flow_payload); + } +} + +fn dispatch_error( + connected: &Connected, + outbound: &OutboundConnections, + tx: &OutboundSender, + connection_id: &str, + path: &str, + message: String, +) { + for flow in route::find_matching_error_flows(&connected.flows(), path) { + let path_params = route::extract_path_params(&flow, route::ERROR_PATH_SETTING, path); + let payload = json!({ + "connection_id": connection_id, + "path": path, + "message": message, + "path_params": path_params, + }); + fire(connected, outbound, tx, flow.flow_id, payload); + } +} + +/// Executes `flow_id` fire-and-forget (cron-action's `execute_flow` style — +/// no reply is awaited), first registering `tx` as this execution's outbound +/// channel so a `send` call from within the flow (functions.rs) can find its +/// way back to this connection. The registration is removed once the +/// execution finishes so `outbound`'s map doesn't grow unbounded. +fn fire( + connected: &Connected, + outbound: &OutboundConnections, + tx: &OutboundSender, + flow_id: i64, + payload: PlainValue, +) { + let execution_id = connected.reserve_execution_id(); + outbound::register(outbound, execution_id.clone(), tx.clone()); + + let connected = connected.clone(); + let outbound = outbound.clone(); + tokio::spawn(async move { + let result = connected + .execute_flow_with_id(execution_id.clone(), flow_id.to_string(), payload) + .await; + outbound::remove(&outbound, &execution_id); + if let Err(err) = result { + log::error!("flow {flow_id} (execution {execution_id}) failed: {err}"); + } + }); +} + +fn close_details(frame: Option, was_clean: bool) -> (bool, Option, Option) { + match frame { + Some(frame) => ( + was_clean, + Some(frame.code.into()), + Some(frame.reason.to_string()), + ), + None => (was_clean, None, None), + } +} + +fn parse_text_payload(text: &str) -> serde_json::Value { + serde_json::from_str(text).unwrap_or_else(|_| serde_json::Value::String(text.to_string())) +} + +fn parse_query(query: Option<&str>) -> HashMap { + let Some(query) = query else { + return HashMap::new(); + }; + form_urlencoded::parse(query.as_bytes()) + .map(|(key, value)| (key.into_owned(), value.into_owned())) + .collect() +} + +fn header_map_to_json(headers: &HeaderMap) -> HashMap { + headers + .iter() + .map(|(name, value)| { + let value = value + .to_str() + .map(str::to_owned) + .unwrap_or_else(|_| String::from_utf8_lossy(value.as_bytes()).into_owned()); + (name.as_str().to_owned(), value) + }) + .collect() +} + +fn reject(status: StatusCode, message: &str) -> ErrorResponse { + Response::builder() + .status(status) + .body(Some(message.to_string())) + .unwrap_or_else(|_| { + let mut response = Response::new(Some(message.to_string())); + *response.status_mut() = status; + response + }) +} From 6507af751bfbed9eff5ce66b2a3d26c6d116acbc Mon Sep 17 00:00:00 2001 From: Raphael Date: Sun, 9 Aug 2026 11:54:56 +0200 Subject: [PATCH 2/2] feat: added websocket client --- actions/websocket-action/Cargo.lock | 79 ++++ actions/websocket-action/Cargo.toml | 2 +- .../websocket-action/src/auth/credentials.rs | 26 ++ actions/websocket-action/src/auth/mod.rs | 34 +- actions/websocket-action/src/auth/settings.rs | 8 +- actions/websocket-action/src/client.rs | 379 ++++++++++++++++++ actions/websocket-action/src/data_types.rs | 75 ++++ actions/websocket-action/src/events.rs | 141 +++++++ actions/websocket-action/src/main.rs | 15 +- actions/websocket-action/src/route.rs | 58 +++ 10 files changed, 811 insertions(+), 6 deletions(-) create mode 100644 actions/websocket-action/src/client.rs diff --git a/actions/websocket-action/Cargo.lock b/actions/websocket-action/Cargo.lock index 563d13d..8739cc3 100644 --- a/actions/websocket-action/Cargo.lock +++ b/actions/websocket-action/Cargo.lock @@ -1074,6 +1074,39 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "once_cell", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.23" @@ -1199,6 +1232,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -1286,6 +1325,16 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.19" @@ -1306,8 +1355,12 @@ checksum = "17a073bfed563fa236697a068031408a93cd9522e08abf9933ead3e73411bd71" dependencies = [ "futures-util", "log", + "rustls", + "rustls-pki-types", "tokio", + "tokio-rustls", "tungstenite", + "webpki-roots 0.26.11", ] [[package]] @@ -1492,6 +1545,8 @@ dependencies = [ "httparse", "log", "rand", + "rustls", + "rustls-pki-types", "sha1", "thiserror", ] @@ -1597,6 +1652,24 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.9", +] + +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "websocket-action" version = "0.0.0" @@ -1709,6 +1782,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zmij" version = "1.0.23" diff --git a/actions/websocket-action/Cargo.toml b/actions/websocket-action/Cargo.toml index 693dc22..063d964 100644 --- a/actions/websocket-action/Cargo.toml +++ b/actions/websocket-action/Cargo.toml @@ -9,7 +9,7 @@ hercules = { git = "https://github.com/code0-tech/hercules.git", branch = "#22-p tucana = { version = "0.0.80", default-features = false } tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time"] } tokio-stream = "0.1" -tokio-tungstenite = "0.30" +tokio-tungstenite = { version = "0.30", features = ["rustls-tls-webpki-roots"] } futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] } http = "1" serde = { version = "1", features = ["derive"] } diff --git a/actions/websocket-action/src/auth/credentials.rs b/actions/websocket-action/src/auth/credentials.rs index d5b7a02..97d7c1f 100644 --- a/actions/websocket-action/src/auth/credentials.rs +++ b/actions/websocket-action/src/auth/credentials.rs @@ -33,6 +33,32 @@ pub(super) fn matches_authorization( } } +/// Builds the `Authorization` header value to *send* on an outbound client +/// handshake, given the flow's configured auth type/value — the inverse of +/// [`matches_authorization`], which *validates* an inbound one. +/// +/// `BearerJwt` and `BearerStatic` are handled identically here: this action +/// doesn't mint JWTs (it only validates HS256 signatures on the inbound +/// side, see `jwt.rs`), so a `Bearer JWT`-configured outbound connection just +/// sends the configured value verbatim as the bearer token, same as +/// `Bearer static`. The operator is expected to supply an already-encoded +/// token in that case. +pub(super) fn build_authorization_header( + auth_type: AuthenticationType, + auth_value: &Value, +) -> Option { + match auth_type { + AuthenticationType::BearerJwt | AuthenticationType::BearerStatic => { + value_as_string(auth_value).map(|token| format!("Bearer {}", token.trim())) + } + AuthenticationType::Basic => { + let credentials = basic_credentials(auth_value)?; + let encoded = base64::engine::general_purpose::STANDARD.encode(credentials.as_bytes()); + Some(format!("Basic {encoded}")) + } + } +} + fn basic_credentials(value: &Value) -> Option { if let Some(credentials) = value_as_string(value) { return Some(credentials.trim().to_string()); diff --git a/actions/websocket-action/src/auth/mod.rs b/actions/websocket-action/src/auth/mod.rs index 8cb9458..19a0c41 100644 --- a/actions/websocket-action/src/auth/mod.rs +++ b/actions/websocket-action/src/auth/mod.rs @@ -8,7 +8,7 @@ use std::collections::HashMap; use http::{HeaderMap, HeaderValue, header::AUTHORIZATION}; use tucana::aquila::ActionFlow; -use self::credentials::matches_authorization; +use self::credentials::{build_authorization_header, matches_authorization}; use self::settings::{FlowAuthConfig, flow_auth_config}; pub use self::types::AuthenticationError; use crate::flow_setting; @@ -24,7 +24,7 @@ pub fn validate_flow_auth( headers: &HeaderMap, query_params: &HashMap, ) -> Result<(), AuthenticationError> { - let auth_type = match flow_auth_config(flow) { + let auth_type = match flow_auth_config(flow, "ws_auth") { FlowAuthConfig::Unauthenticated => return Ok(()), FlowAuthConfig::Invalid => { log::warn!( @@ -63,6 +63,36 @@ pub fn validate_flow_auth( } } +/// Builds the `Authorization` header value to send on an outbound client +/// handshake (`client.rs`) from a flow's `ws_client_auth`/ +/// `ws_client_auth_value` settings, if configured — the outbound-mode +/// counterpart of [`validate_flow_auth`]. Returns `None` if the flow has no +/// (or an invalid) auth configuration, in which case the handshake is simply +/// sent without an `Authorization` header. +pub fn client_authorization_header(flow: &ActionFlow) -> Option { + let auth_type = match flow_auth_config(flow, "ws_client_auth") { + FlowAuthConfig::Unauthenticated => return None, + FlowAuthConfig::Invalid => { + log::warn!( + "client auth config invalid: flow_id={} reason=invalid_ws_client_auth", + flow.flow_id + ); + return None; + } + FlowAuthConfig::Authenticated(auth_type) => auth_type, + }; + + let Some(auth_value) = flow_setting::value(flow, "ws_client_auth_value") else { + log::warn!( + "client auth config missing value: flow_id={} reason=missing_ws_client_auth_value", + flow.flow_id + ); + return None; + }; + + build_authorization_header(auth_type, auth_value) +} + /// The effective `Authorization`-style header value: the real header if /// present, otherwise a synthetic `Bearer ` built from the /// `access_token` query parameter (see the module doc). diff --git a/actions/websocket-action/src/auth/settings.rs b/actions/websocket-action/src/auth/settings.rs index a9d8234..69afa5c 100644 --- a/actions/websocket-action/src/auth/settings.rs +++ b/actions/websocket-action/src/auth/settings.rs @@ -9,8 +9,12 @@ pub(super) enum FlowAuthConfig { Invalid, } -pub(super) fn flow_auth_config(flow: &ActionFlow) -> FlowAuthConfig { - let Some(raw_auth_type) = flow_setting::as_string(flow, "ws_auth") else { +/// `auth_setting_id` is the flow setting carrying the auth type (`"ws_auth"` +/// for the inbound handshake, `"ws_client_auth"` for an outbound client +/// connection's handshake — see `auth/mod.rs`'s `validate_flow_auth` and +/// `client_authorization_header` respectively). +pub(super) fn flow_auth_config(flow: &ActionFlow, auth_setting_id: &str) -> FlowAuthConfig { + let Some(raw_auth_type) = flow_setting::as_string(flow, auth_setting_id) else { return FlowAuthConfig::Unauthenticated; }; diff --git a/actions/websocket-action/src/client.rs b/actions/websocket-action/src/client.rs new file mode 100644 index 0000000..f3f5cdd --- /dev/null +++ b/actions/websocket-action/src/client.rs @@ -0,0 +1,379 @@ +//! Outbound WebSocket client connections: the mirror of `server.rs` for +//! *this action* connecting *out* to another WebSocket server, one +//! long-lived connection per flow that carries a `WEBSOCKET_CLIENT_CONNECT` +//! trigger. +//! +//! Unlike the inbound server side, there's no listener to bind once and +//! route arbitrary connections through — each client connection belongs to +//! exactly one flow (the one whose `ws_client_connect_url` setting names the +//! target), so the connection's lifecycle is driven by that flow's own +//! lifecycle: `main.rs` calls [`ClientManager::reconcile`] on every +//! `HerculesEvent::FlowUpserted` (starting a connection task the first time +//! a flow gets client-connect settings, restarting it if those settings +//! change) and [`ClientManager::remove`] on `HerculesEvent::FlowDeleted`. +//! +//! Once connected, frames are dispatched the same way `server.rs` does +//! (`WEBSOCKET_CLIENT_MESSAGE` per frame, fire-and-forget +//! `execute_flow_with_id`, the triggering execution's outbound channel +//! registered in `outbound.rs` so `functions.rs`'s `send` works unmodified). +//! What's genuinely new here: a client losing its connection should retry +//! rather than just exit, so the per-flow task loops forever, reconnecting +//! with exponential backoff after every disconnect (clean or not) or failed +//! connection attempt. + +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use hercules::{Connected, PlainValue}; +use serde_json::json; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::protocol::CloseFrame; +use tucana::aquila::ActionFlow; + +use crate::flow_setting; +use crate::outbound::{self, OutboundConnections, OutboundSender}; +use crate::{auth, route}; + +/// Backoff applied between reconnect attempts for a single flow's +/// connection: doubles after every failed attempt or dropped connection, up +/// to `RECONNECT_MAX`, and resets to `RECONNECT_INITIAL` after every +/// successful connect (regardless of how long it then stays up). +const RECONNECT_INITIAL: Duration = Duration::from_secs(1); +const RECONNECT_MAX: Duration = Duration::from_secs(30); + +/// Tracks the one outbound-connection task spawned per flow carrying a +/// `WEBSOCKET_CLIENT_CONNECT` trigger, keyed by `flow_id`, so `main.rs` can +/// start/stop them as flows are upserted/deleted. +pub struct ClientManager { + tasks: Mutex>>, +} + +impl ClientManager { + pub fn new() -> Self { + Self { + tasks: Mutex::new(HashMap::new()), + } + } + + /// Re-evaluates whether `flow` should have a running outbound connection + /// task. `flow` may belong to any of the four `WEBSOCKET_CLIENT_*` + /// events (or none of them) — this only acts when it carries + /// `ws_client_connect_url` (i.e. it's a `WEBSOCKET_CLIENT_CONNECT` flow). + /// + /// Always stops any previously-running task for this `flow_id` first: + /// `HerculesEvent::FlowUpserted` only fires for a genuine change (see + /// `hercules::Connected`'s doc), so if this flow already had a task + /// running its settings (the target URL, or auth) may well have changed + /// and the connection needs to be re-established against the new + /// configuration. + pub fn reconcile(&self, flow: &ActionFlow, connected: &Connected, outbound: &OutboundConnections) { + self.stop(flow.flow_id); + + let Some(url) = flow_setting::as_string(flow, route::CLIENT_CONNECT_URL_SETTING) else { + return; + }; + let url = url.to_string(); + + log::info!( + "flow {} carries a client-connect target ({url}); starting outbound connection", + flow.flow_id + ); + + let flow = flow.clone(); + let connected = connected.clone(); + let outbound = outbound.clone(); + let flow_id = flow.flow_id; + let handle = tokio::spawn(async move { + run_connection(flow, url, connected, outbound).await; + }); + + self.tasks().insert(flow_id, handle); + } + + /// Stops (aborts) `flow_id`'s connection task, if one is running. Call + /// on `HerculesEvent::FlowDeleted`. + pub fn remove(&self, flow_id: i64) { + self.stop(flow_id); + } + + fn stop(&self, flow_id: i64) { + if let Some(handle) = self.tasks().remove(&flow_id) { + handle.abort(); + } + } + + fn tasks(&self) -> std::sync::MutexGuard<'_, HashMap>> { + self.tasks.lock().unwrap_or_else(|err| err.into_inner()) + } +} + +impl Default for ClientManager { + fn default() -> Self { + Self::new() + } +} + +/// Runs forever: connects to `url`, serves it until it disconnects (cleanly +/// or not), then reconnects with backoff. Only returns if the task is +/// aborted (`ClientManager::stop`/`Drop` of the `JoinHandle`). +async fn run_connection(flow: ActionFlow, url: String, connected: Connected, outbound: OutboundConnections) { + let mut backoff = RECONNECT_INITIAL; + + loop { + match connect_once(&flow, &url, &connected, &outbound).await { + Ok(()) => { + // Connected, ran, and eventually disconnected — reset the + // backoff since the target was reachable. + backoff = RECONNECT_INITIAL; + } + Err(err) => { + log::warn!( + "flow {}: outbound websocket connection to {url} failed: {err}", + flow.flow_id + ); + } + } + + log::info!( + "flow {}: reconnecting to {url} in {backoff:?}", + flow.flow_id + ); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(RECONNECT_MAX); + } +} + +/// One connection attempt end-to-end: connect, fire `WEBSOCKET_CLIENT_CONNECT`, +/// serve frames until disconnect, fire `WEBSOCKET_CLIENT_DISCONNECT`. +/// +/// Returns `Err` only if the connection attempt itself failed (in which case +/// a `WEBSOCKET_CLIENT_ERROR` — not `_DISCONNECT`, since there was never a +/// connection to disconnect from — is fired first); `Ok(())` covers every +/// other outcome, including a protocol error mid-connection (which still +/// gets a `WEBSOCKET_CLIENT_DISCONNECT` once the read loop gives up). +async fn connect_once( + flow: &ActionFlow, + url: &str, + connected: &Connected, + outbound: &OutboundConnections, +) -> Result<(), String> { + // Reserved before the connection attempt so a failed attempt still has a + // stable id to report in its WEBSOCKET_CLIENT_ERROR payload. + let connection_id = uuid::Uuid::new_v4().to_string(); + + let mut request = url + .into_client_request() + .map_err(|err| format!("invalid target url: {err}"))?; + if let Some(header_value) = auth::client_authorization_header(flow) { + let header_value = header_value + .parse() + .map_err(|err| format!("invalid authorization header value: {err}"))?; + request + .headers_mut() + .insert(http::header::AUTHORIZATION, header_value); + } + + let ws_stream = match tokio_tungstenite::connect_async(request).await { + Ok((stream, _response)) => stream, + Err(err) => { + let (tx, _rx) = mpsc::unbounded_channel::(); + dispatch_error( + connected, + outbound, + &tx, + &connection_id, + &flow.project_slug, + url, + format!("connection failed: {err}"), + ); + return Err(err.to_string()); + } + }; + + log::info!( + "{connection_id}: outbound websocket connected to {url} (flow {})", + flow.flow_id + ); + + let (write, mut read) = ws_stream.split(); + let (tx, mut rx) = mpsc::unbounded_channel::(); + + tokio::spawn(async move { + let mut write = write; + while let Some(message) = rx.recv().await { + if let Err(err) = write.send(message).await { + log::debug!("failed to write outbound websocket frame: {err}"); + break; + } + } + }); + + for target in route::find_matching_client_connect_flows(&connected.flows(), &flow.project_slug, url) { + let payload = json!({ + "connection_id": connection_id, + "url": url, + }); + fire(connected, outbound, &tx, target.flow_id, payload); + } + + let (was_clean, code, reason) = loop { + match read.next().await { + Some(Ok(Message::Text(text))) => { + let payload = parse_text_payload(text.as_str()); + dispatch_message( + connected, + outbound, + &tx, + &connection_id, + &flow.project_slug, + url, + false, + payload, + ); + } + Some(Ok(Message::Binary(bytes))) => { + use base64::Engine; + let encoded = base64::engine::general_purpose::STANDARD.encode(&bytes); + dispatch_message( + connected, + outbound, + &tx, + &connection_id, + &flow.project_slug, + url, + true, + serde_json::Value::String(encoded), + ); + } + Some(Ok(Message::Ping(payload))) => { + let _ = tx.send(Message::Pong(payload)); + } + Some(Ok(Message::Pong(_))) => {} + Some(Ok(Message::Close(frame))) => { + break close_details(frame, true); + } + Some(Ok(Message::Frame(_))) => {} + Some(Err(err)) => { + log::warn!("{connection_id}: outbound websocket protocol error: {err}"); + dispatch_error( + connected, + outbound, + &tx, + &connection_id, + &flow.project_slug, + url, + err.to_string(), + ); + break (false, None, Some(err.to_string())); + } + None => { + break (false, None, None); + } + } + }; + + log::info!( + "{connection_id}: outbound websocket disconnected from {url} (clean={was_clean})" + ); + for target in route::find_matching_client_disconnect_flows(&connected.flows(), &flow.project_slug, url) { + let payload = json!({ + "connection_id": connection_id, + "url": url, + "code": code, + "reason": reason, + "was_clean": was_clean, + }); + fire(connected, outbound, &tx, target.flow_id, payload); + } + + Ok(()) +} + +fn dispatch_message( + connected: &Connected, + outbound: &OutboundConnections, + tx: &OutboundSender, + connection_id: &str, + project_slug: &str, + url: &str, + is_binary: bool, + payload: serde_json::Value, +) { + for target in route::find_matching_client_message_flows(&connected.flows(), project_slug, url) { + let flow_payload = json!({ + "connection_id": connection_id, + "url": url, + "is_binary": is_binary, + "payload": payload, + }); + fire(connected, outbound, tx, target.flow_id, flow_payload); + } +} + +fn dispatch_error( + connected: &Connected, + outbound: &OutboundConnections, + tx: &OutboundSender, + connection_id: &str, + project_slug: &str, + url: &str, + message: String, +) { + for target in route::find_matching_client_error_flows(&connected.flows(), project_slug, url) { + let payload = json!({ + "connection_id": connection_id, + "url": url, + "message": message, + }); + fire(connected, outbound, tx, target.flow_id, payload); + } +} + +/// Executes `flow_id` fire-and-forget (cron-action's `execute_flow` style — +/// no reply is awaited), first registering `tx` as this execution's outbound +/// channel in the same `outbound.rs` registry `server.rs` uses, so a `send` +/// call from within the flow (functions.rs) works identically for a client +/// connection with no changes needed there. The registration is removed once +/// the execution finishes so `outbound`'s map doesn't grow unbounded. +fn fire( + connected: &Connected, + outbound: &OutboundConnections, + tx: &OutboundSender, + flow_id: i64, + payload: PlainValue, +) { + let execution_id = connected.reserve_execution_id(); + outbound::register(outbound, execution_id.clone(), tx.clone()); + + let connected = connected.clone(); + let outbound = outbound.clone(); + tokio::spawn(async move { + let result = connected + .execute_flow_with_id(execution_id.clone(), flow_id.to_string(), payload) + .await; + outbound::remove(&outbound, &execution_id); + if let Err(err) = result { + log::error!("flow {flow_id} (execution {execution_id}) failed: {err}"); + } + }); +} + +fn close_details(frame: Option, was_clean: bool) -> (bool, Option, Option) { + match frame { + Some(frame) => ( + was_clean, + Some(frame.code.into()), + Some(frame.reason.to_string()), + ), + None => (was_clean, None, None), + } +} + +fn parse_text_payload(text: &str) -> serde_json::Value { + serde_json::from_str(text).unwrap_or_else(|_| serde_json::Value::String(text.to_string())) +} diff --git a/actions/websocket-action/src/data_types.rs b/actions/websocket-action/src/data_types.rs index 0265ac3..922657e 100644 --- a/actions/websocket-action/src/data_types.rs +++ b/actions/websocket-action/src/data_types.rs @@ -101,3 +101,78 @@ pub struct WebsocketErrorInput { pub path: String, pub message: String, } + +// -- Outbound (client) data types -------------------------------------------- +// +// Same field shape as the inbound `Websocket*Input` types above, minus +// `path`/`path_params` (a client connection isn't routed by an inbound path; +// see events.rs's module doc). `url` carries the target URL instead, so a +// flow can tell which outbound connection an event came from. + +#[hercules::data_type( + identifier = "WEBSOCKET_CLIENT_CONNECT_INPUT", + name(en_US = "WebSocket Client Connect Input"), + display_message(en_US = "WebSocket Client Connect Input"), + alias(en_US = "websocket;ws;client;connect;input") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketClientConnectInput { + /// A locally-unique id for this connection. Pass it to `send` (via the + /// flow execution it triggers) to push a frame back out on it — see + /// `functions.rs`. + pub connection_id: String, + pub url: String, +} + +#[hercules::data_type( + identifier = "WEBSOCKET_CLIENT_MESSAGE_INPUT", + name(en_US = "WebSocket Client Message Input"), + display_message(en_US = "WebSocket Client Message Input"), + alias(en_US = "websocket;ws;client;message;input"), + generic_keys("T"), + type_override = "{ connection_id: TEXT, url: TEXT, is_binary: BOOLEAN, payload: T }", + linked_data_type_identifiers("TEXT", "BOOLEAN") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketClientMessageInput { + pub connection_id: String, + pub url: String, + /// `true` if the inbound frame was a binary frame. Binary payloads are + /// carried as a base64-encoded string; text payloads are carried as + /// parsed JSON when they parse as JSON, or as a plain string otherwise. + pub is_binary: bool, + pub payload: serde_json::Value, +} + +#[hercules::data_type( + identifier = "WEBSOCKET_CLIENT_DISCONNECT_INPUT", + name(en_US = "WebSocket Client Disconnect Input"), + display_message(en_US = "WebSocket Client Disconnect Input"), + alias(en_US = "websocket;ws;client;disconnect;input") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketClientDisconnectInput { + pub connection_id: String, + pub url: String, + /// The WebSocket close code, if the peer (or this action) sent a close + /// frame. Absent for an ungraceful disconnect (e.g. TCP reset) or a + /// failed connection attempt. + pub code: Option, + pub reason: Option, + /// `true` if the connection closed via a clean close-frame handshake, + /// `false` if it dropped due to an error/reset/failed connection attempt. + pub was_clean: bool, +} + +#[hercules::data_type( + identifier = "WEBSOCKET_CLIENT_ERROR_INPUT", + name(en_US = "WebSocket Client Error Input"), + display_message(en_US = "WebSocket Client Error Input"), + alias(en_US = "websocket;ws;client;error;input") +)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct WebsocketClientErrorInput { + pub connection_id: String, + pub url: String, + pub message: String, +} diff --git a/actions/websocket-action/src/events.rs b/actions/websocket-action/src/events.rs index 869b300..fb188a1 100644 --- a/actions/websocket-action/src/events.rs +++ b/actions/websocket-action/src/events.rs @@ -130,3 +130,144 @@ pub struct WsDisconnectRuntimeEvent; linked_data_type_identifiers("HTTP_URL") )] pub struct WsErrorRuntimeEvent; + +// -- Outbound (client) events ----------------------------------------------- +// +// The four events above are for *inbound* connections (this action as a +// server, routed by path). The four below are the client-mode mirror: this +// action opens and maintains its own *outbound* connection to another +// WebSocket server, one per flow carrying a `WEBSOCKET_CLIENT_CONNECT` +// trigger (see `client.rs`). There's no inbound path to route by, so instead +// of a path setting shared by project, these use a target URL setting +// (`ws_client_connect_url` / `ws_client_message_url` / +// `ws_client_disconnect_url` / `ws_client_error_url`) as the join key: a +// `WEBSOCKET_CLIENT_MESSAGE`/`WEBSOCKET_CLIENT_DISCONNECT`/ +// `WEBSOCKET_CLIENT_ERROR` flow is matched to a given outbound connection by +// having the same URL (within the same project) as the +// `WEBSOCKET_CLIENT_CONNECT` flow that opened it — see `route.rs`'s +// `find_matching_client_*_flows`. Same reasoning as the inbound side applies +// for why each event needs its own uniquely-named setting rather than a +// shared one: `ActionFlow` doesn't carry which runtime event a flow was +// defined against. + +#[hercules::runtime_event( + identifier = "WEBSOCKET_CLIENT_CONNECT", + signature = "(ws_client_connect_url: HTTP_URL, ws_client_auth: A, ws_client_auth_value: WEBSOCKET_AUTH_VALUE): WEBSOCKET_CLIENT_CONNECT_INPUT", + name(en_US = "WebSocket Client Connected"), + description( + en_US = "Fires once this action successfully opens an outbound WebSocket connection to the configured target URL. This action maintains exactly one such connection per flow, automatically reconnecting with backoff if it drops (see 'WebSocket Client Disconnected')." + ), + display_message(en_US = "Outbound WebSocket connected to ${ws_client_connect_url}"), + alias(en_US = "websocket;ws;socket;client;connect;open;outbound"), + display_icon = "tabler:plug-connected", + linked_data_type_identifiers("WEBSOCKET_CLIENT_CONNECT_INPUT"), + editable +)] +#[setting( + identifier = "ws_client_connect_url", + unique = "project", + name(en_US = "Target URL"), + description( + en_US = "The ws:// or wss:// URL of the WebSocket server to connect out to (e.g. wss://example.com/chat)." + ), + linked_data_type_identifiers("HTTP_URL") +)] +#[setting( + identifier = "ws_client_auth", + name(en_US = "Authentication type"), + description( + en_US = "Specifies the Authorization header this action sends on the outbound handshake (e.g. Bearer JWT, Bearer static, Basic)." + ), + linked_data_type_identifiers("WEBSOCKET_AUTH_TYPE"), + optional +)] +#[setting( + identifier = "ws_client_auth_value", + name(en_US = "Authentication value"), + description( + en_US = "Provides the credential value matching the selected authentication type (e.g. token string or username/password pair)." + ), + linked_data_type_identifiers("WEBSOCKET_AUTH_VALUE"), + optional +)] +pub struct WsClientConnectRuntimeEvent; + +#[hercules::runtime_event( + identifier = "WEBSOCKET_CLIENT_MESSAGE", + signature = "(ws_client_message_url: HTTP_URL, input_schema: TYPE): WEBSOCKET_CLIENT_MESSAGE_INPUT", + name(en_US = "WebSocket Client Message Received"), + description( + en_US = "Fires for every inbound WebSocket frame (text or binary) received on this action's outbound connection to the configured target URL. Runs fire-and-forget, one flow execution per message; use the 'send' function to push a reply back out on the same connection." + ), + display_message(en_US = "WebSocket client message received on ${ws_client_message_url}"), + alias(en_US = "websocket;ws;socket;client;message;frame;outbound"), + display_icon = "tabler:message-2", + linked_data_type_identifiers("WEBSOCKET_CLIENT_MESSAGE_INPUT"), + editable +)] +#[setting( + identifier = "ws_client_message_url", + unique = "project", + name(en_US = "Target URL"), + description( + en_US = "The target URL of the outbound WebSocket connection this flow reacts to (matches a 'WebSocket Client Connected' flow's target URL)." + ), + linked_data_type_identifiers("HTTP_URL") +)] +#[setting( + identifier = "input_schema", + name(en_US = "Input schema"), + description( + en_US = "Input schema which defines the expected structure of an incoming message's payload (used only when the message is text and parses as JSON)." + ), + optional +)] +pub struct WsClientMessageRuntimeEvent; + +#[hercules::runtime_event( + identifier = "WEBSOCKET_CLIENT_DISCONNECT", + signature = "(ws_client_disconnect_url: HTTP_URL): WEBSOCKET_CLIENT_DISCONNECT_INPUT", + name(en_US = "WebSocket Client Disconnected"), + description( + en_US = "Fires once this action's outbound connection to the configured target URL closes, whether cleanly (a close frame) or due to an error (the connection dropping). This action will then retry the connection with backoff; a fresh 'WebSocket Client Connected' fires once it reconnects." + ), + display_message(en_US = "Outbound WebSocket disconnected from ${ws_client_disconnect_url}"), + alias(en_US = "websocket;ws;socket;client;disconnect;close;outbound"), + display_icon = "tabler:plug-connected-x", + linked_data_type_identifiers("WEBSOCKET_CLIENT_DISCONNECT_INPUT"), + editable +)] +#[setting( + identifier = "ws_client_disconnect_url", + unique = "project", + name(en_US = "Target URL"), + description( + en_US = "The target URL of the outbound WebSocket connection this flow reacts to (matches a 'WebSocket Client Connected' flow's target URL)." + ), + linked_data_type_identifiers("HTTP_URL") +)] +pub struct WsClientDisconnectRuntimeEvent; + +#[hercules::runtime_event( + identifier = "WEBSOCKET_CLIENT_ERROR", + signature = "(ws_client_error_url: HTTP_URL): WEBSOCKET_CLIENT_ERROR_INPUT", + name(en_US = "WebSocket Client Error"), + description( + en_US = "Fires when a protocol-level error occurs on this action's outbound connection to the configured target URL (e.g. a malformed frame, or a failed connection attempt). This does not necessarily mean the connection was closed; a 'WebSocket Client Disconnected' event follows separately once it actually closes." + ), + display_message(en_US = "Outbound WebSocket error on ${ws_client_error_url}"), + alias(en_US = "websocket;ws;socket;client;error;outbound"), + display_icon = "tabler:alert-triangle", + linked_data_type_identifiers("WEBSOCKET_CLIENT_ERROR_INPUT"), + editable +)] +#[setting( + identifier = "ws_client_error_url", + unique = "project", + name(en_US = "Target URL"), + description( + en_US = "The target URL of the outbound WebSocket connection this flow reacts to (matches a 'WebSocket Client Connected' flow's target URL)." + ), + linked_data_type_identifiers("HTTP_URL") +)] +pub struct WsClientErrorRuntimeEvent; diff --git a/actions/websocket-action/src/main.rs b/actions/websocket-action/src/main.rs index 37e290d..9e51d34 100644 --- a/actions/websocket-action/src/main.rs +++ b/actions/websocket-action/src/main.rs @@ -1,4 +1,5 @@ mod auth; +mod client; mod data_types; mod events; mod flow_setting; @@ -10,6 +11,7 @@ mod server; use hercules::{Action, HerculesEvent, ScalingOption, Translation}; use tokio_stream::StreamExt; +use client::ClientManager; use functions::Send; use outbound::new_outbound_connections; @@ -81,7 +83,16 @@ async fn main() -> hercules::Result<()> { .parse() .unwrap_or_else(|err| panic!("invalid WEBSOCKET_SERVER_HOST/WEBSOCKET_SERVER_PORT: {err}")); - tokio::spawn(server::serve(addr, connected, outbound)); + tokio::spawn(server::serve(addr, connected.clone(), outbound.clone())); + + // Outbound (client) connections are driven purely by the flow event + // stream below rather than a one-time scan of `connected.flows()` at + // startup: every flow that ends up in `connected.flows()` got there via + // a `FlowUpserted` event first (see `hercules::Connected::flows`'s + // doc), and `subscribe()` above was called before `connect()`, so no + // flow — including ones that already existed before this instance + // connected — can be missed here. + let client_manager = ClientManager::new(); while let Some(event) = events.next().await { match event { @@ -89,9 +100,11 @@ async fn main() -> hercules::Result<()> { HerculesEvent::Error(error) => panic!("Aquila stream error: {error}"), HerculesEvent::FlowUpserted(flow) => { log::info!("flow {} was created/updated", flow.flow_id); + client_manager.reconcile(&flow, &connected, &outbound); } HerculesEvent::FlowDeleted(flow_id) => { log::info!("flow {flow_id} was deleted"); + client_manager.remove(flow_id); } _ => {} } diff --git a/actions/websocket-action/src/route.rs b/actions/websocket-action/src/route.rs index 44e7969..60974ed 100644 --- a/actions/websocket-action/src/route.rs +++ b/actions/websocket-action/src/route.rs @@ -43,6 +43,64 @@ pub fn any_flow_matches_path(flows: &[ActionFlow], path: &str) -> bool { || !find_matching_error_flows(flows, path).is_empty() } +pub const CLIENT_CONNECT_URL_SETTING: &str = "ws_client_connect_url"; +pub const CLIENT_MESSAGE_URL_SETTING: &str = "ws_client_message_url"; +pub const CLIENT_DISCONNECT_URL_SETTING: &str = "ws_client_disconnect_url"; +pub const CLIENT_ERROR_URL_SETTING: &str = "ws_client_error_url"; + +/// Client-mode matching (`client.rs`): unlike the inbound side, a client +/// connection isn't reached by an inbound path, so there's no pattern to +/// compile — the target URL is matched literally, scoped to the connecting +/// flow's project (two different projects independently targeting the same +/// external URL shouldn't see each other's events). +pub fn find_matching_client_connect_flows( + flows: &[ActionFlow], + project_slug: &str, + url: &str, +) -> Vec { + find_matching_client_flows(flows, CLIENT_CONNECT_URL_SETTING, project_slug, url) +} + +pub fn find_matching_client_message_flows( + flows: &[ActionFlow], + project_slug: &str, + url: &str, +) -> Vec { + find_matching_client_flows(flows, CLIENT_MESSAGE_URL_SETTING, project_slug, url) +} + +pub fn find_matching_client_disconnect_flows( + flows: &[ActionFlow], + project_slug: &str, + url: &str, +) -> Vec { + find_matching_client_flows(flows, CLIENT_DISCONNECT_URL_SETTING, project_slug, url) +} + +pub fn find_matching_client_error_flows( + flows: &[ActionFlow], + project_slug: &str, + url: &str, +) -> Vec { + find_matching_client_flows(flows, CLIENT_ERROR_URL_SETTING, project_slug, url) +} + +fn find_matching_client_flows( + flows: &[ActionFlow], + url_setting_id: &str, + project_slug: &str, + url: &str, +) -> Vec { + flows + .iter() + .filter(|flow| { + flow.project_slug == project_slug + && flow_setting::as_string(flow, url_setting_id) == Some(url) + }) + .cloned() + .collect() +} + /// Reads back the named `:param` captures for `path` against whichever of /// `flow`'s path settings (`path_setting_id`) matched it — used to populate /// a data type's `path_params` field, mirroring rest-action's