#include "tableview.h"

#include "duckdb.h"
#include "../../../../cpp/vendor/sqlite3/sqlite3.h"

#include <chrono>
#include <cstdio>
#include <cstring>
#include <cstdlib>

namespace ge {

namespace {

// Escape simple para SQL identifiers — solo permite [A-Za-z0-9_]. Usado para
// nombres de tabla / columnas que vienen de metadata. Si encuentra un char
// invalido, lo reemplaza por '_'. NO sustituye al binding de parametros.
std::string sanitize_ident(const char* s) {
    std::string out;
    if (!s) return out;
    for (const char* p = s; *p; ++p) {
        char c = *p;
        if ((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z')
            || (c >= '0' && c <= '9') || c == '_') out += c;
        else out += '_';
    }
    return out;
}

// Escape para literales de string en SQL: dobla las comillas simples.
std::string sql_escape(const char* s) {
    std::string out;
    if (!s) return out;
    for (const char* p = s; *p; ++p) {
        out += *p;
        if (*p == '\'') out += '\'';
    }
    return out;
}

// Helper RAII alrededor de duckdb_database/connection.
struct DuckHandle {
    duckdb_database db   = nullptr;
    duckdb_connection cn = nullptr;
    bool open(const char* path) {
        if (duckdb_open(path, &db) == DuckDBError) return false;
        if (duckdb_connect(db, &cn) == DuckDBError) return false;
        return true;
    }
    ~DuckHandle() {
        if (cn) duckdb_disconnect(&cn);
        if (db) duckdb_close(&db);
    }
};

bool duck_query_silent(duckdb_connection cn, const char* sql) {
    duckdb_result r;
    duckdb_state st = duckdb_query(cn, sql, &r);
    duckdb_destroy_result(&r);
    return st == DuckDBSuccess;
}

} // namespace

bool tableview_smoke_test(const char* duckdb_path) {
    DuckHandle h;
    if (!h.open(duckdb_path)) return false;
    duckdb_result r;
    if (duckdb_query(h.cn, "SELECT 42 AS x", &r) == DuckDBError) {
        duckdb_destroy_result(&r);
        return false;
    }
    bool ok = duckdb_row_count(&r) == 1
              && duckdb_value_int64(&r, 0, 0) == 42;
    duckdb_destroy_result(&r);
    return ok;
}

bool tableview_count(const char* duckdb_path,
                      const char* duck_table,
                      const char* sql_filter,
                      int64_t* out)
{
    if (!duckdb_path || !duck_table || !out) return false;
    *out = 0;
    DuckHandle h;
    if (!h.open(duckdb_path)) return false;
    std::string tname = sanitize_ident(duck_table);
    if (tname.empty()) return false;
    std::string sql = "SELECT COUNT(*) FROM " + tname;
    if (sql_filter && *sql_filter) {
        sql += " WHERE ";
        sql += sql_filter;
    }
    duckdb_result r;
    if (duckdb_query(h.cn, sql.c_str(), &r) == DuckDBError) {
        std::fprintf(stderr, "[tableview_count] %s\n",
                      duckdb_result_error(&r) ? duckdb_result_error(&r) : "?");
        duckdb_destroy_result(&r);
        return false;
    }
    if (duckdb_row_count(&r) > 0) {
        *out = duckdb_value_int64(&r, 0, 0);
    }
    duckdb_destroy_result(&r);
    return true;
}

bool tableview_page(const char* duckdb_path,
                     const char* duck_table,
                     const char* id_column,
                     const std::vector<std::string>& columns,
                     const char* sql_filter,
                     const char* ops_db,
                     const char* row_type,
                     int64_t offset, int64_t limit,
                     std::vector<TablePageRow>* out)
{
    if (!out) return false;
    out->clear();
    if (!duckdb_path || !duck_table || !id_column) return false;
    if (limit < 1)    limit = 1;
    if (limit > 5000) limit = 5000;

    DuckHandle h;
    if (!h.open(duckdb_path)) return false;

    std::string idc = sanitize_ident(id_column);
    std::string tn  = sanitize_ident(duck_table);
    if (idc.empty() || tn.empty()) return false;

    // Si tenemos ops_db y row_type, hacemos LEFT JOIN para detectar promovidas
    // a traves de json_extract sobre entities.metadata.source.row_id.
    bool join_ops = (ops_db && *ops_db && row_type && *row_type);

    if (join_ops) {
        // ATTACH del SQLite. Las attaches viven por conexion; idempotente
        // detectando si ya existe seria mas robusto pero este path se llama
        // por cada page() — abrimos conexion fresca cada vez asi que no.
        std::string attach = "ATTACH '" + sql_escape(ops_db) + "' AS ops (TYPE SQLITE)";
        if (!duck_query_silent(h.cn, attach.c_str())) {
            // sin fallar — sin promovidas, solo perdemos el flag.
            join_ops = false;
        }
    }

    // SELECT: id_column + columns... + (CASE WHEN e.id NULL THEN '' ELSE e.id END).
    std::string sel = "SELECT t." + idc;
    for (const auto& c : columns) {
        std::string cc = sanitize_ident(c.c_str());
        if (!cc.empty()) sel += ", t." + cc;
    }
    if (join_ops) {
        sel += ", COALESCE(e.id, '')";
    } else {
        sel += ", ''";
    }
    sel += " FROM " + tn + " AS t";
    if (join_ops) {
        sel += " LEFT JOIN ops.entities AS e ON ";
        sel += "json_extract_string(e.metadata, '$.source.row_id') = CAST(t." + idc + " AS VARCHAR)";
        sel += " AND e.type_ref = '" + sql_escape(row_type) + "'";
    }
    if (sql_filter && *sql_filter) {
        sel += " WHERE ";
        sel += sql_filter;
    }
    sel += " ORDER BY t." + idc + " ASC";
    sel += " LIMIT " + std::to_string(limit);
    sel += " OFFSET " + std::to_string(offset);

    duckdb_result r;
    if (duckdb_query(h.cn, sel.c_str(), &r) == DuckDBError) {
        std::fprintf(stderr, "[tableview_page] %s\n",
                      duckdb_result_error(&r) ? duckdb_result_error(&r) : "?");
        duckdb_destroy_result(&r);
        return false;
    }
    idx_t rows = duckdb_row_count(&r);
    idx_t cols = duckdb_column_count(&r);
    out->reserve((size_t)rows);
    for (idx_t row = 0; row < rows; ++row) {
        TablePageRow tr;
        // col 0 = id
        if (!duckdb_value_is_null(&r, 0, row)) {
            char* v = duckdb_value_varchar(&r, 0, row);
            tr.id = v ? v : "";
            if (v) duckdb_free(v);
        }
        // cols 1..N-2 = columns[]
        idx_t expected_cols = (idx_t)columns.size();
        tr.values.reserve(expected_cols);
        for (idx_t i = 0; i < expected_cols; ++i) {
            idx_t c = 1 + i;
            if (c >= cols) { tr.values.emplace_back(""); continue; }
            if (duckdb_value_is_null(&r, c, row)) {
                tr.values.emplace_back("");
            } else {
                char* v = duckdb_value_varchar(&r, c, row);
                tr.values.emplace_back(v ? v : "");
                if (v) duckdb_free(v);
            }
        }
        // ultima col = promoted_entity_id
        idx_t prom_col = cols > 0 ? cols - 1 : 0;
        if (cols > 0 && !duckdb_value_is_null(&r, prom_col, row)) {
            char* v = duckdb_value_varchar(&r, prom_col, row);
            tr.promoted_entity_id = v ? v : "";
            if (v) duckdb_free(v);
        }
        out->push_back(std::move(tr));
    }
    duckdb_destroy_result(&r);
    return true;
}

bool tableview_create(const char* ops_db,
                       const char* name,
                       const char* duckdb_path,
                       const char* duck_table,
                       const char* row_type,
                       char* out_id, std::size_t out_id_n)
{
    if (!ops_db || !duckdb_path || !duck_table) return false;
    if (!name || !*name) name = "Table";

    auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                      std::chrono::system_clock::now().time_since_epoch()).count();

