// 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)
//
// `summarize` — fold the input items into a single statistics item.
// Operations: count, sum, avg, min, max, distinct over a $json.<field>.

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

pub struct SummarizeNode;

static META: NodeMetadata = NodeMetadata {
    name: "summarize",
    display_name: "Summarize",
    version: &[1, 2],
    group: &["transform"],
    description: "Compute count/sum/avg/min/max/distinct over a field across all input items",
    inputs: 1,
    outputs: 1,
    properties: &[],
    credentials: &[],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let operation = ctx.node_def.parameters.get("operation")
            .and_then(|v| v.as_str()).unwrap_or("count").to_string();
        let field = ctx.node_def.parameters.get("field")
            .and_then(|v| v.as_str()).map(|s| s.to_string());
        let output_field = ctx.node_def.parameters.get("output_field")
            .and_then(|v| v.as_str()).unwrap_or("result").to_string();

        let pull = |item: &NodeItem| -> Option<Value> {
            field.as_ref().and_then(|f| item.json.get(f).cloned())
        };
        let pull_num = |item: &NodeItem| -> Option<f64> { pull(item).and_then(|v| v.as_f64()) };

        let result: Value = match operation.as_str() {
            "count" => json!(ctx.input.len()),
            "sum" => json!(ctx.input.iter().filter_map(pull_num).sum::<f64>()),
            "avg" => {
                let nums: Vec<f64> = ctx.input.iter().filter_map(pull_num).collect();
                if nums.is_empty() { Value::Null }
                else { json!(nums.iter().sum::<f64>() / nums.len() as f64) }
            }
            "min" => {
                ctx.input.iter().filter_map(pull_num)
                    .fold(None::<f64>, |acc, x| Some(acc.map_or(x, |a| a.min(x))))
                    .map(|v| json!(v)).unwrap_or(Value::Null)
            }
            "max" => {
                ctx.input.iter().filter_map(pull_num)
                    .fold(None::<f64>, |acc, x| Some(acc.map_or(x, |a| a.max(x))))
                    .map(|v| json!(v)).unwrap_or(Value::Null)
            }
            "distinct" => {
                let set: HashSet<String> = ctx.input.iter()
                    .filter_map(|i| pull(i).map(|v| v.to_string()))
                    .collect();
                json!(set.len())
            }
            other => return Err(NodeError::Other(format!("summarize: unknown operation `{}`", other))),
        };

        let mut payload = serde_json::Map::new();
        payload.insert("operation".into(), Value::String(operation));
        if let Some(f) = field { payload.insert("field".into(), Value::String(f)); }
        payload.insert(output_field, result);
        payload.insert("input_count".into(), json!(ctx.input.len()));

        Ok(vec![vec![NodeItem { json: Value::Object(payload), ..Default::default() }]])
    }
}

inventory::submit! { NodeFactory::new("summarize", || Arc::new(SummarizeNode)) }
