// calogDb.c -- calog database library (see calogDb.h).
//
// The four natives are thin dispatchers: dbOpen picks a backend by driver name, and
// dbExec/dbQuery/dbClose recover the backend from the handle's type tag. Each backend
// (SQLite, Postgres, MySQL) is compiled in behind CALOG_WITH_SQLITE / _PG / _MYSQL and
// provides open/exec/query plus a marshalling of its rows to CalogValueT.

#define _POSIX_C_SOURCE 200809L

#include "calogDb.h"
#include "calogInternal.h"

#include "calogHandle.h"

#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef CALOG_WITH_SQLITE
#include "sqlite3.h"
#endif

#ifdef CALOG_WITH_PG
#include "libpq-fe.h"
// A few stable pg_type OIDs so a typed column comes back typed (everything else, including
// numeric/date/json, arrives as its text string).
#define PG_OID_BOOL   16
#define PG_OID_BYTEA  17
#define PG_OID_INT8   20
#define PG_OID_INT2   21
#define PG_OID_INT4   23
#define PG_OID_FLOAT4 700
#define PG_OID_FLOAT8 701
#endif

#ifdef CALOG_WITH_MYSQL
#include "mysql.h"
#endif

// Handle type tags, distinct across the whole registry so a stray handle of the wrong
// backend fails to resolve.
#define DB_TYPE_SQLITE 1u
#define DB_TYPE_PG     2u
#define DB_TYPE_MYSQL  3u

#define DB_NUM_SCRATCH 32

// Cap on a mysql "key=value ..." conninfo string; longer input is rejected outright rather
// than silently truncated mid-token.
#define MYSQL_CONNINFO_CAP 512

// Process-wide DB library state: one connection registry shared by every runtime that
// registers the natives (handles are globally-unique ints, so sharing is safe). refCount is
// one per registered runtime, so the singleton is freed only when the last one shuts down.
typedef struct DbLibT {
    CalogHandleTableT *handles;
    int32_t            refCount;
} DbLibT;

static pthread_mutex_t gDbLibMutex = PTHREAD_MUTEX_INITIALIZER;
static DbLibT         *gDbLib      = NULL;

// A query row's columns are marshalled through this pair of callbacks, shared by every
// backend's row loop (dbMarshalRow): nameFn hands back the column name (never fails), valueFn
// marshals the column into *out (already nil'd), returning an error status on OOM.
typedef void    (*DbColumnNameFnT)(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut);
typedef int32_t (*DbColumnValueFnT)(void *ctx, int32_t column, CalogValueT *out);

