// 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)
//
// `postgres` — execute SQL, emit each row as an item.
// Default driver: tokio-postgres against a real Postgres.
// Test/dev escape hatch: connection strings starting with `sqlite:` route
// to an embedded SQLite backend (same SQL, same row→JSON mapping). Lets
// the test suite exercise the full SQL workflow path without spinning up
// a real Postgres — e.g. connection_string = "sqlite::memory:".
// Credential: postgres = { connection_string = "..." }
// or { host, port, user, password, database, ssl_mode }.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use rusqlite::types::ValueRef;
use serde_json::{json, Value};
use std::sync::Arc;
use tokio_postgres::types::Type;
use tokio_postgres::{NoTls, Row};

pub struct PostgresNode;

static META: NodeMetadata = NodeMetadata {
    name: "postgres",
    display_name: "Postgres",
    version: &[1, 2],
    group: &["database"],
    description: "Execute SQL against a Postgres database",
    inputs: 1,
    outputs: 1,
    properties: &[],
    credentials: &[CredentialRef { name: "postgres", required: true }],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let conn_str = build_conn_str(ctx)?;
        let query = ctx.param_str("query")?;
        let operation = ctx.node_def.parameters.get("operation")
            .and_then(|v| v.as_str()).unwrap_or("execute_query").to_string();

        // SQLite escape hatch: any connection string starting with `sqlite:`
        // routes to the embedded backend instead of opening a Postgres
        // socket. Same operations, same row → JSON shape.
        if let Some(sqlite_path) = conn_str.strip_prefix("sqlite:") {
            return execute_sqlite(sqlite_path, &query, &operation).await;
        }

        let (client, connection) = tokio_postgres::connect(&conn_str, NoTls).await
            .map_err(|e| NodeError::Other(format!("postgres connect: {}", e)))?;
        let conn_handle = tokio::spawn(async move {
            if let Err(e) = connection.await {
                tracing::warn!("postgres connection error: {}", e);
            }
        });

        let result = match operation.as_str() {
            "execute_query" | "select" => {
                let rows = client.query(&query, &[]).await
                    .map_err(|e| NodeError::Other(format!("postgres query: {}", e)))?;
                rows.into_iter().map(row_to_item).collect::<Vec<_>>()
            }
            "execute" | "insert" | "update" | "delete" => {
                let n = client.execute(&query, &[]).await
                    .map_err(|e| NodeError::Other(format!("postgres execute: {}", e)))?;
                vec![NodeItem { json: json!({ "rows_affected": n }), ..Default::default() }]
            }
            other => {
                conn_handle.abort();
                return Err(NodeError::Other(format!("postgres: unknown operation `{}`", other)));
            }
        };

        drop(client);
        conn_handle.abort();
        Ok(vec![result])
    }
}

fn build_conn_str(ctx: &ExecuteCtx<'_>) -> Result<String, NodeError> {
    if let Ok(s) = ctx.credential_field("connection_string") {
        if let Some(s) = s.as_str() { return Ok(s.to_string()); }
    }
    let host = ctx.credential_field("host").ok().and_then(|v| v.as_str()).unwrap_or("localhost").to_string();
    let port = ctx.credential_field("port").ok().and_then(|v| v.as_u64()).unwrap_or(5432);
    let user = ctx.credential_field("user").ok().and_then(|v| v.as_str()).unwrap_or("postgres").to_string();
    let password = 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()).unwrap_or("postgres").to_string();
    let auth = match password {
        Some(p) => format!("{}:{}", user, p),
        None => user,
    };
    Ok(format!("postgres://{}@{}:{}/{}", auth, host, port, database))
}

fn row_to_item(row: Row) -> NodeItem {
    let mut obj = serde_json::Map::new();
    for (i, col) in row.columns().iter().enumerate() {
        let val = pg_value(&row, i, col.type_());
        obj.insert(col.name().to_string(), val);
    }
    NodeItem { json: Value::Object(obj), ..Default::default() }
}

fn pg_value(row: &Row, idx: usize, ty: &Type) -> Value {
    macro_rules! try_get { ($t:ty) => { row.try_get::<usize, Option<$t>>(idx).ok().flatten() } }
    match *ty {
        Type::BOOL => try_get!(bool).map(Value::Bool).unwrap_or(Value::Null),
        Type::INT2 => try_get!(i16).map(|v| Value::Number(v.into())).unwrap_or(Value::Null),
        Type::INT4 => try_get!(i32).map(|v| Value::Number(v.into())).unwrap_or(Value::Null),
        Type::INT8 => try_get!(i64).map(|v| Value::Number(v.into())).unwrap_or(Value::Null),
        Type::FLOAT4 => try_get!(f32).and_then(|v| serde_json::Number::from_f64(v as f64)).map(Value::Number).unwrap_or(Value::Null),
        Type::FLOAT8 => try_get!(f64).and_then(serde_json::Number::from_f64).map(Value::Number).unwrap_or(Value::Null),
        Type::TEXT | Type::VARCHAR | Type::NAME | Type::BPCHAR => try_get!(String).map(Value::String).unwrap_or(Value::Null),
        Type::JSON | Type::JSONB => row.try_get::<usize, Option<Value>>(idx).ok().flatten().unwrap_or(Value::Null),
        _ => try_get!(String).map(Value::String).unwrap_or(Value::Null),
    }
}

async fn execute_sqlite(path: &str, query: &str, operation: &str) -> Result<NodeOutput, NodeError> {
    use tokio_rusqlite::Connection;
    let conn = Connection::open(path).await
        .map_err(|e| NodeError::Other(format!("postgres[sqlite]: open `{}`: {}", path, e)))?;
    let q = query.to_string();
    match operation {
        "execute_query" | "select" => {
            let rows = conn.call(move |c| {
                let mut stmt = c.prepare(&q)?;
                let cols: Vec<String> = stmt.column_names().iter().map(|s| (*s).to_string()).collect();
                let mapped = stmt.query_map([], |row| {
                    let mut obj = serde_json::Map::new();
                    for (i, name) in cols.iter().enumerate() {
                        obj.insert(name.clone(), sqlite_value_to_json(row.get_ref(i)?));
                    }
                    Ok(Value::Object(obj))
                })?;
                let mut out = Vec::new();
                for r in mapped { out.push(r?); }
                Ok(out)
            }).await
            .map_err(|e| NodeError::Other(format!("postgres[sqlite]: query: {}", e)))?;
            Ok(vec![rows.into_iter().map(|j| NodeItem { json: j, ..Default::default() }).collect()])
        }
        "execute" | "insert" | "update" | "delete" => {
            let n = conn.call(move |c| Ok(c.execute(&q, [])?)).await
                .map_err(|e| NodeError::Other(format!("postgres[sqlite]: execute: {}", e)))?;
            Ok(vec![vec![NodeItem { json: json!({ "rows_affected": n }), ..Default::default() }]])
        }
        other => Err(NodeError::Other(format!("postgres[sqlite]: unknown operation `{}`", other))),
    }
}

fn sqlite_value_to_json(v: ValueRef<'_>) -> Value {
    match v {
        ValueRef::Null => Value::Null,
        ValueRef::Integer(i) => Value::Number(i.into()),
        ValueRef::Real(f) => serde_json::Number::from_f64(f).map(Value::Number).unwrap_or(Value::Null),
        ValueRef::Text(b) => Value::String(String::from_utf8_lossy(b).into_owned()),
        ValueRef::Blob(b) => Value::String(format!("<blob:{} bytes>", b.len())),
    }
}

inventory::submit! { NodeFactory::new("postgres", || Arc::new(PostgresNode)) }
