// 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)
//
// `mysql` — execute SQL against a MySQL database. v0.1 ships with the
// sqlite: escape hatch (same dispatch as postgres) for sim coverage; real
// MySQL wire driver (mysql_async) is a v0.2 add — flagged NotYetPorted
// for non-sqlite connection strings until that lands.

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;

pub struct MySqlNode;

static META: NodeMetadata = NodeMetadata {
    name: "mysql",
    display_name: "MySQL",
    version: &[1, 2],
    group: &["database"],
    description: "Execute SQL against MySQL (v0.1 supports the sqlite: escape hatch)",
    inputs: 1,
    outputs: 1,
    properties: &[],
    credentials: &[CredentialRef { name: "mysql", required: true }],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let conn_str = ctx.credential_field("connection_string").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()))
            .ok_or_else(|| NodeError::MissingCredential("mysql.connection_string".into()))?;
        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();

        if let Some(sqlite_path) = conn_str.strip_prefix("sqlite:") {
            return execute_sqlite(sqlite_path, &query, &operation).await;
        }
        Err(NodeError::NotYetPorted {
            node: "mysql",
            upstream: "https://github.com/n8n-io/n8n/blob/master/packages/nodes-base/nodes/MySql/MySql.node.ts",
        })
    }
}

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!("mysql[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!("mysql[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!("mysql[sqlite]: execute: {}", e)))?;
            Ok(vec![vec![NodeItem { json: json!({ "rows_affected": n }), ..Default::default() }]])
        }
        other => Err(NodeError::Other(format!("mysql[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("mysql", || Arc::new(MySqlNode)) }
