Add workflow sink commands and typed sink flags

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 <id>
 rtree workflow update <id> --interval-seconds 300
@@ -271,6 +272,12 @@ rtree workflow runs <id>
 rtree workflow cancel <id> <run-id>
 rtree workflow logs <id> --since 1h
 rtree workflow metrics <id> --since 7d
+
+rtree workflow sink list <id>
+rtree workflow sink create <id> --table analytics.alerts_v2
+rtree workflow sink update <id> <sink-id> --header 'Authorization: Bearer ...'
+rtree workflow sink delete <id> <sink-id>
+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.
 

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":"<sink-id>","settings":{"database":"analytics","table":"alerts"}}
-  {"type":"http","id":"<sink-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 <id>`, then pass every sink to keep:
-  rtree workflow update <id> \
-    --sink '{"type":"http","id":"<http-sink-id>"}' \
-    --sink '{"type":"table","id":"<table-sink-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, @<path> 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,
+  @<path> 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 <id> \
+      --sink '{"type":"http","id":"<http-sink-id>"}' \
+      --sink '{"type":"table","id":"<table-sink-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 <workflow-id>
+  rtree workflow sink create <workflow-id> --table analytics.alerts
+  rtree workflow sink create <workflow-id> --http https://example.com/hook \
+    --header 'Authorization: Bearer ...'
+  rtree workflow sink update <workflow-id> <sink-id> --header 'Authorization: Bearer ...'
+  rtree workflow sink delete <workflow-id> <sink-id>
+  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<String>,
+        /// HTTP sink URL
+        #[arg(long = "sink-http", value_name = "URL")]
+        sink_urls: Vec<String>,
+        /// 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<String>,
+        /// Sink as JSON (object or array), @<path>, or - for stdin
+        #[arg(long = "sink", value_name = "JSON|@PATH|-")]
         sinks: Vec<String>,
     },
     /// 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), @<path>, or - for stdin; replaces all sinks
+        #[arg(
+            long = "sink",
+            value_name = "JSON|@PATH|-",
+            conflicts_with = "clear_sinks"
+        )]
         sinks: Vec<String>,
         /// 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<String>,
+        /// Send results to this webhook URL
+        #[arg(long, value_name = "URL")]
+        http: Option<String>,
+        /// HTTP header as 'Name: value'; repeat for more headers
+        #[arg(
+            long = "header",
+            value_name = "HEADER",
+            requires = "http",
+            conflicts_with = "table"
+        )]
+        headers: Vec<String>,
+    },
+    /// 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<String>,
+        /// HTTP sinks: new URL
+        #[arg(long, value_name = "URL")]
+        url: Option<String>,
+        /// HTTP sinks: add or replace a header, as 'Name: value'; other headers are kept
+        #[arg(long = "header", value_name = "HEADER")]
+        headers: Vec<String>,
+        /// HTTP sinks: remove a header by name
+        #[arg(long = "remove-header", value_name = "NAME")]
+        remove_headers: Vec<String>,
+    },
+    /// 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<SinkTypeArg>,
+    },
+}
+
+#[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([