zebflow 0.7.9

//! Small helpers shared by built-in framework nodes.

use std::hash::{Hash, Hasher};

use serde_json::Value;

use crate::language::{
    COMPILE_TARGET_BACKEND, CompileOptions, ExecutionContext, LanguageEngine, ModuleSource,
    SourceKind,
};
use crate::pipeline::PipelineError;
use crate::pipeline::expr::build_expression_scope_input;

pub fn metadata_scope(metadata: &Value) -> Result<(&str, &str, &str, &str), PipelineError> {
    let owner = metadata
        .get("owner")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            PipelineError::new(
                "FW_NODE_SCOPE",
                "missing metadata.owner for project-scoped node",
            )
        })?;
    let project = metadata
        .get("project")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            PipelineError::new(
                "FW_NODE_SCOPE",
                "missing metadata.project for project-scoped node",
            )
        })?;
    let pipeline = metadata
        .get("pipeline")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let request_id = metadata
        .get("request_id")
        .and_then(Value::as_str)
        .unwrap_or_default();
    Ok((owner, project, pipeline, request_id))
}

pub fn resolve_path<'a>(root: &'a Value, path: &str) -> Option<&'a Value> {
    let path = path.trim();
    if path.is_empty() {
        return Some(root);
    }

    let mut current = root;
    for segment in path.split('.') {
        let segment = segment.trim();
        if segment.is_empty() {
            continue;
        }
        current = match current {
            Value::Object(map) => map.get(segment)?,
            Value::Array(arr) => {
                let idx: usize = segment.parse().ok()?;
                arr.get(idx)?
            }
            _ => return None,
        };
    }
    Some(current)
}

pub fn resolve_path_cloned(root: &Value, path: Option<&str>) -> Option<Value> {
    path.and_then(|p| resolve_path(root, p).cloned())
}

pub fn resolve_array_values(root: &Value, path: Option<&str>) -> Vec<Value> {
    let Some(value) = resolve_path_cloned(root, path) else {
        return Vec::new();
    };
    match value {
        Value::Array(items) => items,
        other => vec![other],
    }
}

pub fn eval_deno_expr(
    language: &dyn LanguageEngine,
    expr: &str,
    input: &Value,
    metadata: &Value,
) -> Result<Value, PipelineError> {
    let expr = expr.trim();
    if expr.is_empty() {
        return Err(PipelineError::new(
            "FW_NODE_BINDING_EXPR",
            "binding expression must not be empty",
        ));
    }

    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    expr.hash(&mut hasher);
    let source = format!(
        "var __scope = input;\n\
         var $input = __scope.$input;\n\
         var $item = __scope.$item;\n\
         var $index = __scope.$index;\n\
         var $count = __scope.$count;\n\
         var $trigger = __scope.$trigger || null;\n\
         var $nodes = __scope.$nodes || {{}};\n\
         return ((function(input, $input, $item, $index, $count, $trigger, $nodes) {{\n\
           return ({expr});\n\
         }})($input, $input, $item, $index, $count, $trigger, $nodes));"
    );
    let module = ModuleSource {
        id: format!("pipeline:binding:{:x}", hasher.finish()),
        source_path: None,
        kind: SourceKind::Tsx,
        code: source,
    };
    let ir = language
        .parse(&module)
        .map_err(|err| PipelineError::new("FW_NODE_BINDING_PARSE", err.to_string()))?;
    let compiled = language
        .compile(
            &ir,
            &CompileOptions {
                target: COMPILE_TARGET_BACKEND.to_string(),
                optimize_level: 1,
                emit_trace_hints: false,
            },
        )
        .map_err(|err| PipelineError::new("FW_NODE_BINDING_COMPILE", err.to_string()))?;
    let ctx = ExecutionContext {
        project: metadata
            .get("project")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string(),
        pipeline: metadata
            .get("pipeline")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string(),
        request_id: metadata
            .get("request_id")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string(),
        trigger: metadata.get("trigger").cloned().unwrap_or(Value::Null),
        metadata: metadata.clone(),
    };
    language
        .run(
            &compiled,
            build_expression_scope_input(input, metadata),
            &ctx,
        )
        .map(|out| out.value)
        .map_err(|err| PipelineError::new("FW_NODE_BINDING_RUN", err.to_string()))
}

pub fn resolve_query_binding(
    language: &dyn LanguageEngine,
    input: &Value,
    metadata: &Value,
    query_expr: Option<&str>,
    query_static: &str,
    error_code: &'static str,
) -> Result<String, PipelineError> {
    let expr = query_expr.map(str::trim).filter(|s| !s.is_empty());
    let static_val = query_static.trim();
    if expr.is_some() && !static_val.is_empty() {
        return Err(PipelineError::new(
            error_code,
            "set either query or query_expr, not both",
        ));
    }
    if let Some(expr) = expr {
        let value = eval_deno_expr(language, expr, input, metadata)?;
        return value
            .as_str()
            .map(ToString::to_string)
            .ok_or_else(|| PipelineError::new(error_code, "query_expr must return a string"));
    }
    if static_val.is_empty() {
        return Err(PipelineError::new(error_code, "query must not be empty"));
    }
    Ok(static_val.to_string())
}