    char id[80];
    std::snprintf(id, sizeof(id), "table_%lld", (long long)now_ms);
    if (out_id && out_id_n > 0) {
        std::snprintf(out_id, out_id_n, "%s", id);
    }

    sqlite3* db = nullptr;
    if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READWRITE, nullptr) != SQLITE_OK) {
        if (db) sqlite3_close(db);
        return false;
    }

    std::string meta = "{";
    meta += "\"duckdb_path\":\""; meta += sql_escape(duckdb_path); meta += "\",";
    meta += "\"table_name\":\"";  meta += sql_escape(duck_table);  meta += "\",";
    meta += "\"row_type\":\"";    meta += sql_escape(row_type ? row_type : ""); meta += "\",";
    meta += "\"id_column\":\"id\",";
    meta += "\"label_column\":\"name\",";
    meta += "\"columns\":[],";
    meta += "\"filter_sql\":\"\",";
    meta += "\"expanded\":false";
    meta += "}";

    const char* ins =
        "INSERT INTO entities(id, name, type_ref, status, tags, source, metadata, "
        "                     created_at, updated_at) "
        "VALUES (?, ?, 'Table', 'active', '[]', 'manual', ?, "
        "        strftime('%Y-%m-%dT%H:%M:%fZ','now'), "
        "        strftime('%Y-%m-%dT%H:%M:%fZ','now'))";
    sqlite3_stmt* st = nullptr;
    if (sqlite3_prepare_v2(db, ins, -1, &st, nullptr) != SQLITE_OK) {
        sqlite3_close(db);
        return false;
    }
    sqlite3_bind_text(st, 1, id,        -1, SQLITE_TRANSIENT);
    sqlite3_bind_text(st, 2, name,      -1, SQLITE_TRANSIENT);
    sqlite3_bind_text(st, 3, meta.c_str(), -1, SQLITE_TRANSIENT);
    bool ok = sqlite3_step(st) == SQLITE_DONE;
    sqlite3_finalize(st);
    sqlite3_close(db);
    return ok;
}

