bnto_core/executor/

mod.rs

// Pipeline Executor — top-level orchestrator for pipeline execution.
//
// Walks nodes in order, chaining outputs to inputs. Split across seven files:
//   - `mod.rs` — public API, node chaining and dispatch, shared types
//   - `auto_iteration.rs` — implicit per-file loops for `iteration: "auto"`
//   - `primitive.rs` — leaf node execution (image compress, file rename, etc.)
//   - `container.rs` — container node execution (loop, group, parallel)
//   - `loop_config.rs` — loop container params (error handling, persistence)
//   - `resolve.rs` — `{{fields.*}}` template resolution
//   - `template.rs` — multi-namespace template resolution (env, ctx, node, item)
//
// Pure Rust — no WASM deps. Works with `cargo test` natively.
// The WASM bridge (`bnto-wasm/src/execute.rs`) handles JS type conversions.

mod auto_iteration;
mod container;
pub mod loop_config;
mod primitive;
pub mod resolve;
pub mod template;

use crate::context::ProcessContext;
use crate::errors::BntoError;
use crate::events::{NodeInfo, PipelineEvent, PipelineReporter};
use crate::pipeline::{
    IterationMode, PipelineDefinition, PipelineFile, PipelineFileResult, PipelineNode,
    PipelineResult, is_container_node, is_io_node,
};
use crate::registry::NodeRegistry;

#[cfg(test)]
use crate::processor::NodeInput;
#[cfg(test)]
use crate::progress::ProgressReporter;

// --- Shared Types ---

type PipelineNodeRef<'a> = &'a PipelineNode;

/// Bundles shared immutable state for executor functions, keeping
/// signatures clean and under clippy's 7-argument limit.
struct PipelineContext<'a, F: Fn() -> u64 + Copy> {
    registry: &'a NodeRegistry,
    reporter: &'a PipelineReporter,
    /// System access boundary for processors (commands, temp files, env vars).
    process_ctx: &'a dyn ProcessContext,
    /// Original input file count — used in progress events so the UI
    /// reports global position even inside loop container iterations.
    pipeline_total_files: usize,
    /// Returns current time in ms (injected for testability in WASM).
    now_ms: F,
    /// Current loop iteration's per-file metadata (CSV row columns, etc.).
    /// Set by loop containers from PipelineFile.metadata each iteration.
    loop_item: Option<serde_json::Map<String, serde_json::Value>>,
    /// If executing inside a container, the container's node ID.
    /// Used to set `parent_node_id` on child `NodeStarted` events.
    parent_node_id: Option<String>,
    /// Per-node output metadata for {{node.<id>.*}} templates.
    /// Seeded with input file metadata before execution begins.
    node_outputs: std::collections::BTreeMap<String, serde_json::Value>,
}

/// Where a node chain sits among its sibling nodes, for `NodeStarted` numbering.
///
/// Auto-iteration runs a pipeline as several chains; each one is a slice of
/// the pipeline and must report positions against the whole of it.
#[derive(Clone, Copy)]
struct ChainPosition {
    /// Position of the chain's first node among its siblings.
    first_index: usize,
    /// Number of sibling processing nodes (pipeline-wide, or within a container).
    total_nodes: usize,
}

impl ChainPosition {
    /// A chain that covers every one of its `total_nodes` siblings.
    fn whole(total_nodes: usize) -> Self {
        Self::starting_at(0, total_nodes)
    }

    /// A chain whose first node is at `first_index` among `total_nodes` siblings.
    fn starting_at(first_index: usize, total_nodes: usize) -> Self {
        Self {
            first_index,
            total_nodes,
        }
    }
}

/// Result of executing a single node or container sub-pipeline.
struct NodeExecutionResult {
    output_files: Vec<PipelineFile>,
    files_processed: usize,
    /// Non-fatal warnings (e.g. skipped loop iterations with continue-on-error).
    warnings: Vec<String>,
}

