zebflow 0.10.11

//! `n.function.call` — invokes a function pipeline by slug.
//!
//! # Pipeline position
//!
//! Middleware node. Calls another active pipeline that starts with `n.trigger.function`.
//! On success routes through `out`; on failure (pipeline not found, execution error)
//! routes through `error`.
//!
//! # User-facing config
//! | Field | Type | Required | Description |
//! |---|---|---|---|
//! | `function` | string | yes | Slug of the function pipeline to call |
//! | `input_value` | any | no | What the function receives — a literal or `{{ expr }}`. Omit to pass the whole payload |
//! | `input` | string | no | Static JSON input (overrides input_value when set) |
//!
//! # DSL
//! ```text
//! | function.call --function my-fn --input-value "{{ input.body }}"
//! | function.call --function my-fn --input '{"user_id": "abc"}'
//! ```
//!
//! # Input/output
//! - **Input:** any payload (or sub-section via `input_path`, or static `input`)
//! - **Output `out`:** the function pipeline's last node output value
//! - **Output `error`:** `{ "error": "..." }` on failure

use std::sync::Arc;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::pipeline::model::NodeCapability;
use crate::pipeline::model::{
    DslFlag, DslFlagKind, NodeFieldDataSource, NodeFieldDef, NodeFieldType,
};
use crate::pipeline::{
    NodeDefinition, PipelineError,
    nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler},
};
use crate::platform::services::PlatformService;

pub const NODE_KIND: &str = "n.function.call";

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Config {
    /// Slug of the function pipeline to call (matches `PipelineMeta.name`).
    pub function: Option<String>,
    /// What the called function receives — a literal or `{{ expr }}`,
    /// arriving final. Null = the whole payload. Ignored when `input` is set.
    #[serde(default)]
    pub input_value: serde_json::Value,
    /// Static JSON input string passed directly to the function pipeline.
    /// When non-empty, takes priority over `input_value`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub input: Option<String>,
}

pub struct Node {
    config: Config,
    platform: Option<Arc<PlatformService>>,
}

impl Node {
    pub fn new(config: Config, platform: Option<Arc<PlatformService>>) -> Self {
        Self { config, platform }
    }
}

