// 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)
//
// `clickhouse` — execute SQL via the ClickHouse HTTP interface.
// Credential: clickhouse = { base_url, username?, password?, database? }

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::sync::Arc;

pub struct ClickHouseNode;

static META: NodeMetadata = NodeMetadata {
    name: "clickhouse",
    display_name: "ClickHouse",
    version: &[1],
    group: &["database"],
    description: "Execute SQL against a ClickHouse server via the HTTP interface",
    inputs: 1,
    outputs: 1,
    properties: &[],
    credentials: &[CredentialRef { name: "clickhouse", required: true }],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let base_url = ctx.credential_field("base_url")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("clickhouse.base_url".into()))?
            .trim_end_matches('/').to_string();
        let user = ctx.credential_field("username").ok().and_then(|v| v.as_str().map(|s| s.to_string()));
        let pass = ctx.credential_field("password").ok().and_then(|v| v.as_str().map(|s| s.to_string()));
        let database = ctx.credential_field("database").ok().and_then(|v| v.as_str().map(|s| s.to_string()));
        let query = ctx.param_str("query")?;
        let with_format = if query.to_uppercase().contains("FORMAT ") { query.clone() }
                          else { format!("{} FORMAT JSON", query) };

        let mut url = format!("{}/?query={}", base_url, urlencode(&with_format));
        if let Some(d) = database { url.push_str(&format!("&database={}", urlencode(&d))); }

        let mut req = reqwest::Client::new().post(&url);
        if let (Some(u), Some(p)) = (user, pass) {
            req = req.basic_auth(u, Some(p));
        }
        let resp = req.send().await.map_err(|e| NodeError::Http(format!("clickhouse: {}", e)))?;
        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            return Err(NodeError::Http(format!("clickhouse {}: {}", status, text)));
        }
        let parsed: Value = serde_json::from_str(&text).unwrap_or(Value::String(text));
        Ok(vec![vec![NodeItem { json: parsed, ..Default::default() }]])
    }
}

fn urlencode(s: &str) -> String {
    s.bytes().flat_map(|b| {
        if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'~') { vec![b as char] }
        else { format!("%{:02X}", b).chars().collect() }
    }).collect()
}

inventory::submit! { NodeFactory::new("clickhouse", || Arc::new(ClickHouseNode)) }