// ----------------------------------------------------------------------------
// Path resolution + counts cache
// ----------------------------------------------------------------------------

namespace {

uint64_t fnv1a64(const char* s) {
    uint64_t h = 1469598103934665603ULL;
    for (; s && *s; ++s) {
        h ^= (uint8_t)*s;
        h *= 1099511628211ULL;
    }
    return h;
}

std::string dirname_of(const char* path) {
    if (!path) return "";
    std::string s = path;
    auto pos = s.find_last_of("/\\");
    if (pos == std::string::npos) return ".";
    return s.substr(0, pos);
}

bool is_absolute(const char* p) {
    if (!p || !*p) return false;
    if (p[0] == '/') return true;
    if (std::strlen(p) >= 2 && p[1] == ':' &&
        ((p[0] >= 'A' && p[0] <= 'Z') || (p[0] >= 'a' && p[0] <= 'z'))) return true;
    return false;
}

} // namespace

std::string tableview_resolve_path(const char* ops_db, const char* maybe_rel) {
    if (!maybe_rel) return "";
    if (is_absolute(maybe_rel)) return maybe_rel;
    std::string base = dirname_of(ops_db);
    if (base.empty()) base = ".";
    return base + "/" + maybe_rel;
}

bool tableview_refresh_counts(const char* ops_db,
                               std::unordered_map<uint64_t, int64_t>* out)
{
    if (!ops_db || !out) return false;
    out->clear();
    sqlite3* db = nullptr;
    if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READONLY, nullptr) != SQLITE_OK) {
        if (db) sqlite3_close(db);
        return false;
    }
    const char* sql =
        "SELECT id, "
        "       json_extract(metadata, '$.duckdb_path'), "
        "       json_extract(metadata, '$.table_name'), "
        "       json_extract(metadata, '$.filter_sql') "
        "FROM entities WHERE type_ref = 'Table'";
    sqlite3_stmt* st = nullptr;
    if (sqlite3_prepare_v2(db, sql, -1, &st, nullptr) != SQLITE_OK) {
        sqlite3_close(db);
        return false;
    }
    while (sqlite3_step(st) == SQLITE_ROW) {
        const unsigned char* id_p   = sqlite3_column_text(st, 0);
        const unsigned char* path_p = sqlite3_column_text(st, 1);
        const unsigned char* tab_p  = sqlite3_column_text(st, 2);
        const unsigned char* flt_p  = sqlite3_column_text(st, 3);
        if (!id_p || !path_p || !tab_p) continue;
        std::string abs = tableview_resolve_path(ops_db, (const char*)path_p);
        int64_t total = 0;
        if (!tableview_count(abs.c_str(), (const char*)tab_p,
                              flt_p ? (const char*)flt_p : nullptr,
                              &total)) {
            std::fprintf(stderr,
                "[tableview_refresh_counts] count failed for id=%s\n", id_p);
            continue;
        }
        out->emplace(fnv1a64((const char*)id_p), total);
    }
    sqlite3_finalize(st);
    sqlite3_close(db);
    return true;
}