// --- Public API ---

/// Filter out I/O marker nodes (input/output) that don't perform processing.
fn filter_processing_nodes(definition: &PipelineDefinition) -> Vec<&PipelineNode> {
    definition
        .nodes
        .iter()
        .filter(|n| !is_io_node(&n.node_type))
        .collect()
}

/// Convert final pipeline files to result format, preserving processor metadata.
fn build_pipeline_result(
    files: Vec<PipelineFile>,
    duration_ms: u64,
    warnings: Vec<String>,
) -> PipelineResult {
    let result_files = files
        .into_iter()
        .map(|f| PipelineFileResult {
            name: f.name,
            data: f.data,
            mime_type: f.mime_type,
            metadata: f.metadata,
        })
        .collect();
    PipelineResult {
        files: result_files,
        duration_ms,
        warnings,
    }
}

/// Execute a complete pipeline: walk nodes, iterate files, chain outputs.
///
/// Main entry point for the engine. `now_ms` is injected for testability
/// (no `std::time::Instant` in WASM — uses `js_sys::Date::now()` instead).
pub fn execute_pipeline(
    definition: &PipelineDefinition,
    files: Vec<PipelineFile>,
    registry: &NodeRegistry,
    reporter: &PipelineReporter,
    process_ctx: &dyn ProcessContext,
    now_ms: impl Fn() -> u64 + Copy,
) -> Result<PipelineResult, BntoError> {
    let start_ms = now_ms();
    let processing_nodes = filter_processing_nodes(definition);
    let ctx = build_root_context(definition, &files, registry, reporter, process_ctx, now_ms);
    emit_pipeline_started(&ctx, &processing_nodes, files.len());
    let result = run_processing_nodes(&ctx, definition, &processing_nodes, files)?;
    Ok(finish_pipeline(&ctx, start_ms, result))
}

/// Build the top-level context, seeding node outputs from the input files.
fn build_root_context<'a, F: Fn() -> u64 + Copy>(
    definition: &PipelineDefinition,
    files: &[PipelineFile],
    registry: &'a NodeRegistry,
    reporter: &'a PipelineReporter,
    process_ctx: &'a dyn ProcessContext,
    now_ms: F,
) -> PipelineContext<'a, F> {
    PipelineContext {
        registry,
        reporter,
        process_ctx,
        pipeline_total_files: files.len(),
        now_ms,
        loop_item: None,
        parent_node_id: None,
        node_outputs: template::build_node_outputs_for_input(&definition.nodes, files),
    }
}

fn emit_pipeline_started<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    processing_nodes: &[&PipelineNode],
    total_files: usize,
) {
    let nodes = processing_nodes
        .iter()
        .map(|n| NodeInfo {
            id: n.id.clone(),
            node_type: n.node_type.clone(),
        })
        .collect();
    ctx.reporter.emit(PipelineEvent::PipelineStarted {
        total_nodes: processing_nodes.len(),
        total_files,
        nodes,
    });
}

/// Run the top-level nodes with the definition's iteration strategy.
fn run_processing_nodes<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    definition: &PipelineDefinition,
    processing_nodes: &[&PipelineNode],
    files: Vec<PipelineFile>,
) -> Result<NodeExecutionResult, BntoError> {
    match definition.resolved_iteration() {
        IterationMode::Explicit => {
            let position = ChainPosition::whole(processing_nodes.len());
            run_node_chain(ctx, processing_nodes, files, 0, position)
        }
        IterationMode::Auto => auto_iteration::run_auto_iteration(ctx, processing_nodes, files),
    }
}

/// Emit PipelineCompleted and convert the final files into the result.
fn finish_pipeline<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    start_ms: u64,
    result: NodeExecutionResult,
) -> PipelineResult {
    let duration_ms = (ctx.now_ms)() - start_ms;
    ctx.reporter.emit(PipelineEvent::PipelineCompleted {
        duration_ms,
        total_files_processed: result.files_processed,
    });
    build_pipeline_result(result.output_files, duration_ms, result.warnings)
}

