zebflow 0.11.14

//! Resolve `{{ expr }}` expressions in a node config Value before the node executes.
//!
//! All expressions found in one config object are batched into a single Deno
//! pool invocation for efficiency.
//!
//! # Expression-mode sandbox security
//!
//! The sandbox runs with **`capabilities: []`** (no `n.*` bridge — no database access,
//! no HTTP, no side effects at all) and a tight **`maxOps: 500`** budget.
//! Pre-existing permanent locks (`eval`, `Function`, `fetch`, timers) remain in effect.
//! This makes expression evaluation hermetically sandboxed to pure value computation only.
//!
//! # Scope variables
//!
//! | Variable       | Contents                                                   |
//! |----------------|-------------------------------------------------------------|
//! | `$input`       | Current node's input payload                               |
//! | `$item`        | Current foreach item (`input.item` when present)           |
//! | `$index`       | Current foreach index (`input.index` when present)         |
//! | `$count`       | Current foreach count (`input.count` when present)         |
//! | `$trigger`     | Immutable trigger snapshot (`body`, `files`, `auth`, `params`, `query`, `headers`) |
//! | `$nodes`       | The answers of this node's ancestors by id; a skipped one is `null`, and a path through it is `null` (`$nodes.a.b` is read as `$nodes.a?.b`) |
//! | `$placeholder` | Opaque credential references for composite/WASM nodes. Values are placeholder strings resolved only by platform consumers (HTTP client, script sandbox), never in traces or logs. |
//!
//! # Type preservation
//!
//! Whole-field expressions (`{{ expr }}` and nothing else) return the native JS type.
//! Interpolated expressions (`"Hello {{ name }}!"`) stringify the result and concatenate.

use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::Arc;

use serde_json::{Value, json};

use super::scanner::{ExprField, Segment, scan};
use crate::language::{
    COMPILE_TARGET_BACKEND, CompileOptions, ExecutionContext, LanguageEngine, ModuleSource,
    SourceKind,
};
use crate::pipeline::PipelineError;

/// Build the flat DSL expression scope used by config expressions and logic nodes.
/// Metadata key: `{ item, index, count }` of the `logic.foreach` frame a node
/// runs in. The engine sets it on every node of the frame, in the metadata,
/// not the payload, so a node in the loop body may replace the payload
/// without cutting a later `$item` off from its element.
pub const FOREACH_METADATA_KEY: &str = "foreach";

/// `$item`, `$index` and `$count` come from the loop run the engine carries in
/// the metadata, so they hold anywhere down a `logic.foreach` branch; the
/// payload's own fields are the fallback for a node run outside the engine.
pub fn build_expression_scope_input(input: &Value, metadata: &Value) -> Value {
    let run = metadata.get(FOREACH_METADATA_KEY).filter(|v| v.is_object());
    let loop_field = |name: &str| {
        run.and_then(|r| r.get(name))
            .or_else(|| input.get(name))
            .cloned()
            .unwrap_or(Value::Null)
    };
    json!({
        "$input":       input,
        "$item":        loop_field("item"),
        "$index":       loop_field("index"),
        "$count":       loop_field("count"),
        "$trigger":     metadata.get("trigger").cloned().unwrap_or(Value::Null),
        "$nodes":       metadata.get("nodes").cloned().unwrap_or_else(|| json!({})),
        "$placeholder": metadata.get("placeholder").cloned().unwrap_or_else(|| json!({})),
    })
}

