// r8r: rust port of n8n.io
// Initial author: Michael Cochran <mcochran@cochranblock.org>
// Built for: Tanner Woodrum, intended cofounder
// License: Unlicense (public domain dedication, see UNLICENSE)

use super::node::ExecuteCtx;
use super::registry::Registry;
use super::state::{ExecuteFrame, RunState, TaskRun, TaskStatus, WaitingState};
use chrono::Utc;
use crate::core::credentials::CredentialStore;
use crate::core::error::FlowError;
use crate::core::types::*;
use crate::core::workflow::Workflow;
use std::time::Instant;
use tracing::{debug, error, info, warn};

pub async fn execute(
    workflow: &Workflow,
    credentials: &CredentialStore,
    initial: Vec<NodeItem>,
) -> Result<RunState, FlowError> {
    let registry = Registry::global();
    let execution_id = uuid::Uuid::new_v4().to_string();
    let mut state = RunState::new();

    for entry in workflow.entry_nodes() {
        state.stack.push_back(ExecuteFrame {
            node: entry.name.clone(),
            input: initial.clone(),
            run_index: 0,
        });
    }

    while let Some(frame) = state.stack.pop_front() {
        let node_def = workflow.node(&frame.node)
            .ok_or_else(|| FlowError::Validation(format!("missing node `{}`", frame.node)))?;
        let node_impl = registry.lookup(&node_def.type_name)
            .ok_or_else(|| FlowError::UnknownNodeType(node_def.type_name.clone()))?;

        info!(node = %frame.node, kind = %node_def.type_name, "executing");
        let started = Instant::now();

        let mut ctx = ExecuteCtx {
            node_def,
            input: frame.input.clone(),
            run_data: &state.run_items,
            credentials,
            workflow_name: &workflow.workflow.name,
            execution_id: &execution_id,
        };

        let result = node_impl.execute(&mut ctx).await;
        let elapsed = started.elapsed().as_millis() as u64;

        match result {
            Ok(outputs) => {
                let flat: Vec<NodeItem> = outputs.iter().flatten().cloned().collect();
                state.record(&frame.node, TaskRun {
                    start_time: Utc::now(),
                    execution_time_ms: elapsed,
                    status: TaskStatus::Success,
                    items: flat,
                    error: None,
                });
                propagate(&frame, outputs, workflow, &mut state);
            }
            Err(e) => {
                let msg = e.to_string();
                error!(node = %frame.node, error = %msg, "node failed");
                state.record(&frame.node, TaskRun {
                    start_time: Utc::now(),
                    execution_time_ms: elapsed,
                    status: TaskStatus::Error,
                    items: vec![],
                    error: Some(msg.clone()),
                });
                if node_def.continue_on_fail {
                    warn!(node = %frame.node, "continue_on_fail set, passing input through");
                    let pass = vec![frame.input.clone()];
                    propagate(&frame, pass, workflow, &mut state);
                    continue;
                }
                if node_def.on_error == OnError::ContinueErrorOutput {
                    let mut err_item = NodeItem::default();
                    err_item.error = Some(msg);
                    let routed = vec![vec![], vec![err_item]];
                    propagate(&frame, routed, workflow, &mut state);
                    continue;
                }
                state.error = Some(format!("{}: {}", frame.node, e));
                return Ok(state);
            }
        }
    }

    debug!(nodes_run = state.run_data.len(), "workflow complete");
    Ok(state)
}

fn propagate(
    frame: &ExecuteFrame,
    outputs: NodeOutput,
    workflow: &Workflow,
    state: &mut RunState,
) {
    for (output_index, items) in outputs.into_iter().enumerate() {
        let output_index = output_index as u8;
        if items.is_empty() { continue; }
        for edge in workflow.edges_from(&frame.node, output_index) {
            let target = match workflow.node(&edge.to) {
                Some(n) => n,
                None => continue,
            };
            let input_count = workflow.input_count(&target.name);
            if input_count > 1 {
                let key = (target.name.clone(), frame.run_index);
                let entry = state.waiting.entry(key.clone()).or_insert_with(|| WaitingState {
                    expected_inputs: input_count as u8,
                    ..Default::default()
                });
                entry.by_input.entry(edge.to_index).or_default().extend(items.clone());
                if entry.is_complete() {
                    let merged = state.waiting.get_mut(&key).unwrap().drain();
                    state.waiting.remove(&key);
                    state.stack.push_back(ExecuteFrame {
                        node: target.name.clone(),
                        input: merged,
                        run_index: frame.run_index,
                    });
                }
            } else {
                state.stack.push_back(ExecuteFrame {
                    node: target.name.clone(),
                    input: items.clone(),
                    run_index: frame.run_index,
                });
            }
        }
    }
}
