Align workflow commands with nested queries and nullable intervals

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rawtree-cli"
-version = "0.6.13"
+version = "0.6.14"
 edition = "2021"
 repository = "https://github.com/rawtreedb/rawtree-cli"
 

README.md

@@ -249,17 +249,19 @@ creation; create a replacement key to change it. There is no `key update` comman
 
 ### Workflows
 
-Workflows run saved SQL on a schedule and deliver results to sinks.
+Workflows run saved SQL on demand or on a schedule and deliver results to sinks.
 
 ```sh
 rtree workflow list
 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 ..."}}}' \
   < alerts.sql
 rtree workflow get <id>
 rtree workflow update <id> --interval-seconds 300
+rtree workflow update <id> --interval-seconds null
 rtree workflow update <id> --disable
 rtree workflow update <id> --clear-sinks
 rtree workflow delete <id>
@@ -274,6 +276,20 @@ rtree workflow metrics <id> --since 7d
 Workflow commands need both an organization and a cluster (`--org`/`--cluster`,
 the environment, or the saved defaults). API keys must have admin permission.
 `create` uses the selected database when `--database` is omitted.
+The CLI sends `--database` and `--sql` inside the API's `query` object. On `update`,
+each flag changes only that query field; the other field stays unchanged.
+
+Creating a workflow without `--interval-seconds` makes it manual-only. The CLI
+explicitly sends `interval_seconds: null`, so the API's default interval is not used.
+Use `--interval-seconds null` to remove an existing schedule, matching the API's
+`interval_seconds: null`. The same value is accepted on `create`.
+An interval restores recurring execution, subject to the enabled state; use
+`--enable` as well if the workflow is paused. Omitting `--interval-seconds` on `update`
+preserves the current mode.
+`--disable` pauses scheduled runs while retaining the interval. Manual runs remain
+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
@@ -367,6 +383,9 @@ database creation, insertion, querying, API key login, and deletion through the
 It also logs in with the local test user's email and password, verifies API access
 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.
+They run through the real CLI and clean up their workflows and databases.
 The Platform repository continues to test API endpoints directly.
 
 The Docker job checks out private `rawtreedb/rawtree-platform` at `main`. It

src/cli.rs

@@ -458,7 +458,7 @@ pub enum WorkflowCommand {
         /// Workflow ID
         id: String,
     },
-    /// Create a workflow that runs SQL on a schedule
+    /// Create a workflow that runs SQL on demand or on a schedule
     #[command(after_help = SINK_HELP)]
     Create {
         /// Workflow name (letters, digits, '-' and '_')
@@ -470,9 +470,9 @@ pub enum WorkflowCommand {
         /// SQL to run: a SELECT or INSERT INTO ... SELECT. Use "-" to read from stdin.
         #[arg(long)]
         sql: String,
-        /// Seconds between runs (server default: 1)
-        #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))]
-        interval_seconds: Option<u32>,
+        /// Seconds between automatic runs (1–86400); null or omitted means manual-only
+        #[arg(long, value_name = "SECONDS|null")]
+        interval_seconds: Option<WorkflowIntervalArg>,
         /// Create the workflow paused
         #[arg(long)]
         disabled: bool,
@@ -494,9 +494,9 @@ pub enum WorkflowCommand {
         /// New SQL. Use "-" to read from stdin.
         #[arg(long)]
         sql: Option<String>,
-        /// Seconds between runs
-        #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))]
-        interval_seconds: Option<u32>,
+        /// Seconds between automatic runs (1–86400); null removes the schedule
+        #[arg(long, value_name = "SECONDS|null")]
+        interval_seconds: Option<WorkflowIntervalArg>,
         /// Resume scheduled runs
         #[arg(long, conflicts_with = "disable")]
         enable: bool,
@@ -563,6 +563,23 @@ pub enum WorkflowCommand {
     },
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct WorkflowIntervalArg(pub Option<u32>);
+
+impl FromStr for WorkflowIntervalArg {
+    type Err = String;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        if value == "null" {
+            return Ok(Self(None));
+        }
+        match value.parse::<u32>() {
+            Ok(seconds) if (1..=86400).contains(&seconds) => Ok(Self(Some(seconds))),
+            _ => Err("expected a whole number from 1 to 86400, or null".into()),
+        }
+    }
+}
+
 /// Time window for workflow logs and metrics. Defaults to the last 24 hours; at most 7 days.
 #[derive(clap::Args, Debug, Default)]
 pub struct WorkflowTimeRange {
@@ -1176,7 +1193,7 @@ mod tests {
                         ..
                     },
             } => {
-                assert_eq!(interval_seconds, Some(60));
+                assert_eq!(interval_seconds.map(|interval| interval.0), Some(Some(60)));
                 assert!(disabled);
                 assert_eq!(sinks.len(), 2);
                 assert!(database.is_none());

src/commands/workflow.rs

@@ -33,21 +33,38 @@ impl WorkflowScope<'_> {
     }
 }
 
+#[derive(Deserialize)]
+struct WorkflowQuery {
+    database: String,
+    sql: String,
+}
+
 #[derive(Deserialize)]
 struct Workflow {
     id: String,
     name: String,
-    database: String,
-    sql: String,
+    query: WorkflowQuery,
     enabled: bool,
     revision: i64,
-    interval_seconds: i64,
+    /// `None` for manual-only workflows, which have no schedule.
+    interval_seconds: Option<i64>,
+    next_run_at: Option<String>,
     created_at: String,
     updated_at: String,
     #[serde(default)]
     sinks: Vec<Value>,
 }
 
+impl Workflow {
+    fn status(&self) -> &'static str {
+        match (self.interval_seconds, self.enabled) {
+            (None, _) => "manual",
+            (Some(_), true) => "active",
+            (Some(_), false) => "paused",
+        }
+    }
+}
+
 #[derive(Deserialize)]
 struct ListWorkflowsResponse {
     workflows: Vec<Workflow>,
@@ -133,7 +150,7 @@ pub struct UpdateWorkflow {
     pub name: Option<String>,
     pub database: Option<String>,
     pub sql: Option<String>,
-    pub interval_seconds: Option<u32>,
+    pub interval_seconds: Option<Option<u32>>,
     pub enabled: Option<bool>,
     pub sinks: Option<Vec<String>>,
 }
@@ -166,29 +183,29 @@ fn parse_sinks(sinks: &[String]) -> Result<Vec<Value>> {
 }
 
 fn create_body(request: CreateWorkflow) -> Result<Value> {
-    let mut body = json!({
+    Ok(json!({
         "name": request.name,
-        "database": request.database,
-        "sql": request.sql,
+        "query": {"database": request.database, "sql": request.sql},
         "enabled": !request.disabled,
+        "interval_seconds": request.interval_seconds,
         "sinks": parse_sinks(&request.sinks)?,
-    });
-    if let Some(interval) = request.interval_seconds {
-        body["interval_seconds"] = json!(interval);
-    }
-    Ok(body)
+    }))
 }
 
 fn update_body(request: UpdateWorkflow) -> Result<Value> {
     let mut body = Map::new();
     if let Some(name) = request.name {
         body.insert("name".into(), json!(name));
     }
+    let mut query = Map::new();
     if let Some(database) = request.database {
-        body.insert("database".into(), json!(database));
+        query.insert("database".into(), json!(database));
     }
     if let Some(sql) = request.sql {
-        body.insert("sql".into(), json!(sql));
+        query.insert("sql".into(), json!(sql));
+    }
+    if !query.is_empty() {
+        body.insert("query".into(), Value::Object(query));
     }
     if let Some(enabled) = request.enabled {
         body.insert("enabled".into(), json!(enabled));
@@ -281,11 +298,12 @@ fn format_ms(ms: i64) -> String {
         .unwrap_or_else(|| ms.to_string())
 }
 
-fn format_interval(seconds: i64) -> String {
+fn format_interval(seconds: Option<i64>) -> String {
     match seconds {
-        s if s % 3600 == 0 => format!("{}h", s / 3600),
-        s if s % 60 == 0 => format!("{}m", s / 60),
-        s => format!("{s}s"),
+        None => "—".into(),
+        Some(s) if s % 3600 == 0 => format!("{}h", s / 3600),
+        Some(s) if s % 60 == 0 => format!("{}m", s / 60),
+        Some(s) => format!("{s}s"),
     }
 }
 
@@ -299,6 +317,7 @@ fn describe_sink(sink: &Value) -> String {
             settings["table"].as_str().unwrap_or("?")
         ),
         Some("http") => {
+            let url = settings["url"].as_str().unwrap_or("URL unavailable");
             let headers = settings["header_names"]
                 .as_array()
                 .map(|names| {
@@ -310,9 +329,9 @@ fn describe_sink(sink: &Value) -> String {
                 })
                 .unwrap_or_default();
             if headers.is_empty() {
-                format!("http  id={id}")
+                format!("http {url}  id={id}")
             } else {
-                format!("http  id={id}  headers={headers}")
+                format!("http {url}  id={id}  headers={headers}")
             }
         }
         _ => sink.to_string(),
@@ -322,16 +341,13 @@ fn describe_sink(sink: &Value) -> String {
 fn print_workflow(workflow: &Workflow) {
     println!("  id:       {}", workflow.id);
     println!("  name:     {}", workflow.name);
-    println!("  database: {}", workflow.database);
+    println!("  database: {}", workflow.query.database);
+    println!("  status:   {}", workflow.status());
+    println!("  interval: {}", format_interval(workflow.interval_seconds));
     println!(
-        "  status:   {}",
-        if workflow.enabled {
-            "enabled"
-        } else {
-            "paused"
-        }
+        "  next run: {}",
+        workflow.next_run_at.as_deref().unwrap_or("—")
     );
-    println!("  interval: {}", format_interval(workflow.interval_seconds));
     println!("  revision: {}", workflow.revision);
     println!("  created:  {}", workflow.created_at);
     println!("  updated:  {}", workflow.updated_at);
@@ -344,7 +360,7 @@ fn print_workflow(workflow: &Workflow) {
         }
     }
     println!("  sql:");
-    for line in workflow.sql.lines() {
+    for line in workflow.query.sql.lines() {
         println!("    {line}");
     }
 }
@@ -358,19 +374,16 @@ pub fn list(client: &ApiClient, scope: &WorkflowScope, json_mode: bool) -> Resul
         }
         let mut table = new_cli_table();
         table.set_header(vec![
-            "name", "database", "status", "interval", "sinks", "updated", "id",
+            "name", "database", "status", "interval", "next run", "sinks", "updated", "id",
         ]);
         for workflow in &resp.workflows {
             table.add_row(vec![
                 Cell::new(&workflow.name),
-                Cell::new(&workflow.database),
-                Cell::new(if workflow.enabled {
-                    "enabled"
-                } else {
-                    "paused"
-                }),
+                Cell::new(&workflow.query.database),
+                Cell::new(workflow.status()),
                 Cell::new(format_interval(workflow.interval_seconds))
                     .set_alignment(CellAlignment::Right),
+                Cell::new(workflow.next_run_at.as_deref().unwrap_or("—")),
                 Cell::new(workflow.sinks.len()).set_alignment(CellAlignment::Right),
                 Cell::new(&workflow.updated_at),
                 Cell::new(&workflow.id),
@@ -652,9 +665,9 @@ mod tests {
             body,
             json!({
                 "name": "alerts",
-                "database": "analytics",
-                "sql": "SELECT 1",
+                "query": {"database": "analytics", "sql": "SELECT 1"},
                 "enabled": false,
+                "interval_seconds": null,
                 "sinks": [{"type": "table", "settings": {"database": "a", "table": "b"}}],
             })
         );
@@ -687,7 +700,7 @@ mod tests {
             name: None,
             database: None,
             sql: None,
-            interval_seconds: Some(30),
+            interval_seconds: Some(Some(30)),
             enabled: Some(false),
             sinks: Some(vec![]),
         })
@@ -696,6 +709,25 @@ mod tests {
             body,
             json!({"interval_seconds": 30, "enabled": false, "sinks": []})
         );
+
+        let body = update_body(UpdateWorkflow {
+            name: None,
+            database: None,
+            sql: Some("SELECT 2".into()),
+            interval_seconds: None,
+            enabled: None,
+            sinks: None,
+        })
+        .unwrap();
+        assert_eq!(body, json!({"query": {"sql": "SELECT 2"}}));
+    }
+
+    #[test]
+    fn format_interval_shows_only_scheduled_durations() {
+        assert_eq!(format_interval(None), "—");
+        assert_eq!(format_interval(Some(7200)), "2h");
+        assert_eq!(format_interval(Some(120)), "2m");
+        assert_eq!(format_interval(Some(45)), "45s");
     }
 
     #[test]
@@ -765,7 +797,7 @@ mod tests {
             describe_sink(
                 &json!({"type": "http", "id": "s2", "settings": {"url_configured": true, "header_names": ["Authorization"]}})
             ),
-            "http  id=s2  headers=Authorization"
+            "http URL unavailable  id=s2  headers=Authorization"
         );
         assert_eq!(
             describe_sink(&json!({"type": "queue"})),

src/main.rs

@@ -531,7 +531,7 @@ fn run(cli: Cli) -> Result<ExitCode> {
                         name,
                         database: resolve_database(database)?,
                         sql: read_sql_arg(sql)?,
-                        interval_seconds,
+                        interval_seconds: interval_seconds.and_then(|interval| interval.0),
                         disabled,
                         sinks,
                     },
@@ -555,7 +555,7 @@ fn run(cli: Cli) -> Result<ExitCode> {
                         name,
                         database,
                         sql: sql.map(read_sql_arg).transpose()?,
-                        interval_seconds,
+                        interval_seconds: interval_seconds.map(|interval| interval.0),
                         enabled: (enable || disable).then_some(enable),
                         sinks: (clear_sinks || !sinks.is_empty()).then_some(sinks),
                     },