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..23ac81b --- /dev/null +++ b/src/commands/workflow/sink.rs @@ -0,0 +1,906 @@ +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)) + }; +} + +/// Documents the platform's `MAX_SINKS`; the API enforces it. +const MAX_SINKS: usize = 5; + +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 (index, arg) in args.iter().enumerate() { + 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 = 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 { + // 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)), + 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 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<()> { + for (index, sink) in sinks.iter().enumerate() { + let kind = sink["type"].as_str().and_then(sink_type); + if let Err(reason) = validate_shape(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." + ), + } + } + } + Ok(()) +} + +fn validate_shape(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(_)) => 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"] { + 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", Some(Value::Object(settings))) => { + reject_unknown(settings, &["url", "headers"], "settings.")?; + match settings.get("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)) => 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()), + } +} + +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(()), + } +} + +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. 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") => Ok(json!({"type": "table", "id": sink["id"], "settings": { + "database": sink["settings"]["database"], + "table": sink["settings"]["table"], + }})), + 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") + ), + } +} + +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 = 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 + .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) { + Ok(replacement.clone()) + } else { + retained(sink, workflow_id) + } + }) + .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(|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}), + 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 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("--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] + fn validation_checks_shape_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": "a", "table": 1}}]), + "settings.table is required and must be a string", + ), + ( + 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", "headers": {"X": 1}}}]), + "header 'X' must be a string", + ), + ( + json!([{"type": "http", "id": 7}]), + r#""id" must be a string"#, + ), + ] { + 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 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(&[ + 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"}}), + "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"]}}), + "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] + 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() {