// 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)
//
// `merge` — concatenates the items from all incoming branches.
// Single-output. Mode: append (default), combine_by_key.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;

pub struct MergeNode;

static META: NodeMetadata = NodeMetadata {
    name: "merge",
    display_name: "Merge",
    version: &[1, 2],
    group: &["transform"],
    description: "Combine items from multiple branches into a single output",
    inputs: 2,
    outputs: 1,
    properties: &[],
    credentials: &[],
};

#[async_trait]
impl Node for MergeNode {
    fn metadata(&self) -> &'static NodeMetadata { &META }

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let mode = ctx.node_def.parameters.get("mode")
            .and_then(|v| v.as_str())
            .unwrap_or("append")
            .to_string();
        match mode.as_str() {
            "append" => Ok(vec![ctx.input.clone()]),
            "combine_by_key" => {
                let key = ctx.node_def.parameters.get("key")
                    .and_then(|v| v.as_str())
                    .ok_or_else(|| NodeError::MissingParameter("key".into()))?
                    .to_string();
                let mut by_key: HashMap<String, Value> = HashMap::new();
                let mut order: Vec<String> = Vec::new();
                for item in &ctx.input {
                    let k = item.json.get(&key).cloned().unwrap_or(Value::Null).to_string();
                    if !by_key.contains_key(&k) { order.push(k.clone()); }
                    let entry = by_key.entry(k).or_insert(Value::Object(Default::default()));
                    if let (Value::Object(dst), Value::Object(src)) = (entry, &item.json) {
                        for (k, v) in src { dst.insert(k.clone(), v.clone()); }
                    }
                }
                let out: Vec<NodeItem> = order.into_iter().map(|k| NodeItem {
                    json: by_key.remove(&k).unwrap_or(Value::Null),
                    ..Default::default()
                }).collect();
                Ok(vec![out])
            }
            other => Err(NodeError::Other(format!("merge: unknown mode `{}`", other))),
        }
    }
}

inventory::submit! { NodeFactory::new("merge", || Arc::new(MergeNode)) }