// ----------------------------------------------------------------------------
// Issue 0011 — UI fase 2 helpers
// ----------------------------------------------------------------------------

namespace {

std::string sanitize_id_part(const char* s) {
    std::string out;
    if (!s) return out;
    for (const char* p = s; *p && out.size() < 60; ++p) {
        char c = *p;
        if ((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z')
            || (c >= '0' && c <= '9') || c == '_') out += c;
        else out += '_';
    }
    return out;
}

// Parser minimo para extraer un array de strings de un JSON tipo
// "[\"a\",\"b\",\"c\"]". No es un JSON parser general — se usa solo sobre
// el campo `columns` que escribe esta misma capa, asi que el formato esta
// controlado.
std::vector<std::string> parse_json_string_array(const char* s) {
    std::vector<std::string> out;
    if (!s) return out;
    const char* p = s;
    while (*p && *p != '[') ++p;
    if (*p != '[') return out;
    ++p;
    while (*p) {
        while (*p && (*p == ' ' || *p == ',' || *p == '\t' || *p == '\n')) ++p;
        if (*p == ']' || !*p) break;
        if (*p != '"') { ++p; continue; }
        ++p;
        std::string v;
        while (*p && *p != '"') {
            if (*p == '\\' && p[1]) { v += p[1]; p += 2; }
            else                     { v += *p++; }
        }
        if (*p == '"') ++p;
        out.push_back(std::move(v));
    }
    return out;
}

} // namespace

bool tableview_list_columns(const char* duckdb_path,
                             const char* duck_table,
                             std::vector<std::string>* out)
{
    if (!duckdb_path || !duck_table || !out) return false;
    out->clear();
    DuckHandle h;
    if (!h.open(duckdb_path)) return false;
    std::string sql = "SELECT * FROM " + sanitize_ident(duck_table) + " LIMIT 0";
    duckdb_result r;
    if (duckdb_query(h.cn, sql.c_str(), &r) == DuckDBError) {
        duckdb_destroy_result(&r);
        return false;
    }
    idx_t cn = duckdb_column_count(&r);
    for (idx_t i = 0; i < cn; ++i) {
        const char* name = duckdb_column_name(&r, i);
        if (name) out->emplace_back(name);
    }
    duckdb_destroy_result(&r);
    return true;
}

bool tableview_get_metadata(const char* ops_db, const char* entity_id,
                             TableMetadata* out)
{
    if (!ops_db || !entity_id || !out) return false;
    *out = TableMetadata{};
    out->entity_id = entity_id;
    sqlite3* db = nullptr;
    if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READONLY, nullptr) != SQLITE_OK) {
        if (db) sqlite3_close(db);
        return false;
    }
    const char* sql =
        "SELECT name, "
        "       json_extract(metadata, '$.duckdb_path'), "
        "       json_extract(metadata, '$.table_name'), "
        "       json_extract(metadata, '$.row_type'), "
        "       json_extract(metadata, '$.id_column'), "
        "       json_extract(metadata, '$.label_column'), "
        "       json_extract(metadata, '$.columns'), "
        "       json_extract(metadata, '$.filter_sql'), "
        "       json_extract(metadata, '$.expanded') "
        "FROM entities WHERE id = ? AND type_ref = 'Table'";
    sqlite3_stmt* st = nullptr;
    if (sqlite3_prepare_v2(db, sql, -1, &st, nullptr) != SQLITE_OK) {
        sqlite3_close(db);
        return false;
    }
    sqlite3_bind_text(st, 1, entity_id, -1, SQLITE_TRANSIENT);
    bool ok = false;
    if (sqlite3_step(st) == SQLITE_ROW) {
        auto col_text = [&](int i) -> std::string {
            const unsigned char* p = sqlite3_column_text(st, i);
            return p ? std::string((const char*)p) : std::string();
        };
        out->name             = col_text(0);
        out->duckdb_path      = col_text(1);
        out->table_name       = col_text(2);
        out->row_type         = col_text(3);
        out->id_column        = col_text(4);
        out->label_column     = col_text(5);
        std::string cols_json = col_text(6);
        out->filter_sql       = col_text(7);
        std::string exp_json  = col_text(8);

        if (out->id_column.empty())    out->id_column    = "id";
        if (out->label_column.empty()) out->label_column = "name";
        out->columns           = parse_json_string_array(cols_json.c_str());
        out->expanded          = (exp_json == "true" || exp_json == "1");
        out->duckdb_path_abs   = tableview_resolve_path(ops_db, out->duckdb_path.c_str());
        ok = true;
    }
    sqlite3_finalize(st);
    sqlite3_close(db);
    return ok;
}

