// 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)
//
// `kafka` — produce a record to a Kafka topic via rskafka (pure-rust).
// Param names lifted from upstream Kafka.node.ts: topic, sendInputData,
// message, jsonParameters, useSchemaRegistry, schemaRegistryUrl, useKey,
// key, eventName, headersUi.
// Credential `kafka`: { clientId, brokers, ssl, authentication, username,
// password, saslMechanism }.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use rskafka::client::ClientBuilder;
use rskafka::client::partition::UnknownTopicHandling;
use rskafka::record::Record;
use serde_json::Value;
use std::sync::Arc;
use std::collections::BTreeMap;
use chrono::Utc;

pub struct KafkaNode;

static META: NodeMetadata = NodeMetadata {
    name: "kafka",
    display_name: "Kafka",
    version: &[1],
    group: &["transform"],
    description: "Send messages to a Kafka topic",
    inputs: 1,
    outputs: 1,
    properties: &[
        PropertyDef { name: "topic", display_name: "Topic", kind: PropertyKind::String, required: true, default: None },
        PropertyDef { name: "sendInputData", display_name: "Send Input Data", kind: PropertyKind::Boolean, required: false, default: Some("true") },
        PropertyDef { name: "message", display_name: "Message", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "jsonParameters", display_name: "JSON Parameters", kind: PropertyKind::Boolean, required: false, default: Some("false") },
        PropertyDef { name: "useKey", display_name: "Use Key", kind: PropertyKind::Boolean, required: false, default: Some("false") },
        PropertyDef { name: "key", display_name: "Key", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "eventName", display_name: "Event Name", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "headersUi", display_name: "Headers", kind: PropertyKind::Json, required: false, default: None },
    ],
    credentials: &[CredentialRef { name: "kafka", required: true }],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let brokers_raw = ctx.credential_field("brokers")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("kafka.brokers".into()))?
            .to_string();
        let brokers: Vec<String> = brokers_raw.split(',')
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .collect();
        if brokers.is_empty() {
            return Err(NodeError::MissingCredential("kafka.brokers".into()));
        }

        let topic = ctx.param_str("topic")?;
        let send_input = ctx.node_def.parameters.get("sendInputData")
            .and_then(|v| v.as_bool()).unwrap_or(true);
        let body_str = if send_input {
            let item = ctx.input.first().cloned().unwrap_or_default();
            serde_json::to_string(&item.json).unwrap_or_default()
        } else {
            ctx.param_opt_str("message").unwrap_or_default()
        };
        let key_opt = if ctx.node_def.parameters.get("useKey").and_then(|v| v.as_bool()).unwrap_or(false) {
            ctx.param_opt_str("key").map(|s| s.into_bytes())
        } else {
            None
        };

        let mut headers = BTreeMap::<String, Vec<u8>>::new();
        if let Some(name) = ctx.param_opt_str("eventName") {
            headers.insert("eventName".to_string(), name.into_bytes());
        }
        if let Some(Value::Object(map)) = ctx.node_def.parameters.get("headersUi") {
            if let Some(Value::Array(arr)) = map.get("headerValues") {
                for entry in arr {
                    let k = entry.get("key").and_then(|v| v.as_str()).unwrap_or("");
                    let v = entry.get("value").and_then(|v| v.as_str()).unwrap_or("");
                    if !k.is_empty() {
                        headers.insert(k.to_string(), v.as_bytes().to_vec());
                    }
                }
            }
        }

        let client = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            ClientBuilder::new(brokers).build(),
        ).await
            .map_err(|_| NodeError::Other("kafka connect: timed out".into()))?
            .map_err(|e| NodeError::Other(format!("kafka connect: {}", e)))?;
        let partition_client = client.partition_client(
            topic.clone(),
            0,
            UnknownTopicHandling::Retry,
        ).await.map_err(|e| NodeError::Other(format!("kafka topic: {}", e)))?;

        let record = Record {
            key: key_opt,
            value: Some(body_str.into_bytes()),
            headers,
            timestamp: Utc::now(),
        };
        let offsets = partition_client.produce(vec![record], rskafka::client::partition::Compression::NoCompression).await
            .map_err(|e| NodeError::Other(format!("kafka produce: {}", e)))?;

        let json = serde_json::json!({
            "topic": topic,
            "partition": 0,
            "offsets": offsets,
        });
        Ok(vec![vec![NodeItem { json, ..Default::default() }]])
    }
}

inventory::submit! { NodeFactory::new("kafka", || Arc::new(KafkaNode)) }
