// tql_duckdb.cpp — DuckDB adapter para ejecutar SQL emitido por tql_to_sql.
// Compilado solo si FN_TQL_DUCKDB define. Ver issue 0080.
#include "tql_duckdb.h"

#ifdef FN_TQL_DUCKDB

#include "duckdb.h"

#include <chrono>
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>

namespace tql_duckdb {

using namespace data_table;

namespace {

// SQL identifier quote (mismo patron que tql_to_sql).
std::string sql_ident(const std::string& name) {
    std::string out;
    out.reserve(name.size() + 4);
    out += '"';
    for (char c : name) {
        if (c == '"') out += "\"\"";
        else          out += c;
    }
    out += '"';
    return out;
}

const char* duckdb_type_for(ColumnType t) {
    switch (t) {
        case ColumnType::Int:   return "BIGINT";
        case ColumnType::Float: return "DOUBLE";
        case ColumnType::Bool:  return "BOOLEAN";
        case ColumnType::Date:  return "VARCHAR";  // se almacena ISO como texto v1
        case ColumnType::Json:  return "VARCHAR";
        default:                return "VARCHAR";
    }
}

// SQL literal escape para string.
std::string lit_str(const char* s) {
    std::string out = "'";
    for (const char* p = s ? s : ""; *p; ++p) {
        if (*p == '\'') out += "''";
        else            out += *p;
    }
    out += "'";
    return out;
}

bool create_and_load(duckdb_connection cn, const TableInput& t, std::string& err) {
    // CREATE TABLE
    std::string ddl = "CREATE TABLE " + sql_ident(t.name) + " (";
    for (size_t i = 0; i < t.headers.size(); ++i) {
        if (i > 0) ddl += ", ";
        ColumnType ct = (i < t.types.size()) ? t.types[i] : ColumnType::String;
        ddl += sql_ident(t.headers[i]) + " " + duckdb_type_for(ct);
    }
    ddl += ");";
    duckdb_result rr;
    if (duckdb_query(cn, ddl.c_str(), &rr) == DuckDBError) {
        err = duckdb_result_error(&rr);
        duckdb_destroy_result(&rr);
        return false;
    }
    duckdb_destroy_result(&rr);

    // INSERT rows via VALUES batches (1000 rows/insert).
    if (t.rows == 0 || t.cols == 0) return true;
    const int batch = 1000;
    for (int r0 = 0; r0 < t.rows; r0 += batch) {
        int r1 = (r0 + batch < t.rows) ? r0 + batch : t.rows;
        std::string ins = "INSERT INTO " + sql_ident(t.name) + " VALUES ";
        for (int r = r0; r < r1; ++r) {
            if (r > r0) ins += ", ";
            ins += "(";
            for (int c = 0; c < t.cols; ++c) {
                if (c > 0) ins += ", ";
                const char* v = t.cells[r * t.cols + c];
                if (!v || !*v) { ins += "NULL"; continue; }
                ColumnType ct = (c < (int)t.types.size()) ? t.types[c] : ColumnType::String;
                if (ct == ColumnType::Int || ct == ColumnType::Float) {
                    // Asumir parseable; sino DuckDB error.
                    ins += v;
                } else if (ct == ColumnType::Bool) {
                    ins += (std::strcmp(v, "true") == 0) ? "TRUE" : "FALSE";
                } else {
                    ins += lit_str(v);
                }
            }
            ins += ")";
        }
        ins += ";";
        if (duckdb_query(cn, ins.c_str(), &rr) == DuckDBError) {
            err = std::string("INSERT into ") + t.name + ": " + duckdb_result_error(&rr);
            duckdb_destroy_result(&rr);
            return false;
        }
        duckdb_destroy_result(&rr);
    }
    return true;
}

ColumnType type_from_duckdb(duckdb_type t) {
    switch (t) {
        case DUCKDB_TYPE_BOOLEAN: return ColumnType::Bool;
        case DUCKDB_TYPE_TINYINT:
        case DUCKDB_TYPE_SMALLINT:
        case DUCKDB_TYPE_INTEGER:
        case DUCKDB_TYPE_BIGINT:
        case DUCKDB_TYPE_HUGEINT:
        case DUCKDB_TYPE_UTINYINT:
        case DUCKDB_TYPE_USMALLINT:
        case DUCKDB_TYPE_UINTEGER:
        case DUCKDB_TYPE_UBIGINT:
            return ColumnType::Int;
        case DUCKDB_TYPE_FLOAT:
        case DUCKDB_TYPE_DOUBLE:
        case DUCKDB_TYPE_DECIMAL:
            return ColumnType::Float;
        case DUCKDB_TYPE_DATE:
        case DUCKDB_TYPE_TIMESTAMP:
            return ColumnType::Date;
        default:
            return ColumnType::String;
    }
}

} // anon

Result execute(const std::string& sql,
                const std::vector<std::string>& params,
                const std::vector<TableInput>& tables) {
    Result out;
    auto t0 = std::chrono::steady_clock::now();

    duckdb_database db = nullptr;
    duckdb_connection cn = nullptr;
    if (duckdb_open(nullptr, &db) == DuckDBError) {
        out.error = "duckdb_open failed";
        return out;
    }
    if (duckdb_connect(db, &cn) == DuckDBError) {
        out.error = "duckdb_connect failed";
        duckdb_close(&db);
        return out;
    }

    // Crear y poblar tablas.
    for (const auto& t : tables) {
        std::string e;
        if (!create_and_load(cn, t, e)) {
            out.error = e;
            duckdb_disconnect(&cn);
            duckdb_close(&db);
            return out;
        }
    }

    // Preparar + bind params.
    duckdb_prepared_statement stmt = nullptr;
    if (duckdb_prepare(cn, sql.c_str(), &stmt) == DuckDBError) {
        out.error = std::string("prepare: ") + duckdb_prepare_error(stmt);
        duckdb_destroy_prepare(&stmt);
        duckdb_disconnect(&cn);
        duckdb_close(&db);
        return out;
    }
    for (size_t i = 0; i < params.size(); ++i) {
        // DuckDB params son 1-based.
        if (duckdb_bind_varchar(stmt, (idx_t)(i + 1), params[i].c_str()) == DuckDBError) {
            out.error = "bind param fail";
            duckdb_destroy_prepare(&stmt);
            duckdb_disconnect(&cn);
            duckdb_close(&db);
            return out;
        }
    }

    duckdb_result res;
    if (duckdb_execute_prepared(stmt, &res) == DuckDBError) {
        out.error = std::string("execute: ") + duckdb_result_error(&res);
        duckdb_destroy_result(&res);
        duckdb_destroy_prepare(&stmt);
        duckdb_disconnect(&cn);
        duckdb_close(&db);
        return out;
    }

    // Materializar resultado en StageOutput.
    idx_t cols = duckdb_column_count(&res);
    idx_t rows = duckdb_row_count(&res);
    out.out.cols = (int)cols;
    out.out.rows = (int)rows;
    out.row_count = (int)rows;

    out.out.headers.reserve(cols);
    out.out.types.reserve(cols);
    for (idx_t c = 0; c < cols; ++c) {
        out.out.headers.emplace_back(duckdb_column_name(&res, c));
        out.out.types.push_back(type_from_duckdb(duckdb_column_type(&res, c)));
    }
    out.out.cell_backing.reserve(rows * cols);
    out.out.cells.reserve(rows * cols);
    for (idx_t r = 0; r < rows; ++r) {
        for (idx_t c = 0; c < cols; ++c) {
            char* v = duckdb_value_varchar(&res, c, r);
            out.out.cell_backing.emplace_back(v ? v : "");
            if (v) duckdb_free(v);
        }
    }
    for (auto& s : out.out.cell_backing) out.out.cells.push_back(s.c_str());

    duckdb_destroy_result(&res);
    duckdb_destroy_prepare(&stmt);
    duckdb_disconnect(&cn);
    duckdb_close(&db);

    auto t1 = std::chrono::steady_clock::now();
    out.duration_ms = std::chrono::duration<double, std::milli>(t1 - t0).count();
    return out;
}

} // namespace tql_duckdb

#endif // FN_TQL_DUCKDB