namespace {

bool exec_metadata_patch(const char* ops_db, const char* entity_id,
                          const char* json_set_clause)
{
    sqlite3* db = nullptr;
    if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READWRITE, nullptr) != SQLITE_OK) {
        if (db) sqlite3_close(db);
        return false;
    }
    std::string sql = "UPDATE entities SET metadata = ";
    sql += json_set_clause;
    sql += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?";
    sqlite3_stmt* st = nullptr;
    if (sqlite3_prepare_v2(db, sql.c_str(), -1, &st, nullptr) != SQLITE_OK) {
        sqlite3_close(db);
        return false;
    }
    sqlite3_bind_text(st, 1, entity_id, -1, SQLITE_TRANSIENT);
    bool ok = sqlite3_step(st) == SQLITE_DONE;
    sqlite3_finalize(st);
    sqlite3_close(db);
    return ok;
}

} // namespace

bool tableview_set_expanded(const char* ops_db, const char* entity_id,
                             bool expanded)
{
    if (!ops_db || !entity_id) return false;
    std::string clause = "json_set(COALESCE(metadata,'{}'), '$.expanded', json('";
    clause += expanded ? "true" : "false";
    clause += "'))";
    return exec_metadata_patch(ops_db, entity_id, clause.c_str());
}

bool tableview_set_columns(const char* ops_db, const char* entity_id,
                            const std::vector<std::string>& columns)
{
    if (!ops_db || !entity_id) return false;
    std::string arr = "[";
    for (size_t i = 0; i < columns.size(); ++i) {
        if (i) arr += ",";
        arr += "\"";
        arr += sql_escape(columns[i].c_str());
        arr += "\"";
    }
    arr += "]";
    std::string clause = "json_set(COALESCE(metadata,'{}'), '$.columns', json('";
    clause += sql_escape(arr.c_str());
    clause += "'))";
    return exec_metadata_patch(ops_db, entity_id, clause.c_str());
}

namespace {

// Busca si ya existe una entidad promovida por (duckdb_path, row_id).
bool find_existing_promotion(const char* ops_db, const char* duckdb_path,
                              const char* row_id, std::string* out_id)
{
    if (!ops_db || !duckdb_path || !row_id || !out_id) return false;
    out_id->clear();
    sqlite3* db = nullptr;
    if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READONLY, nullptr) != SQLITE_OK) {
        if (db) sqlite3_close(db);
        return false;
    }
    const char* sql =
        "SELECT id FROM entities "
        "WHERE json_extract(metadata, '$.source.duckdb') = ? "
        "  AND json_extract(metadata, '$.source.row_id') = ? "
        "LIMIT 1";
    sqlite3_stmt* st = nullptr;
    if (sqlite3_prepare_v2(db, sql, -1, &st, nullptr) != SQLITE_OK) {
        sqlite3_close(db);
        return false;
    }
    sqlite3_bind_text(st, 1, duckdb_path, -1, SQLITE_TRANSIENT);
    sqlite3_bind_text(st, 2, row_id,      -1, SQLITE_TRANSIENT);
    bool found = false;
    if (sqlite3_step(st) == SQLITE_ROW) {
        const unsigned char* p = sqlite3_column_text(st, 0);
        if (p) { *out_id = (const char*)p; found = true; }
    }
    sqlite3_finalize(st);
    sqlite3_close(db);
    return found;
}

} // namespace

