use std::collections::HashMap;
use serde_json::{Value, json};
use crate::pipeline::model::{DslFlag, DslFlagKind, PipelineEdge, PipelineGraph, PipelineNode};
use crate::pipeline::nodes::builtin_node_definitions;
#[derive(Debug, Clone)]
pub enum DslVerb {
Get {
resource: String,
path: Option<String>,
filter: Option<String>,
status: Option<String>,
},
Describe {
kind: String,
name: String,
compact: bool,
},
Read { kind: String, name: String },
Write {
kind: String,
name: String,
body: Option<String>,
},
Delete { kind: String, name: String },
Activate { file_rel_path: String },
Deactivate { file_rel_path: String },
Execute { file_rel_path: String, input: Value },
Register {
file_rel_path: String,
title: String,
description: String,
as_json: bool,
body: String,
},
Patch {
file_rel_path: String,
node_id: String,
flags: HashMap<String, Value>,
body: Option<String>,
},
Run { body: String, dry_run: bool },
Git {
subcommand: String,
args: Vec<String>,
body: Option<String>,
},
NodeHelp { kind: String },
CredentialBlocked { reason: String },
Unknown { raw: String },
}
pub fn tokenize(s: &str) -> Vec<String> {
let mut tokens = Vec::new();
let mut current = String::new();
let mut in_single = false;
let mut in_double = false;
for ch in s.chars() {
match ch {
'\'' if !in_double => {
in_single = !in_single;
}
'"' if !in_single => {
in_double = !in_double;
}
' ' | '\t' | '\n' | '\r' if !in_single && !in_double => {
if !current.is_empty() {
tokens.push(current.clone());
current.clear();
}
}
_ => {
current.push(ch);
}
}
}
if !current.is_empty() {
tokens.push(current);
}
tokens
}
pub fn split_commands(dsl: &str) -> Vec<String> {
let joined = dsl.replace("\\\r\n", "\n").replace("\\\n", "\n");
joined
.split("&&")
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
}
pub fn expand_kind(short: &str) -> Option<&'static str> {
match short {
"trigger.webhook" | "n.trigger.webhook" => Some("n.trigger.webhook"),
"trigger.schedule" | "n.trigger.schedule" => Some("n.trigger.schedule"),
"trigger.manual" | "n.trigger.manual" => Some("n.trigger.manual"),
"pg.query" | "n.pg.query" => Some("n.pg.query"),
"sekejap.query" | "n.sekejap.query" => Some("n.sekejap.query"),
"table.convert" | "n.table.convert" => Some("n.table.convert"),
"table.query" | "n.table.query" => Some("n.table.query"),
"script" | "n.script" => Some("n.script"),
"web.response" | "n.web.response" => Some("n.web.response"),
"web.static.generate" | "n.web.static.generate" => Some("n.web.static.generate"),
"web.docs.generate" | "n.web.docs.generate" => Some("n.web.docs.generate"),
"http.request" | "n.http.request" => Some("n.http.request"),
"zebtune" | "n.zebtune" => Some("n.zebtune"),
"logic.if" | "n.logic.if" => Some("n.logic.if"),
"logic.match" | "n.logic.match" => Some("n.logic.match"),
"logic.collect" | "n.logic.collect" => Some("n.logic.collect"),
"logic.foreach" | "n.logic.foreach" => Some("n.logic.foreach"),
"logic.reduce" | "n.logic.reduce" => Some("n.logic.reduce"),
"logic.retry" | "n.logic.retry" => Some("n.logic.retry"),
"trigger.ws" | "n.trigger.ws" => Some("n.trigger.ws"),
"ws.emit" | "n.ws.emit" => Some("n.ws.emit"),
"ws.sync_state" | "n.ws.sync_state" => Some("n.ws.sync_state"),
"auth.token.create" | "n.auth.token.create" => Some("n.auth.token.create"),
"crypto" | "n.crypto" => Some("n.crypto"),
"ai.agent" | "n.ai.agent" => Some("n.ai.agent"),
"ai.tts" | "n.ai.tts" => Some("n.ai.tts"),
"browser.run" | "n.browser.run" => Some("n.browser.run"),
"trigger.weberror" | "n.trigger.weberror" => Some("n.trigger.weberror"),
"trigger.function" | "n.trigger.function" => Some("n.trigger.function"),
"function.call" | "n.function.call" => Some("n.function.call"),
"fs.save" | "n.fs.save" => Some("n.fs.save"),
"fs.compress" | "n.fs.compress" => Some("n.fs.compress"),
"fs.decompress" | "n.fs.decompress" => Some("n.fs.decompress"),
"fs.pdf.convert" | "n.fs.pdf.convert" => Some("n.fs.pdf.convert"),
"fs.thumbnail" | "n.fs.thumbnail" => Some("n.fs.thumbnail"),
"mem.set" | "n.mem.set" => Some("n.mem.set"),
"mem.get" | "n.mem.get" => Some("n.mem.get"),
"mem.del" | "n.mem.del" => Some("n.mem.del"),
"mem.exists" | "n.mem.exists" => Some("n.mem.exists"),
"mem.expire" | "n.mem.expire" => Some("n.mem.expire"),
"mem.incr" | "n.mem.incr" => Some("n.mem.incr"),
"mem.publish" | "n.mem.publish" => Some("n.mem.publish"),
"trigger.memsubscribe" | "n.trigger.memsubscribe" => Some("n.trigger.memsubscribe"),
_ => None,
}
}
pub fn default_pins(kind: &str) -> (Vec<String>, Vec<String>) {
match kind {
"n.trigger.webhook" | "n.trigger.schedule" | "n.trigger.manual" | "n.trigger.function" => {
(vec![], vec!["out".to_string()])
}
"n.pg.query" | "n.sekejap.query" | "n.table.convert" | "n.table.query" | "n.script"
| "n.http.request" | "n.zebtune" | "n.logic.collect" | "n.ai.tts" => {
(vec!["in".to_string()], vec!["out".to_string()])
}
"n.logic.foreach" => (vec!["in".to_string()], vec!["item".to_string()]),
"n.logic.reduce" => (vec!["in".to_string()], vec!["out".to_string()]),
"n.logic.retry" => (
vec!["in".to_string()],
vec!["retry".to_string(), "failed".to_string()],
),
"n.logic.if" => (
vec!["in".to_string()],
vec!["true".to_string(), "false".to_string()],
),
"n.logic.match" => (vec!["in".to_string()], vec!["default".to_string()]),
"n.web.response" => (vec!["in".to_string()], vec!["out".to_string()]),
"n.trigger.ws" => (vec![], vec!["out".to_string()]),
"n.ws.emit" | "n.ws.sync_state" => (vec!["in".to_string()], vec!["out".to_string()]),
"n.function.call" => (
vec!["in".to_string()],
vec!["out".to_string(), "error".to_string()],
),
_ => (vec!["in".to_string()], vec!["out".to_string()]),
}
}
fn find_first_pipe_in_raw(raw: &str) -> Option<usize> {
let mut in_single = false;
let mut in_double = false;
for (i, ch) in raw.char_indices() {
match ch {
'\'' if !in_double => in_single = !in_single,
'"' if !in_single => in_double = !in_double,
'|' if !in_single && !in_double => return Some(i),
_ => {}
}
}
None
}
fn strip_outer_quotes(s: &str) -> &str {
if s.len() >= 2
&& ((s.starts_with('"') && s.ends_with('"')) || (s.starts_with('\'') && s.ends_with('\'')))
{
&s[1..s.len() - 1]
} else {
s
}
}
fn extract_raw_body_from(raw: &str) -> Option<String> {
raw.find(" -- ")
.map(|pos| {
let after = raw[pos + 4..].trim();
strip_outer_quotes(after).to_string()
})
.filter(|s| !s.is_empty())
}
fn coerce_scalar_value(s: &str) -> Value {
match s {
"true" => json!(true),
"false" => json!(false),
_ => {
if let Ok(n) = s.parse::<i64>() {
json!(n)
} else if let Ok(f) = s.parse::<f64>() {
json!(f)
} else {
json!(s)
}
}
}
}
pub fn parse_node_config(
tokens: &[String],
raw: &str,
dsl_flags: &[DslFlag],
) -> Result<(Value, Option<String>), String> {
let mut config = serde_json::Map::new();
let mut list_modes = HashMap::<String, ListFlagMode>::new();
let mut body: Option<String> = None;
let mut i = 0;
while i < tokens.len() {
let t = &tokens[i];
if t == "--" {
body = extract_raw_body_from(raw);
break;
}
if let Some(key) = t.strip_prefix("--") {
let flag_str = format!("--{key}");
let dsl_flag = dsl_flags
.iter()
.find(|f| f.flag == flag_str)
.ok_or_else(|| {
format!("unknown flag `--{key}` — not declared in this node's dsl_flags")
})?;
match dsl_flag.kind {
DslFlagKind::Scalar => {
let val = tokens.get(i + 1).cloned().unwrap_or_default();
config.insert(dsl_flag.config_key.clone(), coerce_scalar_value(&val));
i += 2;
}
DslFlagKind::CommaSeparatedList | DslFlagKind::RepeatedList => {
let val = tokens.get(i + 1).cloned().unwrap_or_default();
push_list_flag_value(&mut config, &mut list_modes, dsl_flag, &val)?;
i += 2;
}
DslFlagKind::Bool => {
config.insert(dsl_flag.config_key.clone(), json!(true));
i += 1;
}
DslFlagKind::KeyValuePairs => {
let raw = tokens.get(i + 1).cloned().unwrap_or_default();
let (k, v) = if let Some(eq) = raw.find('=') {
(raw[..eq].trim().to_string(), raw[eq + 1..].to_string())
} else {
(raw.trim().to_string(), String::new())
};
let entry = config
.entry(dsl_flag.config_key.clone())
.or_insert_with(|| Value::Object(serde_json::Map::new()));
if let Value::Object(m) = entry {
m.insert(k, json!(v));
}
i += 2;
}
}
} else {
i += 1;
}
}
Ok((Value::Object(config), body))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ListFlagMode {
Compact,
Repeated,
}
fn push_list_flag_value(
config: &mut serde_json::Map<String, Value>,
list_modes: &mut HashMap<String, ListFlagMode>,
flag: &DslFlag,
value: &str,
) -> Result<(), String> {
let has_comma = value.contains(',');
let next_mode = if has_comma {
ListFlagMode::Compact
} else {
ListFlagMode::Repeated
};
let key = flag.config_key.clone();
if let Some(existing) = list_modes.get(&key) {
if *existing != next_mode || has_comma {
return Err(format!(
"list flag `{}` must use one style per node command: either `{0} a,b,c` or `{0} a {0} b`, not both",
flag.flag
));
}
} else {
list_modes.insert(key.clone(), next_mode);
}
let values = if has_comma {
value
.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|s| json!(s))
.collect::<Vec<_>>()
} else if value.trim().is_empty() {
Vec::new()
} else {
vec![json!(value)]
};
let entry = config
.entry(key)
.or_insert_with(|| Value::Array(Vec::new()));
if let Value::Array(existing) = entry {
existing.extend(values);
}
Ok(())
}
fn parse_flags_for_patch(tokens: &[String], cmd: &str) -> (HashMap<String, Value>, Option<String>) {
let mut flags: HashMap<String, Value> = HashMap::new();
let mut body: Option<String> = None;
let mut i = 0;
while i < tokens.len() {
let t = &tokens[i];
if t == "--" {
body = extract_raw_body_from(cmd);
break;
}
if let Some(key) = t.strip_prefix("--") {
let val = tokens.get(i + 1).cloned().unwrap_or_default();
let config_key = key.replace('-', "_");
let new_val = coerce_scalar_value(&val);
match flags.entry(config_key) {
std::collections::hash_map::Entry::Occupied(mut e) => match e.get_mut() {
Value::Array(arr) => arr.push(new_val),
existing => {
let prev = existing.clone();
*existing = Value::Array(vec![prev, new_val]);
}
},
std::collections::hash_map::Entry::Vacant(e) => {
e.insert(new_val);
}
}
i += 2;
} else {
i += 1;
}
}
(flags, body)
}
fn parse_register(tokens: &[String], cmd: &str) -> DslVerb {
let file_rel_path = tokens.get(1).cloned().unwrap_or_default();
let mut title = String::new();
let mut description = String::new();
let mut as_json = false;
let mut i = 2;
while i < tokens.len() {
match tokens[i].as_str() {
"--title" => {
title = tokens.get(i + 1).cloned().unwrap_or_default();
i += 2;
}
"--description" => {
description = tokens.get(i + 1).cloned().unwrap_or_default();
i += 2;
}
"--as-json" => {
as_json = true;
i += 1;
}
_ => break,
}
}
let body = match find_first_pipe_in_raw(cmd) {
Some(pos) => cmd[pos..].to_string(),
None => cmd
.find('[')
.map(|pos| cmd[pos..].to_string())
.unwrap_or_default(),
};
DslVerb::Register {
file_rel_path,
title,
description,
as_json,
body,
}
}
fn parse_patch(tokens: &[String], cmd: &str) -> DslVerb {
let file_rel_path = tokens.get(2).cloned().unwrap_or_default();
let node_id = tokens.get(4).cloned().unwrap_or_default();
let flag_tokens = if tokens.len() > 5 {
tokens[5..].to_vec()
} else {
vec![]
};
let (flags, body) = parse_flags_for_patch(&flag_tokens, cmd);
DslVerb::Patch {
file_rel_path,
node_id,
flags,
body,
}
}
fn extract_flag(tokens: &[String], flag: &str) -> Option<String> {
let pos = tokens.iter().position(|t| t == flag)?;
tokens.get(pos + 1).cloned()
}
fn extract_body(tokens: &[String]) -> Option<String> {
let pos = tokens.iter().position(|t| t == "--")?;
let rest = tokens[pos + 1..].join(" ");
if rest.is_empty() { None } else { Some(rest) }
}
pub fn parse_one_command(cmd: &str) -> DslVerb {
let tokens = tokenize(cmd);
if tokens.is_empty() {
return DslVerb::Unknown {
raw: cmd.to_string(),
};
}
match tokens[0].to_lowercase().as_str() {
"get" => {
let resource = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let path = extract_flag(&tokens, "--path");
let filter = extract_flag(&tokens, "--filter");
let status = extract_flag(&tokens, "--status");
DslVerb::Get { resource, path, filter, status }
}
"describe" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
let compact = tokens.iter().any(|t| t == "--compact");
DslVerb::Describe { kind, name, compact }
}
"read" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
DslVerb::Read { kind, name }
}
"write" | "create" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
let body = extract_body(&tokens);
DslVerb::Write { kind, name, body }
}
"delete" | "rm" | "remove" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
DslVerb::Delete { kind, name }
}
"activate" => {
let mut idx = 1;
if tokens.get(1).map(|s| s.as_str()) == Some("pipeline") { idx = 2; }
let file_rel_path = tokens.get(idx).cloned().unwrap_or_default();
DslVerb::Activate { file_rel_path }
}
"deactivate" => {
let mut idx = 1;
if tokens.get(1).map(|s| s.as_str()) == Some("pipeline") { idx = 2; }
let file_rel_path = tokens.get(idx).cloned().unwrap_or_default();
DslVerb::Deactivate { file_rel_path }
}
"execute" | "exec" => {
let mut idx = 1;
if tokens.get(1).map(|s| s.as_str()) == Some("pipeline") { idx = 2; }
let file_rel_path = tokens.get(idx).cloned().unwrap_or_default();
let input_str = extract_flag(&tokens, "--input").unwrap_or_default();
let input = serde_json::from_str(&input_str).unwrap_or(json!({}));
DslVerb::Execute { file_rel_path, input }
}
"register" | "reg" => parse_register(&tokens, cmd),
"patch" => parse_patch(&tokens, cmd),
"run" => {
let dry_run = tokens.iter().any(|t| t == "--dry-run");
let body = match find_first_pipe_in_raw(cmd) {
Some(pos) => cmd[pos..].to_string(),
None => cmd
.find('[')
.map(|pos| cmd[pos..].to_string())
.unwrap_or_default(),
};
DslVerb::Run { body, dry_run }
}
"git" => {
let subcommand = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let dash_pos = tokens.iter().position(|t| t == "--");
let args = if let Some(pos) = dash_pos {
tokens[2..pos].to_vec()
} else {
tokens[2..].to_vec()
};
let body = dash_pos.and_then(|pos| {
let b = tokens[pos + 1..].join(" ");
if b.is_empty() { None } else { Some(b) }
});
DslVerb::Git { subcommand, args, body }
}
"node" => {
let sub = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
if sub == "help" {
let kind = tokens.get(2).cloned().unwrap_or_default();
DslVerb::NodeHelp { kind }
} else {
DslVerb::Unknown { raw: cmd.to_string() }
}
}
"credential" | "credentials" | "secret" | "secrets"
| "set-secret" | "set-credential" | "set-env" => {
DslVerb::CredentialBlocked {
reason: "Credential writes are blocked via DSL for security. Use the Credentials UI at /projects/{owner}/{project}/credentials".to_string(),
}
}
_ => DslVerb::Unknown { raw: cmd.to_string() },
}
}
pub fn build_pipeline_graph(id: &str, body: &str) -> Result<PipelineGraph, String> {
let body = body.trim();
if body.is_empty() {
return Err("Pipeline body is empty".to_string());
}
if body.contains('[') && body.contains("] ->") {
build_graph_mode(id, body)
} else {
build_pipe_mode(id, body)
}
}
fn split_pipe_segments(body: &str) -> Vec<&str> {
let mut segments = Vec::new();
let mut start = 0usize;
let mut in_single = false;
let mut in_double = false;
for (byte_pos, ch) in body.char_indices() {
match ch {
'\'' if !in_double => in_single = !in_single,
'"' if !in_single => in_double = !in_double,
'|' if !in_single && !in_double => {
let seg = body[start..byte_pos].trim();
if !seg.is_empty() {
segments.push(seg);
}
start = byte_pos + ch.len_utf8();
}
_ => {}
}
}
let last = body[start..].trim();
if !last.is_empty() {
segments.push(last);
}
segments
}
fn build_graph_mode(id: &str, body: &str) -> Result<PipelineGraph, String> {
let mut nodes: Vec<PipelineNode> = Vec::new();
let mut edges: Vec<PipelineEdge> = Vec::new();
let joined = body.replace("\\\n", " ").replace("\\\r\n", " ");
let lines: Vec<&str> = joined
.lines()
.map(str::trim)
.filter(|s| !s.is_empty())
.collect();
for line in &lines {
if !line.starts_with('[') {
continue;
}
if line.contains("->") {
parse_graph_edge(line, &mut edges)?;
} else {
parse_graph_node(line, &mut nodes)?;
}
}
let to_nodes: std::collections::HashSet<&str> =
edges.iter().map(|e| e.to_node.as_str()).collect();
let entry_nodes: Vec<String> = nodes
.iter()
.filter(|n| !to_nodes.contains(n.id.as_str()))
.map(|n| n.id.clone())
.collect();
Ok(PipelineGraph {
kind: "zebflow.pipeline".to_string(),
version: "0.1".to_string(),
id: id.to_string(),
description: None,
metadata: None,
entry_nodes,
nodes,
edges,
})
}
#[cfg(test)]
mod tests {
use crate::pipeline::model::PipelineNode;
use serde_json::json;
use super::{
DslVerb, build_pipeline_graph, node_to_segment_no_body, parse_one_command, split_commands,
};
#[test]
fn logic_match_cases_accept_compact_list_form() {
let dsl = r#"
[a] trigger.manual
[b] logic.match --expr "$input.type" --cases create,update,delete --default unknown
[a] -> [b]
"#;
let graph = build_pipeline_graph("parser-match-cases-test", dsl).expect("graph");
let node = graph
.nodes
.iter()
.find(|node| node.id == "b")
.expect("match node");
assert_eq!(
node.config.get("cases"),
Some(&json!(["create", "update", "delete"]))
);
assert_eq!(
node.output_pins,
vec![
"create".to_string(),
"update".to_string(),
"delete".to_string(),
"unknown".to_string()
]
);
}
#[test]
fn logic_match_cases_accept_repeated_list_form() {
let dsl = r#"
[a] trigger.manual
[b] logic.match --expr "$input.type" --cases create --cases update --cases delete --default unknown
[a] -> [b]
"#;
let graph = build_pipeline_graph("parser-match-cases-repeated-test", dsl).expect("graph");
let node = graph
.nodes
.iter()
.find(|node| node.id == "b")
.expect("match node");
assert_eq!(
node.config.get("cases"),
Some(&json!(["create", "update", "delete"]))
);
}
#[test]
fn list_flags_reject_mixed_compact_and_repeated_forms() {
let dsl = r#"
[a] trigger.manual
[b] logic.match --expr "$input.type" --cases create,update --cases delete --default unknown
[a] -> [b]
"#;
let err = build_pipeline_graph("parser-match-cases-mixed-test", dsl)
.expect_err("mixed list syntax must fail");
assert!(err.contains("must use one style"));
}
#[test]
fn repeated_list_reconstructs_source_binding_objects() {
let node = PipelineNode {
id: "b".to_string(),
kind: "n.table.query".to_string(),
config: json!({
"sources": [
{ "source": "datasets/posts.csv", "alias": "posts" },
{ "source": "$input.rows", "alias": "rows" }
],
"sql": "select * from posts",
"to_json": true
}),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string()],
};
let segment = node_to_segment_no_body(&node);
assert!(segment.contains("--from \"datasets/posts.csv as posts\""));
assert!(segment.contains("--from \"$input.rows as rows\""));
assert!(segment.contains("--to-json"));
}
#[test]
fn split_commands_preserves_graph_mode_lines_for_run() {
let dsl = r#"
run \
[a] trigger.manual \
[b] script -- "return { ok: true };" \
[a] -> [b]
"#;
let commands = split_commands(dsl);
assert_eq!(commands.len(), 1);
let verb = parse_one_command(&commands[0]);
match verb {
DslVerb::Run { body, dry_run } => {
assert!(!dry_run);
let graph = build_pipeline_graph("graph-run-test", &body).expect("graph");
assert_eq!(graph.nodes.len(), 2);
assert_eq!(graph.edges.len(), 1);
}
other => panic!("expected run verb, got {other:?}"),
}
}
}
fn parse_graph_node(line: &str, nodes: &mut Vec<PipelineNode>) -> Result<(), String> {
let rest = line
.strip_prefix('[')
.ok_or("expected '[' at start of node declaration")?;
let (label, rest) = rest
.split_once(']')
.ok_or("expected ']' in node declaration")?;
let label = label.trim();
if label.is_empty() {
return Err("node label must not be empty".to_string());
}
let rest = rest.trim();
let tokens = tokenize(rest);
if tokens.is_empty() {
return Err(format!("node '[{label}]' has no kind"));
}
let raw_kind = &tokens[0];
let full_kind =
expand_kind(raw_kind).ok_or_else(|| format!("Unknown node kind: '{raw_kind}'"))?;
let (input_pins, mut output_pins) = default_pins(full_kind);
let all_defs = builtin_node_definitions();
let dsl_flags = all_defs
.iter()
.find(|d| d.kind == full_kind)
.map(|d| d.dsl_flags.as_slice())
.unwrap_or(&[]);
let (mut config, body_val) = parse_node_config(&tokens[1..], rest, dsl_flags)?;
if let Some(bval) = body_val {
let body_key = match full_kind {
"n.pg.query" => "query",
"n.sekejap.query" => "query",
"n.table.query" => "sql",
"n.script" => "source",
"n.logic.match" | "n.logic.if" => "expression",
"n.browser.run" => "code",
_ => "body",
};
if let Value::Object(ref mut map) = config {
map.insert(body_key.to_string(), json!(bval));
}
}
if full_kind == "n.logic.match" {
if let Value::Object(ref map) = config {
let cases: Vec<String> = map
.get("cases")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_default();
let default_pin = map
.get("default")
.and_then(|v| v.as_str())
.unwrap_or("default")
.to_string();
let mut pins = cases;
if !pins.contains(&default_pin) {
pins.push(default_pin);
}
if !pins.is_empty() {
output_pins = pins;
}
}
}
nodes.push(PipelineNode {
id: label.to_string(),
kind: full_kind.to_string(),
input_pins,
output_pins,
config,
});
Ok(())
}
fn parse_graph_edge(line: &str, edges: &mut Vec<PipelineEdge>) -> Result<(), String> {
let arrow_pos = line.find("->").ok_or("expected '->' in edge declaration")?;
let from_part = line[..arrow_pos].trim();
let to_part = line[arrow_pos + 2..].trim();
let (from_node, from_pin) = parse_node_pin_part(from_part, "out")?;
let (to_node, to_pin) = parse_node_pin_part(to_part, "in")?;
edges.push(PipelineEdge {
from_node,
from_pin,
to_node,
to_pin,
});
Ok(())
}
fn parse_node_pin_part(s: &str, default_pin: &str) -> Result<(String, String), String> {
let s = s.trim();
let inner = s
.strip_prefix('[')
.ok_or_else(|| format!("expected '[' in edge endpoint: '{s}'"))?;
let (label, rest) = inner
.split_once(']')
.ok_or_else(|| format!("expected ']' in edge endpoint: '{s}'"))?;
let label = label.trim().to_string();
let pin = rest
.trim()
.strip_prefix(':')
.map(|p| p.trim().to_string())
.unwrap_or_else(|| default_pin.to_string());
Ok((label, pin))
}
pub fn graph_to_dsl(graph: &PipelineGraph) -> String {
if is_linear_graph(graph) {
graph_to_pipe_mode(graph)
} else {
graph_to_graph_mode(graph)
}
}
fn is_linear_graph(graph: &PipelineGraph) -> bool {
let mut in_count: HashMap<&str, usize> = HashMap::new();
let mut out_count: HashMap<&str, usize> = HashMap::new();
for edge in &graph.edges {
if edge.from_pin != "out" || edge.to_pin != "in" {
return false;
}
*in_count.entry(edge.to_node.as_str()).or_insert(0) += 1;
*out_count.entry(edge.from_node.as_str()).or_insert(0) += 1;
}
graph.nodes.iter().all(|n| {
*in_count.get(n.id.as_str()).unwrap_or(&0) <= 1
&& *out_count.get(n.id.as_str()).unwrap_or(&0) <= 1
})
}
fn node_to_segment(node: &PipelineNode) -> String {
let all_defs = builtin_node_definitions();
let dsl_flags = all_defs
.iter()
.find(|d| d.kind == node.kind)
.map(|d| d.dsl_flags.as_slice())
.unwrap_or(&[]);
let kind = node.kind.strip_prefix("n.").unwrap_or(&node.kind);
let mut parts = vec![kind.to_string()];
for flag in dsl_flags {
let Some(val) = node.config.get(&flag.config_key) else {
continue;
};
match &flag.kind {
DslFlagKind::Bool => {
if val.as_bool().unwrap_or(false) {
parts.push(flag.flag.clone());
}
}
DslFlagKind::CommaSeparatedList => {
if let Some(arr) = val.as_array() {
let csv = arr
.iter()
.filter_map(|v| v.as_str())
.collect::<Vec<_>>()
.join(",");
if !csv.is_empty() {
parts.push(flag.flag.clone());
parts.push(csv);
}
}
}
DslFlagKind::RepeatedList => {
if let Some(arr) = val.as_array() {
for item in arr.iter().filter_map(repeated_list_item_to_string) {
if item.is_empty() {
continue;
}
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&item));
}
}
}
DslFlagKind::Scalar => {
let s = match val {
Value::String(s) if !s.is_empty() => s.clone(),
Value::Number(n) => n.to_string(),
Value::Bool(b) => b.to_string(),
_ => continue,
};
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&s));
}
DslFlagKind::KeyValuePairs => {
if let Some(map) = val.as_object() {
for (k, v) in map {
let v_str = match v {
Value::String(s) => s.clone(),
other => other.to_string(),
};
parts.push(flag.flag.clone());
parts.push(format!("{}={}", k, v_str));
}
}
}
}
}
let body_key = match node.kind.as_str() {
"n.pg.query" => "query",
"n.sekejap.query" => "query",
"n.table.query" => "sql",
"n.script" => "source",
"n.logic.match" | "n.logic.if" => "expression",
_ => "body",
};
if let Some(body) = node.config.get(body_key).and_then(|v| v.as_str()) {
let body = body.trim();
if !body.is_empty() {
let inline: String = body
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.collect::<Vec<_>>()
.join(" ");
parts.push("--".to_string());
parts.push(inline);
}
}
parts.join(" ")
}
pub fn node_to_segment_no_body(node: &PipelineNode) -> String {
let all_defs = builtin_node_definitions();
let dsl_flags = all_defs
.iter()
.find(|d| d.kind == node.kind)
.map(|d| d.dsl_flags.as_slice())
.unwrap_or(&[]);
let kind = node.kind.strip_prefix("n.").unwrap_or(&node.kind);
let mut parts = vec![kind.to_string()];
for flag in dsl_flags {
let body_key = match node.kind.as_str() {
"n.pg.query" => "query",
"n.sekejap.query" => "query",
"n.table.query" => "sql",
"n.script" => "source",
"n.logic.match" | "n.logic.if" => "expression",
_ => "body",
};
if flag.config_key == body_key {
continue;
}
let Some(val) = node.config.get(&flag.config_key) else {
continue;
};
match &flag.kind {
DslFlagKind::Bool => {
if val.as_bool().unwrap_or(false) {
parts.push(flag.flag.clone());
}
}
DslFlagKind::CommaSeparatedList => {
if let Some(arr) = val.as_array() {
let csv = arr
.iter()
.filter_map(|v| v.as_str())
.collect::<Vec<_>>()
.join(",");
if !csv.is_empty() {
parts.push(flag.flag.clone());
parts.push(csv);
}
}
}
DslFlagKind::RepeatedList => {
if let Some(arr) = val.as_array() {
for item in arr.iter().filter_map(repeated_list_item_to_string) {
if item.is_empty() {
continue;
}
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&item));
}
}
}
DslFlagKind::Scalar => {
let s = match val {
Value::String(s) if !s.is_empty() => s.clone(),
Value::Number(n) => n.to_string(),
Value::Bool(b) => b.to_string(),
_ => continue,
};
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&s));
}
DslFlagKind::KeyValuePairs => {
if let Some(map) = val.as_object() {
for (k, v) in map {
let v_str = match v {
Value::String(s) => s.clone(),
other => other.to_string(),
};
parts.push(flag.flag.clone());
parts.push(format!("{}={}", k, v_str));
}
}
}
}
}
parts.join(" ")
}
fn quote_dsl_arg(value: &str) -> String {
if value.chars().any(char::is_whitespace) || value.contains('"') {
format!("\"{}\"", value.replace('"', "\\\""))
} else {
value.to_string()
}
}
fn repeated_list_item_to_string(value: &Value) -> Option<String> {
if let Some(s) = value.as_str() {
return Some(s.to_string());
}
let obj = value.as_object()?;
let source = obj.get("source").and_then(Value::as_str)?.trim();
let alias = obj.get("alias").and_then(Value::as_str)?.trim();
if source.is_empty() || alias.is_empty() {
return None;
}
Some(format!("{source} as {alias}"))
}
fn graph_to_pipe_mode(graph: &PipelineGraph) -> String {
let node_map: HashMap<&str, &PipelineNode> =
graph.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
let next_map: HashMap<&str, &str> = graph
.edges
.iter()
.map(|e| (e.from_node.as_str(), e.to_node.as_str()))
.collect();
let mut ordered: Vec<&PipelineNode> = Vec::new();
if let Some(first_id) = graph.entry_nodes.first() {
let mut cur = first_id.as_str();
let mut visited = std::collections::HashSet::new();
loop {
if !visited.insert(cur) {
break; }
if let Some(&node) = node_map.get(cur) {
ordered.push(node);
}
match next_map.get(cur) {
Some(&next) => cur = next,
None => break,
}
}
}
ordered
.iter()
.map(|n| format!("| {}", node_to_segment(n)))
.collect::<Vec<_>>()
.join("\n")
}
fn graph_to_graph_mode(graph: &PipelineGraph) -> String {
let mut lines: Vec<String> = Vec::new();
for node in &graph.nodes {
lines.push(format!("[{}] {}", node.id, node_to_segment(node)));
}
if !graph.edges.is_empty() {
lines.push(String::new());
for edge in &graph.edges {
if edge.from_pin == "out" && edge.to_pin == "in" {
lines.push(format!("[{}] -> [{}]", edge.from_node, edge.to_node));
} else {
lines.push(format!(
"[{}]:{} -> [{}]:{}",
edge.from_node, edge.from_pin, edge.to_node, edge.to_pin
));
}
}
}
lines.join("\n")
}
fn build_pipe_mode(id: &str, body: &str) -> Result<PipelineGraph, String> {
let body = body.trim_start_matches('|').trim();
let segments: Vec<&str> = split_pipe_segments(body);
if segments.is_empty() {
return Err("No nodes in pipeline body".to_string());
}
let mut nodes: Vec<PipelineNode> = Vec::new();
let mut edges: Vec<PipelineEdge> = Vec::new();
let first_tokens = tokenize(segments[0]);
let first_raw_kind = first_tokens.first().map(|s| s.as_str()).unwrap_or("");
let first_full_kind = expand_kind(first_raw_kind).unwrap_or(first_raw_kind);
let has_trigger_first = first_full_kind.starts_with("n.trigger.");
if !has_trigger_first {
nodes.push(PipelineNode {
id: "trigger".to_string(),
kind: "n.trigger.manual".to_string(),
input_pins: vec![],
output_pins: vec!["out".to_string()],
config: json!({}),
});
}
let all_defs = builtin_node_definitions();
for (idx, segment) in segments.iter().enumerate() {
let seg_tokens = tokenize(segment);
if seg_tokens.is_empty() {
continue;
}
let raw_kind = &seg_tokens[0];
let full_kind =
expand_kind(raw_kind).ok_or_else(|| format!("Unknown node kind: '{raw_kind}'"))?;
let node_id = format!("n{idx}");
let (input_pins, mut output_pins) = default_pins(full_kind);
let dsl_flags = all_defs
.iter()
.find(|d| d.kind == full_kind)
.map(|d| d.dsl_flags.as_slice())
.unwrap_or(&[]);
let (mut config, body_val) = parse_node_config(&seg_tokens[1..], segment, dsl_flags)?;
if let Some(bval) = body_val {
let body_key = match full_kind {
"n.pg.query" => "query",
"n.table.query" => "sql",
"n.script" => "source",
"n.browser.run" => "code",
_ => "body",
};
if let Value::Object(ref mut map) = config {
map.insert(body_key.to_string(), json!(bval));
}
}
if full_kind == "n.logic.match" {
if let Value::Object(ref map) = config {
let cases: Vec<String> = map
.get("cases")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_default();
let default_pin = map
.get("default")
.and_then(|v| v.as_str())
.unwrap_or("default")
.to_string();
let mut pins = cases;
if !pins.contains(&default_pin) {
pins.push(default_pin);
}
if !pins.is_empty() {
output_pins = pins;
}
}
}
let from_node = if idx == 0 {
if !has_trigger_first {
Some("trigger".to_string())
} else {
None }
} else {
Some(format!("n{}", idx - 1))
};
if let Some(from_id) = from_node {
edges.push(PipelineEdge {
from_node: from_id,
from_pin: "out".to_string(),
to_node: node_id.clone(),
to_pin: "in".to_string(),
});
}
nodes.push(PipelineNode {
id: node_id,
kind: full_kind.to_string(),
input_pins,
output_pins,
config,
});
}
let entry_nodes = nodes
.first()
.map(|n| vec![n.id.clone()])
.unwrap_or_default();
Ok(PipelineGraph {
kind: "zebflow.pipeline".to_string(),
version: "0.1".to_string(),
id: id.to_string(),
description: None,
metadata: None,
entry_nodes,
nodes,
edges,
})
}