/// Chain nodes sequentially, passing each node's output as the next node's input.
///
/// `position` places the chain among its siblings, so a chain that is a
/// slice of the pipeline still reports pipeline-wide progress.
fn run_node_chain<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    nodes: &[&PipelineNode],
    files: Vec<PipelineFile>,
    file_offset: usize,
    position: ChainPosition,
) -> Result<NodeExecutionResult, BntoError> {
    let mut current_files = files;
    let mut total_files_processed: usize = 0;
    let mut all_warnings: Vec<String> = Vec::new();

    for (i, node) in nodes.iter().enumerate() {
        let node_index = position.first_index + i;
        let result = execute_node(
            ctx,
            node,
            current_files,
            node_index,
            position.total_nodes,
            file_offset,
        )?;
        total_files_processed += result.files_processed;
        all_warnings.extend(result.warnings);
        current_files = result.output_files;
    }

    Ok(NodeExecutionResult {
        output_files: current_files,
        files_processed: total_files_processed,
        warnings: all_warnings,
    })
}

// --- Internal: Node Dispatch ---

/// Emit failure events (NodeFailed + PipelineFailed) and return the error.
fn emit_node_failure<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    node_id: &str,
    error: BntoError,
) -> BntoError {
    let error_msg = error.to_string();
    ctx.reporter.emit(PipelineEvent::NodeFailed {
        node_id: node_id.to_string(),
        error: error_msg.clone(),
        parent_node_id: ctx.parent_node_id.clone(),
    });
    ctx.reporter.emit(PipelineEvent::PipelineFailed {
        node_id: node_id.to_string(),
        error: error_msg,
    });
    error
}

/// Execute a single node — dispatches to primitive or container handler.
///
/// Emits NodeStarted, delegates to the appropriate handler, then emits
/// NodeCompleted or NodeFailed. Container children call back into this
/// for recursive execution.
fn execute_node<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    node: &PipelineNode,
    files: Vec<PipelineFile>,
    node_index: usize,
    total_nodes: usize,
    file_offset: usize,
) -> Result<NodeExecutionResult, BntoError> {
    let node_start = (ctx.now_ms)();
    emit_node_started(ctx, node, node_index, total_nodes);

    let result = dispatch_node(ctx, node, files, file_offset);

    match result {
        Ok(exec_result) => {
            emit_node_completed(ctx, &node.id, node_start, exec_result.files_processed);
            Ok(exec_result)
        }
        Err(error) => Err(emit_node_failure(ctx, &node.id, error)),
    }
}

fn emit_node_started<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    node: &PipelineNode,
    node_index: usize,
    total_nodes: usize,
) {
    ctx.reporter.emit(PipelineEvent::NodeStarted {
        node_id: node.id.clone(),
        node_index,
        total_nodes,
        node_type: node.node_type.clone(),
        parent_node_id: ctx.parent_node_id.clone(),
    });
}

fn emit_node_completed<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    node_id: &str,
    node_start: u64,
    files_processed: usize,
) {
    let duration_ms = (ctx.now_ms)() - node_start;
    ctx.reporter.emit(PipelineEvent::NodeCompleted {
        node_id: node_id.to_string(),
        duration_ms,
        files_processed,
        parent_node_id: ctx.parent_node_id.clone(),
    });
}

fn dispatch_node<F: Fn() -> u64 + Copy>(
    ctx: &PipelineContext<F>,
    node: &PipelineNode,
    files: Vec<PipelineFile>,
    file_offset: usize,
) -> Result<NodeExecutionResult, BntoError> {
    if is_container_node(&node.node_type) {
        container::execute_container_node(ctx, node, files, file_offset)
    } else {
        primitive::execute_primitive_node(ctx, node, files, file_offset)
    }
}

#[cfg(test)]
mod tests;