static int32_t  dbClose(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static void     dbCloser(uint32_t type, void *resource);
static int32_t  dbExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t  dbFinishOpen(DbLibT *lib, uint32_t type, void *conn, CalogValueT *result);
static int32_t  dbMarshalRow(CalogValueT *result, CalogAggT *rows, void *ctx, int32_t columnCount, DbColumnNameFnT nameFn, DbColumnValueFnT valueFn);
static int32_t  dbOpen(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t  dbQuery(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static uint32_t dbResolve(DbLibT *lib, int64_t handle, void **connOut);
#ifdef CALOG_WITH_MYSQL
static void        mysqlCleanup(MYSQL_BIND *binds, char **buffers, unsigned long *lengths, my_bool *isNull, unsigned int columnCount, MYSQL_RES *meta, MYSQL_STMT *stmt);
static int32_t     mysqlColumn(enum enum_field_types type, const char *bytes, unsigned long length, CalogValueT *out);
static int32_t     mysqlExec(MYSQL *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static int32_t     mysqlOpen(DbLibT *lib, const char *conninfo, CalogValueT *result);
static MYSQL_STMT *mysqlPrepare(MYSQL *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static int32_t     mysqlQuery(MYSQL *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static void        mysqlRowName(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut);
static int32_t     mysqlRowValue(void *ctx, int32_t column, CalogValueT *out);
#endif
#ifdef CALOG_WITH_PG
static int32_t   pgColumn(PGresult *res, int row, int column, CalogValueT *out);
static int32_t   pgExec(PGconn *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static int32_t   pgOpen(DbLibT *lib, const char *conninfo, CalogValueT *result);
static int32_t   pgQuery(PGconn *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static void      pgRowName(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut);
static int32_t   pgRowValue(void *ctx, int32_t column, CalogValueT *out);
static PGresult *pgRun(PGconn *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
#endif
#ifdef CALOG_WITH_SQLITE
static int32_t sqliteColumn(sqlite3_stmt *stmt, int column, CalogValueT *out);
static int32_t sqliteExec(sqlite3 *db, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static int32_t sqliteOpen(DbLibT *lib, const char *path, CalogValueT *result);
static int32_t sqlitePrepare(sqlite3 *db, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, sqlite3_stmt **out, CalogValueT *result);
static int32_t sqliteQuery(sqlite3 *db, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result);
static void    sqliteRowName(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut);
static int32_t sqliteRowValue(void *ctx, int32_t column, CalogValueT *out);
#endif


int32_t calogDbRegister(CalogT *calog) {
    int32_t status;

    pthread_mutex_lock(&gDbLibMutex);
    if (gDbLib == NULL) {
        DbLibT *lib;
        lib = (DbLibT *)calloc(1, sizeof(*lib));
        if (lib == NULL) {
            pthread_mutex_unlock(&gDbLibMutex);
            return calogErrOomE;
        }
        lib->handles = calogHandleTableCreate();
        if (lib->handles == NULL) {
            free(lib);
            pthread_mutex_unlock(&gDbLibMutex);
            return calogErrOomE;
        }
        gDbLib = lib;
    }
    gDbLib->refCount++;
    pthread_mutex_unlock(&gDbLibMutex);
    status = calogRegisterInline(calog, "dbOpen", dbOpen, gDbLib);
    if (status == calogOkE) {
        status = calogRegisterInline(calog, "dbExec", dbExec, gDbLib);
    }
    if (status == calogOkE) {
        status = calogRegisterInline(calog, "dbQuery", dbQuery, gDbLib);
    }
    if (status == calogOkE) {
        status = calogRegisterInline(calog, "dbClose", dbClose, gDbLib);
    }
    if (status != calogOkE) {
        // Roll back the refCount taken above so a partially-registered runtime does not
        // keep the process-wide registry alive past its own shutdown.
        calogDbShutdown();
        return status;
    }
    return calogAtDestroy(calog, calogDbShutdown, calogDestroyAfterContextsE);
}


void calogDbShutdown(void) {
    pthread_mutex_lock(&gDbLibMutex);
    if (gDbLib == NULL) {
        pthread_mutex_unlock(&gDbLibMutex);
        return;
    }
    gDbLib->refCount--;
    if (gDbLib->refCount <= 0) {
        calogHandleTableDestroy(gDbLib->handles, dbCloser);
        free(gDbLib);
        gDbLib = NULL;
    }
    pthread_mutex_unlock(&gDbLibMutex);
}


static int32_t dbClose(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    DbLibT  *lib;
    void    *conn;
    uint32_t backend;

    lib = (DbLibT *)userData;
    calogValueNil(result);
    if (argCount != 1 || args[0].type != calogIntE) {
        return calogFail(result, calogErrArgE, "dbClose expects a connection handle");
    }
    // A single atomic remove yields the connection and its backend tag; a concurrent dbClose
    // of the same handle gets NULL here and must not double-close.
    conn = calogHandleRemoveAny(lib->handles, args[0].as.i, &backend);
    if (conn == NULL) {
        return calogFail(result, calogErrArgE, "dbClose: invalid connection handle");
    }
    dbCloser(backend, conn);
    return calogOkE;
}


static void dbCloser(uint32_t type, void *resource) {
    switch (type) {
#ifdef CALOG_WITH_SQLITE
        case DB_TYPE_SQLITE:
            sqlite3_close((sqlite3 *)resource);
            break;
#endif
#ifdef CALOG_WITH_PG
        case DB_TYPE_PG:
            PQfinish((PGconn *)resource);
            break;
#endif
#ifdef CALOG_WITH_MYSQL
        case DB_TYPE_MYSQL:
            mysql_close((MYSQL *)resource);
            break;
#endif
        default:
            break;
    }
}


static int32_t dbExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    DbLibT  *lib;
    void    *conn;
    uint32_t backend;

    lib = (DbLibT *)userData;
    calogValueNil(result);
    if (argCount < 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "dbExec expects (handle, sql, ...params)");
    }
    backend = dbResolve(lib, args[0].as.i, &conn);
    switch (backend) {
#ifdef CALOG_WITH_SQLITE
        case DB_TYPE_SQLITE:
            return sqliteExec((sqlite3 *)conn, &args[1], &args[2], argCount - 2, result);
#endif
#ifdef CALOG_WITH_PG
        case DB_TYPE_PG:
            return pgExec((PGconn *)conn, &args[1], &args[2], argCount - 2, result);
#endif
#ifdef CALOG_WITH_MYSQL
        case DB_TYPE_MYSQL:
            return mysqlExec((MYSQL *)conn, &args[1], &args[2], argCount - 2, result);
#endif
        default:
            return calogFail(result, calogErrArgE, "dbExec: invalid connection handle");
    }
}


// Finish a successful backend connect: register it in the handle table (or close it back
// out on out-of-memory) and set result to the new handle. Shared tail of sqliteOpen,
// pgOpen, and mysqlOpen.
static int32_t dbFinishOpen(DbLibT *lib, uint32_t type, void *conn, CalogValueT *result) {
    int64_t handle;

    handle = calogHandleAdd(lib->handles, type, conn);
    if (handle == 0) {
        dbCloser(type, conn);
        return calogFail(result, calogErrOomE, "dbOpen: out of memory");
    }
    calogValueInt(result, handle);
    return calogOkE;
}


// Marshal one result row into a {columnName: value} map and push it onto rows. Shared row
// loop body of mysqlQuery/pgQuery/sqliteQuery: nameFn/valueFn hide the backend-specific
// column access behind ctx, so all three copies of this ~20-line block (and the error-unwind
// paths that came with them, including the uninitialized-value free bug the copies had
// drifted into) collapse to one.
static int32_t dbMarshalRow(CalogValueT *result, CalogAggT *rows, void *ctx, int32_t columnCount, DbColumnNameFnT nameFn, DbColumnValueFnT valueFn) {
    CalogAggT  *rowMap;
    CalogValueT rowValue;
    int32_t     column;
    int32_t     status;

    status = calogAggCreate(&rowMap, calogMapE);
    if (status != calogOkE) {
        return calogFail(result, status, "dbQuery: out of memory");
    }
    for (column = 0; column < columnCount; column++) {
        CalogValueT key;
        CalogValueT value;
        const char *name;
        int64_t     nameLength;

        calogValueNil(&value);
        nameFn(ctx, column, &name, &nameLength);
        status = calogValueString(&key, name, nameLength);
        if (status == calogOkE) {
            status = valueFn(ctx, column, &value);
        }
        if (status == calogOkE) {
            status = calogAggSet(rowMap, &key, &value);
        }
        if (status != calogOkE) {
            calogValueFree(&key);
            calogValueFree(&value);
            calogAggFree(rowMap);
            return calogFail(result, status, "dbQuery: failed to build a row");
        }
    }
    calogValueAgg(&rowValue, rowMap);
    status = calogAggPush(rows, &rowValue);
    if (status != calogOkE) {
        calogValueFree(&rowValue);
        return calogFail(result, status, "dbQuery: out of memory");
    }
    return calogOkE;
}


static int32_t dbOpen(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    DbLibT     *lib;
    const char *driver;

    lib = (DbLibT *)userData;
    calogValueNil(result);
    if (argCount != 2 || args[0].type != calogStringE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "dbOpen expects (driver, connection)");
    }
    driver = args[0].as.s.bytes;
#ifdef CALOG_WITH_SQLITE
    if (strcmp(driver, "sqlite") == 0) {
        return sqliteOpen(lib, args[1].as.s.bytes, result);
    }
#endif
#ifdef CALOG_WITH_PG
    if (strcmp(driver, "postgres") == 0) {
        return pgOpen(lib, args[1].as.s.bytes, result);
    }
#endif
#ifdef CALOG_WITH_MYSQL
    if (strcmp(driver, "mysql") == 0) {
        return mysqlOpen(lib, args[1].as.s.bytes, result);
    }
#endif
    return calogFail(result, calogErrUnsupportedE, "dbOpen: unknown or uncompiled driver");
}


static int32_t dbQuery(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
    DbLibT  *lib;
    void    *conn;
    uint32_t backend;

    lib = (DbLibT *)userData;
    calogValueNil(result);
    if (argCount < 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
        return calogFail(result, calogErrArgE, "dbQuery expects (handle, sql, ...params)");
    }
    backend = dbResolve(lib, args[0].as.i, &conn);
    switch (backend) {
#ifdef CALOG_WITH_SQLITE
        case DB_TYPE_SQLITE:
            return sqliteQuery((sqlite3 *)conn, &args[1], &args[2], argCount - 2, result);
#endif
#ifdef CALOG_WITH_PG
        case DB_TYPE_PG:
            return pgQuery((PGconn *)conn, &args[1], &args[2], argCount - 2, result);
#endif
#ifdef CALOG_WITH_MYSQL
        case DB_TYPE_MYSQL:
            return mysqlQuery((MYSQL *)conn, &args[1], &args[2], argCount - 2, result);
#endif
        default:
            return calogFail(result, calogErrArgE, "dbQuery: invalid connection handle");
    }
}


// Resolve a handle to its backend type tag and connection pointer, or 0 if not found.
static uint32_t dbResolve(DbLibT *lib, int64_t handle, void **connOut) {
    uint32_t type;

    // One locked scan yields both the connection and its backend tag; this table holds only
    // this library's own connections, so the returned tag is trustworthy.
    *connOut = calogHandleGetAny(lib->handles, handle, &type);
    return type;
}


#ifdef CALOG_WITH_MYSQL
// Stable storage for a bound scalar parameter during mysql_stmt_execute.
typedef union MysqlScalarT {
    long long i;
    double    d;
    char      b;
} MysqlScalarT;


// Row context bound to mysqlRowName/mysqlRowValue for one mysqlQuery call.
typedef struct MysqlRowCtxT {
    MYSQL_FIELD   *fields;
    char         **buffers;
    unsigned long *lengths;
    my_bool       *isNull;
} MysqlRowCtxT;


// Free everything a query allocated: the per-column buffers, the three parallel arrays, the
// result metadata, and the statement. Each argument may be NULL.
static void mysqlCleanup(MYSQL_BIND *binds, char **buffers, unsigned long *lengths, my_bool *isNull, unsigned int columnCount, MYSQL_RES *meta, MYSQL_STMT *stmt) {
    unsigned int column;

    if (buffers != NULL) {
        for (column = 0; column < columnCount; column++) {
            free(buffers[column]);
        }
    }
    free(binds);
    free(buffers);
    free(lengths);
    free(isNull);
    if (meta != NULL) {
        mysql_free_result(meta);
    }
    mysql_stmt_close(stmt);
}


// Marshal a MySQL cell (fetched as its string form) to a CalogValueT, typed by the field's
// SQL type: integer types -> int, float/double -> real, everything else (text, blob,
// decimal, dates) -> a binary-safe string. bytes is NUL-terminated at [length].
static int32_t mysqlColumn(enum enum_field_types type, const char *bytes, unsigned long length, CalogValueT *out) {
    calogValueNil(out);
    switch (type) {
        case MYSQL_TYPE_TINY:
        case MYSQL_TYPE_SHORT:
        case MYSQL_TYPE_LONG:
        case MYSQL_TYPE_LONGLONG:
        case MYSQL_TYPE_INT24:
        case MYSQL_TYPE_YEAR:
            calogValueInt(out, (int64_t)strtoll(bytes, NULL, 10));
            return calogOkE;
        case MYSQL_TYPE_FLOAT:
        case MYSQL_TYPE_DOUBLE:
            calogValueReal(out, strtod(bytes, NULL));
            return calogOkE;
        default:
            return calogValueString(out, bytes, (int64_t)length);
    }
}


static int32_t mysqlExec(MYSQL *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    MYSQL_STMT *stmt;

    stmt = mysqlPrepare(conn, sqlValue, params, paramCount, result);
    if (stmt == NULL) {
        return calogErrArgE;
    }
    calogValueInt(result, (int64_t)mysql_stmt_affected_rows(stmt));
    mysql_stmt_close(stmt);
    return calogOkE;
}


// A libpq-style "key=value key=value" conninfo, so the driver's connection string is
// uniform: host, user, password, dbname, port, socket.
static int32_t mysqlOpen(DbLibT *lib, const char *conninfo, CalogValueT *result) {
    MYSQL       *conn;
    char         buffer[MYSQL_CONNINFO_CAP];
    char        *host;
    char        *user;
    char        *password;
    char        *dbname;
    char        *socket;
    char        *sslmode;
    char        *token;
    char        *saveptr;
    unsigned int port;
    my_bool      sslEnforce;
    my_bool      sslVerify;

    host     = NULL;
    user     = NULL;
    password = NULL;
    dbname   = NULL;
    socket   = NULL;
    sslmode  = NULL;
    saveptr  = NULL;
    port     = 0;
    if (strlen(conninfo) >= sizeof(buffer)) {
        return calogFail(result, calogErrArgE, "dbOpen: mysql connection string too long");
    }
    snprintf(buffer, sizeof(buffer), "%s", conninfo);
    token = strtok_r(buffer, " ", &saveptr);
    while (token != NULL) {
        char *equals;
        equals = strchr(token, '=');
        if (equals != NULL) {
            *equals = '\0';
            if (strcmp(token, "host") == 0) {
                host = equals + 1;
            } else if (strcmp(token, "user") == 0) {
                user = equals + 1;
            } else if (strcmp(token, "password") == 0) {
                password = equals + 1;
            } else if (strcmp(token, "dbname") == 0) {
                dbname = equals + 1;
            } else if (strcmp(token, "socket") == 0) {
                socket = equals + 1;
            } else if (strcmp(token, "sslmode") == 0) {
                sslmode = equals + 1;
            } else if (strcmp(token, "port") == 0) {
                port = (unsigned int)strtoul(equals + 1, NULL, 10);
            }
        }
        token = strtok_r(NULL, " ", &saveptr);
    }
    conn = mysql_init(NULL);
    if (conn == NULL) {
        return calogFail(result, calogErrOomE, "dbOpen: mysql_init failed");
    }
    // The connector otherwise defaults to require-TLS + verify, which fails against a plain
    // server. sslmode selects the posture (default "prefer"):
    //   "disable" -> plain, no TLS.
    //   "prefer"  -> use TLS if the server offers it, else fall back to plain; no certificate
    //                verification (matches libpq's default sslmode=prefer).
    //   "require" -> require TLS and verify the server certificate.
    // Any other value is REJECTED, so an explicit stronger request is never silently
    // downgraded to the unverified default.
    if (sslmode == NULL || strcmp(sslmode, "prefer") == 0) {
        sslEnforce = 1;
        sslVerify  = 0;
    } else if (strcmp(sslmode, "disable") == 0) {
        sslEnforce = 0;
        sslVerify  = 0;
    } else if (strcmp(sslmode, "require") == 0) {
        sslEnforce = 1;
        sslVerify  = 1;
    } else {
        mysql_close(conn);
        return calogFail(result, calogErrArgE, "dbOpen: mysql sslmode must be 'disable', 'prefer', or 'require'");
    }
    mysql_options(conn, MYSQL_OPT_SSL_ENFORCE, &sslEnforce);
    mysql_options(conn, MYSQL_OPT_SSL_VERIFY_SERVER_CERT, &sslVerify);
    if (mysql_real_connect(conn, host, user, password, dbname, port, socket, 0) == NULL) {
        int32_t status;
        status = calogFail(result, calogErrArgE, mysql_error(conn));
        mysql_close(conn);
        return status;
    }
    return dbFinishOpen(lib, DB_TYPE_MYSQL, conn, result);
}


// Init, prepare, bind params as ?, and execute. Returns the executed statement (the caller
// closes it) or NULL with the error set into result.
static MYSQL_STMT *mysqlPrepare(MYSQL *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    MYSQL_STMT    *stmt;
    MYSQL_BIND    *binds;
    MysqlScalarT  *scalars;
    unsigned long *lengths;
    int32_t        index;
    int            rc;

    stmt = mysql_stmt_init(conn);
    if (stmt == NULL) {
        calogFail(result, calogErrOomE, "dbExec/dbQuery: mysql_stmt_init failed");
        return NULL;
    }
    if (mysql_stmt_prepare(stmt, sqlValue->as.s.bytes, (unsigned long)sqlValue->as.s.length) != 0) {
        calogFail(result, calogErrArgE, mysql_stmt_error(stmt));
        mysql_stmt_close(stmt);
        return NULL;
    }
    if (paramCount == 0) {
        if (mysql_stmt_execute(stmt) != 0) {
            calogFail(result, calogErrArgE, mysql_stmt_error(stmt));
            mysql_stmt_close(stmt);
            return NULL;
        }
        return stmt;
    }
    binds   = (MYSQL_BIND *)calloc((size_t)paramCount, sizeof(MYSQL_BIND));
    scalars = (MysqlScalarT *)calloc((size_t)paramCount, sizeof(MysqlScalarT));
    lengths = (unsigned long *)calloc((size_t)paramCount, sizeof(unsigned long));
    if (binds == NULL || scalars == NULL || lengths == NULL) {
        free(binds);
        free(scalars);
        free(lengths);
        mysql_stmt_close(stmt);
        calogFail(result, calogErrOomE, "dbExec/dbQuery: out of memory");
        return NULL;
    }
    for (index = 0; index < paramCount; index++) {
        const CalogValueT *param;
        param = &params[index];
        switch (param->type) {
            case calogNilE:
                binds[index].buffer_type = MYSQL_TYPE_NULL;
                break;
            case calogBoolE:
                scalars[index].b         = param->as.b ? 1 : 0;
                binds[index].buffer_type = MYSQL_TYPE_TINY;
                binds[index].buffer      = &scalars[index].b;
                break;
            case calogIntE:
                scalars[index].i         = (long long)param->as.i;
                binds[index].buffer_type = MYSQL_TYPE_LONGLONG;
                binds[index].buffer      = &scalars[index].i;
                break;
            case calogRealE:
                scalars[index].d         = param->as.r;
                binds[index].buffer_type = MYSQL_TYPE_DOUBLE;
                binds[index].buffer      = &scalars[index].d;
                break;
            case calogStringE:
                lengths[index]             = (unsigned long)param->as.s.length;
                binds[index].buffer_type   = MYSQL_TYPE_STRING;
                binds[index].buffer        = (void *)param->as.s.bytes;
                binds[index].buffer_length = lengths[index];
                binds[index].length        = &lengths[index];
                break;
            default:
                free(binds);
                free(scalars);
                free(lengths);
                mysql_stmt_close(stmt);
                calogFail(result, calogErrTypeE, "dbExec/dbQuery: a bound parameter must be nil, bool, int, real, or string");
                return NULL;
        }
    }
    rc = mysql_stmt_bind_param(stmt, binds);
    if (rc == 0) {
        rc = mysql_stmt_execute(stmt);
    }
    free(binds);
    free(scalars);
    free(lengths);
    if (rc != 0) {
        calogFail(result, calogErrArgE, mysql_stmt_error(stmt));
        mysql_stmt_close(stmt);
        return NULL;
    }
    return stmt;
}


static int32_t mysqlQuery(MYSQL *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    MYSQL_STMT    *stmt;
    MYSQL_RES     *meta;
    MYSQL_FIELD   *fields;
    MYSQL_BIND    *binds;
    char         **buffers;
    unsigned long *lengths;
    my_bool       *isNull;
    my_bool        updateMaxLength;
    CalogAggT     *rows;
    MysqlRowCtxT   rowCtx;
    unsigned int   columnCount;
    unsigned int   column;
    int32_t        status;
    int            rc;

    stmt = mysqlPrepare(conn, sqlValue, params, paramCount, result);
    if (stmt == NULL) {
        return calogErrArgE;
    }
    meta = mysql_stmt_result_metadata(stmt);
    if (meta == NULL) {
        // No result set; yield an empty list.
        status = calogAggCreate(&rows, calogListE);
        if (status != calogOkE) {
            mysql_stmt_close(stmt);
            return calogFail(result, status, "dbQuery: out of memory");
        }
        calogValueAgg(result, rows);
        mysql_stmt_close(stmt);
        return calogOkE;
    }
    // Buffer the full result client-side with max_length computed, so each column can bind a
    // real, correctly-sized string buffer -- the connector then converts every value
    // (including numerics) to its string form in a single fetch.
    updateMaxLength = 1;
    mysql_stmt_attr_set(stmt, STMT_ATTR_UPDATE_MAX_LENGTH, &updateMaxLength);
    if (mysql_stmt_store_result(stmt) != 0) {
        status = calogFail(result, calogErrArgE, mysql_stmt_error(stmt));
        mysqlCleanup(NULL, NULL, NULL, NULL, 0, meta, stmt);
        return status;
    }
    columnCount = mysql_num_fields(meta);
    fields      = mysql_fetch_fields(meta);
    binds   = (MYSQL_BIND *)calloc(columnCount, sizeof(MYSQL_BIND));
    buffers = (char **)calloc(columnCount, sizeof(char *));
    lengths = (unsigned long *)calloc(columnCount, sizeof(unsigned long));
    isNull  = (my_bool *)calloc(columnCount, sizeof(my_bool));
    if (binds == NULL || buffers == NULL || lengths == NULL || isNull == NULL) {
        mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
        return calogFail(result, calogErrOomE, "dbQuery: out of memory");
    }
    for (column = 0; column < columnCount; column++) {
        unsigned long size;
        size             = (unsigned long)fields[column].max_length + 1;
        buffers[column]  = (char *)malloc(size);
        if (buffers[column] == NULL) {
            mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
            return calogFail(result, calogErrOomE, "dbQuery: out of memory");
        }
        binds[column].buffer_type   = MYSQL_TYPE_STRING;
        binds[column].buffer        = buffers[column];
        binds[column].buffer_length = size;
        binds[column].length        = &lengths[column];
        binds[column].is_null       = &isNull[column];
    }
    if (mysql_stmt_bind_result(stmt, binds) != 0) {
        status = calogFail(result, calogErrArgE, mysql_stmt_error(stmt));
        mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
        return status;
    }
    status = calogAggCreate(&rows, calogListE);
    if (status != calogOkE) {
        mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
        return calogFail(result, status, "dbQuery: out of memory");
    }
    rowCtx.fields  = fields;
    rowCtx.buffers = buffers;
    rowCtx.lengths = lengths;
    rowCtx.isNull  = isNull;
    while ((rc = mysql_stmt_fetch(stmt)) == 0 || rc == MYSQL_DATA_TRUNCATED) {
        status = dbMarshalRow(result, rows, &rowCtx, (int32_t)columnCount, mysqlRowName, mysqlRowValue);
        if (status != calogOkE) {
            calogAggFree(rows);
            mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
            return status;
        }
    }
    if (rc != MYSQL_NO_DATA) {
        status = calogFail(result, calogErrArgE, mysql_stmt_error(stmt));
        calogAggFree(rows);
        mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
        return status;
    }
    mysqlCleanup(binds, buffers, lengths, isNull, columnCount, meta, stmt);
    calogValueAgg(result, rows);
    return calogOkE;
}


// dbMarshalRow callbacks bound to a MysqlRowCtxT for one mysqlQuery call.
static void mysqlRowName(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut) {
    MysqlRowCtxT *rowCtx;

    rowCtx         = (MysqlRowCtxT *)ctx;
    *nameOut       = rowCtx->fields[column].name;
    *nameLengthOut = (int64_t)strlen(rowCtx->fields[column].name);
}


static int32_t mysqlRowValue(void *ctx, int32_t column, CalogValueT *out) {
    MysqlRowCtxT *rowCtx;

    rowCtx = (MysqlRowCtxT *)ctx;
    if (rowCtx->isNull[column]) {
        return calogOkE;
    }
    rowCtx->buffers[column][rowCtx->lengths[column]] = '\0';
    return mysqlColumn(rowCtx->fields[column].type, rowCtx->buffers[column], rowCtx->lengths[column], out);
}
#endif


#ifdef CALOG_WITH_PG
// Row context bound to pgRowName/pgRowValue for one pgQuery call.
typedef struct PgRowCtxT {
    PGresult *res;
    int       row;
} PgRowCtxT;


// Marshal a Postgres cell (text result format) to a CalogValueT: typed for the common
// numeric/bool OIDs, bytea decoded from its "\x.." hex, everything else as its text bytes.
static int32_t pgColumn(PGresult *res, int row, int column, CalogValueT *out) {
    const char *text;
    int         length;
    Oid         oid;

    calogValueNil(out);
    if (PQgetisnull(res, row, column)) {
        return calogOkE;
    }
    text   = PQgetvalue(res, row, column);
    length = PQgetlength(res, row, column);
    oid    = PQftype(res, column);
    switch (oid) {
        case PG_OID_INT2:
        case PG_OID_INT4:
        case PG_OID_INT8:
            calogValueInt(out, (int64_t)strtoll(text, NULL, 10));
            return calogOkE;
        case PG_OID_FLOAT4:
        case PG_OID_FLOAT8:
            calogValueReal(out, strtod(text, NULL));
            return calogOkE;
        case PG_OID_BOOL:
            calogValueBool(out, text[0] == 't');
            return calogOkE;
        case PG_OID_BYTEA:
            if (length >= 2 && text[0] == '\\' && text[1] == 'x') {
                int32_t  status;
                char    *bytes;
                int      count;
                int      index;
                count = (length - 2) / 2;
                bytes = (char *)malloc((size_t)(count > 0 ? count : 1));
                if (bytes == NULL) {
                    return calogErrOomE;
                }
                for (index = 0; index < count; index++) {
                    bytes[index] = (char)(strtol((char[]){ text[2 + index * 2], text[3 + index * 2], 0 }, NULL, 16));
                }
                status = calogValueString(out, bytes, (int64_t)count);
                free(bytes);
                return status;
            }
            return calogValueString(out, text, (int64_t)length);
        default:
            return calogValueString(out, text, (int64_t)length);
    }
}


static int32_t pgExec(PGconn *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    PGresult *res;

    res = pgRun(conn, sqlValue, params, paramCount, result);
    if (res == NULL) {
        return calogErrArgE;
    }
    calogValueInt(result, (int64_t)strtoll(PQcmdTuples(res), NULL, 10));
    PQclear(res);
    return calogOkE;
}


static int32_t pgOpen(DbLibT *lib, const char *conninfo, CalogValueT *result) {
    PGconn *conn;

    conn = PQconnectdb(conninfo);
    if (PQstatus(conn) != CONNECTION_OK) {
        int32_t status;
        status = calogFail(result, calogErrArgE, PQerrorMessage(conn));
        PQfinish(conn);
        return status;
    }
    return dbFinishOpen(lib, DB_TYPE_PG, conn, result);
}


static int32_t pgQuery(PGconn *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    PGresult  *res;
    CalogAggT *rows;
    PgRowCtxT  rowCtx;
    int        rowCount;
    int        columnCount;
    int        row;
    int32_t    status;

    res = pgRun(conn, sqlValue, params, paramCount, result);
    if (res == NULL) {
        return calogErrArgE;
    }
    status = calogAggCreate(&rows, calogListE);
    if (status != calogOkE) {
        PQclear(res);
        return calogFail(result, status, "dbQuery: out of memory");
    }
    rowCount    = PQntuples(res);
    columnCount = PQnfields(res);
    rowCtx.res  = res;
    for (row = 0; row < rowCount; row++) {
        rowCtx.row = row;
        status     = dbMarshalRow(result, rows, &rowCtx, columnCount, pgRowName, pgRowValue);
        if (status != calogOkE) {
            calogAggFree(rows);
            PQclear(res);
            return status;
        }
    }
    calogValueAgg(result, rows);
    PQclear(res);
    return calogOkE;
}


// dbMarshalRow callbacks bound to a PgRowCtxT for one pgQuery call.
static void pgRowName(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut) {
    PgRowCtxT  *rowCtx;
    const char *name;

    rowCtx         = (PgRowCtxT *)ctx;
    name           = PQfname(rowCtx->res, column);
    *nameOut       = name;
    *nameLengthOut = (int64_t)strlen(name);
}


static int32_t pgRowValue(void *ctx, int32_t column, CalogValueT *out) {
    PgRowCtxT *rowCtx;

    rowCtx = (PgRowCtxT *)ctx;
    return pgColumn(rowCtx->res, rowCtx->row, column, out);
}


// Run a parameterized statement ($1..$N): nil -> SQL NULL; bool as 't'/'f'; ints/reals
// formatted as text; strings bound in BINARY format with an explicit length, so they are
// binary-safe (embedded NULs survive, unlike NUL-terminated text params). Always requests a
// text-format result (pgColumn only ever decodes text). On any error the message is copied
// into result via calogFail and NULL is returned.
static PGresult *pgRun(PGconn *conn, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    const char **values;
    int         *lengths;
    int         *formats;
    char        *scratch;
    PGresult    *res;
    int32_t      index;

    values  = NULL;
    lengths = NULL;
    formats = NULL;
    scratch = NULL;
    if (paramCount > 0) {
        values  = (const char **)calloc((size_t)paramCount, sizeof(char *));
        lengths = (int *)calloc((size_t)paramCount, sizeof(int));
        formats = (int *)calloc((size_t)paramCount, sizeof(int));
        scratch = (char *)malloc((size_t)paramCount * DB_NUM_SCRATCH);
        if (values == NULL || lengths == NULL || formats == NULL || scratch == NULL) {
            free((void *)values);
            free(lengths);
            free(formats);
            free(scratch);
            calogFail(result, calogErrOomE, "dbExec/dbQuery: out of memory");
            return NULL;
        }
    }
    for (index = 0; index < paramCount; index++) {
        const CalogValueT *param;
        char              *buffer;
        param  = &params[index];
        buffer = &scratch[index * DB_NUM_SCRATCH];
        switch (param->type) {
            case calogNilE:
                values[index] = NULL;
                break;
            case calogBoolE:
                values[index] = param->as.b ? "t" : "f";
                break;
            case calogIntE:
                snprintf(buffer, DB_NUM_SCRATCH, "%lld", (long long)param->as.i);
                values[index] = buffer;
                break;
            case calogRealE:
                snprintf(buffer, DB_NUM_SCRATCH, "%.17g", param->as.r);
                values[index] = buffer;
                break;
            case calogStringE:
                if (param->as.s.length > INT32_MAX) {
                    free((void *)values);
                    free(lengths);
                    free(formats);
                    free(scratch);
                    calogFail(result, calogErrRangeE, "dbExec/dbQuery: a bound string is too long");
                    return NULL;
                }
                values[index]  = param->as.s.bytes;
                lengths[index] = (int)param->as.s.length;
                formats[index] = 1;
                break;
            default:
                free((void *)values);
                free(lengths);
                free(formats);
                free(scratch);
                calogFail(result, calogErrTypeE, "dbExec/dbQuery: a bound parameter must be nil, bool, int, real, or string");
                return NULL;
        }
    }
    // Always text-format result (0): pgColumn only decodes text.
    res = PQexecParams(conn, sqlValue->as.s.bytes, paramCount, NULL, values, lengths, formats, 0);
    free((void *)values);
    free(lengths);
    free(formats);
    free(scratch);
    if (PQresultStatus(res) != PGRES_COMMAND_OK && PQresultStatus(res) != PGRES_TUPLES_OK) {
        calogFail(result, calogErrArgE, PQerrorMessage(conn));
        PQclear(res);
        return NULL;
    }
    return res;
}
#endif


#ifdef CALOG_WITH_SQLITE
static int32_t sqliteColumn(sqlite3_stmt *stmt, int column, CalogValueT *out) {
    calogValueNil(out);
    switch (sqlite3_column_type(stmt, column)) {
        case SQLITE_INTEGER:
            calogValueInt(out, (int64_t)sqlite3_column_int64(stmt, column));
            return calogOkE;
        case SQLITE_FLOAT:
            calogValueReal(out, sqlite3_column_double(stmt, column));
            return calogOkE;
        case SQLITE_TEXT:
            return calogValueString(out, (const char *)sqlite3_column_text(stmt, column), (int64_t)sqlite3_column_bytes(stmt, column));
        case SQLITE_BLOB:
            return calogValueString(out, (const char *)sqlite3_column_blob(stmt, column), (int64_t)sqlite3_column_bytes(stmt, column));
        case SQLITE_NULL:
        default:
            return calogOkE;
    }
}


static int32_t sqliteExec(sqlite3 *db, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    sqlite3_stmt *stmt;
    int32_t       status;
    int           rc;

    status = sqlitePrepare(db, sqlValue, params, paramCount, &stmt, result);
    if (status != calogOkE) {
        return status;
    }
    do {
        rc = sqlite3_step(stmt);
    } while (rc == SQLITE_ROW);
    if (rc != SQLITE_DONE) {
        status = calogFail(result, calogErrArgE, sqlite3_errmsg(db));
        sqlite3_finalize(stmt);
        return status;
    }
    calogValueInt(result, (int64_t)sqlite3_changes(db));
    sqlite3_finalize(stmt);
    return calogOkE;
}


static int32_t sqliteOpen(DbLibT *lib, const char *path, CalogValueT *result) {
    sqlite3 *db;

    db = NULL;
    if (sqlite3_open(path, &db) != SQLITE_OK) {
        int32_t status;
        status = calogFail(result, calogErrArgE, sqlite3_errmsg(db));
        sqlite3_close(db);
        return status;
    }
    return dbFinishOpen(lib, DB_TYPE_SQLITE, db, result);
}


// Prepare sql and bind params[0..paramCount) as ?1..?N. Binary-safe strings bind as text
// with an explicit length (preserving embedded NULs); an aggregate/function param errors.
static int32_t sqlitePrepare(sqlite3 *db, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, sqlite3_stmt **out, CalogValueT *result) {
    sqlite3_stmt *stmt;
    int32_t       index;

    *out = NULL;
    if (sqlValue->as.s.length > INT32_MAX) {
        return calogFail(result, calogErrRangeE, "dbExec/dbQuery: sql text is too long");
    }
    if (sqlite3_prepare_v2(db, sqlValue->as.s.bytes, (int)sqlValue->as.s.length, &stmt, NULL) != SQLITE_OK) {
        return calogFail(result, calogErrArgE, sqlite3_errmsg(db));
    }
    for (index = 0; index < paramCount; index++) {
        const CalogValueT *param;
        int                rc;
        param = &params[index];
        switch (param->type) {
            case calogNilE:
                rc = sqlite3_bind_null(stmt, index + 1);
                break;
            case calogBoolE:
                rc = sqlite3_bind_int(stmt, index + 1, param->as.b ? 1 : 0);
                break;
            case calogIntE:
                rc = sqlite3_bind_int64(stmt, index + 1, param->as.i);
                break;
            case calogRealE:
                rc = sqlite3_bind_double(stmt, index + 1, param->as.r);
                break;
            case calogStringE:
                if (param->as.s.length > INT32_MAX) {
                    sqlite3_finalize(stmt);
                    return calogFail(result, calogErrRangeE, "dbExec/dbQuery: a bound string is too long");
                }
                rc = sqlite3_bind_text(stmt, index + 1, param->as.s.bytes, (int)param->as.s.length, SQLITE_TRANSIENT);
                break;
            default:
                sqlite3_finalize(stmt);
                return calogFail(result, calogErrTypeE, "dbExec/dbQuery: a bound parameter must be nil, bool, int, real, or string");
        }
        if (rc != SQLITE_OK) {
            int32_t status;
            status = calogFail(result, calogErrArgE, sqlite3_errmsg(db));
            sqlite3_finalize(stmt);
            return status;
        }
    }
    *out = stmt;
    return calogOkE;
}


static int32_t sqliteQuery(sqlite3 *db, const CalogValueT *sqlValue, const CalogValueT *params, int32_t paramCount, CalogValueT *result) {
    sqlite3_stmt *stmt;
    CalogAggT    *rows;
    int32_t       columns;
    int32_t       status;
    int           rc;

    status = sqlitePrepare(db, sqlValue, params, paramCount, &stmt, result);
    if (status != calogOkE) {
        return status;
    }
    status = calogAggCreate(&rows, calogListE);
    if (status != calogOkE) {
        sqlite3_finalize(stmt);
        return calogFail(result, status, "dbQuery: out of memory");
    }
    columns = sqlite3_column_count(stmt);
    while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
        status = dbMarshalRow(result, rows, stmt, columns, sqliteRowName, sqliteRowValue);
        if (status != calogOkE) {
            calogAggFree(rows);
            sqlite3_finalize(stmt);
            return status;
        }
    }
    if (rc != SQLITE_DONE) {
        status = calogFail(result, calogErrArgE, sqlite3_errmsg(db));
        calogAggFree(rows);
        sqlite3_finalize(stmt);
        return status;
    }
    calogValueAgg(result, rows);
    sqlite3_finalize(stmt);
    return calogOkE;
}


// dbMarshalRow callbacks bound directly to the sqlite3_stmt for one sqliteQuery call.
static void sqliteRowName(void *ctx, int32_t column, const char **nameOut, int64_t *nameLengthOut) {
    const char *name;

    name           = sqlite3_column_name((sqlite3_stmt *)ctx, column);
    *nameOut       = name;
    *nameLengthOut = (int64_t)strlen(name);
}


static int32_t sqliteRowValue(void *ctx, int32_t column, CalogValueT *out) {
    return sqliteColumn((sqlite3_stmt *)ctx, column, out);
}
#endif