diff --git a/Cargo.lock b/Cargo.lock index b43c16f4d..a39715850 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -934,6 +934,15 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" +[[package]] +name = "convert_case" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "633458d4ef8c78b72454de2d54fd6ab2e60f9e02be22f3c6104cdc8a4e0fceb9" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -977,6 +986,7 @@ dependencies = [ "diom-operations", "diom-proto", "diom-rate-limit", + "diom-transformations", "ctor", "fjall", "fjall-utils", @@ -1359,6 +1369,7 @@ dependencies = [ "clap", "diom", "diom-core", + "diom-transformations", "dotenvy", "mimalloc", "opentelemetry", @@ -1372,6 +1383,30 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "diom-transformations" +version = "0.0.1" +dependencies = [ + "anyhow", + "bytes", + "diom-error", + "futures-util", + "hex", + "libc", + "rand 0.9.2", + "rmp-serde", + "rquickjs", + "seccompiler", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tracing", + "tracing-test", + "uuid", +] + [[package]] name = "cpufeatures" version = "0.2.17" @@ -4227,6 +4262,15 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "relative-path" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bca40a312222d8ba74837cb474edef44b37f561da5f773981007a10bbaa992b0" +dependencies = [ + "serde", +] + [[package]] name = "rend" version = "0.4.2" @@ -4401,6 +4445,54 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "rquickjs" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c50dc6d6c587c339edb4769cf705867497a2baf0eca8b4645fa6ecd22f02c77a" +dependencies = [ + "rquickjs-core", + "rquickjs-macro", +] + +[[package]] +name = "rquickjs-core" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8bf7840285c321c3ab20e752a9afb95548c75cd7f4632a0627cea3507e310c1" +dependencies = [ + "async-lock", + "hashbrown 0.16.1", + "relative-path", + "rquickjs-sys", +] + +[[package]] +name = "rquickjs-macro" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7106215ff41a5677b104906a13e1a440b880f4b6362b5dc4f3978c267fad2b80" +dependencies = [ + "convert_case", + "fnv", + "ident_case", + "indexmap 2.13.0", + "proc-macro-crate", + "proc-macro2", + "quote", + "rquickjs-core", + "syn 2.0.117", +] + +[[package]] +name = "rquickjs-sys" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27344601ef27460e82d6a4e1ecb9e7e99f518122095f3c51296da8e9be2b9d83" +dependencies = [ + "cc", +] + [[package]] name = "rust_decimal" version = "1.40.0" @@ -4618,6 +4710,15 @@ version = "4.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" +[[package]] +name = "seccompiler" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "345a3e4dddf721a478089d4697b83c6c0a8f5bf16086f6c13397e4534eb6e2e5" +dependencies = [ + "libc", +] + [[package]] name = "secrecy" version = "0.10.3" diff --git a/Cargo.toml b/Cargo.toml index 77c95625a..d7b521fa2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,6 +43,7 @@ diom-namespace.workspace = true diom-operations.workspace = true diom-proto.workspace = true diom-rate-limit.workspace = true +diom-transformations.workspace = true fjall.workspace = true fjall-utils.workspace = true fs-err.workspace = true @@ -87,6 +88,7 @@ zip.workspace = true [dev-dependencies] assert_matches.workspace = true +diom-transformations.workspace = true ctor.workspace = true rand.workspace = true rmp-serde.workspace = true @@ -149,6 +151,7 @@ diom-namespace.path = "crates/diom-namespace" diom-operations.path = "crates/diom-operations" diom-proto.path = "crates/diom-proto" diom-rate-limit.path = "crates/rate-limit" +diom-transformations.path = "crates/diom-transformations" criterion = "0.8.2" ctor = "0.6.3" dirs = "6.0.0" @@ -207,7 +210,7 @@ tap = "1" tempfile = "3.24.0" test-utils.path = "crates/test-utils" thiserror = "2.0.18" -tokio = { version = "1.24.2", features = ["rt", "rt-multi-thread", "signal", "fs"] } +tokio = { version = "1.24.2", features = ["rt", "rt-multi-thread", "signal", "fs", "process", "sync", "io-util", "io-std"] } tokio-util = "0.7" toml = { version = "1.0.3", default-features = false, features = ["parse", "serde"] } tower = "0.5.1" diff --git a/crates/diom-transformations/Cargo.toml b/crates/diom-transformations/Cargo.toml new file mode 100644 index 000000000..66fab7136 --- /dev/null +++ b/crates/diom-transformations/Cargo.toml @@ -0,0 +1,34 @@ +[package] +name = "diom-transformations" +publish = false +version.workspace = true +license.workspace = true +rust-version.workspace = true +edition = "2024" + +[target.'cfg(target_os = "linux")'.dependencies] +libc = "0.2" +seccompiler = "0.4" + +[dependencies] +anyhow.workspace = true +bytes.workspace = true +diom-error.workspace = true +futures-util = { workspace = true, features = ["sink"] } +hex.workspace = true +rand.workspace = true +rmp-serde.workspace = true +rquickjs = { version = "0.11.0", features = ["futures", "macro", "array-buffer", "parallel"] } +serde.workspace = true +serde_json.workspace = true +thiserror = "2" +tokio.workspace = true +tokio-util = { workspace = true, features = ["codec"] } +tracing.workspace = true +uuid = { version = "1", features = ["v4"] } + +[dev-dependencies] +tracing-test.workspace = true + +[lints] +workspace = true diff --git a/crates/diom-transformations/src/engine.rs b/crates/diom-transformations/src/engine.rs new file mode 100644 index 000000000..e503dcace --- /dev/null +++ b/crates/diom-transformations/src/engine.rs @@ -0,0 +1,417 @@ +use std::time::{Duration, Instant}; + +use rand::RngCore; +use rquickjs::{ + CatchResultExt, Ctx, JsLifetime, Promise, Value, class::Trace, context::EvalOptions, + function::Rest, +}; +use thiserror::Error; +use tracing::{Instrument, field::Empty}; + +/// Max heap allocations: 5MiB +const MAX_RAM_BYTES: usize = 5_242_880; + +#[derive(Error, Debug)] +pub enum ScriptError { + #[error("internal error setting up the JavaScript interpreter: {0}")] + InternalError(rquickjs::Error), + #[error("unable to parse input into QuickJS: {0}")] + InvalidInputEncoding(rquickjs::Error), + #[error("unable to serialize output: {0}")] + InvalidOutputDecoding(rquickjs::Error), + #[error("script failed to return an output")] + NoOutput, + #[error("exception raised by JavaScript code: {message:?} {}", stack.as_deref().unwrap_or(""))] + ExecutionException { + message: String, + stack: Option, + }, + #[error("Evaluating the top-level module failed to return a valid promise")] + InvalidPromise, + #[error("maximum processing time exceeded")] + ProcessTimeout, + #[error("maximum RAM use exceeded")] + OutOfMemory, +} + +impl From for diom_error::Error { + fn from(err: ScriptError) -> Self { + match err { + ScriptError::InternalError(_) | ScriptError::InvalidPromise => { + diom_error::Error::internal(err) + } + ScriptError::InvalidInputEncoding(_) + | ScriptError::InvalidOutputDecoding(_) + | ScriptError::NoOutput => diom_error::Error::bad_request("content_validation", err), + ScriptError::ExecutionException { .. } | ScriptError::ProcessTimeout => { + diom_error::Error::bad_request("execution_timeout", err) + } + ScriptError::OutOfMemory => diom_error::Error::bad_request("out_of_memory", err), + } + } +} + +#[derive(Clone, Trace, JsLifetime)] +#[rquickjs::class(frozen)] +struct DiomCrypto {} + +#[rquickjs::methods(rename_all = "camelCase")] +impl DiomCrypto { + #[qjs(rename = "randomUUID")] + pub(crate) fn random_uuid(&self) -> String { + uuid::Uuid::new_v4().to_string() + } + + pub(crate) fn get_random_bytes<'js>( + &self, + ctx: Ctx<'js>, + num_bytes: usize, + ) -> rquickjs::Result> { + let mut vec = vec![0u8; num_bytes]; + let mut rng = rand::rng(); + rng.fill_bytes(&mut vec); + rquickjs::TypedArray::new(ctx, vec) + } + + pub(crate) fn array_to_hex<'js>(&self, bytes: rquickjs::TypedArray<'js, u8>) -> String { + hex::encode(bytes) + } +} + +#[derive(Clone, Trace, JsLifetime)] +#[rquickjs::class(frozen)] +struct Console {} + +#[rquickjs::methods] +impl Console { + fn debug(&self, _values: Rest>) -> rquickjs::Result<()> { + tracing::debug!("debug called from script"); + Ok(()) + } + + fn log(&self, _values: Rest>) -> rquickjs::Result<()> { + tracing::debug!("log called from script"); + Ok(()) + } + + fn warn(&self, _values: Rest>) -> rquickjs::Result<()> { + tracing::debug!("warn called from script"); + Ok(()) + } + + fn error(&self, _values: Rest>) -> rquickjs::Result<()> { + tracing::debug!("error called from script"); + Ok(()) + } +} + +fn handle_caught_error(e: rquickjs::CaughtError<'_>) -> ScriptError { + match e { + rquickjs::CaughtError::Exception(exc) => { + if exc.message().as_deref() == Some("interrupted") { + return ScriptError::ProcessTimeout; + } + if exc.message().as_deref() == Some("out of memory") { + ScriptError::OutOfMemory + } else { + ScriptError::ExecutionException { + message: exc + .message() + .unwrap_or_else(|| "".to_string()), + stack: exc.stack(), + } + } + } + rquickjs::CaughtError::Error(err) => ScriptError::ExecutionException { + message: format!("{err:?}"), + stack: None, + }, + rquickjs::CaughtError::Value(v) => { + // OOM is represented by raising `null` as an exception. who knows why. + if v.is_null() { + ScriptError::OutOfMemory + } else { + ScriptError::ExecutionException { + message: format!("{v:?}"), + stack: None, + } + } + } + } +} + +#[tracing::instrument(skip_all, level = "debug", fields(runtime_us = Empty, max_memory_bytes = Empty))] +pub(crate) async fn run_script( + script: impl Into, + payload_json: impl Into, + max_duration: Duration, +) -> Result { + let start = Instant::now(); + let result = run_script_inner(script.into(), payload_json.into(), max_duration) + .instrument(tracing::Span::current()) + .await; + if let Err(e) = &result { + tracing::warn!( + err = ?e, + runtime_us = start.elapsed().as_micros(), + "error executing transformation" + ); + } + result +} + +pub(crate) async fn run_script_inner( + script: String, + payload_json: String, + max_duration: Duration, +) -> Result { + let runtime = rquickjs::AsyncRuntime::new().map_err(ScriptError::InternalError)?; + runtime.set_memory_limit(MAX_RAM_BYTES).await; + + let start = Instant::now(); + runtime + .set_interrupt_handler(Some(Box::new(move || start.elapsed() > max_duration))) + .await; + + let context = rquickjs::AsyncContext::full(&runtime) + .await + .map_err(ScriptError::InternalError)?; + + context + .with(|ctx| -> Result<(), ScriptError> { + let globals = ctx.globals(); + globals + .set("crypto", DiomCrypto {}) + .map_err(ScriptError::InternalError)?; + globals + .set("console", Console {}) + .map_err(ScriptError::InternalError)?; + globals + .set("script_input", payload_json.as_str()) + .map_err(ScriptError::InternalError)?; + Ok(()) + }) + .await?; + + let mut full_script = script; + let value = rquickjs::async_with!(context => |ctx| { + let mut options = EvalOptions::default(); + options.global = true; + options.strict = false; + options.promise = true; + + full_script.push_str(";\n handler(JSON.parse(script_input))"); + let value = ctx + .eval_with_options::, String>(full_script, options) + .catch(&ctx) + .map_err(handle_caught_error)? + .into_future::>() + .await + .catch(&ctx) + .map_err(handle_caught_error)?; + // https://github.com/DelSkayn/rquickjs/issues/360 + let object = value.as_object().ok_or(ScriptError::InvalidPromise)?; + let value: Value<'_> = object.get("value").map_err(|_| ScriptError::InvalidPromise)?; + + ctx.json_stringify(value) + .and_then(|opt| match opt { + Some(val) => val.to_string().map(Some), + None => Ok(None), + }) + .map_err(ScriptError::InvalidOutputDecoding)? + .ok_or(ScriptError::NoOutput) + }) + .await?; + + let span = tracing::Span::current(); + span.record("runtime_us", start.elapsed().as_micros()); + + let usage = runtime.memory_usage().await; + span.record("max_memory_bytes", usage.memory_used_size); + + Ok(value) +} + +#[cfg(test)] +mod tests { + #[allow(clippy::disallowed_types)] + use serde_json::Value; + use std::{ + str::FromStr, + time::{Duration, Instant}, + }; + + use super::{ScriptError, run_script}; + use tracing_test::traced_test; + + async fn exec(script: &str, data: &str) -> Result { + run_script(script, data, Duration::from_millis(10)).await + } + + #[tokio::test] + async fn test_random_uuid() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { return globalThis.crypto.randomUUID(); }"#, + "null", + ) + .await?; + // response is a JSON string, e.g. `"xxxxxxxx-..."` + let s: String = serde_json::from_str(&response)?; + let parsed = uuid::Uuid::from_str(&s)?; + assert_eq!(parsed.get_version_num(), 4); + Ok(()) + } + + #[tokio::test] + async fn test_get_random_bytes() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { + return { "small": Array.from(globalThis.crypto.getRandomBytes(32)) } + }"#, + "null", + ) + .await?; + #[allow(clippy::disallowed_types)] + let parsed: Value = serde_json::from_str(&response)?; + let Some(Value::Array(ints)) = parsed.pointer("/small") else { + anyhow::bail!("Unexpected return type, got {parsed:?}"); + }; + assert_eq!(ints.len(), 32); + assert!(ints.iter().all(|i| matches!(i, Value::Number(_)))); + let numbers = ints.iter().map(|i| i.as_i64().unwrap()).collect::>(); + assert!(numbers.iter().all(|i| (0..=256).contains(i))); + assert!(numbers.iter().any(|i| *i != 0 && *i != 256)); + Ok(()) + } + + #[tokio::test] + async fn test_array_to_hex() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { + const array = new Uint8Array([222, 173, 190, 239]); + return globalThis.crypto.arrayToHex(array); + }"#, + "null", + ) + .await?; + assert_eq!(response, r#""deadbeef""#); + Ok(()) + } + + #[tokio::test] + async fn test_slow_gets_killed() -> anyhow::Result<()> { + let start = Instant::now(); + let response = exec( + r#"function handler(input) { + var j = 0; + for (var i = 0 ; i < 10000000000; ++i) {} + }"#, + "null", + ) + .await; + let dur = start.elapsed(); + let Err(ScriptError::ProcessTimeout) = response else { + anyhow::bail!("expected an interrupted exception; got {response:?}"); + }; + assert!(dur >= Duration::from_millis(9)); + assert!(dur <= Duration::from_millis(100)); + Ok(()) + } + + #[tokio::test] + async fn test_leaky_gets_killed() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { + let array = [1, 2, 3, 4]; + for (var i = 0 ; i < 10000; ++i) { + array.push(JSON.parse(JSON.stringify(array))); + } + return 1; + }"#, + "null", + ) + .await; + let Err(ScriptError::OutOfMemory) = response else { + anyhow::bail!("expected an OOM error; got {response:?}"); + }; + Ok(()) + } + + #[tokio::test] + #[traced_test] + async fn test_console_log_goes_to_tracing() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { + console.log("logged at info"); + console.error("logged at error"); + return 42 + }"#, + "null", + ) + .await?; + assert_eq!(response, "42"); + assert!(logs_contain("log called from script")); + assert!(logs_contain("error called from script")); + Ok(()) + } + + #[tokio::test] + async fn test_allows_globals() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { + newOutput = 42; + return newOutput; + }"#, + "null", + ) + .await?; + assert_eq!(response, "42"); + Ok(()) + } + + #[tokio::test] + async fn test_error_message() -> anyhow::Result<()> { + let response = exec( + r#"function handler(input) { + input.map((x) => x + 1); + return input; + }"#, + "{}", + ) + .await; + let err = response.expect_err("should return an error"); + let ScriptError::ExecutionException { message, stack } = err else { + panic!("Unexpected error") + }; + assert_eq!(message, "not a function"); + assert_eq!( + stack.as_deref(), + Some(" at handler (eval_script:2:23)\n at (eval_script:5:21)\n") + ); + Ok(()) + } + + #[tokio::test] + async fn test_payload_is_accessible() -> anyhow::Result<()> { + let result = run_script( + r#"function handler(input) { return input.x + 1; }"#, + r#"{"x":41}"#, + Duration::from_millis(100), + ) + .await?; + assert_eq!(result, "42"); + Ok(()) + } + + #[tokio::test] + async fn test_return_map() -> anyhow::Result<()> { + let result = run_script( + r#"function handler(input) { return {"foo": {"bar": 123, "x": true}}; }"#, + r#"{"x":41}"#, + Duration::from_millis(100), + ) + .await?; + assert_eq!(result, "{\"foo\":{\"bar\":123,\"x\":true}}"); + Ok(()) + } +} diff --git a/crates/diom-transformations/src/lib.rs b/crates/diom-transformations/src/lib.rs new file mode 100644 index 000000000..bdc400f6e --- /dev/null +++ b/crates/diom-transformations/src/lib.rs @@ -0,0 +1,5 @@ +mod engine; +mod worker; + +pub use engine::ScriptError; +pub use worker::{run_as_worker, run_script}; diff --git a/crates/diom-transformations/src/worker.rs b/crates/diom-transformations/src/worker.rs new file mode 100644 index 000000000..f17fed102 --- /dev/null +++ b/crates/diom-transformations/src/worker.rs @@ -0,0 +1,477 @@ +use std::{ + collections::HashMap, + path::PathBuf, + sync::{ + Arc, OnceLock, + atomic::{AtomicU64, Ordering}, + }, + thread::available_parallelism, + time::Duration, +}; + +use bytes::Bytes; +use futures_util::{SinkExt, StreamExt}; +use serde::{Deserialize, Serialize}; +use tokio::{ + process::{Child, ChildStdin, ChildStdout, Command}, + sync::{Mutex, Semaphore, mpsc, oneshot}, + time::timeout, +}; +use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec}; + +use crate::engine::ScriptError; + +// Types for the std::io IPC protocol + +#[derive(Serialize, Deserialize)] +struct WorkerRequestMsg { + id: u64, + script: String, + input: String, + timeout_ms: u64, +} + +#[derive(Serialize, Deserialize)] +struct WorkerResponseMsg { + id: u64, + response: WorkerResponse, +} + +#[derive(Serialize, Deserialize)] +enum WorkerResponse { + Ok(String), + Err(WorkerError), +} + +#[derive(Serialize, Deserialize)] +#[serde(tag = "kind")] +enum WorkerError { + InternalError { + message: String, + }, + InvalidInputEncoding { + message: String, + }, + InvalidOutputDecoding { + message: String, + }, + NoOutput, + ExecutionException { + message: String, + stack: Option, + }, + InvalidPromise, + ProcessTimeout, + OutOfMemory, +} + +impl From for WorkerError { + fn from(e: ScriptError) -> Self { + match e { + ScriptError::InternalError(err) => WorkerError::InternalError { + message: err.to_string(), + }, + ScriptError::InvalidInputEncoding(err) => WorkerError::InvalidInputEncoding { + message: err.to_string(), + }, + ScriptError::InvalidOutputDecoding(err) => WorkerError::InvalidOutputDecoding { + message: err.to_string(), + }, + ScriptError::NoOutput => WorkerError::NoOutput, + ScriptError::ExecutionException { message, stack } => { + WorkerError::ExecutionException { message, stack } + } + ScriptError::InvalidPromise => WorkerError::InvalidPromise, + ScriptError::ProcessTimeout => WorkerError::ProcessTimeout, + ScriptError::OutOfMemory => WorkerError::OutOfMemory, + } + } +} + +impl From for ScriptError { + fn from(e: WorkerError) -> Self { + match e { + WorkerError::InternalError { .. } => { + ScriptError::InternalError(rquickjs::Error::Unknown) + } + WorkerError::InvalidInputEncoding { .. } => { + ScriptError::InvalidInputEncoding(rquickjs::Error::Unknown) + } + WorkerError::InvalidOutputDecoding { .. } => { + ScriptError::InvalidOutputDecoding(rquickjs::Error::Unknown) + } + WorkerError::NoOutput => ScriptError::NoOutput, + WorkerError::ExecutionException { message, stack } => { + ScriptError::ExecutionException { message, stack } + } + WorkerError::InvalidPromise => ScriptError::InvalidPromise, + WorkerError::ProcessTimeout => ScriptError::ProcessTimeout, + WorkerError::OutOfMemory => ScriptError::OutOfMemory, + } + } +} + +fn encode(msg: &T) -> anyhow::Result { + Ok(rmp_serde::to_vec_named(msg)?.into()) +} + +fn decode Deserialize<'de>>(bytes: &[u8]) -> anyhow::Result { + Ok(rmp_serde::from_slice(bytes)?) +} + +/// This handle manages an executed subprocess that can handle multiple concurrent transformations. +struct WorkerProcess { + /// Send requests to the dedicated writer task. + request_tx: mpsc::UnboundedSender, + /// Oneshot senders waiting for their response, keyed by request ID. + pending: Arc>>>, + next_id: AtomicU64, + /// Keep the child alive for kill-on-drop. + _child: Mutex, +} + +impl WorkerProcess { + fn spawn(exe_path: &PathBuf) -> anyhow::Result> { + let mut child = Command::new(exe_path) + .arg("transform-worker") + .env_clear() + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::inherit()) + .kill_on_drop(true) + .spawn()?; + + let stdin = FramedWrite::new( + child.stdin.take().expect("stdin piped"), + LengthDelimitedCodec::new(), + ); + let stdout = FramedRead::new( + child.stdout.take().expect("stdout piped"), + LengthDelimitedCodec::new(), + ); + + let pending: Arc>>> = + Arc::new(Mutex::new(HashMap::new())); + + let (request_tx, request_rx) = mpsc::unbounded_channel(); + + tokio::spawn(Self::writer_task(stdin, request_rx)); + tokio::spawn(Self::reader_task(stdout, Arc::clone(&pending))); + + Ok(Arc::new(WorkerProcess { + request_tx, + pending, + next_id: AtomicU64::new(0), + _child: Mutex::new(child), + })) + } + + fn is_alive(&self) -> bool { + !self.request_tx.is_closed() + } + + async fn send(&self, req: WorkerRequestMsg) -> WorkerResponse { + let id = req.id; + let (tx, rx) = oneshot::channel(); + self.pending.lock().await.insert(id, tx); + + if self.request_tx.send(req).is_err() { + self.pending.lock().await.remove(&id); + return WorkerResponse::Err(WorkerError::InternalError { + message: "worker subprocess exited".into(), + }); + } + + match rx.await { + Ok(response) => response, + Err(_) => WorkerResponse::Err(WorkerError::InternalError { + message: "worker subprocess exited".into(), + }), + } + } + + async fn writer_task( + mut stdin: FramedWrite, + mut rx: mpsc::UnboundedReceiver, + ) { + while let Some(msg) = rx.recv().await { + let Ok(bytes) = encode(&msg) else { break }; + if stdin.send(bytes).await.is_err() { + break; + } + } + } + + async fn reader_task( + mut stdout: FramedRead, + pending: Arc>>>, + ) { + while let Some(frame) = stdout.next().await { + match frame { + Err(_) => break, + Ok(bytes) => { + let Ok(msg) = decode::(&bytes) else { + continue; + }; + let mut map = pending.lock().await; + if let Some(tx) = map.remove(&msg.id) { + let _ = tx.send(msg.response); + } + } + } + } + // Stream ended (subprocess exited); fail all in-flight requests. + let mut map = pending.lock().await; + for (_, tx) in map.drain() { + let _ = tx.send(WorkerResponse::Err(WorkerError::InternalError { + message: "worker subprocess exited unexpectedly".into(), + })); + } + } +} + +struct Worker { + permits: Arc, + /// The single subprocess handle. Replaced when the process dies. + process: Mutex>>, + exe_path: PathBuf, +} + +impl Worker { + fn new(max_workers: usize, exe_path: PathBuf) -> Self { + Worker { + permits: Arc::new(Semaphore::new(max_workers)), + process: Mutex::new(None), + exe_path, + } + } + + async fn run_script( + &self, + script: String, + input: String, + max_duration: Duration, + ) -> Result { + let _permit = self + .permits + .acquire() + .await + .expect("semaphore never closed"); + + let handle = self.get_or_spawn().await?; + + let id = handle.next_id.fetch_add(1, Ordering::Relaxed); + let req = WorkerRequestMsg { + id, + script, + input, + timeout_ms: max_duration.as_millis() as u64, + }; + + let deadline = max_duration + Duration::from_secs(1); + match timeout(deadline, handle.send(req)).await { + Ok(WorkerResponse::Ok(output)) => Ok(output), + Ok(WorkerResponse::Err(e)) => Err(ScriptError::from(e)), + Err(_elapsed) => { + tracing::warn!("worker subprocess timed out, replacing it"); + *self.process.lock().await = None; + Err(ScriptError::ProcessTimeout) + } + } + } + + async fn get_or_spawn(&self) -> Result, ScriptError> { + let mut guard = self.process.lock().await; + if let Some(handle) = guard.as_ref() + && handle.is_alive() + { + return Ok(Arc::clone(handle)); + } + let handle = WorkerProcess::spawn(&self.exe_path).map_err(|e| { + tracing::error!(err = ?e, "failed to spawn transform worker"); + ScriptError::InternalError(rquickjs::Error::Unknown) + })?; + *guard = Some(Arc::clone(&handle)); + Ok(handle) + } +} + +// --------------------------------------------------------------------------- +// Global singleton +// --------------------------------------------------------------------------- + +static WORKER: OnceLock = OnceLock::new(); + +fn get_worker() -> &'static Worker { + WORKER.get_or_init(|| { + let max_workers = available_parallelism().map(|n| n.get()).unwrap_or(4); + let exe_path = std::env::current_exe().expect("cannot determine current exe path"); + Worker::new(max_workers, exe_path) + }) +} + +/// This function lets you run a specific script +pub async fn run_script( + script: impl Into, + input: impl Into, + max_duration: Duration, +) -> Result { + get_worker() + .run_script(script.into(), input.into(), max_duration) + .await +} + +#[cfg(target_os = "linux")] +fn apply_sandboxing() -> anyhow::Result<()> { + apply_seccomp()?; + Ok(()) +} + +#[cfg(not(target_os = "linux"))] +fn apply_sandboxing() -> anyhow::Result<()> { + Ok(()) +} + +#[cfg(target_os = "linux")] +fn apply_seccomp() -> anyhow::Result<()> { + use seccompiler::{BpfProgram, SeccompAction, SeccompFilter, TargetArch}; + use std::collections::BTreeMap; + + let arch = match std::env::consts::ARCH { + "x86_64" => TargetArch::x86_64, + "aarch64" => TargetArch::aarch64, + other => { + tracing::warn!( + arch = other, + "seccomp BPF not supported on this architecture, skipping" + ); + return Ok(()); + } + }; + + // Allowlist: only the syscalls listed here are permitted inside the worker + // subprocess. Everything else (socket, open/openat, execve, fork, …) causes + // the kernel to send SIGSYS and kill the process immediately. + // + // Notably absent (blocked): + // file access – open, openat, creat, unlink, rename, … + // networking – socket, connect, bind, listen, sendto, recvfrom, … + // new processes – fork, vfork, execve, execveat + // (clone is allowed; it is needed for tokio/rquickjs thread pools) + #[rustfmt::skip] + let allowed: &[libc::c_long] = &[ + // I/O on already-open file descriptors (stdin / stdout are pre-opened) + libc::SYS_read, libc::SYS_write, + libc::SYS_readv, libc::SYS_writev, + libc::SYS_pread64, libc::SYS_pwrite64, + libc::SYS_close, + libc::SYS_fstat, libc::SYS_newfstatat, + libc::SYS_fcntl, // non-blocking / close-on-exec flags + libc::SYS_ioctl, // terminal/pipe detection by libc/tracing + + // Memory management + libc::SYS_mmap, libc::SYS_mprotect, libc::SYS_munmap, + libc::SYS_brk, libc::SYS_mremap, libc::SYS_madvise, + + // Threads and synchronisation (clone = thread creation; no fork/exec) + libc::SYS_clone, + libc::SYS_clone3, + libc::SYS_futex, + libc::SYS_set_robust_list, libc::SYS_get_robust_list, + libc::SYS_set_tid_address, + libc::SYS_sched_yield, libc::SYS_sched_getaffinity, + + // Signals + libc::SYS_rt_sigaction, libc::SYS_rt_sigprocmask, + libc::SYS_rt_sigreturn, libc::SYS_sigaltstack, + + // Time + libc::SYS_nanosleep, libc::SYS_clock_gettime, + libc::SYS_clock_getres, libc::SYS_clock_nanosleep, + libc::SYS_gettimeofday, + libc::SYS_timerfd_settime, libc::SYS_timerfd_gettime, + libc::SYS_timerfd_create, + + // Async I/O multiplexing (tokio epoll reactor) + libc::SYS_epoll_create1, libc::SYS_epoll_ctl, + libc::SYS_epoll_wait, libc::SYS_epoll_pwait, + libc::SYS_epoll_pwait2, + libc::SYS_eventfd2, + libc::SYS_poll, libc::SYS_ppoll, + libc::SYS_select, libc::SYS_pselect6, + libc::SYS_pipe2, // waker pipe used by some tokio internals + + // Process / thread identity (read-only) + libc::SYS_getpid, libc::SYS_gettid, + libc::SYS_getuid, libc::SYS_geteuid, + libc::SYS_getgid, libc::SYS_getegid, + libc::SYS_prlimit64, + + // Miscellaneous (allocator / runtime init) + libc::SYS_prctl, // thread naming, PR_SET_NAME, etc. + libc::SYS_arch_prctl, // x86-64 TLS segment setup + libc::SYS_getrandom, // entropy for hash maps, UUIDs + libc::SYS_rseq, // glibc 2.35+ restartable sequences + + // Exit + libc::SYS_exit, libc::SYS_exit_group, + libc::SYS_restart_syscall, + + // Extra needed by tokio + libc::SYS_openat + ]; + + let rules: BTreeMap> = + allowed.iter().map(|&nr| (nr, vec![])).collect(); + + let filter = SeccompFilter::new( + rules, + SeccompAction::KillProcess, + SeccompAction::Allow, + arch, + )?; + + let prog: BpfProgram = filter.try_into()?; + if let Err(e) = seccompiler::apply_filter(&prog) { + tracing::warn!(error = %e, "failed to apply seccomp BPF filter, worker will run unsandboxed"); + return Ok(()); + } + + tracing::debug!("seccomp BPF filter applied to worker subprocess"); + Ok(()) +} + +/// Worker event loop - this is what's run in the worker subprocess. +pub async fn run_as_worker() -> anyhow::Result<()> { + apply_sandboxing()?; + + let stdout = Arc::new(Mutex::new(FramedWrite::new( + tokio::io::stdout(), + LengthDelimitedCodec::new(), + ))); + let mut reader = FramedRead::new(tokio::io::stdin(), LengthDelimitedCodec::new()); + while let Some(frame) = reader.next().await { + let bytes = frame?; + let req: WorkerRequestMsg = rmp_serde::from_slice(&bytes)?; + + let stdout = Arc::clone(&stdout); + tokio::spawn(async move { + let duration = Duration::from_millis(req.timeout_ms); + let result = crate::engine::run_script(req.script, req.input, duration).await; + let response = match result { + Ok(output) => WorkerResponse::Ok(output), + Err(e) => WorkerResponse::Err(WorkerError::from(e)), + }; + let msg = WorkerResponseMsg { + id: req.id, + response, + }; + if let Ok(bytes) = encode(&msg) { + let mut w = stdout.lock().await; + let _ = w.send(bytes).await; + } + }); + } + + Ok(()) +} diff --git a/openapi.json b/openapi.json index 2d6c4f7e7..bade28415 100644 --- a/openapi.json +++ b/openapi.json @@ -4004,6 +4004,83 @@ ] } }, + "/api/v1.transformations.execute": { + "post": { + "tags": [ + "Transformations" + ], + "summary": "Execute", + "description": "Execute a JavaScript transformation script against a payload and return the result.", + "operationId": "v1.transformations.execute", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TransformIn" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TransformOut" + } + } + } + }, + "400": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StandardErrorBody" + } + } + } + }, + "401": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StandardErrorBody" + } + } + } + }, + "403": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StandardErrorBody" + } + } + } + }, + "422": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ValidationErrorBody" + } + } + } + } + }, + "security": [ + { + "HTTPBearer": [] + } + ] + } + }, "/api/v1.health.ping": { "get": { "tags": [ @@ -6604,6 +6681,41 @@ "timestamp" ] }, + "TransformIn": { + "type": "object", + "properties": { + "input": { + "description": "JSON-encoded payload passed to the script as `input`.", + "type": "string" + }, + "script": { + "description": "JavaScript source. Must define a `handler(input)` function.", + "type": "string" + }, + "max_duration_ms": { + "description": "How long to let the script run before being killed.", + "type": "integer", + "format": "uint64", + "default": 500 + } + }, + "required": [ + "input", + "script" + ] + }, + "TransformOut": { + "type": "object", + "properties": { + "output": { + "description": "JSON-encoded value returned by the script's `handler` function.", + "type": "string" + } + }, + "required": [ + "output" + ] + }, "ValidationErrorBody": { "type": "object", "properties": { diff --git a/server/Cargo.toml b/server/Cargo.toml index 7bf007bae..5a9e744de 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -9,6 +9,7 @@ anyhow.workspace = true clap.workspace = true diom.workspace = true diom-core.workspace = true +diom-transformations.workspace = true dotenvy.workspace = true mimalloc.workspace = true opentelemetry.workspace = true diff --git a/server/src/main.rs b/server/src/main.rs index fc9d73d77..8c71a3ec0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -37,14 +37,21 @@ enum Commands { }, /// Run the server (this is also the default if no subcommand is passed) Server, + #[clap(hide = true)] + TransformWorker, } #[tokio::main] async fn main() -> anyhow::Result<()> { - _ = dotenv(); - let args = Args::parse(); + if matches!(args.command, Some(Commands::TransformWorker)) { + diom_transformations::run_as_worker().await?; + return Ok(()); + } + + _ = dotenv(); + // Handle commands that don't need configuration first if let Some(Commands::Healthcheck { server_url }) = args.command { let client = reqwest::Client::new(); @@ -79,6 +86,9 @@ async fn main() -> anyhow::Result<()> { otel::setup_metrics(&cfg); run(cfg).await } + Some(Commands::TransformWorker) => { + unreachable!("TransformWorker is handled before config loading") + } }; #[allow(clippy::disallowed_methods)] diff --git a/src/v1/endpoints/mod.rs b/src/v1/endpoints/mod.rs index 1ecc2a550..1ae0de80d 100644 --- a/src/v1/endpoints/mod.rs +++ b/src/v1/endpoints/mod.rs @@ -19,3 +19,4 @@ pub mod idempotency; pub mod kv; pub mod msgs; pub mod rate_limit; +pub mod transformations; diff --git a/src/v1/endpoints/transformations.rs b/src/v1/endpoints/transformations.rs new file mode 100644 index 000000000..0b5c58f99 --- /dev/null +++ b/src/v1/endpoints/transformations.rs @@ -0,0 +1,55 @@ +use std::time::Duration; + +use aide::axum::{ApiRouter, routing::post_with}; +use axum::extract::State; +use diom_core::types::DurationMs; +use diom_derive::aide_annotate; +use diom_proto::MsgPackOrJson; +use diom_transformations::run_script; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use validator::Validate; + +use crate::{AppState, error::Result, v1::utils::openapi_tag}; + +fn default_max_duration_ms() -> DurationMs { + DurationMs::from(500u64) +} + +#[derive(Deserialize, Validate, JsonSchema)] +pub struct TransformIn { + /// JSON-encoded payload passed to the script as `input`. + pub input: String, + /// JavaScript source. Must define a `handler(input)` function. + pub script: String, + /// How long to let the script run before being killed. + #[serde(default = "default_max_duration_ms")] + pub max_duration_ms: DurationMs, +} + +#[derive(Serialize, JsonSchema)] +pub struct TransformOut { + /// JSON-encoded value returned by the script's `handler` function. + pub output: String, +} + +/// Execute a JavaScript transformation script against a payload and return the result. +#[aide_annotate(op_id = "v1.transformations.execute")] +async fn execute( + State(_state): State, + MsgPackOrJson(data): MsgPackOrJson, +) -> Result> { + let output = run_script( + &data.script, + &data.input, + Duration::from_millis(data.max_duration_ms.as_millis()), + ) + .await + .map_err(diom_error::Error::from)?; + Ok(MsgPackOrJson(TransformOut { output })) +} + +pub fn router() -> ApiRouter { + let tag = openapi_tag("Transformations"); + ApiRouter::new().api_route_with(execute_path, post_with(execute, execute_operation), &tag) +} diff --git a/src/v1/mod.rs b/src/v1/mod.rs index b8c0ae18b..29a98cc4c 100644 --- a/src/v1/mod.rs +++ b/src/v1/mod.rs @@ -21,7 +21,8 @@ pub fn router(state: Option) -> ApiRouter { .merge(endpoints::kv::router()) .merge(endpoints::rate_limit::router()) .merge(endpoints::idempotency::router()) - .merge(endpoints::msgs::router()); + .merge(endpoints::msgs::router()) + .merge(endpoints::transformations::router()); if let Some(state) = state { authenticated_router = diff --git a/tests/it/main.rs b/tests/it/main.rs index 0eb637747..1e0369bd5 100644 --- a/tests/it/main.rs +++ b/tests/it/main.rs @@ -7,8 +7,17 @@ mod kv; mod msgpack; mod msgs; mod rate_limit; +mod transformations; #[ctor::ctor] fn test_setup() { + // If this process was re-exec'd as a transform worker subprocess, run the + // worker event loop and exit — never reaching the test harness. + if std::env::args().nth(1).as_deref() == Some("transform-worker") { + let rt = tokio::runtime::Runtime::new().expect("tokio runtime"); + rt.block_on(diom_transformations::run_as_worker()) + .expect("worker exited with error"); + std::process::exit(0); + } diom::setup_tracing_for_tests(); } diff --git a/tests/it/transformations.rs b/tests/it/transformations.rs new file mode 100644 index 000000000..87590f816 --- /dev/null +++ b/tests/it/transformations.rs @@ -0,0 +1,111 @@ +use serde_json::json; +use test_utils::{ + StatusCode, TestResult, + server::{TestContext, start_server}, +}; + +#[tokio::test] +async fn test_execute_identity() -> TestResult { + let TestContext { + client, + handle: _handle, + .. + } = start_server().await; + + let response = client + .post("v1.transformations.execute") + .json(json!({ + "input": r#"{"x": 42}"#, + "script": "function handler(input) { return input; }" + })) + .await? + .expect(StatusCode::OK) + .json(); + + assert_eq!(response["output"], json!(r#"{"x":42}"#)); + + Ok(()) +} + +#[tokio::test] +async fn test_execute_transforms_input() -> TestResult { + let TestContext { + client, + handle: _handle, + .. + } = start_server().await; + + let response = client + .post("v1.transformations.execute") + .json(json!({ + "input": r#"{"x": 1}"#, + "script": "function handler(input) { return { result: input.x + 41 }; }" + })) + .await? + .expect(StatusCode::OK) + .json(); + + assert_eq!(response["output"], json!(r#"{"result":42}"#)); + + Ok(()) +} + +#[tokio::test] +async fn test_execute_script_exception_is_bad_request() -> TestResult { + let TestContext { + client, + handle: _handle, + .. + } = start_server().await; + + client + .post("v1.transformations.execute") + .json(json!({ + "input": "null", + "script": r#"function handler(input) { throw new Error("intentional failure"); }"# + })) + .await? + .expect(StatusCode::BAD_REQUEST); + + Ok(()) +} + +#[tokio::test] +async fn test_execute_timeout_is_bad_request() -> TestResult { + let TestContext { + client, + handle: _handle, + .. + } = start_server().await; + + client + .post("v1.transformations.execute") + .json(json!({ + "input": "null", + "script": r#"function handler(input) { for (var i = 0; i < 10000000000; ++i) {} }"# + })) + .await? + .expect(StatusCode::BAD_REQUEST); + + Ok(()) +} + +#[tokio::test] +async fn test_execute_missing_handler_is_bad_request() -> TestResult { + let TestContext { + client, + handle: _handle, + .. + } = start_server().await; + + client + .post("v1.transformations.execute") + .json(json!({ + "input": "null", + "script": "var x = 1;" + })) + .await? + .expect(StatusCode::BAD_REQUEST); + + Ok(()) +} diff --git a/z-clients/cli/src/cmds/api/mod.rs b/z-clients/cli/src/cmds/api/mod.rs index 885056ef3..2117ca84d 100644 --- a/z-clients/cli/src/cmds/api/mod.rs +++ b/z-clients/cli/src/cmds/api/mod.rs @@ -18,6 +18,7 @@ mod msgs_stream; mod msgs_topic; mod rate_limit; mod rate_limit_namespace; +mod transformations; pub(crate) use self::{ admin::AdminArgs, admin_auth_token::AdminAuthTokenArgs, admin_cluster::AdminClusterArgs, diff --git a/z-clients/cli/src/cmds/api/transformations.rs b/z-clients/cli/src/cmds/api/transformations.rs new file mode 100644 index 000000000..308343963 --- /dev/null +++ b/z-clients/cli/src/cmds/api/transformations.rs @@ -0,0 +1,38 @@ +// this file is @generated +use clap::{Args, Subcommand}; +use diom_client::DiomClient; + +#[derive(Args)] +#[command(args_conflicts_with_subcommands = true, flatten_help = true)] +pub struct TransformationsArgs { + #[command(subcommand)] + pub command: TransformationsCommands, +} + +#[derive(Subcommand)] +pub enum TransformationsCommands { + /// Execute a JavaScript transformation script against a payload and return the result. + Execute { + transform_in: crate::json::JsonOf, + }, +} + +impl TransformationsCommands { + pub async fn exec( + self, + client: &DiomClient, + color_mode: colored_json::ColorMode, + ) -> anyhow::Result<()> { + match self { + Self::Execute { transform_in } => { + let resp = client + .transformations() + .execute(transform_in.into_inner()) + .await?; + crate::json::print_json_output(&resp, color_mode)?; + } + } + + Ok(()) + } +} diff --git a/z-clients/go/apis.go b/z-clients/go/apis.go index cebcc78a0..49d81f8c3 100644 --- a/z-clients/go/apis.go +++ b/z-clients/go/apis.go @@ -53,3 +53,9 @@ type RateLimit = diom_apis.RateLimit func (diom Diom) RateLimit() RateLimit { return diom_apis.NewRateLimit(&diom.inner) } + +type Transformations = diom_apis.Transformations + +func (diom Diom) Transformations() Transformations { + return diom_apis.NewTransformations(&diom.inner) +} diff --git a/z-clients/go/internal/apis/transformations.go b/z-clients/go/internal/apis/transformations.go new file mode 100644 index 000000000..a4fb781ce --- /dev/null +++ b/z-clients/go/internal/apis/transformations.go @@ -0,0 +1,32 @@ +package diom_apis + +// This file is @generated DO NOT EDIT + +import ( + "context" + + diom_models "github.com/svix/diom/z-clients/go/internal/models" + diom_proto "github.com/svix/diom/z-clients/go/internal/proto" +) + +type Transformations struct { + client *diom_proto.HttpClient +} + +func NewTransformations(client *diom_proto.HttpClient) Transformations { + return Transformations{client} +} + +// Execute a JavaScript transformation script against a payload and return the result. +func (transformations Transformations) Execute( + ctx context.Context, + transformIn diom_models.TransformIn, +) (*diom_models.TransformOut, error) { + return diom_proto.ExecuteRequest[diom_models.TransformIn, diom_models.TransformOut]( + ctx, + transformations.client, + "POST", + "/api/v1.transformations.execute", + &transformIn, + ) +} diff --git a/z-clients/go/internal/models/transform_in.go b/z-clients/go/internal/models/transform_in.go new file mode 100644 index 000000000..5388c4b97 --- /dev/null +++ b/z-clients/go/internal/models/transform_in.go @@ -0,0 +1,9 @@ +package diom_models + +// This file is @generated DO NOT EDIT + +type TransformIn struct { + Input string `msgpack:"input"` // JSON-encoded payload passed to the script as `input`. + Script string `msgpack:"script"` // JavaScript source. Must define a `handler(input)` function. + MaxDurationMs *uint64 `msgpack:"max_duration_ms,omitempty"` // How long to let the script run before being killed. +} diff --git a/z-clients/go/internal/models/transform_out.go b/z-clients/go/internal/models/transform_out.go new file mode 100644 index 000000000..30efb5d45 --- /dev/null +++ b/z-clients/go/internal/models/transform_out.go @@ -0,0 +1,7 @@ +package diom_models + +// This file is @generated DO NOT EDIT + +type TransformOut struct { + Output string `msgpack:"output"` // JSON-encoded value returned by the script's `handler` function. +} diff --git a/z-clients/go/models.go b/z-clients/go/models.go index 8a2cf5e99..bbaf63a38 100644 --- a/z-clients/go/models.go +++ b/z-clients/go/models.go @@ -124,4 +124,6 @@ type ( SeekPosition = diom_models.SeekPosition ServerState = diom_models.ServerState StreamMsgOut = diom_models.StreamMsgOut + TransformIn = diom_models.TransformIn + TransformOut = diom_models.TransformOut ) diff --git a/z-clients/java/src/main/java/com/svix/diom/Diom.java b/z-clients/java/src/main/java/com/svix/diom/Diom.java index 769c2f4df..709cdd81e 100644 --- a/z-clients/java/src/main/java/com/svix/diom/Diom.java +++ b/z-clients/java/src/main/java/com/svix/diom/Diom.java @@ -12,6 +12,7 @@ import com.svix.diom.apis.Kv; import com.svix.diom.apis.Msgs; import com.svix.diom.apis.RateLimit; +import com.svix.diom.apis.Transformations; import okhttp3.HttpUrl; @@ -71,4 +72,8 @@ public Msgs getMsgs() { public RateLimit getRateLimit() { return new RateLimit(this.httpClient); } + + public Transformations getTransformations() { + return new Transformations(this.httpClient); + } } \ No newline at end of file diff --git a/z-clients/java/src/main/java/com/svix/diom/apis/Transformations.java b/z-clients/java/src/main/java/com/svix/diom/apis/Transformations.java new file mode 100644 index 000000000..ee2ef50e1 --- /dev/null +++ b/z-clients/java/src/main/java/com/svix/diom/apis/Transformations.java @@ -0,0 +1,40 @@ +// this file is @generated +package com.svix.diom.apis; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.svix.diom.ApiException; +import com.svix.diom.HttpClient; +import com.svix.diom.Utils; +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import okhttp3.Headers; +import okhttp3.HttpUrl; +import com.svix.diom.models.TransformIn; +import com.svix.diom.models.TransformOut; + +public class Transformations { + private final HttpClient client; + + public Transformations(HttpClient client) { + this.client = client; + } + + /** Execute a JavaScript transformation script against a payload and return the result. */ + public TransformOut execute( + final TransformIn transformIn + ) throws IOException, ApiException { + HttpUrl.Builder url = this.client.newUrlBuilder().encodedPath("/api/v1.transformations.execute"); + + return this.client.executeRequest( + "POST", + url.build(), + null, + transformIn, + TransformOut.class + ); + } +} \ No newline at end of file diff --git a/z-clients/java/src/main/java/com/svix/diom/models/TransformIn.java b/z-clients/java/src/main/java/com/svix/diom/models/TransformIn.java new file mode 100644 index 000000000..f60cabde0 --- /dev/null +++ b/z-clients/java/src/main/java/com/svix/diom/models/TransformIn.java @@ -0,0 +1,112 @@ +// this file is @generated +package com.svix.diom.models; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonValue; +import com.fasterxml.jackson.annotation.JsonFilter; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.svix.diom.Utils; +import java.util.Map; +import java.util.Set; +import java.util.List; +import java.util.Optional; +import java.util.HashMap; +import java.time.OffsetDateTime; +import java.util.LinkedHashSet; +import java.util.ArrayList; +import java.net.URI; +import java.util.Objects; +import lombok.EqualsAndHashCode; +import lombok.ToString; + +@ToString +@EqualsAndHashCode +@JsonInclude(JsonInclude.Include.NON_NULL) +@JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE) +public class TransformIn { + @JsonProperty private String input; + @JsonProperty private String script; + @JsonProperty("max_duration_ms") private Long maxDurationMs; + public TransformIn() {} + + public TransformIn input(String input) { + this.input = input; + return this; + } + + /** + * JSON-encoded payload passed to the script as `input`. + * + * @return input + */ + @javax.annotation.Nonnull + public String getInput() { + return input; + } + + public void setInput(String input) { + this.input = input; + } + + public TransformIn script(String script) { + this.script = script; + return this; + } + + /** + * JavaScript source. Must define a `handler(input)` function. + * + * @return script + */ + @javax.annotation.Nonnull + public String getScript() { + return script; + } + + public void setScript(String script) { + this.script = script; + } + + public TransformIn maxDurationMs(Long maxDurationMs) { + this.maxDurationMs = maxDurationMs; + return this; + } + + /** + * How long to let the script run before being killed. + * + * @return maxDurationMs + */ + @javax.annotation.Nullable + public Long getMaxDurationMs() { + return maxDurationMs; + } + + public void setMaxDurationMs(Long maxDurationMs) { + this.maxDurationMs = maxDurationMs; + } + + /** + * Create an instance of TransformIn given a JSON string + * + * @param jsonString JSON string + * @return An instance of TransformIn + * @throws JsonProcessingException if the JSON string is invalid with respect to TransformIn + */ + public static TransformIn fromJson(String jsonString) throws JsonProcessingException { + return Utils.getObjectMapper().readValue(jsonString, TransformIn.class); + } + + /** + * Convert an instance of TransformIn to a JSON string + * + * @return JSON string + */ + public String toJson() throws JsonProcessingException { + return Utils.getObjectMapper().writeValueAsString(this); + } +} \ No newline at end of file diff --git a/z-clients/java/src/main/java/com/svix/diom/models/TransformOut.java b/z-clients/java/src/main/java/com/svix/diom/models/TransformOut.java new file mode 100644 index 000000000..1fd7f0403 --- /dev/null +++ b/z-clients/java/src/main/java/com/svix/diom/models/TransformOut.java @@ -0,0 +1,72 @@ +// this file is @generated +package com.svix.diom.models; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonValue; +import com.fasterxml.jackson.annotation.JsonFilter; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.svix.diom.Utils; +import java.util.Map; +import java.util.Set; +import java.util.List; +import java.util.Optional; +import java.util.HashMap; +import java.time.OffsetDateTime; +import java.util.LinkedHashSet; +import java.util.ArrayList; +import java.net.URI; +import java.util.Objects; +import lombok.EqualsAndHashCode; +import lombok.ToString; + +@ToString +@EqualsAndHashCode +@JsonInclude(JsonInclude.Include.NON_NULL) +@JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE) +public class TransformOut { + @JsonProperty private String output; + public TransformOut() {} + + public TransformOut output(String output) { + this.output = output; + return this; + } + + /** + * JSON-encoded value returned by the script's `handler` function. + * + * @return output + */ + @javax.annotation.Nonnull + public String getOutput() { + return output; + } + + public void setOutput(String output) { + this.output = output; + } + + /** + * Create an instance of TransformOut given a JSON string + * + * @param jsonString JSON string + * @return An instance of TransformOut + * @throws JsonProcessingException if the JSON string is invalid with respect to TransformOut + */ + public static TransformOut fromJson(String jsonString) throws JsonProcessingException { + return Utils.getObjectMapper().readValue(jsonString, TransformOut.class); + } + + /** + * Convert an instance of TransformOut to a JSON string + * + * @return JSON string + */ + public String toJson() throws JsonProcessingException { + return Utils.getObjectMapper().writeValueAsString(this); + } +} \ No newline at end of file diff --git a/z-clients/javascript/src/apis/transformations.ts b/z-clients/javascript/src/apis/transformations.ts new file mode 100644 index 000000000..33f71e3f7 --- /dev/null +++ b/z-clients/javascript/src/apis/transformations.ts @@ -0,0 +1,32 @@ +// this file is @generated + +import { + type TransformIn, + TransformInSerializer, +} from '../models/transformIn'; +import { + type TransformOut, + TransformOutSerializer, +} from '../models/transformOut'; +import { HttpMethod, DiomRequest, type DiomRequestContext } from "../request"; + +export class Transformations { + public constructor(private readonly requestCtx: DiomRequestContext) {} + + /** Execute a JavaScript transformation script against a payload and return the result. */ + public execute( + transformIn: TransformIn, + ): Promise { + const request = new DiomRequest(HttpMethod.POST, "/api/v1.transformations.execute"); + + request.setBody( + TransformInSerializer._toJsonObject(transformIn) + ); + + return request.send( + this.requestCtx, + TransformOutSerializer._fromJsonObject, + ); + } +} + diff --git a/z-clients/javascript/src/client.ts b/z-clients/javascript/src/client.ts index 3df96de94..75e724345 100644 --- a/z-clients/javascript/src/client.ts +++ b/z-clients/javascript/src/client.ts @@ -9,6 +9,7 @@ import { Idempotency } from "./apis/idempotency"; import { Kv } from "./apis/kv"; import { Msgs } from "./apis/msgs"; import { RateLimit } from "./apis/rateLimit"; +import { Transformations } from "./apis/transformations"; export { Admin, @@ -19,6 +20,7 @@ export { Kv, Msgs, RateLimit, + Transformations, }; export class Diom { @@ -59,4 +61,8 @@ export class Diom { public get rateLimit(){ return new RateLimit(this.requestCtx); } + + public get transformations(){ + return new Transformations(this.requestCtx); + } } \ No newline at end of file diff --git a/z-clients/javascript/src/models/transformIn.ts b/z-clients/javascript/src/models/transformIn.ts new file mode 100644 index 000000000..4d7e0031a --- /dev/null +++ b/z-clients/javascript/src/models/transformIn.ts @@ -0,0 +1,30 @@ +// this file is @generated + +export interface TransformIn { + /** JSON-encoded payload passed to the script as `input`. */ + input: string; + /** JavaScript source. Must define a `handler(input)` function. */ + script: string; + /** How long to let the script run before being killed. */ + maxDurationMs?: number; +} + +export const TransformInSerializer = { + // biome-ignore lint/suspicious/noExplicitAny: intentional any + _fromJsonObject(object: any): TransformIn { + return { + input: object['input'], + script: object['script'], + maxDurationMs: object['max_duration_ms'], + }; + }, + + // biome-ignore lint/suspicious/noExplicitAny: intentional any + _toJsonObject(self: TransformIn): any { + return { + 'input': self.input, + 'script': self.script, + 'max_duration_ms': self.maxDurationMs, + }; + } +} \ No newline at end of file diff --git a/z-clients/javascript/src/models/transformOut.ts b/z-clients/javascript/src/models/transformOut.ts new file mode 100644 index 000000000..9cbed0627 --- /dev/null +++ b/z-clients/javascript/src/models/transformOut.ts @@ -0,0 +1,22 @@ +// this file is @generated + +export interface TransformOut { + /** JSON-encoded value returned by the script's `handler` function. */ + output: string; +} + +export const TransformOutSerializer = { + // biome-ignore lint/suspicious/noExplicitAny: intentional any + _fromJsonObject(object: any): TransformOut { + return { + output: object['output'], + }; + }, + + // biome-ignore lint/suspicious/noExplicitAny: intentional any + _toJsonObject(self: TransformOut): any { + return { + 'output': self.output, + }; + } +} \ No newline at end of file diff --git a/z-clients/python/diom/__init__.py b/z-clients/python/diom/__init__.py index 25ebc2ec4..4c95e84a0 100644 --- a/z-clients/python/diom/__init__.py +++ b/z-clients/python/diom/__init__.py @@ -17,6 +17,8 @@ MsgsAsync, RateLimit, RateLimitAsync, + Transformations, + TransformationsAsync, ) from .client_base import ClientBase from .options import DiomOptions @@ -55,6 +57,10 @@ def msgs(self) -> Msgs: def rate_limit(self) -> RateLimit: return RateLimit(self._client) + @property + def transformations(self) -> Transformations: + return Transformations(self._client) + class DiomAsync(ClientBase): @property @@ -89,5 +95,9 @@ def msgs(self) -> MsgsAsync: def rate_limit(self) -> RateLimitAsync: return RateLimitAsync(self._client) + @property + def transformations(self) -> TransformationsAsync: + return TransformationsAsync(self._client) + __all__ = ["Diom", "DiomAsync", "DiomOptions"] diff --git a/z-clients/python/diom/apis/__init__.py b/z-clients/python/diom/apis/__init__.py index 1d5568659..1098abd9e 100644 --- a/z-clients/python/diom/apis/__init__.py +++ b/z-clients/python/diom/apis/__init__.py @@ -18,6 +18,7 @@ from .msgs_topic import MsgsTopic, MsgsTopicAsync from .rate_limit import RateLimit, RateLimitAsync from .rate_limit_namespace import RateLimitNamespace, RateLimitNamespaceAsync +from .transformations import Transformations, TransformationsAsync __all__ = [ @@ -59,4 +60,6 @@ "RateLimitAsync", "RateLimitNamespace", "RateLimitNamespaceAsync", + "Transformations", + "TransformationsAsync", ] diff --git a/z-clients/python/diom/apis/transformations.py b/z-clients/python/diom/apis/transformations.py new file mode 100644 index 000000000..899263ded --- /dev/null +++ b/z-clients/python/diom/apis/transformations.py @@ -0,0 +1,39 @@ +# This file is @generated + +from ..internal.api_common import ApiBase +from ..models import ( + TransformIn, + TransformOut, +) + + +class TransformationsAsync(ApiBase): + async def execute( + self, + transform_in: TransformIn, + ) -> TransformOut: + """Execute a JavaScript transformation script against a payload and return the result.""" + body = transform_in.model_dump(exclude_none=True) + + return await self._request_asyncio( + method="post", + path="/api/v1.transformations.execute", + body=body, + response_type=TransformOut, + ) + + +class Transformations(ApiBase): + def execute( + self, + transform_in: TransformIn, + ) -> TransformOut: + """Execute a JavaScript transformation script against a payload and return the result.""" + body = transform_in.model_dump(exclude_none=True) + + return self._request_sync( + method="post", + path="/api/v1.transformations.execute", + body=body, + response_type=TransformOut, + ) diff --git a/z-clients/python/diom/models/__init__.py b/z-clients/python/diom/models/__init__.py index 8ce9ab2bf..327b67d43 100644 --- a/z-clients/python/diom/models/__init__.py +++ b/z-clients/python/diom/models/__init__.py @@ -118,6 +118,8 @@ from .seek_position import SeekPosition from .server_state import ServerState from .stream_msg_out import StreamMsgOut +from .transform_in import TransformIn +from .transform_out import TransformOut __all__ = [ @@ -240,4 +242,6 @@ "SeekPosition", "ServerState", "StreamMsgOut", + "TransformIn", + "TransformOut", ] diff --git a/z-clients/python/diom/models/transform_in.py b/z-clients/python/diom/models/transform_in.py new file mode 100644 index 000000000..58ccb86e8 --- /dev/null +++ b/z-clients/python/diom/models/transform_in.py @@ -0,0 +1,16 @@ +# this file is @generated +import typing as t +from pydantic import Field + +from ..internal.base_model import BaseModel + + +class TransformIn(BaseModel): + input: str + """JSON-encoded payload passed to the script as `input`.""" + + script: str + """JavaScript source. Must define a `handler(input)` function.""" + + max_duration_ms: t.Optional[int] = Field(default=None, alias="max_duration_ms") + """How long to let the script run before being killed.""" diff --git a/z-clients/python/diom/models/transform_out.py b/z-clients/python/diom/models/transform_out.py new file mode 100644 index 000000000..77bf97d4f --- /dev/null +++ b/z-clients/python/diom/models/transform_out.py @@ -0,0 +1,8 @@ +# this file is @generated + +from ..internal.base_model import BaseModel + + +class TransformOut(BaseModel): + output: str + """JSON-encoded value returned by the script's `handler` function.""" diff --git a/z-clients/rust/src/api/mod.rs b/z-clients/rust/src/api/mod.rs index 794b1de4d..5fcd4463a 100644 --- a/z-clients/rust/src/api/mod.rs +++ b/z-clients/rust/src/api/mod.rs @@ -20,6 +20,7 @@ mod msgs_stream; mod msgs_topic; mod rate_limit; mod rate_limit_namespace; +mod transformations; pub use self::{ admin::Admin, admin_auth_token::AdminAuthToken, admin_cluster::AdminCluster, @@ -28,6 +29,7 @@ pub use self::{ idempotency_namespace::IdempotencyNamespace, kv::Kv, kv_namespace::KvNamespace, msgs::Msgs, msgs_namespace::MsgsNamespace, msgs_queue::MsgsQueue, msgs_stream::MsgsStream, msgs_topic::MsgsTopic, rate_limit::RateLimit, rate_limit_namespace::RateLimitNamespace, + transformations::Transformations, }; impl DiomClient { @@ -62,4 +64,8 @@ impl DiomClient { pub fn rate_limit(&self) -> RateLimit<'_> { RateLimit::new(&self.cfg) } + + pub fn transformations(&self) -> Transformations<'_> { + Transformations::new(&self.cfg) + } } diff --git a/z-clients/rust/src/api/transformations.rs b/z-clients/rust/src/api/transformations.rs new file mode 100644 index 000000000..a95f36e3f --- /dev/null +++ b/z-clients/rust/src/api/transformations.rs @@ -0,0 +1,20 @@ +// this file is @generated +use crate::{Configuration, error::Result, models::*}; + +pub struct Transformations<'a> { + cfg: &'a Configuration, +} + +impl<'a> Transformations<'a> { + pub(super) fn new(cfg: &'a Configuration) -> Self { + Self { cfg } + } + + /// Execute a JavaScript transformation script against a payload and return the result. + pub async fn execute(&self, transform_in: TransformIn) -> Result { + crate::request::Request::new(http::Method::POST, "/api/v1.transformations.execute") + .with_body(transform_in) + .execute(self.cfg) + .await + } +} diff --git a/z-clients/rust/src/models/mod.rs b/z-clients/rust/src/models/mod.rs index dd7682ae1..0103bf84a 100644 --- a/z-clients/rust/src/models/mod.rs +++ b/z-clients/rust/src/models/mod.rs @@ -120,6 +120,8 @@ mod retention; mod seek_position; mod server_state; mod stream_msg_out; +mod transform_in; +mod transform_out; pub use self::{ admin_auth_token_create_in::AdminAuthTokenCreateIn, @@ -193,6 +195,7 @@ pub use self::{ rate_limit_reset_out::RateLimitResetOut, rate_limit_token_bucket_config::RateLimitTokenBucketConfig, retention::Retention, seek_position::SeekPosition, server_state::ServerState, stream_msg_out::StreamMsgOut, + transform_in::TransformIn, transform_out::TransformOut, }; pub(crate) use self::{ diff --git a/z-clients/rust/src/models/transform_in.rs b/z-clients/rust/src/models/transform_in.rs new file mode 100644 index 000000000..33bd20141 --- /dev/null +++ b/z-clients/rust/src/models/transform_in.rs @@ -0,0 +1,30 @@ +// this file is @generated +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct TransformIn { + /// JSON-encoded payload passed to the script as `input`. + pub input: String, + + /// JavaScript source. Must define a `handler(input)` function. + pub script: String, + + /// How long to let the script run before being killed. + #[serde(skip_serializing_if = "Option::is_none")] + pub max_duration_ms: Option, +} + +impl TransformIn { + pub fn new(input: String, script: String) -> Self { + Self { + input, + script, + max_duration_ms: None, + } + } + + pub fn with_max_duration_ms(mut self, value: impl Into>) -> Self { + self.max_duration_ms = value.into(); + self + } +} diff --git a/z-clients/rust/src/models/transform_out.rs b/z-clients/rust/src/models/transform_out.rs new file mode 100644 index 000000000..c0c95b519 --- /dev/null +++ b/z-clients/rust/src/models/transform_out.rs @@ -0,0 +1,14 @@ +// this file is @generated +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct TransformOut { + /// JSON-encoded value returned by the script's `handler` function. + pub output: String, +} + +impl TransformOut { + pub fn new(output: String) -> Self { + Self { output } + } +}