use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use crate::pipeline::model::{DslFlag, DslFlagKind, NodeFieldDef, NodeFieldType};
use crate::pipeline::{
NodeDefinition, PipelineError,
nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler},
};
pub const NODE_KIND: &str = "trigger.function";
pub const ANSWER_KEY: &str = "function";
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Config {
#[serde(default)]
pub description: String,
#[serde(default)]
pub schema: Value,
#[serde(default)]
pub result_schema: Value,
#[serde(default)]
pub examples: Value,
}
pub struct Node {
#[allow(dead_code)]
config: Config,
}
impl Node {
pub fn new(config: Config) -> Self {
Self { config }
}
}
pub fn definition() -> NodeDefinition {
NodeDefinition {
kind: NODE_KIND.to_string(),
title: "Function Trigger".to_string(),
description: "Makes this pipeline a function other pipelines call. They call it with `function.result.call --function <slug>`; the slug is the file's \
stem (`jobs/send-welcome` → `send-welcome`). Answers one key, `function`: exactly the arguments the caller passed \
(`input.function.<name>`, `$trigger.<name>`); the function's result is what its last node answered (a closing `javascript.script.run`'s `script`). Declare the contract with \
`--parameter name:type! \"doc\"` / `--result name:type!` (or full `--schema` / `--result-schema` JSON) so callers and the \
assistant see typed fields. Also the trigger for one-off runs through `pipeline_run` (`| trigger.function | …`), where the \
`input` argument is what it answers."
.to_string(),
input_pins: vec![],
output_pins: vec!["out".to_string()],
config_schema: serde_json::json!({
"type": "object",
"required": ["description"],
"properties": {
"description": {
"type": "string",
"description": "What this function does and when callers should use it."
},
"schema": {
"type": "object",
"description": "JSON Schema object describing the function's arguments."
},
"result_schema": {
"type": "object",
"description": "JSON Schema object describing the function's result."
},
"examples": {
"type": "array",
"description": "Optional examples with input and output objects.",
"items": { "type": "object" }
}
}
}),
dsl_flags: vec![
DslFlag {
flag: "--description".to_string(),
config_key: "description".to_string(),
description: "What this function does and when callers should use it.".to_string(),
kind: DslFlagKind::Scalar,
required: true,
value: "text".to_string(),
..Default::default()
},
DslFlag {
flag: "--parameter".to_string(),
config_key: "schema".to_string(),
description: "One parameter the function declares, repeated: name:type! plus an optional description. \
A caller fills it with `function.result.call --argument name=value`. \
Example: --parameter columns:string[]! \"Source column names.\""
.to_string(),
kind: DslFlagKind::SchemaField,
required: false,
value: "text".to_string(),
..Default::default()
},
DslFlag {
flag: "--result".to_string(),
config_key: "result_schema".to_string(),
description: "One field of the function's result, repeated: name:type! plus an optional description. \
Example: --result ok:boolean! \"Whether the function succeeded.\""
.to_string(),
kind: DslFlagKind::SchemaField,
required: false,
value: "text".to_string(),
..Default::default()
},
DslFlag {
flag: "--schema".to_string(),
config_key: "schema".to_string(),
description: "The parameters as one full JSON Schema object (instead of --parameter).".to_string(),
kind: DslFlagKind::Scalar,
required: false,
value: "json".to_string(),
..Default::default()
},
DslFlag {
flag: "--result-schema".to_string(),
config_key: "result_schema".to_string(),
description: "The result as one full JSON Schema object (instead of --result).".to_string(),
kind: DslFlagKind::Scalar,
required: false,
value: "json".to_string(),
..Default::default()
},
DslFlag {
flag: "--example".to_string(),
config_key: "examples".to_string(),
description: "Optional JSON example with input and output. Repeat for 1-3 examples."
.to_string(),
kind: DslFlagKind::RepeatedList,
required: false,
value: "json".to_string(),
..Default::default()
},
],
fields: vec![
NodeFieldDef {
name: "description".to_string(),
label: "Description".to_string(),
field_type: NodeFieldType::Textarea,
help: Some("What this function does and when callers should use it.".to_string()),
rows: Some(3),
span: Some("full".to_string()),
..Default::default()
},
NodeFieldDef {
name: "schema".to_string(),
label: "Arguments".to_string(),
field_type: NodeFieldType::ParamsBuilder,
help: Some(
"Define the arguments this function accepts. Required fields are enforced before execution.".to_string(),
),
default_value: Some(json!({ "type": "object", "required": [], "properties": {} })),
span: Some("full".to_string()),
..Default::default()
},
NodeFieldDef {
name: "result_schema".to_string(),
label: "Result".to_string(),
field_type: NodeFieldType::ParamsBuilder,
help: Some(
"Define the structured result this function returns. Include ok:boolean for tool-friendly results.".to_string(),
),
default_value: Some(json!({
"type": "object",
"required": ["ok"],
"properties": {
"ok": {
"type": "boolean",
"description": "Whether the function succeeded."
}
}
})),
span: Some("full".to_string()),
..Default::default()
},
NodeFieldDef {
name: "examples".to_string(),
label: "Examples".to_string(),
field_type: NodeFieldType::CodeEditor,
language: Some("json".to_string()),
help: Some(
"Optional JSON array of examples: [{\"input\": {...}, \"output\": {...}}].".to_string(),
),
default_value: Some(json!("[]")),
span: Some("full".to_string()),
..Default::default()
},
],
examples: vec![
crate::pipeline::model::NodeExample::dsl("A reusable lookup", r#"trigger.function --description "Find one user by email." --parameter email:string! "Address to look up." --result user:object "The row, or null.""#)
.input(serde_json::json!({ "email": "a@example.com" }))
.output(serde_json::json!({ "function": { "email": "a@example.com" } }))
.note("Registered as `jobs/find-user`; called with `function.result.call --function find-user --argument \"email={{ $trigger.body.email }}\"`; the next node reads `input.function.email`."),
],
..Default::default()
}
}
pub fn input_schema_from_config(config: &Value) -> Value {
schema_from_config(config, "schema").unwrap_or_else(empty_object_schema)
}
pub fn output_schema_from_config(config: &Value) -> Value {
schema_from_config(config, "result_schema").unwrap_or_else(|| {
json!({
"type": "object",
"required": ["ok"],
"properties": {
"ok": {
"type": "boolean",
"description": "Whether the function succeeded."
}
}
})
})
}
pub fn examples_from_config(config: &Value) -> Value {
let value = config
.get("examples")
.and_then(parse_json_value)
.filter(|v| v.is_array())
.unwrap_or_else(|| json!([]));
let items = value
.as_array()
.map(|arr| {
arr.iter()
.map(|item| parse_json_value(item).unwrap_or_else(|| item.clone()))
.collect::<Vec<_>>()
})
.unwrap_or_default();
Value::Array(items)
}
pub fn tool_description_from_config(slug: &str, config: &Value) -> String {
let title = config
.get("title")
.and_then(Value::as_str)
.filter(|s| !s.trim().is_empty())
.unwrap_or(slug);
let description = config
.get("description")
.and_then(Value::as_str)
.unwrap_or("")
.trim();
let output_schema = output_schema_from_config(config);
let examples = examples_from_config(config);
let mut parts = vec![format!("{}: {}", title.trim(), description)];
if let Ok(pretty) = serde_json::to_string_pretty(&output_schema) {
parts.push(format!("Output schema:\n{pretty}"));
}
if examples.as_array().map(|a| !a.is_empty()).unwrap_or(false)
&& let Ok(pretty) = serde_json::to_string_pretty(&examples)
{
parts.push(format!("Examples:\n{pretty}"));
}
parts.join("\n\n")
}
pub fn validate_function_input(schema: &Value, input: &Value) -> Result<(), Value> {
let Some(obj) = input.as_object() else {
return Err(validation_error(
"invalid_input_type",
None,
"Function input must be a JSON object.",
));
};
if let Some(required) = schema.get("required").and_then(Value::as_array) {
for name in required.iter().filter_map(Value::as_str) {
if !obj.contains_key(name) {
return Err(validation_error(
"missing_required_arg",
Some(name),
&format!("Call this function again with `{name}`."),
));
}
}
}
let Some(properties) = schema.get("properties").and_then(Value::as_object) else {
return Ok(());
};
for (name, prop_schema) in properties {
if let Some(value) = obj.get(name)
&& !schema_type_matches(prop_schema, value)
{
return Err(validation_error(
"invalid_arg_type",
Some(name),
&format!("Argument `{name}` does not match the declared schema."),
));
}
}
Ok(())
}
fn schema_from_config(config: &Value, key: &str) -> Option<Value> {
config
.get(key)
.and_then(parse_json_value)
.map(normalize_schema_object)
}
fn parse_json_value(value: &Value) -> Option<Value> {
match value {
Value::String(s) => serde_json::from_str::<Value>(s).ok(),
other => Some(other.clone()),
}
}
fn normalize_schema_object(schema: Value) -> Value {
if schema.get("properties").is_some() {
let mut schema = schema;
if let Some(obj) = schema.as_object_mut() {
obj.entry("type".to_string())
.or_insert_with(|| json!("object"));
if !obj.contains_key("required")
&& let Some(props) = obj.get("properties")
{
obj.insert("required".to_string(), required_from_properties(props));
}
}
schema
} else if schema.is_object() {
json!({
"type": "object",
"properties": schema,
"required": required_from_properties(&schema)
})
} else {
empty_object_schema()
}
}
fn required_from_properties(properties: &Value) -> Value {
let required = properties
.as_object()
.map(|props| {
props
.iter()
.filter_map(|(name, prop)| {
prop.get("required")
.and_then(Value::as_bool)
.unwrap_or(false)
.then(|| json!(name))
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
Value::Array(required)
}
fn empty_object_schema() -> Value {
json!({ "type": "object", "properties": {}, "required": [] })
}
fn schema_type_matches(schema: &Value, value: &Value) -> bool {
if let Some(zeb_type) = schema.get("x-zebflow-type").and_then(Value::as_str) {
return match zeb_type {
"file" | "blob" => {
value.is_object() && value.get("uri").and_then(Value::as_str).is_some()
}
"bytes" => {
value.is_object()
&& value.get("encoding").and_then(Value::as_str) == Some("base64")
&& value.get("data").and_then(Value::as_str).is_some()
}
_ => true,
};
}
match schema.get("type").and_then(Value::as_str) {
None => true,
Some("string") => value.is_string(),
Some("number") => value.is_number(),
Some("integer") => value.as_i64().is_some() || value.as_u64().is_some(),
Some("boolean") => value.is_boolean(),
Some("object") => value.is_object(),
Some("array") => {
let Some(arr) = value.as_array() else {
return false;
};
if let Some(item_schema) = schema.get("items") {
arr.iter()
.all(|item| schema_type_matches(item_schema, item))
} else {
true
}
}
_ => true,
}
}
fn validation_error(code: &str, arg: Option<&str>, retry_hint: &str) -> Value {
let mut obj = Map::new();
obj.insert("ok".to_string(), json!(false));
obj.insert("code".to_string(), json!(code));
if let Some(arg) = arg {
obj.insert("arg".to_string(), json!(arg));
}
obj.insert("retry_hint".to_string(), json!(retry_hint));
Value::Object(obj)
}
#[async_trait]
impl NodeHandler for Node {
fn kind(&self) -> &'static str {
NODE_KIND
}
fn input_pins(&self) -> &'static [&'static str] {
&[]
}
fn output_pins(&self) -> &'static [&'static str] {
&["out"]
}
async fn execute_async(
&self,
input: NodeExecutionInput,
) -> Result<NodeExecutionOutput, PipelineError> {
Ok(NodeExecutionOutput {
output_pins: vec!["out".to_string()],
payload: super::answer_under(ANSWER_KEY, input.payload),
trace: vec![format!("node_kind={NODE_KIND}")],
})
}
}
#[cfg(test)]
mod tests {
use super::{input_schema_from_config, validate_function_input};
use serde_json::json;
#[test]
fn function_input_validation_checks_required_and_special_types() {
let config = json!({
"schema": {
"type": "object",
"required": ["source", "image"],
"properties": {
"source": { "type": "object", "x-zebflow-type": "file" },
"image": { "type": "object", "x-zebflow-type": "bytes" },
"options": {}
}
}
});
let schema = input_schema_from_config(&config);
let missing = validate_function_input(
&schema,
&json!({
"source": { "uri": "zebfs://project/uploads/a.csv" }
}),
)
.expect_err("missing bytes should fail");
assert_eq!(missing["code"], json!("missing_required_arg"));
assert_eq!(missing["arg"], json!("image"));
let invalid_file = validate_function_input(
&schema,
&json!({
"source": { "name": "a.csv" },
"image": { "encoding": "base64", "data": "AA==" }
}),
)
.expect_err("file ref without uri should fail");
assert_eq!(invalid_file["code"], json!("invalid_arg_type"));
assert_eq!(invalid_file["arg"], json!("source"));
validate_function_input(
&schema,
&json!({
"source": { "uri": "zebfs://project/uploads/a.csv", "name": "a.csv" },
"image": { "encoding": "base64", "data": "AA==", "mime": "image/png" },
"options": ["any", { "json": true }]
}),
)
.expect("valid input");
}
}