From 71e645beb2aaee343c0fd7e5e3eaf98119419ad7 Mon Sep 17 00:00:00 2001 From: jordiori Date: Thu, 8 Oct 2026 18:20:12 +0200 Subject: [PATCH 1/3] Add workflow sink commands and typed sink flags Add `rtree workflow sink list|create|update|delete|schema` to manage one sink at a time, `--sink-table`, `--sink-http`, and `--sink-header` on `workflow create`, and accept arrays, @path, and stdin for `--sink`. Validate sinks locally against the platform's rules and point errors at `rtree workflow sink schema`. --- README.md | 37 +- src/cli.rs | 238 ++++- src/commands/{workflow.rs => workflow/mod.rs} | 38 +- src/commands/workflow/sink.rs | 965 ++++++++++++++++++ src/main.rs | 143 ++- tests/api_contracts.rs | 183 ++++ tests/live_platform/workflows.rs | 32 + 7 files changed, 1565 insertions(+), 71 deletions(-) rename src/commands/{workflow.rs => workflow/mod.rs} (97%) create mode 100644 src/commands/workflow/sink.rs diff --git a/README.md b/README.md index e2e6eb2..e7da3fd 100644 --- a/README.md +++ b/README.md @@ -257,7 +257,8 @@ rtree workflow create --name errors --database analytics --interval-seconds 60 \ --sql "INSERT INTO error_counts SELECT count() FROM events WHERE level = 'error'" rtree workflow create --name on-demand --database analytics --sql "SELECT 1" rtree workflow create --name alerts --disabled --sql - \ - --sink '{"type":"http","settings":{"url":"https://example.com/hook","headers":{"Authorization":"Bearer ..."}}}' \ + --sink-table analytics.alerts \ + --sink-http https://example.com/hook --sink-header 'Authorization: Bearer ...' \ < alerts.sql rtree workflow get rtree workflow update --interval-seconds 300 @@ -271,6 +272,12 @@ rtree workflow runs rtree workflow cancel rtree workflow logs --since 1h rtree workflow metrics --since 7d + +rtree workflow sink list +rtree workflow sink create --table analytics.alerts_v2 +rtree workflow sink update --header 'Authorization: Bearer ...' +rtree workflow sink delete +rtree workflow sink schema --type http ``` Workflow commands need both an organization and a cluster (`--org`/`--cluster`, @@ -291,11 +298,26 @@ available in either mode, including while paused. Text output shows status as `active`, `paused`, or `manual`, the next scheduled run when available, and HTTP sink URLs. -Sinks are JSON objects passed with `--sink`, up to five. Each has a `type` and a -nested `settings` object: `{"database","table"}` for `table` sinks and -`{"url","headers"}` for `http` sinks. On `update`, `--sink` replaces the whole -list. Include each existing sink's `id` to keep it; an existing HTTP sink may omit -`settings`, or its `url` or `headers`, to retain them. +A workflow has up to five sinks. `rtree workflow sink schema` lists each sink type's +fields, rules, and examples, in JSON with `--json`. + +On `create`, `--sink-table [DATABASE.]TABLE` adds a table sink (the database defaults +to the workflow's) and `--sink-http URL` an HTTP sink; `--sink-header 'Name: value'` +applies when there is exactly one `--sink-http`. `--sink` takes the API's JSON for +anything else: an object, an array of objects, `@path` to read a file, or `-` to +read stdin. Sink files keep header credentials out of shell history. +The CLI validates sinks before sending them and names the failing field. + +`rtree workflow sink` changes one sink at a time. `create` takes `--table` or +`--http` with `--header`; `update` takes `--table` for table sinks, or `--url`, +`--header`, and `--remove-header` for HTTP sinks, keeping the other stored headers. +The API replaces the whole list on each change, so the CLI reads the current sinks +and sends the others back by ID; a concurrent edit to the same workflow's sinks can +be overwritten. + +On `update`, `--sink` replaces the whole list. Include each existing sink's `id` to +keep it; an existing HTTP sink may omit `settings`, or its `url` or `headers`, to +retain them. `run` returns once the run is accepted. Reusing an `--idempotency-key` returns the same run instead of starting another. Logs and metrics cover the last 24 hours by @@ -384,7 +406,8 @@ It also logs in with the local test user's email and password, verifies API acce using the saved session, logs out, and checks that credentials are cleared and API access requires authentication again. Workflow tests cover definition updates, manual and scheduled execution, -pause/resume, cancellation, idempotency, table sink delivery, logs, and metrics. +pause/resume, cancellation, idempotency, table sink delivery, per-sink changes, +logs, and metrics. They run through the real CLI and clean up their workflows and databases. The Platform repository continues to test API endpoints directly. diff --git a/src/cli.rs b/src/cli.rs index d386dec..91fa8b2 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -433,21 +433,42 @@ pub enum TableCommand { }, } -const SINK_HELP: &str = "Sinks are JSON objects with a type and its settings:\n {\"type\":\"table\",\"settings\":{\"database\":\"analytics\",\"table\":\"alerts\"}}\n {\"type\":\"http\",\"settings\":{\"url\":\"https://example.com/hook\",\"headers\":{\"Authorization\":\"Bearer ...\"}}}"; - -const SINK_UPDATE_HELP: &str = r#"Sinks are JSON objects with a type and its settings. --sink replaces the whole -list; include the "id" of existing sinks to keep them. An existing HTTP sink may -omit "settings", or its "url" or "headers", to retain them, and a null header -value keeps the stored value: - {"type":"table","id":"","settings":{"database":"analytics","table":"alerts"}} - {"type":"http","id":""} - {"type":"http","settings":{"url":"https://example.com/hook"}} - -Example: keep an HTTP sink unchanged and point a table sink at another table. -Find the sink IDs with `rtree workflow get `, then pass every sink to keep: - rtree workflow update \ - --sink '{"type":"http","id":""}' \ - --sink '{"type":"table","id":"","settings":{"database":"analytics","table":"alerts_v2"}}'"#; +const SINK_HELP: &str = r#"Sinks: + --sink-table and --sink-http cover the common cases. --sink takes JSON for + anything else: an object, an array of objects, @ to read a file, or - + to read stdin. A workflow has at most five sinks in total. + + rtree workflow create --name alerts --sql - --sink-table analytics.alerts < alerts.sql + rtree workflow create --name hook --sql 'SELECT 1' \ + --sink-http https://example.com/hook --sink-header 'Authorization: Bearer ...' + rtree workflow create --name alerts --sql 'SELECT 1' --sink @sinks.json + +Run `rtree workflow sink schema` for each sink type's fields."#; + +const SINK_UPDATE_HELP: &str = r#"Sinks: + To add, change, or remove a single sink, use `rtree workflow sink`. + + --sink replaces the whole list. It takes a JSON object, an array of objects, + @ to read a file, or - to read stdin. Include the "id" of existing sinks + to keep them; an existing HTTP sink may omit "settings", or its "url" or + "headers", to retain them: + rtree workflow update \ + --sink '{"type":"http","id":""}' \ + --sink '{"type":"table","id":"","settings":{"database":"analytics","table":"alerts_v2"}}' + +Run `rtree workflow sink schema` for each sink type's fields."#; + +const SINK_COMMAND_HELP: &str = r#"Each change reads the workflow's current sinks and writes back the full list, +so a concurrent edit to the same workflow's sinks can be overwritten. + +Examples: + rtree workflow sink list + rtree workflow sink create --table analytics.alerts + rtree workflow sink create --http https://example.com/hook \ + --header 'Authorization: Bearer ...' + rtree workflow sink update --header 'Authorization: Bearer ...' + rtree workflow sink delete + rtree workflow sink schema --type http"#; #[derive(Subcommand)] pub enum WorkflowCommand { @@ -476,8 +497,17 @@ pub enum WorkflowCommand { /// Create the workflow paused #[arg(long)] disabled: bool, - /// Sink as a JSON object; repeat for up to five sinks - #[arg(long = "sink", value_name = "JSON")] + /// Table sink as [DATABASE.]TABLE; the database defaults to the workflow's + #[arg(long = "sink-table", value_name = "[DATABASE.]TABLE")] + sink_tables: Vec, + /// HTTP sink URL + #[arg(long = "sink-http", value_name = "URL")] + sink_urls: Vec, + /// Header for the HTTP sink, as 'Name: value'; requires exactly one --sink-http + #[arg(long = "sink-header", value_name = "HEADER", requires = "sink_urls")] + sink_headers: Vec, + /// Sink as JSON (object or array), @, or - for stdin + #[arg(long = "sink", value_name = "JSON|@PATH|-")] sinks: Vec, }, /// Update a workflow; omitted fields are left unchanged @@ -503,8 +533,12 @@ pub enum WorkflowCommand { /// Pause scheduled runs #[arg(long)] disable: bool, - /// Sink as a JSON object; replaces all sinks. Repeat for up to five. - #[arg(long = "sink", value_name = "JSON", conflicts_with = "clear_sinks")] + /// Sink as JSON (object or array), @, or - for stdin; replaces all sinks + #[arg( + long = "sink", + value_name = "JSON|@PATH|-", + conflicts_with = "clear_sinks" + )] sinks: Vec, /// Remove all sinks #[arg(long)] @@ -561,6 +595,79 @@ pub enum WorkflowCommand { #[command(flatten)] range: WorkflowTimeRange, }, + /// List, add, change, or remove a workflow's sinks + #[command(after_help = SINK_COMMAND_HELP)] + Sink { + #[command(subcommand)] + action: WorkflowSinkCommand, + }, +} + +#[derive(Subcommand)] +pub enum WorkflowSinkCommand { + /// List a workflow's sinks + List { + /// Workflow ID + workflow_id: String, + }, + /// Add a sink to a workflow + #[command(group(clap::ArgGroup::new("target").required(true).args(["table", "http"])))] + Create { + /// Workflow ID + workflow_id: String, + /// Write results to a table, as [DATABASE.]TABLE; the database defaults to the workflow's + #[arg(long, value_name = "[DATABASE.]TABLE")] + table: Option, + /// Send results to this webhook URL + #[arg(long, value_name = "URL")] + http: Option, + /// HTTP header as 'Name: value'; repeat for more headers + #[arg( + long = "header", + value_name = "HEADER", + requires = "http", + conflicts_with = "table" + )] + headers: Vec, + }, + /// Change one sink; everything not passed is left unchanged + Update { + /// Workflow ID + workflow_id: String, + /// Sink ID, from `rtree workflow sink list` + sink_id: String, + /// Table sinks: new target, as [DATABASE.]TABLE; the database defaults to the sink's current one + #[arg(long, value_name = "[DATABASE.]TABLE", conflicts_with_all = ["url", "headers", "remove_headers"])] + table: Option, + /// HTTP sinks: new URL + #[arg(long, value_name = "URL")] + url: Option, + /// HTTP sinks: add or replace a header, as 'Name: value'; other headers are kept + #[arg(long = "header", value_name = "HEADER")] + headers: Vec, + /// HTTP sinks: remove a header by name + #[arg(long = "remove-header", value_name = "NAME")] + remove_headers: Vec, + }, + /// Remove a sink from a workflow + Delete { + /// Workflow ID + workflow_id: String, + /// Sink ID, from `rtree workflow sink list` + sink_id: String, + }, + /// Show each sink type's fields, rules, and examples + Schema { + /// Sink type to describe; omit to describe every type + #[arg(long = "type", value_enum)] + sink_type: Option, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum SinkTypeArg { + Table, + Http, } #[derive(Clone, Copy, Debug)] @@ -1214,6 +1321,99 @@ mod tests { } } + #[test] + fn workflow_sink_flags_enforce_their_target_type() { + for (accepted, args) in [ + ( + false, + vec![ + "create", + "--name", + "a", + "--sql", + "x", + "--sink-header", + "A: b", + ], + ), + ( + true, + vec![ + "create", + "--name", + "a", + "--sql", + "x", + "--sink-http", + "https://a.example", + "--sink-header", + "A: b", + ], + ), + (false, vec!["sink", "create", "wf-1"]), + ( + false, + vec![ + "sink", + "create", + "wf-1", + "--table", + "t", + "--http", + "https://a.example", + ], + ), + ( + false, + vec!["sink", "create", "wf-1", "--table", "t", "--header", "A: b"], + ), + ( + true, + vec![ + "sink", + "create", + "wf-1", + "--http", + "https://a.example", + "--header", + "A: b", + ], + ), + ( + false, + vec![ + "sink", + "update", + "wf-1", + "s-1", + "--table", + "t", + "--url", + "https://a.example", + ], + ), + ( + true, + vec![ + "sink", + "update", + "wf-1", + "s-1", + "--header", + "A: b", + "--remove-header", + "C", + ], + ), + (true, vec!["sink", "schema", "--type", "http"]), + (false, vec!["sink", "schema", "--type", "queue"]), + ] { + let mut command = vec!["rtree", "workflow"]; + command.extend(&args); + assert_eq!(Cli::try_parse_from(command).is_ok(), accepted, "{args:?}"); + } + } + #[test] fn workflow_interval_and_time_ranges_are_validated() { assert!(Cli::try_parse_from([ diff --git a/src/commands/workflow.rs b/src/commands/workflow/mod.rs similarity index 97% rename from src/commands/workflow.rs rename to src/commands/workflow/mod.rs index 779738e..8abcc3c 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow/mod.rs @@ -1,3 +1,5 @@ +pub mod sink; + use anyhow::{bail, Context, Result}; use chrono::{DateTime, SecondsFormat, Utc}; use comfy_table::{Cell, CellAlignment}; @@ -5,6 +7,7 @@ use serde::de::DeserializeOwned; use serde::Deserialize; use serde_json::{json, Map, Value}; +use self::sink::SinkArgs; use super::logs::parse_duration; use super::table_output::new_cli_table; use crate::cli::WorkflowTimeRange; @@ -143,7 +146,7 @@ pub struct CreateWorkflow { pub sql: String, pub interval_seconds: Option, pub disabled: bool, - pub sinks: Vec, + pub sinks: SinkArgs, } pub struct UpdateWorkflow { @@ -170,25 +173,14 @@ fn print_response( Ok(()) } -fn parse_sinks(sinks: &[String]) -> Result> { - sinks - .iter() - .map(|sink| match serde_json::from_str::(sink) { - Ok(value @ Value::Object(_)) => Ok(value), - _ => bail!( - "invalid --sink '{sink}': expected a JSON object such as {{\"type\":\"table\",\"settings\":{{\"database\":\"analytics\",\"table\":\"alerts\"}}}}" - ), - }) - .collect() -} - fn create_body(request: CreateWorkflow) -> Result { + let sinks = sink::build(&request.sinks, &request.database)?; Ok(json!({ "name": request.name, "query": {"database": request.database, "sql": request.sql}, "enabled": !request.disabled, "interval_seconds": request.interval_seconds, - "sinks": parse_sinks(&request.sinks)?, + "sinks": sinks, })) } @@ -214,7 +206,9 @@ fn update_body(request: UpdateWorkflow) -> Result { body.insert("interval_seconds".into(), json!(interval)); } if let Some(sinks) = request.sinks { - body.insert("sinks".into(), Value::Array(parse_sinks(&sinks)?)); + let sinks = sink::read_json_args(&sinks)?; + sink::validate(&sinks)?; + body.insert("sinks".into(), Value::Array(sinks)); } if body.is_empty() { bail!("nothing to update: pass at least one of --name, --database, --sql, --interval-seconds, --enable, --disable, --sink, or --clear-sinks"); @@ -658,7 +652,12 @@ mod tests { sql: "SELECT 1".into(), interval_seconds: None, disabled: true, - sinks: vec![r#"{"type":"table","settings":{"database":"a","table":"b"}}"#.into()], + sinks: SinkArgs { + tables: vec![], + urls: vec![], + headers: vec![], + json: vec![r#"{"type":"table","settings":{"database":"a","table":"b"}}"#.into()], + }, }) .unwrap(); assert_eq!( @@ -678,7 +677,12 @@ mod tests { sql: "SELECT 1".into(), interval_seconds: Some(60), disabled: false, - sinks: vec!["[]".into()], + sinks: SinkArgs { + tables: vec![], + urls: vec![], + headers: vec![], + json: vec!["42".into()], + }, }) .unwrap_err(); assert!(err.to_string().contains("invalid --sink")); diff --git a/src/commands/workflow/sink.rs b/src/commands/workflow/sink.rs new file mode 100644 index 0000000..4341466 --- /dev/null +++ b/src/commands/workflow/sink.rs @@ -0,0 +1,965 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::io::Read; + +use anyhow::{Context, Result}; +use comfy_table::Cell; +use serde_json::{json, Map, Value}; + +use super::super::table_output::new_cli_table; +use super::{describe_sink, Workflow, WorkflowScope}; +use crate::cli::SinkTypeArg; +use crate::client::ApiClient; +use crate::output; + +/// Sink input errors exit as validation errors, whatever their wording. +macro_rules! invalid { + ($($arg:tt)*) => { + return Err(output::coded_error("validation_error", format!($($arg)*), 2)) + }; +} + +/// Mirrors the platform's `MAX_SINKS` and the HTTP sink header limit. +const MAX_SINKS: usize = 5; +const MAX_HEADERS: usize = 20; + +struct SinkField { + name: &'static str, + kind: &'static str, + required: &'static str, + description: &'static str, +} + +struct SinkType { + name: &'static str, + description: &'static str, + fields: &'static [SinkField], + rules: &'static [&'static str], + example: &'static str, + flags: &'static [&'static str], +} + +const IDENTIFIER_RULE: &str = + "Letters, digits, and '_', starting with a letter; at most 64 characters"; + +const SINK_TYPES: [SinkType; 2] = [ + SinkType { + name: "table", + description: "Inserts the rows each run returns into a RawTree table.", + fields: &[ + SinkField { + name: "settings.database", + kind: "string", + required: "always", + description: IDENTIFIER_RULE, + }, + SinkField { + name: "settings.table", + kind: "string", + required: "always", + description: IDENTIFIER_RULE, + }, + ], + rules: &[ + "Each table sink in a workflow must target a different database and table.", + "Existing table sinks are sent with their id and full settings.", + ], + example: r#"{"type":"table","settings":{"database":"analytics","table":"alerts"}}"#, + flags: &[ + "rtree workflow create ... --sink-table analytics.alerts", + "rtree workflow sink create --table analytics.alerts", + ], + }, + SinkType { + name: "http", + description: "Sends the rows each run returns to a webhook, as one JSON array.", + fields: &[ + SinkField { + name: "settings.url", + kind: "string", + required: "when adding", + description: "HTTP or HTTPS URL without user info or a fragment; at most 2048 bytes. The server rejects destinations it does not allow.", + }, + SinkField { + name: "settings.headers", + kind: "object", + required: "no", + description: "Header names to values, at most 20. Values are write-only: responses list header_names only.", + }, + ], + rules: &[ + "An existing HTTP sink (sent with its id) may omit settings, settings.url, or settings.headers to keep them.", + "In a supplied headers map, null keeps the stored value and omitted names are removed.", + "URLs and header values cannot contain {{, %, or ${; percent-encoded URLs are not supported.", + ], + example: r#"{"type":"http","settings":{"url":"https://example.com/hook","headers":{"Authorization":"Bearer ..."}}}"#, + flags: &[ + "rtree workflow create ... --sink-http https://example.com/hook --sink-header 'Authorization: Bearer ...'", + "rtree workflow sink create --http https://example.com/hook --header 'Authorization: Bearer ...'", + ], + }, +]; + +fn sink_type(name: &str) -> Option<&'static SinkType> { + SINK_TYPES.iter().find(|sink_type| sink_type.name == name) +} + +pub fn schema(sink_type_arg: Option, json_mode: bool) -> Result<()> { + let types: Vec<&SinkType> = match sink_type_arg { + Some(SinkTypeArg::Table) => vec![&SINK_TYPES[0]], + Some(SinkTypeArg::Http) => vec![&SINK_TYPES[1]], + None => SINK_TYPES.iter().collect(), + }; + let value = json!({ + "max_sinks": MAX_SINKS, + "common_fields": [ + {"name": "type", "type": "string", "required": "always", "description": "Sink type: table or http"}, + {"name": "id", "type": "string", "required": "no", "description": "Existing sink ID. Omit when adding a sink; include it to keep a sink when replacing the list."}, + ], + "types": types.iter().map(|sink_type| json!({ + "type": sink_type.name, + "description": sink_type.description, + "fields": sink_type.fields.iter().map(|field| json!({ + "name": field.name, + "type": field.kind, + "required": field.required, + "description": field.description, + })).collect::>(), + "rules": sink_type.rules, + "example": serde_json::from_str::(sink_type.example).expect("valid example"), + "flags": sink_type.flags, + })).collect::>(), + }); + output::print_result(&value, json_mode, |_| { + println!( + "Each sink is a JSON object with a \"type\", its \"settings\", and an \"id\" when it already exists." + ); + println!("A workflow has at most {MAX_SINKS} sinks."); + for sink_type in &types { + println!(); + println!("{}: {}", sink_type.name, sink_type.description); + let mut table = new_cli_table(); + table.set_header(vec!["field", "type", "required", "description"]); + for field in sink_type.fields { + table.add_row(vec![ + Cell::new(field.name), + Cell::new(field.kind), + Cell::new(field.required), + Cell::new(field.description), + ]); + } + println!("{table}"); + for rule in sink_type.rules { + println!(" - {rule}"); + } + println!(" JSON:"); + println!(" {}", sink_type.example); + println!(" Flags:"); + for flag in sink_type.flags { + println!(" {flag}"); + } + } + }); + Ok(()) +} + +/// Sink flags of `rtree workflow create`, resolved once the workflow's +/// database is known. +pub struct SinkArgs { + pub tables: Vec, + pub urls: Vec, + pub headers: Vec, + pub json: Vec, +} + +pub(super) fn build(args: &SinkArgs, default_database: &str) -> Result> { + if !args.headers.is_empty() && args.urls.len() != 1 { + invalid!("--sink-header applies to a single --sink-http; pass several HTTP sinks with headers as --sink JSON"); + } + let mut sinks = args + .tables + .iter() + .map(|target| Ok(table_sink(target, default_database))) + .collect::>>()?; + for url in &args.urls { + sinks.push(http_sink(url, &args.headers)?); + } + sinks.extend(read_json_args(&args.json)?); + validate(&sinks)?; + Ok(sinks) +} + +fn table_sink(target: &str, default_database: &str) -> Value { + let (database, table) = target.split_once('.').unwrap_or((default_database, target)); + json!({"type": "table", "settings": {"database": database, "table": table}}) +} + +fn http_sink(url: &str, headers: &[String]) -> Result { + let mut settings = Map::new(); + settings.insert("url".into(), json!(url)); + if !headers.is_empty() { + let mut map = Map::new(); + for raw in headers { + let (name, value) = parse_header(raw)?; + set_header(&mut map, name, json!(value))?; + } + settings.insert("headers".into(), Value::Object(map)); + } + Ok(json!({"type": "http", "settings": settings})) +} + +/// Errors never echo the argument because it usually carries a credential. +fn parse_header(raw: &str) -> Result<(String, String)> { + match raw.split_once(':') { + Some((name, value)) if !name.trim().is_empty() => { + Ok((name.trim().to_string(), value.trim().to_string())) + } + _ => { + invalid!("invalid header: expected 'Name: value', e.g. 'Authorization: Bearer '") + } + } +} + +fn set_header(headers: &mut Map, name: String, value: Value) -> Result<()> { + if headers + .keys() + .any(|existing| existing.eq_ignore_ascii_case(&name)) + { + invalid!("header '{name}' is passed more than once"); + } + headers.insert(name, value); + Ok(()) +} + +/// Accepts inline JSON, `@path`, or `-` for stdin; each holds a sink object or +/// an array of them. +pub(super) fn read_json_args(args: &[String]) -> Result> { + let mut sinks = Vec::new(); + for arg in args { + let (label, text) = if arg == "-" { + let mut text = String::new(); + std::io::stdin() + .read_to_string(&mut text) + .context("failed to read sinks from stdin")?; + ("from stdin".to_string(), text) + } else if let Some(path) = arg.strip_prefix('@') { + let text = std::fs::read_to_string(path) + .with_context(|| format!("failed to read sink file '{path}'"))?; + (format!("in '{path}'"), text) + } else { + (format!("'{arg}'"), arg.clone()) + }; + match serde_json::from_str::(&text) { + Ok(Value::Object(sink)) => sinks.push(Value::Object(sink)), + Ok(Value::Array(items)) => sinks.extend(items), + Ok(_) => invalid!( + "invalid --sink {label}: expected a JSON object or an array of objects, e.g. {}. Run `rtree workflow sink schema` for every field.", + SINK_TYPES[0].example + ), + Err(error) => invalid!( + "invalid --sink {label}: not valid JSON ({error}). Run `rtree workflow sink schema` for the expected shape." + ), + } + } + Ok(sinks) +} + +/// Checks the shape the platform accepts so mistakes fail before a request. +/// Types this CLI does not know are left for the server to judge. +pub(super) fn validate(sinks: &[Value]) -> Result<()> { + if sinks.len() > MAX_SINKS { + invalid!( + "a workflow supports at most {MAX_SINKS} sinks; got {}", + sinks.len() + ); + } + let mut ids = BTreeSet::new(); + let mut targets = BTreeSet::new(); + for (index, sink) in sinks.iter().enumerate() { + let kind = sink["type"].as_str().and_then(sink_type); + if let Err(reason) = validate_sink(sink, kind) { + let position = index + 1; + match kind { + Some(kind) => invalid!( + "invalid {} sink #{position}: {reason}.\nExpected, e.g.: {}\nRun `rtree workflow sink schema --type {}` for every field.", + kind.name, kind.example, kind.name + ), + None => invalid!( + "invalid sink #{position}: {reason}.\nRun `rtree workflow sink schema` for every sink type." + ), + } + } + if let Some(id) = sink["id"].as_str() { + if !ids.insert(id) { + invalid!("sink ID '{id}' appears more than once"); + } + } + if kind.is_some_and(|kind| kind.name == "table") { + let settings = &sink["settings"]; + let target = ( + settings["database"].as_str().unwrap_or_default(), + settings["table"].as_str().unwrap_or_default(), + ); + if !targets.insert(target) { + invalid!( + "each table sink must target a different table; {}.{} appears more than once", + target.0, + target.1 + ); + } + } + } + Ok(()) +} + +fn validate_sink(sink: &Value, kind: Option<&SinkType>) -> Result<(), String> { + let Value::Object(fields) = sink else { + return Err("expected a JSON object".into()); + }; + let Some(kind) = kind else { + return match fields.get("type") { + Some(Value::String(_)) => Ok(()), + _ => Err(r#"missing "type"; expected "table" or "http""#.into()), + }; + }; + reject_unknown(fields, &["type", "id", "settings"], "")?; + let existing = match fields.get("id") { + None => false, + Some(Value::String(id)) if !id.trim().is_empty() => true, + Some(_) => return Err(r#""id" must be a sink ID string"#.into()), + }; + match (kind.name, fields.get("settings")) { + ("table", Some(Value::Object(settings))) => { + reject_unknown(settings, &["database", "table"], "settings.")?; + for name in ["database", "table"] { + match settings.get(name).and_then(Value::as_str) { + Some(value) if is_identifier(value) => {} + Some(value) => { + return Err(format!( + "settings.{name} '{value}' is not a valid identifier ({})", + IDENTIFIER_RULE.to_lowercase() + )) + } + None => return Err(format!("settings.{name} is required")), + } + } + Ok(()) + } + ("table", _) => Err("settings must be an object with database and table".into()), + ("http", None) if existing => Ok(()), + ("http", None) => Err("settings.url is required when adding an HTTP sink".into()), + ("http", Some(Value::Object(settings))) => { + reject_unknown(settings, &["url", "headers"], "settings.")?; + match settings.get("url") { + Some(Value::String(url)) => validate_url(url)?, + Some(_) => return Err("settings.url must be a string".into()), + None if existing => {} + None => return Err("settings.url is required when adding an HTTP sink".into()), + } + match settings.get("headers") { + None => Ok(()), + Some(Value::Object(headers)) => { + if headers.len() > MAX_HEADERS { + return Err(format!( + "settings.headers allows at most {MAX_HEADERS} headers" + )); + } + for (name, value) in headers { + match value { + Value::String(value) if has_reserved_sequence(value) => { + return Err(format!( + "header '{name}' contains {{{{, %, or ${{, which are reserved" + )) + } + Value::String(_) => {} + Value::Null if existing => {} + Value::Null => { + return Err(format!( + "header '{name}' is null; null keeps a stored value and only applies to existing sinks" + )) + } + _ => return Err(format!("header '{name}' must be a string")), + } + } + Ok(()) + } + Some(_) => { + Err("settings.headers must be an object of header names to values".into()) + } + } + } + _ => Err("settings must be an object".into()), + } +} + +fn reject_unknown( + fields: &Map, + allowed: &[&str], + prefix: &str, +) -> Result<(), String> { + match fields.keys().find(|key| !allowed.contains(&key.as_str())) { + Some(key) => Err(format!( + "unknown field \"{prefix}{key}\"; expected {}", + allowed + .iter() + .map(|name| format!("\"{prefix}{name}\"")) + .collect::>() + .join(", ") + )), + None => Ok(()), + } +} + +/// The platform's delivery configuration reserves these sequences. +fn has_reserved_sequence(value: &str) -> bool { + ["{{", "%", "${"] + .iter() + .any(|reserved| value.contains(reserved)) +} + +fn validate_url(url: &str) -> Result<(), String> { + if has_reserved_sequence(url) { + return Err(format!( + "settings.url '{url}' contains {{{{, %, or ${{, which are reserved; percent-encoded URLs are not supported" + )); + } + match reqwest::Url::parse(url.trim()) { + Ok(parsed) if matches!(parsed.scheme(), "http" | "https") && parsed.has_host() => Ok(()), + _ => Err(format!( + "settings.url '{url}' must be an absolute HTTP or HTTPS URL" + )), + } +} + +fn is_identifier(value: &str) -> bool { + value.len() <= 64 + && value.starts_with(|c: char| c.is_ascii_alphabetic()) + && value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') +} + +fn load(client: &ApiClient, scope: &WorkflowScope, workflow_id: &str) -> Result { + let value: Value = client.get(&scope.workflow_path(workflow_id, ""))?; + serde_json::from_value(value).context("failed to parse server response") +} + +/// The API replaces the whole list, so sinks that are not being changed are +/// sent back by ID. HTTP sinks omit settings, which keeps their URL and +/// write-only headers. +fn retained(sink: &Value) -> Value { + match sink["type"].as_str() { + Some("table") => json!({"type": "table", "id": sink["id"], "settings": { + "database": sink["settings"]["database"], + "table": sink["settings"]["table"], + }}), + _ => json!({"type": sink["type"], "id": sink["id"]}), + } +} + +fn find_sink<'a>(workflow: &'a Workflow, workflow_id: &str, sink_id: &str) -> Result<&'a Value> { + workflow + .sinks + .iter() + .find(|sink| sink["id"].as_str() == Some(sink_id)) + .ok_or_else(|| { + output::coded_error( + "not_found", + format!( + "workflow '{}' has no sink '{sink_id}'. Run `rtree workflow sink list {workflow_id}` to see its sinks.", + workflow.name + ), + 4, + ) + }) +} + +fn replace_sinks( + client: &ApiClient, + scope: &WorkflowScope, + workflow_id: &str, + sinks: Vec, +) -> Result { + validate(&sinks)?; + let value: Value = client.patch( + &scope.workflow_path(workflow_id, ""), + &json!({"sinks": sinks}), + )?; + serde_json::from_value(value).context("failed to parse server response") +} + +pub fn list( + client: &ApiClient, + scope: &WorkflowScope, + workflow_id: &str, + json_mode: bool, +) -> Result<()> { + let workflow = load(client, scope, workflow_id)?; + output::print_result(&json!({"sinks": workflow.sinks}), json_mode, |_| { + if workflow.sinks.is_empty() { + println!( + "Workflow '{}' has no sinks. Add one with `rtree workflow sink create {workflow_id} --table `.", + workflow.name + ); + return; + } + let mut table = new_cli_table(); + table.set_header(vec!["type", "target", "headers", "id"]); + for sink in &workflow.sinks { + let settings = &sink["settings"]; + let (target, headers) = match sink["type"].as_str() { + Some("table") => ( + format!( + "{}.{}", + settings["database"].as_str().unwrap_or("?"), + settings["table"].as_str().unwrap_or("?") + ), + String::new(), + ), + Some("http") => ( + settings["url"] + .as_str() + .unwrap_or("URL unavailable") + .to_string(), + header_names(sink).join(", "), + ), + _ => (settings.to_string(), String::new()), + }; + table.add_row(vec![ + Cell::new(sink["type"].as_str().unwrap_or("?")), + Cell::new(target), + Cell::new(headers), + Cell::new(sink["id"].as_str().unwrap_or("—")), + ]); + } + println!("{table}"); + }); + Ok(()) +} + +fn header_names(sink: &Value) -> Vec<&str> { + sink["settings"]["header_names"] + .as_array() + .map(|names| names.iter().filter_map(Value::as_str).collect()) + .unwrap_or_default() +} + +pub enum NewSink { + Table(String), + Http { url: String, headers: Vec }, +} + +pub fn create( + client: &ApiClient, + scope: &WorkflowScope, + workflow_id: &str, + new_sink: NewSink, + json_mode: bool, +) -> Result<()> { + let workflow = load(client, scope, workflow_id)?; + let new_sink = match new_sink { + NewSink::Table(target) => table_sink(&target, &workflow.query.database), + NewSink::Http { url, headers } => http_sink(&url, &headers)?, + }; + let previous: BTreeSet<&str> = workflow + .sinks + .iter() + .filter_map(|sink| sink["id"].as_str()) + .collect(); + let mut sinks: Vec = workflow.sinks.iter().map(retained).collect(); + sinks.push(new_sink); + let updated = replace_sinks(client, scope, workflow_id, sinks)?; + let created = updated + .sinks + .iter() + .find(|sink| sink["id"].as_str().is_some_and(|id| !previous.contains(id))) + .context("the server response did not include the new sink")?; + output::print_result(created, json_mode, |sink| { + println!("Sink added to workflow '{}':", updated.name); + println!(" {}", describe_sink(sink)); + }); + Ok(()) +} + +pub struct SinkChanges { + pub table: Option, + pub url: Option, + pub headers: Vec, + pub remove_headers: Vec, +} + +pub fn update( + client: &ApiClient, + scope: &WorkflowScope, + workflow_id: &str, + sink_id: &str, + changes: SinkChanges, + json_mode: bool, +) -> Result<()> { + let workflow = load(client, scope, workflow_id)?; + let current = find_sink(&workflow, workflow_id, sink_id)?; + let replacement = changed_sink(current, changes)?; + let sinks = workflow + .sinks + .iter() + .map(|sink| { + if sink["id"].as_str() == Some(sink_id) { + replacement.clone() + } else { + retained(sink) + } + }) + .collect(); + let updated = replace_sinks(client, scope, workflow_id, sinks)?; + let sink = updated + .sinks + .iter() + .find(|sink| sink["id"].as_str() == Some(sink_id)) + .context("the server response did not include the updated sink")?; + output::print_result(sink, json_mode, |sink| { + println!("Sink updated on workflow '{}':", updated.name); + println!(" {}", describe_sink(sink)); + }); + Ok(()) +} + +fn changed_sink(current: &Value, changes: SinkChanges) -> Result { + let id = current["id"].as_str().unwrap_or_default(); + let http_changes = + changes.url.is_some() || !changes.headers.is_empty() || !changes.remove_headers.is_empty(); + match current["type"].as_str() { + Some("table") if http_changes => { + invalid!("sink {id} is a table sink; --url, --header, and --remove-header apply to HTTP sinks") + } + Some("table") => { + let Some(target) = changes.table else { + invalid!("nothing to update: pass --table for a table sink"); + }; + let database = current["settings"]["database"].as_str().unwrap_or_default(); + let mut sink = table_sink(&target, database); + sink["id"] = json!(id); + Ok(sink) + } + Some("http") if changes.table.is_some() => { + invalid!("sink {id} is an HTTP sink; --table applies to table sinks") + } + Some("http") => { + if !http_changes { + invalid!("nothing to update: pass --url, --header, or --remove-header for an HTTP sink"); + } + let mut settings = Map::new(); + if let Some(url) = changes.url { + settings.insert("url".into(), json!(url)); + } + if !changes.headers.is_empty() || !changes.remove_headers.is_empty() { + settings.insert( + "headers".into(), + Value::Object(merged_headers( + &header_names(current), + &changes.headers, + &changes.remove_headers, + )?), + ); + } + Ok(json!({"type": "http", "id": id, "settings": settings})) + } + _ => invalid!("sink {id} has a type this CLI cannot edit; replace the list with `rtree workflow update --sink`"), + } +} + +/// A supplied headers map removes omitted names, so every stored header that +/// should survive is sent as null, which keeps its write-only value. +fn merged_headers( + stored: &[&str], + set: &[String], + remove: &[String], +) -> Result> { + let stored_name = |name: &str| { + stored + .iter() + .find(|stored| stored.eq_ignore_ascii_case(name)) + .map(|stored| stored.to_string()) + }; + for name in remove { + if stored_name(name).is_none() { + invalid!( + "the sink has no header '{name}'; its headers are: {}", + if stored.is_empty() { + "none".to_string() + } else { + stored.join(", ") + } + ); + } + } + let mut parsed = BTreeMap::new(); + for raw in set { + let (name, value) = parse_header(raw)?; + let name = stored_name(&name).unwrap_or(name); + if parsed + .keys() + .any(|existing: &String| existing.eq_ignore_ascii_case(&name)) + { + invalid!("header '{name}' is passed more than once"); + } + parsed.insert(name, value); + } + let mut headers = Map::new(); + for name in stored { + let replaced = parsed.keys().any(|new| new.eq_ignore_ascii_case(name)); + let removed = remove.iter().any(|gone| gone.eq_ignore_ascii_case(name)); + if !replaced && !removed { + headers.insert(name.to_string(), Value::Null); + } + } + for (name, value) in parsed { + headers.insert(name, json!(value)); + } + Ok(headers) +} + +pub fn delete( + client: &ApiClient, + scope: &WorkflowScope, + workflow_id: &str, + sink_id: &str, + json_mode: bool, +) -> Result<()> { + let workflow = load(client, scope, workflow_id)?; + find_sink(&workflow, workflow_id, sink_id)?; + let sinks = workflow + .sinks + .iter() + .filter(|sink| sink["id"].as_str() != Some(sink_id)) + .map(retained) + .collect(); + let updated = replace_sinks(client, scope, workflow_id, sinks)?; + output::print_result( + &json!({"workflow_id": workflow_id, "sink_id": sink_id, "deleted": true}), + json_mode, + |_| println!("Sink {sink_id} removed from workflow '{}'.", updated.name), + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn args(tables: &[&str], urls: &[&str], headers: &[&str], json: &[&str]) -> SinkArgs { + let owned = |values: &[&str]| values.iter().map(|value| value.to_string()).collect(); + SinkArgs { + tables: owned(tables), + urls: owned(urls), + headers: owned(headers), + json: owned(json), + } + } + + fn error(sinks: Value) -> String { + validate(sinks.as_array().unwrap()).unwrap_err().to_string() + } + + #[test] + fn typed_flags_build_sinks_with_the_workflow_database() { + let sinks = build( + &args( + &["alerts", "other.copy"], + &["https://example.com/hook"], + &["Authorization: Bearer a:b"], + &[r#"[{"type":"table","settings":{"database":"x","table":"y"}}]"#], + ), + "analytics", + ) + .unwrap(); + assert_eq!( + sinks, + vec![ + json!({"type": "table", "settings": {"database": "analytics", "table": "alerts"}}), + json!({"type": "table", "settings": {"database": "other", "table": "copy"}}), + json!({"type": "http", "settings": {"url": "https://example.com/hook", + "headers": {"Authorization": "Bearer a:b"}}}), + json!({"type": "table", "settings": {"database": "x", "table": "y"}}), + ] + ); + } + + #[test] + fn headers_need_exactly_one_http_sink_and_never_echo_values() { + let err = build( + &args( + &[], + &["https://a.example", "https://b.example"], + &["X-Key: 1"], + &[], + ), + "db", + ) + .unwrap_err(); + assert!(err.to_string().contains("single --sink-http"), "{err}"); + + let err = build( + &args(&[], &["https://a.example"], &["secret-token"], &[]), + "db", + ) + .unwrap_err() + .to_string(); + assert!(!err.contains("secret-token"), "{err}"); + } + + #[test] + fn sink_files_and_inline_json_accept_objects_and_arrays() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("sinks.json"); + std::fs::write( + &path, + r#"[{"type":"http","settings":{"url":"https://example.com"}}]"#, + ) + .unwrap(); + let sinks = read_json_args(&[ + format!("@{}", path.display()), + r#"{"type":"table","settings":{"database":"a","table":"b"}}"#.into(), + ]) + .unwrap(); + assert_eq!(sinks.len(), 2); + + let err = read_json_args(&["[".into()]).unwrap_err().to_string(); + assert!(err.contains("not valid JSON"), "{err}"); + let err = read_json_args(&["42".into()]).unwrap_err().to_string(); + assert!(err.contains("object or an array"), "{err}"); + let err = read_json_args(&["@missing.json".into()]) + .unwrap_err() + .to_string(); + assert!(err.contains("missing.json"), "{err}"); + } + + #[test] + fn validation_matches_platform_rules_and_points_to_the_schema() { + for (sinks, expected) in [ + (json!([{"settings": {}}]), r#"missing "type""#), + (json!(["table"]), "expected a JSON object"), + ( + json!([{"type": "table", "settings": {"database": "a"}}]), + "settings.table is required", + ), + ( + json!([{"type": "table", "settings": {"database": "1a", "table": "b"}}]), + "not a valid identifier", + ), + ( + json!([{"type": "table", "name": "x", "settings": {"database": "a", "table": "b"}}]), + r#"unknown field "name""#, + ), + ( + json!([{"type": "http", "settings": {"uri": "https://a.example"}}]), + r#"unknown field "settings.uri""#, + ), + (json!([{"type": "http"}]), "settings.url is required"), + ( + json!([{"type": "http", "settings": {"url": "https://a.example/a%20b"}}]), + "which are reserved", + ), + ( + json!([{"type": "http", "settings": {"url": "https://a.example", "headers": {"X": "${TOKEN}"}}}]), + "header 'X' contains", + ), + ( + json!([{"type": "http", "settings": {"url": "ftp://a.example"}}]), + "absolute HTTP or HTTPS URL", + ), + ( + json!([{"type": "http", "settings": {"url": "https://a.example", "headers": {"X": null}}}]), + "only applies to existing sinks", + ), + ( + json!([ + {"type": "table", "settings": {"database": "a", "table": "b"}}, + {"type": "table", "settings": {"database": "a", "table": "b"}}, + ]), + "a.b appears more than once", + ), + ( + json!([{"type": "http", "id": "s1"}, {"type": "http", "id": "s1"}]), + "appears more than once", + ), + ( + Value::Array(vec![json!({"type": "http", "id": "s"}); 6]), + "at most 5 sinks", + ), + ] { + let err = error(sinks); + assert!(err.contains(expected), "expected {expected:?} in {err}"); + } + assert!(error(json!([{"type": "http"}])).contains("rtree workflow sink schema --type http")); + } + + #[test] + fn existing_sinks_may_omit_http_settings_and_unknown_types_pass_through() { + validate(&[ + json!({"type": "http", "id": "s1"}), + json!({"type": "http", "id": "s2", "settings": {"headers": {"Authorization": null}}}), + json!({"type": "queue", "settings": {"anything": true}}), + ]) + .unwrap(); + } + + #[test] + fn retained_sinks_keep_table_targets_and_omit_http_settings() { + assert_eq!( + retained( + &json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "b"}}) + ), + json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "b"}}) + ); + assert_eq!( + retained( + &json!({"type": "http", "id": "h", "settings": {"url": "https://a.example", + "url_configured": true, "header_names": ["Authorization"]}}) + ), + json!({"type": "http", "id": "h"}) + ); + } + + #[test] + fn header_changes_keep_other_stored_headers() { + let headers = merged_headers( + &["Authorization", "X-Team", "X-Old"], + &["authorization: Bearer new".into(), "X-New: 1".into()], + &["x-old".into()], + ) + .unwrap(); + assert_eq!( + Value::Object(headers), + json!({"X-Team": null, "Authorization": "Bearer new", "X-New": "1"}) + ); + assert!(merged_headers(&["A"], &[], &["B".into()]).is_err()); + } + + #[test] + fn sink_updates_reject_flags_for_the_other_type() { + let table = + json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "b"}}); + let http = json!({"type": "http", "id": "h", "settings": {"header_names": []}}); + let changes = |table: Option<&str>, url: Option<&str>| SinkChanges { + table: table.map(str::to_string), + url: url.map(str::to_string), + headers: vec![], + remove_headers: vec![], + }; + assert_eq!( + changed_sink(&table, changes(Some("c"), None)).unwrap(), + json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "c"}}) + ); + assert!(changed_sink(&table, changes(None, Some("https://a.example"))).is_err()); + assert!(changed_sink(&http, changes(Some("c"), None)).is_err()); + assert!(changed_sink(&http, changes(None, None)).is_err()); + assert_eq!( + changed_sink(&http, changes(None, Some("https://a.example"))).unwrap(), + json!({"type": "http", "id": "h", "settings": {"url": "https://a.example"}}) + ); + } + + #[test] + fn schema_examples_pass_validation() { + for sink_type in &SINK_TYPES { + validate(&[serde_json::from_str(sink_type.example).unwrap()]).unwrap(); + } + } +} diff --git a/src/main.rs b/src/main.rs index 1f8ba85..92e0d81 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,7 +15,7 @@ use clap_complete::generate; use cli::{ Cli, ClusterCommand, Command, DatabaseCommand, KeyCommand, OrganizationCommand, ShellType, - TableCommand, WorkflowCommand, + TableCommand, WorkflowCommand, WorkflowSinkCommand, }; use client::ApiClient; use constants::DEFAULT_API_URL; @@ -164,6 +164,17 @@ fn read_sql_arg(sql: String) -> Result { } } +fn ensure_single_stdin_reader(sql: Option<&str>, sinks: &[String]) -> Result<()> { + if sql == Some("-") && sinks.iter().any(|sink| sink == "-") { + return Err(output::coded_error( + "validation_error", + "--sql - and --sink - both read stdin; pass the sinks with --sink @ instead", + 2, + )); + } + Ok(()) +} + fn require_workflow_scope<'a>( organization: Option<&'a str>, cluster: Option<&'a str>, @@ -510,6 +521,12 @@ fn run(cli: Cli) -> Result { } } } + Command::Workflow { + action: + WorkflowCommand::Sink { + action: WorkflowSinkCommand::Schema { sink_type }, + }, + } => commands::workflow::sink::schema(sink_type, json), Command::Workflow { action } => { let effective_org = resolve_effective_org(&client, cli_org.clone()); let scope = @@ -523,20 +540,31 @@ fn run(cli: Cli) -> Result { sql, interval_seconds, disabled, + sink_tables, + sink_urls, + sink_headers, sinks, - } => commands::workflow::create( - &client, - &scope, - commands::workflow::CreateWorkflow { - name, - database: resolve_database(database)?, - sql: read_sql_arg(sql)?, - interval_seconds: interval_seconds.and_then(|interval| interval.0), - disabled, - sinks, - }, - json, - ), + } => { + ensure_single_stdin_reader(Some(&sql), &sinks)?; + commands::workflow::create( + &client, + &scope, + commands::workflow::CreateWorkflow { + name, + database: resolve_database(database)?, + sql: read_sql_arg(sql)?, + interval_seconds: interval_seconds.and_then(|interval| interval.0), + disabled, + sinks: commands::workflow::sink::SinkArgs { + tables: sink_tables, + urls: sink_urls, + headers: sink_headers, + json: sinks, + }, + }, + json, + ) + } WorkflowCommand::Update { id, name, @@ -547,20 +575,23 @@ fn run(cli: Cli) -> Result { disable, sinks, clear_sinks, - } => commands::workflow::update( - &client, - &scope, - &id, - commands::workflow::UpdateWorkflow { - name, - database, - sql: sql.map(read_sql_arg).transpose()?, - interval_seconds: interval_seconds.map(|interval| interval.0), - enabled: (enable || disable).then_some(enable), - sinks: (clear_sinks || !sinks.is_empty()).then_some(sinks), - }, - json, - ), + } => { + ensure_single_stdin_reader(sql.as_deref(), &sinks)?; + commands::workflow::update( + &client, + &scope, + &id, + commands::workflow::UpdateWorkflow { + name, + database, + sql: sql.map(read_sql_arg).transpose()?, + interval_seconds: interval_seconds.map(|interval| interval.0), + enabled: (enable || disable).then_some(enable), + sinks: (clear_sinks || !sinks.is_empty()).then_some(sinks), + }, + json, + ) + } WorkflowCommand::Delete { id } => { commands::workflow::delete(&client, &scope, &id, json) } @@ -593,6 +624,62 @@ fn run(cli: Cli) -> Result { WorkflowCommand::Metrics { id, range } => { commands::workflow::metrics(&client, &scope, &id, &range, json) } + WorkflowCommand::Sink { action } => match action { + WorkflowSinkCommand::List { workflow_id } => { + commands::workflow::sink::list(&client, &scope, &workflow_id, json) + } + WorkflowSinkCommand::Create { + workflow_id, + table, + http, + headers, + } => commands::workflow::sink::create( + &client, + &scope, + &workflow_id, + match (table, http) { + (Some(table), _) => commands::workflow::sink::NewSink::Table(table), + (None, Some(url)) => { + commands::workflow::sink::NewSink::Http { url, headers } + } + (None, None) => unreachable!("clap requires --table or --http"), + }, + json, + ), + WorkflowSinkCommand::Update { + workflow_id, + sink_id, + table, + url, + headers, + remove_headers, + } => commands::workflow::sink::update( + &client, + &scope, + &workflow_id, + &sink_id, + commands::workflow::sink::SinkChanges { + table, + url, + headers, + remove_headers, + }, + json, + ), + WorkflowSinkCommand::Delete { + workflow_id, + sink_id, + } => commands::workflow::sink::delete( + &client, + &scope, + &workflow_id, + &sink_id, + json, + ), + WorkflowSinkCommand::Schema { .. } => { + unreachable!("sink schema needs no workflow scope") + } + }, } } Command::Logs { diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 6d49495..47b94a6 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -420,6 +420,189 @@ fn workflow_update_sends_only_changed_fields() { assert_eq!(body, json!({"enabled": false, "sinks": []})); } +#[test] +fn workflow_create_builds_typed_sinks_with_the_workflow_database() { + let (_, _, body) = run( + &[ + "workflow", + "create", + "--name", + "alerts", + "--database", + "analytics", + "--sql", + "SELECT 1", + "--sink-table", + "alerts", + "--sink-http", + "https://example.com/hook", + "--sink-header", + "Authorization: Bearer token", + ], + "POST", + &format!("/v1/workflows?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({}), + ); + assert_eq!( + body["sinks"], + json!([ + {"type": "table", "settings": {"database": "analytics", "table": "alerts"}}, + {"type": "http", "settings": {"url": "https://example.com/hook", + "headers": {"Authorization": "Bearer token"}}}, + ]) + ); +} + +#[test] +fn invalid_sinks_fail_before_any_request() { + let (output, _, _) = common::run_cli( + &[], + &[ + "--org", + "team alpha", + "--cluster", + "prod/eu", + "workflow", + "create", + "--name", + "alerts", + "--database", + "analytics", + "--sql", + "SELECT 1", + "--sink", + r#"{"type":"http","settings":{"headers":{}}}"#, + ], + &json!({}), + ); + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("settings.url is required"), "{stderr}"); + assert!( + stderr.contains("rtree workflow sink schema --type http"), + "{stderr}" + ); +} + +fn workflow_with_sinks() -> Value { + let mut workflow = workflow_response(); + workflow["sinks"] = json!([ + {"type": "table", "id": "sink-table", "settings": {"database": "analytics", "table": "out"}}, + {"type": "http", "id": "sink-http", "settings": {"url": "https://example.com/hook", + "url_configured": true, "header_names": ["Authorization", "X-Team"]}}, + ]); + workflow +} + +/// Runs a `workflow sink` command that reads the workflow and then patches its +/// sinks, returning the JSON output and the PATCH body. +fn run_sink_change(args: &[&str], patched: Value) -> (Value, Value) { + let path = format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"); + let mut command = vec!["--org", "team alpha", "--cluster", "prod/eu", "--json"]; + command.extend_from_slice(args); + let (output, _, bodies) = common::run_cli( + &[ + (format!("GET {path}"), "200 OK", workflow_with_sinks()), + (format!("PATCH {path}"), "200 OK", patched), + ], + &command, + &json!({}), + ); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + ( + serde_json::from_slice(&output.stdout).unwrap(), + bodies[1].clone(), + ) +} + +#[test] +fn sink_create_keeps_existing_sinks_and_returns_the_new_one() { + let mut patched = workflow_with_sinks(); + let added = json!({"type": "table", "id": "sink-new", + "settings": {"database": "analytics", "table": "copy"}}); + patched["sinks"].as_array_mut().unwrap().push(added.clone()); + let (result, body) = run_sink_change( + &["workflow", "sink", "create", "wf-1", "--table", "copy"], + patched, + ); + assert_eq!(result, added); + assert_eq!( + body, + json!({"sinks": [ + {"type": "table", "id": "sink-table", "settings": {"database": "analytics", "table": "out"}}, + {"type": "http", "id": "sink-http"}, + {"type": "table", "settings": {"database": "analytics", "table": "copy"}}, + ]}) + ); +} + +#[test] +fn sink_update_changes_one_header_and_keeps_the_rest() { + let (result, body) = run_sink_change( + &[ + "workflow", + "sink", + "update", + "wf-1", + "sink-http", + "--header", + "Authorization: Bearer new", + ], + workflow_with_sinks(), + ); + assert_eq!(result["id"], "sink-http"); + assert_eq!( + body, + json!({"sinks": [ + {"type": "table", "id": "sink-table", "settings": {"database": "analytics", "table": "out"}}, + {"type": "http", "id": "sink-http", "settings": {"headers": { + "Authorization": "Bearer new", "X-Team": null}}}, + ]}) + ); +} + +#[test] +fn sink_delete_sends_the_remaining_sinks() { + let mut patched = workflow_with_sinks(); + patched["sinks"].as_array_mut().unwrap().remove(0); + let (result, body) = run_sink_change( + &["workflow", "sink", "delete", "wf-1", "sink-table"], + patched, + ); + assert_eq!( + result, + json!({"workflow_id": "wf-1", "sink_id": "sink-table", "deleted": true}) + ); + assert_eq!( + body, + json!({"sinks": [{"type": "http", "id": "sink-http"}]}) + ); +} + +#[test] +fn sink_list_prints_targets_and_header_names() { + let stdout = run_workflow_human( + &["workflow", "sink", "list", "wf-1"], + "GET", + &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), + workflow_with_sinks(), + ); + for expected in [ + "analytics.out", + "sink-table", + "https://example.com/hook", + "Authorization, X-Team", + "sink-http", + ] { + assert!(stdout.contains(expected), "missing {expected}: {stdout}"); + } +} + #[test] fn workflow_delete_and_cancel_accept_empty_responses() { let (result, _, _) = run( diff --git a/tests/live_platform/workflows.rs b/tests/live_platform/workflows.rs index 7d95b68..daec378 100644 --- a/tests/live_platform/workflows.rs +++ b/tests/live_platform/workflows.rs @@ -135,6 +135,38 @@ fn workflow_manual_runs_deliver_rows_and_expose_logs_and_metrics() { }); } +#[test] +#[ignore = "requires a bootstrapped Platform Docker Compose stack"] +fn workflow_sinks_can_be_added_changed_and_removed_one_at_a_time() { + with_workflow("SELECT 1 AS value", |cli, database, id| { + for table in ["results", "results_v2", "audit"] { + cli.json(&["table", "create", table, "--database", database]); + } + let first = cli.json(&["workflow", "sink", "create", id, "--table", "results"]); + assert_eq!(first["settings"]["database"], database); + let first_id = first["id"].as_str().unwrap(); + let audit_target = format!("{database}.audit"); + let second = cli.json(&["workflow", "sink", "create", id, "--table", &audit_target]); + + let changed = cli.json(&[ + "workflow", + "sink", + "update", + id, + first_id, + "--table", + "results_v2", + ]); + assert_eq!(changed["id"], first_id); + assert_eq!(changed["settings"]["table"], "results_v2"); + + let deleted = cli.json(&["workflow", "sink", "delete", id, first_id]); + assert_eq!(deleted["deleted"], true); + let listed = cli.json(&["workflow", "sink", "list", id]); + assert_eq!(listed["sinks"], json!([second])); + }); +} + #[test] #[ignore = "requires a bootstrapped Platform Docker Compose stack"] fn workflow_scheduled_runs_can_be_paused_and_switched_to_manual() { From 9368a178f2c3fd85d8ed44890d2f5727b26ffb6e Mon Sep 17 00:00:00 2001 From: jordiori Date: Thu, 8 Oct 2026 18:36:17 +0200 Subject: [PATCH 2/3] Validate only the shape of sinks and leave values to the API --- src/commands/workflow/sink.rs | 166 ++++++++-------------------------- 1 file changed, 38 insertions(+), 128 deletions(-) diff --git a/src/commands/workflow/sink.rs b/src/commands/workflow/sink.rs index 4341466..36fe5d6 100644 --- a/src/commands/workflow/sink.rs +++ b/src/commands/workflow/sink.rs @@ -18,9 +18,8 @@ macro_rules! invalid { }; } -/// Mirrors the platform's `MAX_SINKS` and the HTTP sink header limit. +/// Documents the platform's `MAX_SINKS`; the API enforces it. const MAX_SINKS: usize = 5; -const MAX_HEADERS: usize = 20; struct SinkField { name: &'static str, @@ -263,20 +262,13 @@ pub(super) fn read_json_args(args: &[String]) -> Result> { Ok(sinks) } -/// Checks the shape the platform accepts so mistakes fail before a request. -/// Types this CLI does not know are left for the server to judge. +/// Checks only the shape of each sink so malformed input fails before a +/// request; values and limits are left to the API. Types this CLI does not +/// know are passed through. pub(super) fn validate(sinks: &[Value]) -> Result<()> { - if sinks.len() > MAX_SINKS { - invalid!( - "a workflow supports at most {MAX_SINKS} sinks; got {}", - sinks.len() - ); - } - let mut ids = BTreeSet::new(); - let mut targets = BTreeSet::new(); for (index, sink) in sinks.iter().enumerate() { let kind = sink["type"].as_str().and_then(sink_type); - if let Err(reason) = validate_sink(sink, kind) { + if let Err(reason) = validate_shape(sink, kind) { let position = index + 1; match kind { Some(kind) => invalid!( @@ -288,30 +280,11 @@ pub(super) fn validate(sinks: &[Value]) -> Result<()> { ), } } - if let Some(id) = sink["id"].as_str() { - if !ids.insert(id) { - invalid!("sink ID '{id}' appears more than once"); - } - } - if kind.is_some_and(|kind| kind.name == "table") { - let settings = &sink["settings"]; - let target = ( - settings["database"].as_str().unwrap_or_default(), - settings["table"].as_str().unwrap_or_default(), - ); - if !targets.insert(target) { - invalid!( - "each table sink must target a different table; {}.{} appears more than once", - target.0, - target.1 - ); - } - } } Ok(()) } -fn validate_sink(sink: &Value, kind: Option<&SinkType>) -> Result<(), String> { +fn validate_shape(sink: &Value, kind: Option<&SinkType>) -> Result<(), String> { let Value::Object(fields) = sink else { return Err("expected a JSON object".into()); }; @@ -324,69 +297,43 @@ fn validate_sink(sink: &Value, kind: Option<&SinkType>) -> Result<(), String> { reject_unknown(fields, &["type", "id", "settings"], "")?; let existing = match fields.get("id") { None => false, - Some(Value::String(id)) if !id.trim().is_empty() => true, - Some(_) => return Err(r#""id" must be a sink ID string"#.into()), + Some(Value::String(_)) => true, + Some(_) => return Err(r#""id" must be a string"#.into()), }; match (kind.name, fields.get("settings")) { ("table", Some(Value::Object(settings))) => { reject_unknown(settings, &["database", "table"], "settings.")?; for name in ["database", "table"] { - match settings.get(name).and_then(Value::as_str) { - Some(value) if is_identifier(value) => {} - Some(value) => { - return Err(format!( - "settings.{name} '{value}' is not a valid identifier ({})", - IDENTIFIER_RULE.to_lowercase() - )) - } - None => return Err(format!("settings.{name} is required")), + if !settings.get(name).is_some_and(Value::is_string) { + return Err(format!("settings.{name} is required and must be a string")); } } Ok(()) } ("table", _) => Err("settings must be an object with database and table".into()), ("http", None) if existing => Ok(()), - ("http", None) => Err("settings.url is required when adding an HTTP sink".into()), ("http", Some(Value::Object(settings))) => { reject_unknown(settings, &["url", "headers"], "settings.")?; match settings.get("url") { - Some(Value::String(url)) => validate_url(url)?, + Some(Value::String(_)) => {} Some(_) => return Err("settings.url must be a string".into()), None if existing => {} None => return Err("settings.url is required when adding an HTTP sink".into()), } match settings.get("headers") { None => Ok(()), - Some(Value::Object(headers)) => { - if headers.len() > MAX_HEADERS { - return Err(format!( - "settings.headers allows at most {MAX_HEADERS} headers" - )); - } - for (name, value) in headers { - match value { - Value::String(value) if has_reserved_sequence(value) => { - return Err(format!( - "header '{name}' contains {{{{, %, or ${{, which are reserved" - )) - } - Value::String(_) => {} - Value::Null if existing => {} - Value::Null => { - return Err(format!( - "header '{name}' is null; null keeps a stored value and only applies to existing sinks" - )) - } - _ => return Err(format!("header '{name}' must be a string")), - } - } - Ok(()) - } + Some(Value::Object(headers)) => headers + .iter() + .find(|(_, value)| !(value.is_string() || value.is_null())) + .map_or(Ok(()), |(name, _)| { + Err(format!("header '{name}' must be a string")) + }), Some(_) => { Err("settings.headers must be an object of header names to values".into()) } } } + ("http", None) => Err("settings.url is required when adding an HTTP sink".into()), _ => Err("settings must be an object".into()), } } @@ -409,33 +356,6 @@ fn reject_unknown( } } -/// The platform's delivery configuration reserves these sequences. -fn has_reserved_sequence(value: &str) -> bool { - ["{{", "%", "${"] - .iter() - .any(|reserved| value.contains(reserved)) -} - -fn validate_url(url: &str) -> Result<(), String> { - if has_reserved_sequence(url) { - return Err(format!( - "settings.url '{url}' contains {{{{, %, or ${{, which are reserved; percent-encoded URLs are not supported" - )); - } - match reqwest::Url::parse(url.trim()) { - Ok(parsed) if matches!(parsed.scheme(), "http" | "https") && parsed.has_host() => Ok(()), - _ => Err(format!( - "settings.url '{url}' must be an absolute HTTP or HTTPS URL" - )), - } -} - -fn is_identifier(value: &str) -> bool { - value.len() <= 64 - && value.starts_with(|c: char| c.is_ascii_alphabetic()) - && value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') -} - fn load(client: &ApiClient, scope: &WorkflowScope, workflow_id: &str) -> Result { let value: Value = client.get(&scope.workflow_path(workflow_id, ""))?; serde_json::from_value(value).context("failed to parse server response") @@ -831,7 +751,7 @@ mod tests { } #[test] - fn validation_matches_platform_rules_and_points_to_the_schema() { + fn validation_checks_shape_and_points_to_the_schema() { for (sinks, expected) in [ (json!([{"settings": {}}]), r#"missing "type""#), (json!(["table"]), "expected a JSON object"), @@ -840,8 +760,8 @@ mod tests { "settings.table is required", ), ( - json!([{"type": "table", "settings": {"database": "1a", "table": "b"}}]), - "not a valid identifier", + json!([{"type": "table", "settings": {"database": "a", "table": 1}}]), + "settings.table is required and must be a string", ), ( json!([{"type": "table", "name": "x", "settings": {"database": "a", "table": "b"}}]), @@ -853,35 +773,12 @@ mod tests { ), (json!([{"type": "http"}]), "settings.url is required"), ( - json!([{"type": "http", "settings": {"url": "https://a.example/a%20b"}}]), - "which are reserved", - ), - ( - json!([{"type": "http", "settings": {"url": "https://a.example", "headers": {"X": "${TOKEN}"}}}]), - "header 'X' contains", + json!([{"type": "http", "settings": {"url": "https://a.example", "headers": {"X": 1}}}]), + "header 'X' must be a string", ), ( - json!([{"type": "http", "settings": {"url": "ftp://a.example"}}]), - "absolute HTTP or HTTPS URL", - ), - ( - json!([{"type": "http", "settings": {"url": "https://a.example", "headers": {"X": null}}}]), - "only applies to existing sinks", - ), - ( - json!([ - {"type": "table", "settings": {"database": "a", "table": "b"}}, - {"type": "table", "settings": {"database": "a", "table": "b"}}, - ]), - "a.b appears more than once", - ), - ( - json!([{"type": "http", "id": "s1"}, {"type": "http", "id": "s1"}]), - "appears more than once", - ), - ( - Value::Array(vec![json!({"type": "http", "id": "s"}); 6]), - "at most 5 sinks", + json!([{"type": "http", "id": 7}]), + r#""id" must be a string"#, ), ] { let err = error(sinks); @@ -890,6 +787,19 @@ mod tests { assert!(error(json!([{"type": "http"}])).contains("rtree workflow sink schema --type http")); } + #[test] + fn values_and_limits_are_left_to_the_api() { + validate(&[ + json!({"type": "table", "settings": {"database": "1bad", "table": "a.b"}}), + json!({"type": "table", "settings": {"database": "1bad", "table": "a.b"}}), + json!({"type": "http", "settings": {"url": "not a url", "headers": {"X": null}}}), + json!({"type": "http", "id": "s"}), + json!({"type": "http", "id": "s"}), + json!({"type": "http", "id": "t"}), + ]) + .unwrap(); + } + #[test] fn existing_sinks_may_omit_http_settings_and_unknown_types_pass_through() { validate(&[ From 7afccec8e4077c23289a33fbfc679984d93ba521 Mon Sep 17 00:00:00 2001 From: jordiori Date: Tue, 13 Oct 2026 15:06:50 +0200 Subject: [PATCH 3/3] Keep sink credentials out of errors and refuse to drop unknown sink types --- src/commands/workflow/sink.rs | 81 ++++++++++++++++++++++++----------- 1 file changed, 56 insertions(+), 25 deletions(-) diff --git a/src/commands/workflow/sink.rs b/src/commands/workflow/sink.rs index 36fe5d6..23ac81b 100644 --- a/src/commands/workflow/sink.rs +++ b/src/commands/workflow/sink.rs @@ -233,7 +233,7 @@ fn set_header(headers: &mut Map, name: String, value: Value) -> R /// an array of them. pub(super) fn read_json_args(args: &[String]) -> Result> { let mut sinks = Vec::new(); - for arg in args { + for (index, arg) in args.iter().enumerate() { let (label, text) = if arg == "-" { let mut text = String::new(); std::io::stdin() @@ -241,11 +241,15 @@ pub(super) fn read_json_args(args: &[String]) -> Result> { .context("failed to read sinks from stdin")?; ("from stdin".to_string(), text) } else if let Some(path) = arg.strip_prefix('@') { - let text = std::fs::read_to_string(path) - .with_context(|| format!("failed to read sink file '{path}'"))?; + let text = match std::fs::read_to_string(path) { + Ok(text) => text, + Err(error) => invalid!("failed to read sink file '{path}': {error}"), + }; (format!("in '{path}'"), text) } else { - (format!("'{arg}'"), arg.clone()) + // Inline JSON often carries header credentials, so errors name its + // position instead of echoing it. + (format!("#{}", index + 1), arg.clone()) }; match serde_json::from_str::(&text) { Ok(Value::Object(sink)) => sinks.push(Value::Object(sink)), @@ -363,14 +367,20 @@ fn load(client: &ApiClient, scope: &WorkflowScope, workflow_id: &str) -> Result< /// The API replaces the whole list, so sinks that are not being changed are /// sent back by ID. HTTP sinks omit settings, which keeps their URL and -/// write-only headers. -fn retained(sink: &Value) -> Value { +/// write-only headers. Other types are refused because their settings cannot +/// be sent back safely. +fn retained(sink: &Value, workflow_id: &str) -> Result { match sink["type"].as_str() { - Some("table") => json!({"type": "table", "id": sink["id"], "settings": { + Some("table") => Ok(json!({"type": "table", "id": sink["id"], "settings": { "database": sink["settings"]["database"], "table": sink["settings"]["table"], - }}), - _ => json!({"type": sink["type"], "id": sink["id"]}), + }})), + Some("http") => Ok(json!({"type": "http", "id": sink["id"]})), + other => invalid!( + "sink {} has type '{}', which this version of rtree cannot preserve; upgrade rtree or replace the whole list with `rtree workflow update {workflow_id} --sink`", + sink["id"].as_str().unwrap_or("?"), + other.unwrap_or("unknown") + ), } } @@ -483,7 +493,11 @@ pub fn create( .iter() .filter_map(|sink| sink["id"].as_str()) .collect(); - let mut sinks: Vec = workflow.sinks.iter().map(retained).collect(); + let mut sinks = workflow + .sinks + .iter() + .map(|sink| retained(sink, workflow_id)) + .collect::>>()?; sinks.push(new_sink); let updated = replace_sinks(client, scope, workflow_id, sinks)?; let created = updated @@ -521,12 +535,12 @@ pub fn update( .iter() .map(|sink| { if sink["id"].as_str() == Some(sink_id) { - replacement.clone() + Ok(replacement.clone()) } else { - retained(sink) + retained(sink, workflow_id) } }) - .collect(); + .collect::>()?; let updated = replace_sinks(client, scope, workflow_id, sinks)?; let sink = updated .sinks @@ -648,8 +662,8 @@ pub fn delete( .sinks .iter() .filter(|sink| sink["id"].as_str() != Some(sink_id)) - .map(retained) - .collect(); + .map(|sink| retained(sink, workflow_id)) + .collect::>()?; let updated = replace_sinks(client, scope, workflow_id, sinks)?; output::print_result( &json!({"workflow_id": workflow_id, "sink_id": sink_id, "deleted": true}), @@ -740,14 +754,19 @@ mod tests { .unwrap(); assert_eq!(sinks.len(), 2); - let err = read_json_args(&["[".into()]).unwrap_err().to_string(); - assert!(err.contains("not valid JSON"), "{err}"); - let err = read_json_args(&["42".into()]).unwrap_err().to_string(); - assert!(err.contains("object or an array"), "{err}"); - let err = read_json_args(&["@missing.json".into()]) + let secret = + r#"{"type":"http","settings":{"headers":{"Authorization":"Bearer secret-token"}"#; + let err = read_json_args(&["{}".into(), secret.into()]) .unwrap_err() .to_string(); - assert!(err.contains("missing.json"), "{err}"); + assert!(err.contains("--sink #2: not valid JSON"), "{err}"); + assert!(!err.contains("secret-token"), "{err}"); + let err = read_json_args(&["42".into()]).unwrap_err().to_string(); + assert!(err.contains("object or an array"), "{err}"); + let err = read_json_args(&["@missing.json".into()]).unwrap_err(); + assert!(err.to_string().contains("missing.json"), "{err}"); + let coded = err.downcast_ref::().unwrap(); + assert_eq!(coded.exit_code(), 2); } #[test] @@ -814,17 +833,29 @@ mod tests { fn retained_sinks_keep_table_targets_and_omit_http_settings() { assert_eq!( retained( - &json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "b"}}) - ), + &json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "b"}}), + "wf" + ) + .unwrap(), json!({"type": "table", "id": "t", "settings": {"database": "a", "table": "b"}}) ); assert_eq!( retained( &json!({"type": "http", "id": "h", "settings": {"url": "https://a.example", - "url_configured": true, "header_names": ["Authorization"]}}) - ), + "url_configured": true, "header_names": ["Authorization"]}}), + "wf" + ) + .unwrap(), json!({"type": "http", "id": "h"}) ); + let err = retained( + &json!({"type": "queue", "id": "q", "settings": {"topic": "t"}}), + "wf", + ) + .unwrap_err() + .to_string(); + assert!(err.contains("type 'queue'"), "{err}"); + assert!(err.contains("rtree workflow update wf --sink"), "{err}"); } #[test]