pub fn definition() -> NodeDefinition {
    NodeDefinition {
        kind: NODE_KIND.to_string(),
        capabilities: vec![NodeCapability::Process],
        title: "Call Function".to_string(),
        description: "Calls another active pipeline that starts with `trigger.function`. `--function` is that pipeline's slug — the file stem, \
            `send-welcome` for `jobs/send-welcome`, not the path. What it receives: `--input-value` (a literal or `{{ expr }}`, e.g. \
            `\"{{ { email: input.body.email } }}\"`), else `--input` (a static JSON string), else the whole current payload. The `out` \
            payload is the function's last node's payload — the caller's payload is replaced, so keep what you still need in \
            `$nodes.<id>`. `error` carries `{ error }` when the function is missing, inactive or failed."
            .to_string(),
        input_pins: vec!["in".to_string()],
        output_pins: vec!["out".to_string(), "error".to_string()],
        output_schema: serde_json::json!({
            "description": "On `out`: the called function's last node payload, whatever shape it declares. On `error`: { error: string }."
        }),
        config_schema: serde_json::json!({
            "type": "object",
            "required": ["function"],
            "properties": {
                "function": {
                    "type": "string",
                    "description": "Slug of the function pipeline to call."
                },
                "input_value": {
                    "type": "string",
                    "description": "JSON Pointer into the flowing payload to use as function input. Ignored when input is set."
                },
                "input": {
                    "type": "string",
                    "description": "Static JSON input passed directly to the function. Overrides input_path when set."
                }
            }
        }),
        dsl_flags: vec![
            DslFlag {
                flag: "--function".to_string(),
                config_key: "function".to_string(),
                description: "Slug of the function pipeline to call.".to_string(),
                kind: DslFlagKind::Scalar,
                required: true,
            },
            DslFlag {
                flag: "--input-value".to_string(),
                config_key: "input_value".to_string(),
                description: "What the function receives — a literal or {{ expr }}, e.g. \"{{ { email: input.body.email } }}\". Omit to pass the whole payload. Ignored when --input is set.".to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
            },
            DslFlag {
                flag: "--input".to_string(),
                config_key: "input".to_string(),
                description: "Static JSON input passed directly to the function (overrides --input-value).".to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
            },
        ],
        fields: vec![
            NodeFieldDef {
                name: "function".to_string(),
                label: "Function".to_string(),
                field_type: NodeFieldType::Datalist,
                data_source: Some(NodeFieldDataSource::FunctionPipelines),
                placeholder: Some("select or type a function slug".to_string()),
                help: Some("Slug of the function pipeline to invoke.".to_string()),
                ..Default::default()
            },
            NodeFieldDef {
                name: "input_value".to_string(),
                label: "Payload Input Path".to_string(),
                field_type: NodeFieldType::Text,
                placeholder: Some("/user  (leave empty for full payload)".to_string()),
                help: Some(
                    "JSON Pointer to extract from the flowing payload as function input. \
                     Leave empty to pass the full payload. Ignored when Static Input is set."
                        .to_string(),
                ),
                ..Default::default()
            },
        ],
        examples: vec![
            crate::pipeline::model::NodeExample::dsl("Reuse a lookup function", r#"function.call --function find-user --input-value "{{ { email: input.body.email } }}""#)
                .output(serde_json::json!({ "user": { "_key": "u_1", "email": "a@x.io" } }))
                .note("Whatever `jobs/find-user`'s last node produced."),
        ],
        ..Default::default()
    }
}

/// What the called function receives.
///
/// `input_value` arrives final — a whole `{{ }}` carries its typed value
/// (NodeIO §Value resolution). Null means "not set", so the whole payload
/// goes, which is the old empty-path default.
///
/// The pointer this replaces had a trap worth naming: a path that matched
/// nothing fell back to the *entire payload*, so a typo silently handed the
/// function everything instead of the one field it asked for. An expression
/// that matches nothing is null, and null is visible.
fn extract_payload_input(
    input_value: &serde_json::Value,
    payload: serde_json::Value,
) -> serde_json::Value {
    if input_value.is_null() {
        payload
    } else {
        input_value.clone()
    }
}

#[async_trait]
impl NodeHandler for Node {
    fn kind(&self) -> &'static str {
        NODE_KIND
    }

    fn input_pins(&self) -> &'static [&'static str] {
        &["in"]
    }

    fn output_pins(&self) -> &'static [&'static str] {
        &["out", "error"]
    }

    async fn execute_async(
        &self,
        input: NodeExecutionInput,
    ) -> Result<NodeExecutionOutput, PipelineError> {
        let platform = match &self.platform {
            Some(p) => p.clone(),
            None => {
                return Err(PipelineError::new(
                    "FW_NODE_FUNCTION_CALL_NO_PLATFORM",
                    "function.call: platform not injected into engine",
                ));
            }
        };

        let slug = match &self.config.function {
            Some(s) if !s.is_empty() => s.clone(),
            _ => {
                return Ok(NodeExecutionOutput {
                    output_pins: vec!["error".to_string()],
                    payload: serde_json::json!({"error": "no function slug configured"}),
                    trace: vec!["function.call: no slug configured".to_string()],
                });
            }
        };

        let owner = input
            .metadata
            .get("owner")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        let project = input
            .metadata
            .get("project")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();

        // Resolve function input: static input > input_path > full payload.
        let payload = input.payload;
        let call_input = if let Some(raw) = self.config.input.as_deref() {
            let trimmed = raw.trim();
            if trimmed.is_empty() {
                extract_payload_input(&self.config.input_value, payload)
            } else {
                serde_json::from_str(trimmed).unwrap_or(payload)
            }
        } else {
            extract_payload_input(&self.config.input_value, payload)
        };

        match platform
            .execute_function_pipeline(&owner, &project, &slug, call_input)
            .await
        {
            Ok(result) => Ok(NodeExecutionOutput {
                output_pins: vec!["out".to_string()],
                payload: result,
                trace: vec![format!("function.call: '{}' ok", slug)],
            }),
            Err(e) => Ok(NodeExecutionOutput {
                output_pins: vec!["error".to_string()],
                payload: serde_json::json!({"error": format!("{}: {}", e.code, e.message)}),
                trace: vec![format!(
                    "function.call: '{}' error: {} — {}",
                    slug, e.code, e.message
                )],
            }),
        }
    }
}