/// `$nodes.a.b.c` → `$nodes.a?.b?.c` (and `$nodes["a"].b` → `$nodes["a"]?.b`).
///
/// A node that was skipped — a branch not taken — or has not run is `null`
/// in `$nodes` (`node-conventions.md` §4), and a path through it is `null`,
/// never an error: `{{ $nodes.big.text ?? $nodes.small.text }}` joins two
/// branches. Only the member accesses after the node's name are made
/// optional; text inside string literals is left alone.
pub fn optional_nodes_paths(expr: &str) -> String {
    const MARKER: &str = "$nodes";
    let bytes = expr.as_bytes();
    let ident = |b: u8| b == b'_' || b == b'$' || b.is_ascii_alphanumeric();
    let mut out = String::with_capacity(expr.len() + 8);
    let mut i = 0;
    // The end of a `[ … ]` group starting at `start`, quotes respected.
    let bracket_end = |start: usize| -> usize {
        let mut depth = 0usize;
        let mut j = start;
        let mut quote: Option<u8> = None;
        while j < bytes.len() {
            let b = bytes[j];
            match quote {
                Some(_) if b == b'\\' => j += 1,
                Some(q) if b == q => quote = None,
                Some(_) => {}
                None if b == b'\'' || b == b'"' || b == b'`' => quote = Some(b),
                None if b == b'[' => depth += 1,
                None if b == b']' => {
                    depth -= 1;
                    if depth == 0 {
                        return j + 1;
                    }
                }
                None => {}
            }
            j += 1;
        }
        bytes.len()
    };
    while i < bytes.len() {
        let b = bytes[i];
        if b == b'\'' || b == b'"' || b == b'`' {
            let start = i;
            i += 1;
            while i < bytes.len() && bytes[i] != b {
                if bytes[i] == b'\\' {
                    i += 1;
                }
                i += 1;
            }
            i = (i + 1).min(bytes.len());
            out.push_str(&expr[start..i]);
            continue;
        }
        let at_marker = expr[i..].starts_with(MARKER)
            && (i == 0 || !(ident(bytes[i - 1]) || bytes[i - 1] == b'.'))
            && !bytes.get(i + MARKER.len()).is_some_and(|&n| ident(n));
        if !at_marker {
            let ch = expr[i..].chars().next().unwrap_or_default();
            out.push(ch);
            i += ch.len_utf8().max(1);
            continue;
        }
        out.push_str(MARKER);
        i += MARKER.len();
        // The node's own name stays as written.
        if bytes.get(i) == Some(&b'.') {
            let start = i;
            i += 1;
            while i < bytes.len() && ident(bytes[i]) {
                i += 1;
            }
            out.push_str(&expr[start..i]);
        } else if bytes.get(i) == Some(&b'[') {
            let end = bracket_end(i);
            out.push_str(&expr[i..end]);
            i = end;
        } else {
            continue;
        }
        // Every access after it is optional.
        loop {
            if expr[i..].starts_with("?.") {
                out.push_str("?.");
                i += 2;
                continue;
            }
            match bytes.get(i) {
                Some(b'.') if bytes.get(i + 1).is_some_and(|&n| ident(n)) => {
                    out.push_str("?.");
                    i += 1;
                    let start = i;
                    while i < bytes.len() && ident(bytes[i]) {
                        i += 1;
                    }
                    out.push_str(&expr[start..i]);
                }
                Some(b'[') => {
                    let end = bracket_end(i);
                    if !out.ends_with("?.") {
                        out.push_str("?.");
                    }
                    out.push_str(&expr[i..end]);
                    i = end;
                }
                Some(&n) if ident(n) && out.ends_with("?.") => {
                    let start = i;
                    while i < bytes.len() && ident(bytes[i]) {
                        i += 1;
                    }
                    out.push_str(&expr[start..i]);
                }
                _ => break,
            }
        }
    }
    out
}

