// 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)
//
// `split_in_batches` — slice the input items into chunks of size N and
// emit one chunk's worth on each call. v0.1: emits the first batch only;
// the executor loop-back required for full pagination is v0.2.

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

pub struct SplitInBatchesNode;

static META: NodeMetadata = NodeMetadata {
    name: "split_in_batches",
    display_name: "Split In Batches",
    version: &[1, 2, 3],
    group: &["transform"],
    description: "Take the first N items from the input. Pagination loop-back is v0.2.",
    inputs: 1,
    outputs: 2,
    properties: &[],
    credentials: &[],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let batch_size = ctx.node_def.parameters.get("batch_size")
            .and_then(|v| v.as_u64())
            .unwrap_or(10) as usize;
        let total = ctx.input.len();
        let take = batch_size.min(total);
        let batch: Vec<NodeItem> = ctx.input.iter().take(take).cloned().collect();
        let remaining: Vec<NodeItem> = ctx.input.iter().skip(take).cloned().collect();

        // Output 0: this batch (potentially augmented with batch metadata)
        let mut batch_out = batch;
        for item in &mut batch_out {
            if let Some(obj) = item.json.as_object_mut() {
                obj.insert("__batch_index".into(), json!(0));
                obj.insert("__batch_size".into(), json!(take));
                obj.insert("__remaining".into(), json!(remaining.len()));
            }
        }
        // Output 1: remaining items (a "done" branch when empty)
        Ok(vec![batch_out, remaining])
    }
}

inventory::submit! { NodeFactory::new("split_in_batches", || Arc::new(SplitInBatchesNode)) }