bool tableview_promote_row(const char* ops_db,
                            const char* duckdb_path,
                            const char* duck_table,
                            const char* row_id,
                            const char* row_type,
                            const char* label_column,
                            char* out_entity_id, std::size_t out_id_n)
{
    if (!ops_db || !duckdb_path || !duck_table || !row_id) return false;
    if (!row_type || !*row_type) row_type = "Row";
    if (!label_column || !*label_column) label_column = "name";

    // Idempotencia: si ya esta promovida, devuelve el id existente.
    {
        std::string existing;
        if (find_existing_promotion(ops_db, duckdb_path, row_id, &existing)) {
            if (out_entity_id && out_id_n > 0) {
                std::snprintf(out_entity_id, out_id_n, "%s", existing.c_str());
            }
            return true;
        }
    }

    // Lee la fila + label desde DuckDB.
    DuckHandle h;
    if (!h.open(duckdb_path)) return false;
    std::string tn  = sanitize_ident(duck_table);
    std::string idc = "id";  // por convencion. Si la tabla usa otro PK, el caller lo sabe.
    {
        // Best-effort: si label_column existe en la tabla, lo seleccionamos.
        std::string sel = "SELECT * FROM " + tn + " WHERE CAST(" + idc + " AS VARCHAR) = '"
                          + sql_escape(row_id) + "' LIMIT 1";
        duckdb_result r;
        if (duckdb_query(h.cn, sel.c_str(), &r) == DuckDBError) {
            std::fprintf(stderr, "[promote] query fail: %s\n",
                          duckdb_result_error(&r) ? duckdb_result_error(&r) : "?");
            duckdb_destroy_result(&r);
            return false;
        }
        if (duckdb_row_count(&r) == 0) {
            duckdb_destroy_result(&r);
            return false;
        }
        // Construir name = valor del label_column si existe; fallback row_id.
        std::string display_name = row_id;
        idx_t ncols = duckdb_column_count(&r);
        for (idx_t i = 0; i < ncols; ++i) {
            const char* cname = duckdb_column_name(&r, i);
            if (cname && std::strcmp(cname, label_column) == 0) {
                if (!duckdb_value_is_null(&r, i, 0)) {
                    char* v = duckdb_value_varchar(&r, i, 0);
                    if (v) { display_name = v; duckdb_free(v); }
                }
                break;
            }
        }

        // Construir metadata JSON con source + columnas.
        std::string meta = "{\"source\":{";
        meta += "\"duckdb\":\""; meta += sql_escape(duckdb_path); meta += "\",";
        meta += "\"table\":\"";  meta += sql_escape(duck_table);  meta += "\",";
        meta += "\"row_id\":\""; meta += sql_escape(row_id);      meta += "\"";
        meta += "}";
        for (idx_t i = 0; i < ncols; ++i) {
            const char* cname = duckdb_column_name(&r, i);
            if (!cname || !*cname) continue;
            meta += ",\"";
            meta += sql_escape(cname);
            meta += "\":";
            if (duckdb_value_is_null(&r, i, 0)) {
                meta += "null";
            } else {
                char* v = duckdb_value_varchar(&r, i, 0);
                meta += "\"";
                if (v) { meta += sql_escape(v); duckdb_free(v); }
                meta += "\"";
            }
        }
        meta += "}";
        duckdb_destroy_result(&r);

        // Genera id estable.
        std::string entity_id = "prom_";
        entity_id += sanitize_id_part(row_type);
        entity_id += "_";
        entity_id += sanitize_id_part(row_id);

        sqlite3* db = nullptr;
        if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READWRITE, nullptr) != SQLITE_OK) {
            if (db) sqlite3_close(db);
            return false;
        }
        const char* ins =
            "INSERT OR IGNORE INTO entities("
            "  id, name, type_ref, status, tags, source, metadata, "
            "  created_at, updated_at) "
            "VALUES (?, ?, ?, 'active', '[]', 'tableview', ?, "
            "        strftime('%Y-%m-%dT%H:%M:%fZ','now'), "
            "        strftime('%Y-%m-%dT%H:%M:%fZ','now'))";
        sqlite3_stmt* st = nullptr;
        if (sqlite3_prepare_v2(db, ins, -1, &st, nullptr) != SQLITE_OK) {
            sqlite3_close(db);
            return false;
        }
        sqlite3_bind_text(st, 1, entity_id.c_str(),    -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(st, 2, display_name.c_str(), -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(st, 3, row_type,             -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(st, 4, meta.c_str(),         -1, SQLITE_TRANSIENT);
        bool ok = sqlite3_step(st) == SQLITE_DONE;
        sqlite3_finalize(st);
        sqlite3_close(db);
        if (ok && out_entity_id && out_id_n > 0) {
            std::snprintf(out_entity_id, out_id_n, "%s", entity_id.c_str());
        }
        return ok;
    }
}

bool tableview_demote_row(const char* ops_db, const char* entity_id) {
    if (!ops_db || !entity_id) return false;
    sqlite3* db = nullptr;
    if (sqlite3_open_v2(ops_db, &db, SQLITE_OPEN_READWRITE, nullptr) != SQLITE_OK) {
        if (db) sqlite3_close(db);
        return false;
    }
    sqlite3_stmt* st = nullptr;
    if (sqlite3_prepare_v2(db, "DELETE FROM entities WHERE id = ?",
                            -1, &st, nullptr) != SQLITE_OK) {
        sqlite3_close(db);
        return false;
    }
    sqlite3_bind_text(st, 1, entity_id, -1, SQLITE_TRANSIENT);
    bool ok = sqlite3_step(st) == SQLITE_DONE;
    sqlite3_finalize(st);
    sqlite3_close(db);
    return ok;
}

namespace {

IngestKind detect_ingest_kind(const char* path) {
    if (!path) return INGEST_CSV;
    const char* dot = std::strrchr(path, '.');
    if (!dot) return INGEST_CSV;
    if (std::strcmp(dot, ".csv") == 0)     return INGEST_CSV;
    if (std::strcmp(dot, ".tsv") == 0)     return INGEST_CSV;
    if (std::strcmp(dot, ".parquet") == 0) return INGEST_PARQUET;
    if (std::strcmp(dot, ".pq") == 0)      return INGEST_PARQUET;
    if (std::strcmp(dot, ".json") == 0)    return INGEST_JSON;
    if (std::strcmp(dot, ".ndjson") == 0)  return INGEST_JSON;
    return INGEST_CSV;
}

const char* ingest_func_name(IngestKind k) {
    switch (k) {
        case INGEST_PARQUET: return "read_parquet";
        case INGEST_JSON:    return "read_json_auto";
        default:             return "read_csv_auto";
    }
}

} // namespace

bool tableview_ingest_file(const char* duckdb_path,
                            const char* file_path,
                            const char* dest_table,
                            IngestKind  kind,
                            std::string* out_error)
{
    if (!duckdb_path || !file_path || !dest_table) {
        if (out_error) *out_error = "args nulos";
        return false;
    }
    if (kind == INGEST_AUTO) kind = detect_ingest_kind(file_path);

    DuckHandle h;
    if (!h.open(duckdb_path)) {
        if (out_error) *out_error = "no se pudo abrir el .duckdb";
        return false;
    }
    std::string tn = sanitize_ident(dest_table);
    std::string sql = "CREATE TABLE " + tn + " AS SELECT * FROM "
                       + ingest_func_name(kind) + "('" + sql_escape(file_path) + "')";
    duckdb_result r;
    if (duckdb_query(h.cn, sql.c_str(), &r) == DuckDBError) {
        if (out_error) {
            const char* err = duckdb_result_error(&r);
            *out_error = err ? err : "ingest fail";
        }
        duckdb_destroy_result(&r);
        return false;
    }
    duckdb_destroy_result(&r);
    return true;
}

} // namespace ge