/// Resolve all `{{ expr }}` expressions in `config` and return the mutated copy.
///
/// Returns the original `config` unchanged (zero mutation) if no expressions are found.
pub fn resolve_config_expressions(
    mut config: Value,
    input: &Value,
    metadata: &Value,
    language: &Arc<dyn LanguageEngine>,
) -> Result<Value, PipelineError> {
    let fields = scan(&config);
    if fields.is_empty() {
        return Ok(config);
    }

    // Deduplicate expressions and assign stable indices.
    let mut exprs: Vec<String> = Vec::new();
    for field in &fields {
        for seg in &field.segments {
            if let Segment::Expr(e) = seg {
                if !exprs.contains(e) {
                    exprs.push(e.clone());
                }
            }
        }
    }

    // Build the batch JS function body.
    // Each expression is assigned to a plain var inside a try/catch so one error doesn't
    // abort the batch.  Flat try/catch (no IIFEs) avoids any interaction with loop-guard
    // instrumentation.
    let mut body = String::from(
        // The scope object arrives as `input`; alias it first so the payload
        // can take that name. `input` is what every doc, every example and
        // the script node call the payload — binding only `$input` was why
        // `{{ input.field }}` silently evaluated to undefined everywhere it
        // was used as a whole value.
        "var __scope = input;\n\
         var input = __scope.$input;\n\
         var $input = 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\
         var $placeholder = __scope.$placeholder || {};\n",
    );
    // Each expression evaluates into its own slot so one failure does not abort
    // the batch — but the failure is *kept*, not discarded. An empty catch here
    // turned every throwing expression into `null`, which the pipeline then
    // wrote onward as if it were the author's intended value: a typo in
    // `{{ input.custmer.id }}` inserted null and reported success.
    for (i, expr) in exprs.iter().enumerate() {
        let expr = optional_nodes_paths(expr);
        body.push_str(&format!("var _e{i} = null; var _x{i} = null;\n"));
        body.push_str(&format!(
            "try {{ _e{i} = ({expr}); }} catch (_zfe) {{ _x{i} = String((_zfe && _zfe.message) || _zfe); }}\n"
        ));
    }
    body.push_str("return {\n");
    for (i, _) in exprs.iter().enumerate() {
        body.push_str(&format!("  \"e{i}\": _e{i},\n"));
        let comma = if i + 1 < exprs.len() { "," } else { "" };
        body.push_str(&format!("  \"x{i}\": _x{i}{comma}\n"));
    }
    body.push_str("};\n");

    // Compile the batch body using the language engine.
    let module = ModuleSource {
        id: format!("pipeline:expr:{}", hash_str(&body)),
        source_path: None,
        kind: SourceKind::Tsx,
        code: body,
    };
    let ir = language
        .parse(&module)
        .map_err(|e| PipelineError::new("FW_EXPR_PARSE", e.to_string()))?;
    let compiled = language
        .compile(
            &ir,
            &CompileOptions {
                target: COMPILE_TARGET_BACKEND.to_string(),
                optimize_level: 1,
                emit_trace_hints: false,
            },
        )
        .map_err(|e| PipelineError::new("FW_EXPR_COMPILE", e.to_string()))?;

    // Build the scope input: all variables are namespaced under `$` keys so
    // they don't collide with any `input` fields the user might have.
    let scope_input = build_expression_scope_input(input, metadata);

    // Build execution context with expression-mode sandbox patch.
    // capabilities: [] → n is Object.freeze({}) — no n.* bridge at all.
    // maxOps: 500 → tight budget for pure value expressions.
    let mut patched_metadata = metadata.clone();
    if let Value::Object(ref mut map) = patched_metadata {
        map.insert(
            "languageRunPatch".to_string(),
            json!({ "capabilities": [], "maxOps": 500 }),
        );
    }

    let expr_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: patched_metadata,
    };

    let run_output = language
        .run(&compiled, scope_input, &expr_ctx)
        .map_err(|e| {
            // FW_EXPR_RUN is Refused, which is right for an expression that
            // misbehaved and wrong for a worker that died — that mistake made
            // genuine infrastructure faults permanently unretryable.
            PipelineError::new(
                crate::pipeline::error_class::wrapper_for(
                    &e.code,
                    "FW_EXPR_RUN",
                    "FW_EXPR_ENGINE",
                ),
                e.to_string(),
            )
        })?;

    let result_map = match run_output.value {
        Value::Object(m) => m,
        _ => return Ok(config), // Unexpected result shape — leave config unchanged.
    };

    // An expression that threw is the author's mistake, and it is reported as
    // one. Silently substituting null makes a wrong pipeline look like a
    // working one, which is the most expensive failure mode a platform has.
    // Authors who genuinely want a missing value write `?.` or `??`, both of
    // which the sandbox supports.
    for (i, expr) in exprs.iter().enumerate() {
        let Some(message) = result_map.get(&format!("x{i}")).and_then(Value::as_str) else {
            continue;
        };
        let where_used = fields
            .iter()
            .find(|f| {
                f.segments
                    .iter()
                    .any(|seg| matches!(seg, Segment::Expr(e) if e == expr))
            })
            .map(|f| f.ptr.clone())
            .unwrap_or_default();
        return Err(PipelineError::new(
            "FW_EXPR_EVAL",
            format!("expression {{{{ {expr} }}}} at config{where_used} failed: {message}"),
        ));
    }

    // Substitute evaluation results back into the config.
    for field in &fields {
        let replacement = build_replacement(field, &exprs, &result_map);
        apply_json_ptr(&mut config, &field.ptr, replacement);
    }

    Ok(config)
}

/// Build the replacement value for one `ExprField`.
fn build_replacement(
    field: &ExprField,
    exprs: &[String],
    results: &serde_json::Map<String, Value>,
) -> Value {
    if field.is_whole {
        // Single whole-field expression → return native JS type.
        let Segment::Expr(expr) = &field.segments[0] else {
            return Value::Null;
        };
        let idx = exprs.iter().position(|e| e == expr).unwrap_or(0);
        results
            .get(&format!("e{idx}"))
            .cloned()
            .unwrap_or(Value::Null)
    } else {
        // Interpolated — concatenate all parts as strings.
        let mut s = String::new();
        for seg in &field.segments {
            match seg {
                Segment::Literal(lit) => s.push_str(lit),
                Segment::Expr(expr) => {
                    let idx = exprs.iter().position(|e| e == expr).unwrap_or(0);
                    let val = results.get(&format!("e{idx}")).unwrap_or(&Value::Null);
                    let rendered = match val {
                        Value::String(sv) => sv.clone(),
                        Value::Null => String::new(),
                        other => other.to_string(),
                    };
                    s.push_str(&rendered);
                }
            }
        }
        Value::String(s)
    }
}

/// Write `value` at the location identified by RFC 6901 JSON Pointer `ptr`.
///
/// Intermediate objects are auto-vivified (created as `null` if missing).
/// If navigation fails (non-object/array encountered mid-path, or array index
/// out of bounds) the write is silently skipped.
fn apply_json_ptr(root: &mut Value, ptr: &str, value: Value) {
    if ptr.is_empty() {
        return;
    }
    let tokens: Vec<String> = ptr
        .trim_start_matches('/')
        .split('/')
        .map(|s| s.replace("~1", "/").replace("~0", "~"))
        .collect();

    if tokens.is_empty() {
        return;
    }
    let Some((last, parents)) = tokens.split_last() else {
        return;
    };

    // Thread a mutable reference through the parent path (same pattern as serde_json's
    // `pointer_mut`).
    let parent = parents
        .iter()
        .try_fold(root as &mut Value, |node, tok| match node {
            Value::Object(map) => Some(map.entry(tok.clone()).or_insert(Value::Null)),
            Value::Array(arr) => tok.parse::<usize>().ok().and_then(|i| arr.get_mut(i)),
            _ => None,
        });

    let Some(parent) = parent else {
        return;
    };

    match parent {
        Value::Object(map) => {
            map.insert(last.clone(), value);
        }
        Value::Array(arr) => {
            if let Ok(i) = last.parse::<usize>() {
                if i < arr.len() {
                    arr[i] = value;
                }
            }
        }
        _ => {}
    }
}

fn hash_str(s: &str) -> String {
    let mut h = DefaultHasher::new();
    s.hash(&mut h);
    format!("{:x}", h.finish())
}

#[cfg(test)]
mod tests {
    use super::optional_nodes_paths;

    /// Every access after the node's name becomes optional; the name, other
    /// scopes and string literals are left as written.
    #[test]
    fn a_path_through_nodes_is_optional_after_the_node_name() {
        for (expr, want) in [
            ("$nodes.big.text ?? $nodes.small.text", "$nodes.big?.text ?? $nodes.small?.text"),
            ("$nodes.a.b.c", "$nodes.a?.b?.c"),
            ("$nodes['geo-convert'].ref", "$nodes['geo-convert']?.ref"),
            ("$nodes.q.query.rows[0].id", "$nodes.q?.query?.rows?.[0]?.id"),
            ("$nodes.a?.b.c", "$nodes.a?.b?.c"),
            ("$nodes.a", "$nodes.a"),
            ("$nodes.q.query.rows.map(r => r.id)", "$nodes.q?.query?.rows?.map(r => r.id)"),
            ("input.a.b + '$nodes.x.y'", "input.a.b + '$nodes.x.y'"),
            ("my$nodes.a.b", "my$nodes.a.b"),
        ] {
            assert_eq!(optional_nodes_paths(expr), want, "{expr}");
        }
    }
}