mx_mysql.c

#define _GNU_SOURCE

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <assert.h>
#include <stdarg.h>

#include <unistd.h>

#include <mysql.h>
#include <mysqld_error.h>
#include <errmsg.h>

#include <errno.h>

#include <time.h>
#include <limits.h>

#include "mx_mysql.h"
#include "mx_util.h"
#include "mx_log.h"

static struct mx_mysql_stmt *mx_mysql_statement_prepare_with_bindings(struct mx_mysql *mysql, char *statement, struct mx_mysql_bind *param, struct mx_mysql_bind *result);

/**********************************************************************/

static char* mx_mysql_last_error;

__attribute__((destructor))
static void mx_mysql_exit(void) {
    free(mx_mysql_last_error);
}

static void mx_mysql_save_error(const char *err) {
    size_t len = strlen(err);
    mx_mysql_last_error = realloc(mx_mysql_last_error, len+1);
    if (mx_mysql_last_error == NULL) {
        mx_log_crit("out of memory");
        return;
    }
    strncpy(mx_mysql_last_error, err, len+1);
}

__attribute__ ((format (printf, 1, 2)))
static void mx_mysql_save_error_va(const char *restrict fmt, ...) {
    char *s;
    va_list ap;
    va_start(ap, fmt);
    int res = vasprintf(&s, fmt, ap);
    va_end(ap);
    if (res == -1)
        return;
    mx_mysql_save_error(s);
    free(s);
}

char *mx_mysql_error(void) {
    return mx_mysql_last_error == NULL ? "no error information" : mx_mysql_last_error;
}

static int mx__mysql_init(struct mx_mysql *mysql)
{
    assert(mysql);
    assert(!mysql->mysql);

    mysql->mysql = mysql_init(NULL);
    if (mysql->mysql)
        return 0;
    mx_mysql_save_error("out of memory");
    return -ENOMEM;
}

static int mx__mysql_options(struct mx_mysql *mysql, enum mysql_option option, const void *arg)
{
    int res;

    assert(mysql);
    assert(mysql->mysql);

    res = mysql_options(mysql->mysql, option, arg);
    if (res == 0)
        return 0;
    mx_mysql_save_error(mysql_error(mysql->mysql));
    return -EBADE;
}

static int mx__mysql_real_connect(struct mx_mysql *mysql, const char *host, const char *user, const char *passwd, const char *db, unsigned int port, const char *unix_socket, unsigned long client_flag)
{
    MYSQL *m;

    assert(mysql);
    assert(mysql->mysql);

    m = mysql_real_connect(mysql->mysql, host, user, passwd, db, port, unix_socket, client_flag);
    if (m)
        return 0;

    mx_mysql_save_error(mysql_error(mysql->mysql));

    switch (mysql_errno(mysql->mysql)) {
        case CR_ALREADY_CONNECTED:
            return -EALREADY;

        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_CONN_HOST_ERROR:
        case CR_CONNECTION_ERROR:
        case CR_IPSOCK_ERROR:
        case CR_SOCKET_CREATE_ERROR:
        case CR_UNKNOWN_HOST:
        case CR_VERSION_ERROR:
        case CR_SERVER_LOST:
        case CR_SERVER_GONE_ERROR:
            return -EAGAIN;
    }
    return -EBADE;
}

static int mx__mysql_ping(struct mx_mysql *mysql)
{
    assert(mysql);
    assert(mysql->mysql);

    int res;

    res = mysql_ping(mysql->mysql);
    if (res == 0)
        return 0;

    mx_mysql_save_error(mysql_error(mysql->mysql));

    switch (mysql_errno(mysql->mysql)) {
        case CR_COMMANDS_OUT_OF_SYNC:
            return -EPROTO;

        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_CONN_HOST_ERROR:
        case CR_CONNECTION_ERROR:
        case CR_IPSOCK_ERROR:
        case CR_SOCKET_CREATE_ERROR:
        case CR_UNKNOWN_HOST:
        case CR_VERSION_ERROR:
        case CR_SERVER_LOST:
        case CR_SERVER_GONE_ERROR:
            return -EAGAIN;

        case CR_UNKNOWN_ERROR:
            return -EIO;
    }
    return -EBADE;
}

static int mx__mysql_stmt_init(struct mx_mysql_stmt *stmt)
{
    assert(stmt);
    assert(!stmt->stmt);
    assert(stmt->mysql);

    stmt->stmt = mysql_stmt_init(stmt->mysql->mysql);
    if (stmt->stmt)
        return 0;
    mx_mysql_save_error("out of memory");
    return -ENOMEM;
}

static int mx__mysql_stmt_prepare(struct mx_mysql_stmt *stmt, char *statement)
{
    int res;

    assert(stmt);
    assert(statement);
    assert(*statement);

    assert(!stmt->statement);

    assert(stmt->mysql);
    assert(stmt->stmt);

    res = mysql_stmt_prepare(stmt->stmt, statement, strlen(statement));
    if (res == 0) {
        stmt->statement = statement;
        return 0;
    }
    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_COMMANDS_OUT_OF_SYNC:
            return -EPROTO;

        case CR_SERVER_GONE_ERROR:
        case CR_SERVER_LOST:
            return -EAGAIN;

        case CR_UNKNOWN_ERROR:
            return -EIO;

        case ER_PARSE_ERROR:
            return -EBADRQC;

    }
    return -EBADE;
}

static int mx__mysql_stmt_bind_param(struct mx_mysql_stmt *stmt)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = (int)mysql_stmt_bind_param(stmt->stmt, stmt->param.bind);
    if (res == 0)
        return 0;

    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_UNSUPPORTED_PARAM_TYPE:
            return -EPROTO;

        case CR_UNKNOWN_ERROR:
            return -EIO;
    }
    return -EBADE;
}

static int mx__mysql_stmt_bind_result(struct mx_mysql_stmt *stmt)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = (int)mysql_stmt_bind_result(stmt->stmt, stmt->result.bind);
    if (res == 0)
        return 0;

    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_UNSUPPORTED_PARAM_TYPE:
            return -EPROTO;

        case CR_UNKNOWN_ERROR:
            return -EIO;
    }
    return -EBADE;
}

static int mx__mysql_stmt_execute(struct mx_mysql_stmt *stmt)
{
    assert(stmt);
    assert(stmt->stmt);

    do {
        if (mysql_stmt_execute(stmt->stmt) == 0)
            return 0;
    } while (mysql_stmt_errno(stmt->stmt) == ER_LOCK_DEADLOCK);

    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_COMMANDS_OUT_OF_SYNC:
            return -EPROTO;

        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_SERVER_GONE_ERROR:
        case CR_SERVER_LOST:
            return -EAGAIN;

        case CR_UNKNOWN_ERROR:
            return -EIO;
    }
    return -EBADE;
}

static int mx__mysql_stmt_store_result(struct mx_mysql_stmt *stmt)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = mysql_stmt_store_result(stmt->stmt);
    if (res == 0)
        return 0;

    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_COMMANDS_OUT_OF_SYNC:
            return -EPROTO;

        case CR_OUT_OF_MEMORY:
            return -ENOMEM;

        case CR_SERVER_GONE_ERROR:
        case CR_SERVER_LOST:
            return -EAGAIN;

        case CR_UNKNOWN_ERROR:
            return -EIO;
    }
    return -EBADE;
}

static int mx__mysql_stmt_free_result(struct mx_mysql_stmt *stmt)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = (int)mysql_stmt_free_result(stmt->stmt);
    if (res == 0)
        return 0;

    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));
    return -EBADE;
}

static int mx__mysql_stmt_fetch(struct mx_mysql_stmt *stmt)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = mysql_stmt_fetch(stmt->stmt);
    if (res == 0)
        return 0;

    if (res == 1) {
        mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

        switch (mysql_stmt_errno(stmt->stmt)) {
            case CR_OUT_OF_MEMORY:
                return -ENOMEM;

            case CR_SERVER_GONE_ERROR:
            case CR_SERVER_LOST:
                return -EAGAIN;

            case CR_UNKNOWN_ERROR:
                return -EIO;

            case CR_COMMANDS_OUT_OF_SYNC:
            case CR_UNSUPPORTED_PARAM_TYPE:
                return -EPROTO;
        }
        return -EBADE;
    }

    switch (res) {
        case MYSQL_NO_DATA:
            return -ENOENT;

        case MYSQL_DATA_TRUNCATED:
            return -ERANGE;
    }
    return -EBADE;
}

static int mx__mysql_stmt_fetch_column(struct mx_mysql_stmt *stmt, unsigned int column, unsigned long offset)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = mysql_stmt_fetch_column(stmt->stmt, &(stmt->result.bind[column]), column, offset);
    if (res == 0)
        return 0;
    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_INVALID_PARAMETER_NO:
            return -EPROTO;

        case CR_NO_DATA:
            return -ENOENT;
    }
    return -EBADE;
}

static int mx__mysql_stmt_param_count(struct mx_mysql_stmt *stmt)
{
    unsigned long count;

    assert(stmt);
    assert(stmt->stmt);

    count = mysql_stmt_param_count(stmt->stmt);
    assert(count <= INT_MAX);

    /* no mysql errors possible */
    return (int)count;
}

static int mx__mysql_stmt_field_count(struct mx_mysql_stmt *stmt)
{
    unsigned long count;

    assert(stmt);
    assert(stmt->stmt);

    count = mysql_stmt_field_count(stmt->stmt);
    assert(count <= INT_MAX);

    /* no mysql errors possible */
    return (int)count;
}

static int mx__mysql_stmt_affected_rows(struct mx_mysql_stmt *stmt, unsigned long long *count)
{
    my_ulonglong c;

    assert(stmt);
    assert(stmt->stmt);

    c = mysql_stmt_affected_rows(stmt->stmt);

    *count = (unsigned long long)c;

    /* no mysql errors possible */
    return 0;
}

static int mx__mysql_stmt_insert_id(struct mx_mysql_stmt *stmt, unsigned long long *count)
{
    my_ulonglong c;

    assert(stmt);
    assert(stmt->stmt);

    c = mysql_stmt_insert_id(stmt->stmt);

    *count = (unsigned long long)c;

    /* no mysql errors possible */
    return 0;
}

static int mx__mysql_stmt_close(struct mx_mysql_stmt *stmt)
{
    bool res;

    assert(stmt);

    res = mysql_stmt_close(stmt->stmt);
    if (res == 0) {
        stmt->stmt = NULL;
        return 0;
    }
    mx_mysql_save_error(mysql_stmt_error(stmt->stmt));

    switch (mysql_stmt_errno(stmt->stmt)) {
        case CR_SERVER_GONE_ERROR:
            return -ECONNABORTED;

        case CR_UNKNOWN_ERROR :
            return -EIO;
    }
    return -EBADE;
}

static int mx__mysql_close(struct mx_mysql *mysql) {
    assert(mysql);

    if (mysql->mysql) {
        mysql_close(mysql->mysql);
        mysql->mysql = NULL;
    }

    /* no mysql errors possible */
    return 0;
}

static int mx__mysql_library_end(void) {
    mysql_library_end();
    /* no mysql errors possible */
    return 0;
}

/**********************************************************************/

static int _mx_mysql_bind_integer(struct mx_mysql_bind *b, unsigned int index, void *value, int type, int is_unsigned)
{
    assert(b);
    assert(value);

    assert(index < b->count);

    assert(!(b->data[index].flags));

    memset(&(b->bind[index]), 0, sizeof(b->bind[index]));

    b->bind[index].buffer_type = (enum enum_field_types)type;
    b->bind[index].buffer      = value;
    b->bind[index].is_unsigned = (bool)is_unsigned;
    b->bind[index].length      = &(b->data[index].length);
    b->bind[index].is_null     = &(b->data[index].is_null);
    b->bind[index].error       = &(b->data[index].is_error);

    b->data[index].flags = 1;

    return 0;
}

void mx_mysql_bind_string(struct mx_mysql_bind *b, unsigned int index, char **value)
{
    assert(b);
    assert(value);

    assert(index < b->count);

    assert(!(b->data[index].flags));

    assert((*value && b->type == MX_MYSQL_BIND_TYPE_PARAM) || (!*value && b->type == MX_MYSQL_BIND_TYPE_RESULT));

    memset(&(b->bind[index]), 0, sizeof(b->bind[index]));

    if (b->type == MX_MYSQL_BIND_TYPE_PARAM) {
        b->data[index].string_ptr = value;
        b->data[index].length     = strlen(*value);

        b->bind[index].buffer_type   = MYSQL_TYPE_STRING;
        b->bind[index].buffer        = *(b->data[index].string_ptr);
        b->bind[index].buffer_length = b->data[index].length;
        b->bind[index].length        = &(b->data[index].length);
        b->bind[index].is_null       = &(b->data[index].is_null);
        b->bind[index].error         = &(b->data[index].is_error);
    } else {
        b->data[index].string_ptr = value;
        b->data[index].length     = 0;

        b->bind[index].buffer_type   = MYSQL_TYPE_STRING;
        b->bind[index].buffer        = NULL;
        b->bind[index].buffer_length = 0;
        b->bind[index].length        = &(b->data[index].length);
        b->bind[index].is_null       = &(b->data[index].is_null);
        b->bind[index].error         = &(b->data[index].is_error);
    }

    b->data[index].flags = 1;
}

static int _mx_mysql_bind_validate(struct mx_mysql_bind *b)
{
    assert(b);

    for (unsigned long i=0; i < b->count; i++) {
        if (!(b->data[i].flags)) {
            mx_mysql_save_error("binding not fully initialized");
            return -EBADSLT;
        }
    }

    return 0;
}

/**********************************************************************/

int mx_mysql_initialize(struct mx_mysql **mysql)
{
    struct mx_mysql *m;

    assert(mysql);
    assert(!(*mysql));

    m = mx_calloc_forever(1, sizeof(*m));

    *mysql = m;

    return 0;
}

static int mx_mysql_init(struct mx_mysql *mysql)
{
    int res;

    assert(mysql);

    do {
        res = mx__mysql_init(mysql);
        if (res == 0)
            break;

        if (res != -ENOMEM)
            return res;

        mx_log_debug("mx__mysql_init() failed: %s - retrying (forever) in %d second(s).", strerror(-res), MX_MYSQL_FAIL_WAIT_DEFAULT);
        mx_sleep(MX_MYSQL_FAIL_WAIT_DEFAULT);

    } while (1);

    return 0;
}

int mx_mysql_option_set_default_file(struct mx_mysql *mysql, char *fname)
{
    assert(mysql);

    if (fname && (*fname == '/') && (euidaccess(fname, R_OK) != 0)) {
        mx_mysql_save_error_va("%s: %m - falling back to mysql default config search path", fname);
        return -errno;
    }

    if (fname && !(*fname))
        fname = NULL;

    mysql->default_file = fname;

    return 0;
}

void mx_mysql_option_set_default_group(struct mx_mysql *mysql, char *group)
{
    assert(mysql);

    if (group && !(*group))
        group = NULL;

    mysql->default_group = group;
}

void mx_mysql_option_set_reconnect(struct mx_mysql *mysql, int reconnect)
{
    assert(mysql);

    mysql->reconnect = (bool)!!reconnect;
}

static int mx_mysql_real_connect(struct mx_mysql *mysql, const char *host, const char *user, const char *passwd, const char *db, unsigned int port, const char *unix_socket, unsigned long client_flag)
{
    int res;

    assert(mysql);
    assert(mysql->mysql);

    if (mysql->default_file) {
        res = mx__mysql_options(mysql, MYSQL_READ_DEFAULT_FILE, mysql->default_file);
        mx_mysql_assert_usage_ok(res);
    }

    if (mysql->default_group) {
        res = mx__mysql_options(mysql, MYSQL_READ_DEFAULT_GROUP, mysql->default_group);
        mx_mysql_assert_usage_ok(res);
    } else {
        res = mx__mysql_options(mysql, MYSQL_READ_DEFAULT_GROUP, program_invocation_short_name);
        mx_mysql_assert_usage_ok(res);
    }
    res = mx__mysql_options(mysql, MYSQL_OPT_RECONNECT, &mysql->reconnect);
    mx_mysql_assert_usage_ok(res);

    res = mx__mysql_real_connect(mysql, host, user, passwd, db, port, unix_socket, client_flag);
    mx_mysql_assert_usage_ok(res);

    if (res == 0)
        return 0;

    if (res == -EALREADY) {
        mx_log_debug("WARNING: %s", mysql_error(mysql->mysql));
        return 0;
    }

    return res;
}

int mx_mysql_connect(struct mx_mysql **mysql)
{
    int res;

    assert(mysql);

    if (!(*mysql)) {
        res = mx_mysql_initialize(mysql);
        if (res < 0)
            return res;
    }
    if (!(*mysql)->mysql) {
        res = mx_mysql_init(*mysql);
        if (res < 0)
            return res;
    }

    res = mx_mysql_real_connect(*mysql, NULL, NULL, NULL, NULL, 0, NULL, 0);
    return res;
}

int mx_mysql_connect_forever_sec(struct mx_mysql **mysql, unsigned int seconds)
{
    int res;

    while ((res = mx_mysql_connect(mysql)) < 0) {
        mx_mysql_assert_usage_ok(res);
        mx_log_warning("mx_mysql_connect() failed: %s - retrying (forever) in %d second(s).", mx_mysql_error(), seconds);
        mx_sleep(seconds);
    }

    return 0;
}

int mx_mysql_disconnect(struct mx_mysql *mysql) {
    assert(mysql);

    return mx__mysql_close(mysql);
}

static int mx_mysql_end(void) {
    return mx__mysql_library_end();
}

static int mx_mysql_free(struct mx_mysql **mysql)
{
    assert(mysql);
    assert(*mysql);
    assert(!((*mysql)->mysql));

    mx_free_null(*mysql);

    return 0;
}

int mx_mysql_finish(struct mx_mysql **mysql)
{
    int res = 0;
    int res1 = 0, res2 = 0, res3 = 0;

    if (mysql && *mysql) {
        res1 = mx_mysql_disconnect(*mysql);
        if (res1 < 0)
            res = res1;

        res2 = mx_mysql_free(mysql);
        if (!res && res2 < 0)
            res = res2;
    }
    res3 = mx_mysql_end();
    if (!res && res3 < 0)
        res = res3;

    return res;
}


static int mx_mysql_ping(struct mx_mysql *mysql)
{
    assert(mysql);

    return mx__mysql_ping(mysql);
}

static int mx_mysql_ping_forever(struct mx_mysql *mysql)
{
    int res;
    int fail = 0;

    assert(mysql);

    while (1) {
        res = mx_mysql_ping(mysql);
        if (res == 0)
            break;

        fail++;

        mx_mysql_assert_usage_ok(res);
        mx_log_warning("mx_mysql_ping() failed: %s - retrying again (forever) in %d second(s).", mx_mysql_error(), MX_MYSQL_FAIL_WAIT_DEFAULT);
        mx_sleep(MX_MYSQL_FAIL_WAIT_DEFAULT);
    }

    if (fail)
        mx_log_info("mx_mysql_ping_forever() recovered from previous errors (%d tries). Yippieh! Back to work!", fail);

    return res;
}

static int mx_mysql_statement_init(struct mx_mysql *mysql, struct mx_mysql_stmt **stmt)
{
    struct mx_mysql_stmt *s;
    int res;

    assert(stmt);
    assert(mysql);
    assert(!(*stmt));

    s = mx_calloc_forever(1, sizeof(*s));

    s->mysql       = mysql;

    do {
        res = mx__mysql_stmt_init(s);
        if (res == 0)
            break;

        if (res != -ENOMEM)
            return res;

        mx_log_debug("mx__mysql_stmt_init() failed: %s - retrying (forever) in %d second(s).", strerror(-res), MX_MYSQL_FAIL_WAIT_DEFAULT);
        mx_sleep(MX_MYSQL_FAIL_WAIT_DEFAULT);
    } while (1);

    *stmt = s;

    return 0;
}

int mx_mysql_statement_execute(struct mx_mysql_stmt *stmt, unsigned long long *count)
{
    int res;

    assert(stmt);
    assert(stmt->stmt);

    res = _mx_mysql_bind_validate(&stmt->param);
    if (res < 0) {
        return res;
    }

    res = mx__mysql_stmt_bind_param(stmt);
    if (res < 0) {
        mx_log_debug("ERROR: mx__mysql_stmt_bind_param: %s", strerror(-res));
        return res;
    }

    res = mx__mysql_stmt_execute(stmt);
    if (res < 0) {
        mx_log_debug("ERROR: mx__mysql_stmt_execute: %s", strerror(-res));
        return res;
    }

    res = mx__mysql_stmt_store_result(stmt);
    if (res < 0) {
        mx_log_debug("ERROR: mx__mysql_stmt_store_result: %s", strerror(-res));
        return res;
    }

    if (count) {
        res = mx__mysql_stmt_affected_rows(stmt, count);
        if (res < 0) {
            mx_log_debug("ERROR: mx__mysql_stmt_affected_rows(): %s", strerror(-res));
            return res;
        }
    }

    return 0;
}

int mx_mysql_statement_insert_id(struct mx_mysql_stmt *stmt, unsigned long long int *id) {
    return mx__mysql_stmt_insert_id(stmt, id);
}

int mx_mysql_statement_fetch(struct mx_mysql_stmt *stmt)
{
    struct mx_mysql_bind *r;
    int res;
    char *str;
    int no_error = 1;

    assert(stmt);
    assert(stmt->stmt);

    res = _mx_mysql_bind_validate(&stmt->result);
    if (res < 0) {
        mx_log_debug("ERROR: result not initialized completely.");
        return res;
    }

    res = mx__mysql_stmt_bind_result(stmt);
    if (res < 0) {
        mx_log_debug("ERROR: mx__mysql_stmt_bind_result: %s", strerror(-res));
        return res;
    }

    res = mx__mysql_stmt_fetch(stmt);
    if (res == -ENOENT)
        return 0;

    if (res < 0 && res != -ERANGE) {
        mx_log_debug("ERROR: mx__mysql_stmt_fetch: %s", strerror(-res));
        return res;
    }

    r = &stmt->result;
    for (unsigned long col = 0; col < r->count; col++) {
        if (r->bind[col].buffer_type == MYSQL_TYPE_STRING) {
            str = mx_calloc_forever(r->data[col].length + 1, sizeof(*str));

            *(r->data[col].string_ptr) = str;
            r->bind[col].buffer        = *(r->data[col].string_ptr);
            r->bind[col].buffer_length =   r->data[col].length;

            mx__mysql_stmt_fetch_column(stmt, col, 0);

            r->data[col].length        = 0;
            r->bind[col].buffer        = NULL;
            r->bind[col].buffer_length = 0;
            continue;
        }

        if (!(r->data[col].is_error))
            continue;

        mx_log_debug("WARNING: result data returned in column with index %lu was truncated. query was:", col);
        mx_log_debug("       \\ %s", stmt->statement);
        no_error = 0;
    }

    if (!no_error)
        return -ERANGE;

    return 0;
}

int mx_mysql_statement_param_count(struct mx_mysql_stmt *stmt)
{
    assert(stmt);

    return mx__mysql_stmt_param_count(stmt);
}

int mx_mysql_statement_field_count(struct mx_mysql_stmt *stmt)
{
    assert(stmt);
    return mx__mysql_stmt_field_count(stmt);
}

static int mx_mysql_stmt_field_count_set(struct mx_mysql_stmt *stmt)
{
    assert(stmt);
    assert(stmt->stmt);

    stmt->field_count = mysql_stmt_field_count(stmt->stmt);

    return 0;
}

static int mx_mysql_stmt_param_count_set(struct mx_mysql_stmt *stmt)
{
    assert(stmt);
    assert(stmt->stmt);

    stmt->param_count = mysql_stmt_param_count(stmt->stmt);

    return 0;
}

static int mx_mysql_bind_cleanup(struct mx_mysql_bind *bind)
{
    if (!bind)
        return 0;

    mx_free_null(bind->bind);
    mx_free_null(bind->data);

    bind->count = 0;

    return 0;
}

static void mx_mysql_bind_init_from(struct mx_mysql_bind *bind, unsigned long count, enum mx_mysql_bind_type type, struct mx_mysql_bind *from)
{
    assert(bind);

    assert(!bind->count);
    assert(!bind->bind);
    assert(!bind->data);

    if (from) {
        assert(count == from->count);
        assert(type  == from->type);
        assert(from->bind);
        assert(from->data);
        memcpy(bind, from, sizeof(*bind));
        return;
    }

    mx_mysql_bind_init(bind, count, type);
}

void mx_mysql_bind_init(struct mx_mysql_bind *bind, unsigned long count, enum mx_mysql_bind_type type)
{
    assert(bind);

    assert(!bind->count);
    assert(!bind->bind);
    assert(!bind->data);

    bind->type  = type;
    bind->count = count;

    if (!count)
        return;

    bind->bind  = mx_calloc_forever(bind->count, sizeof(*bind->bind));
    bind->data  = mx_calloc_forever(bind->count, sizeof(*bind->data));
}

static int _mx_mysql_do_statement(struct mx_mysql *mysql, char *query, struct mx_mysql_bind *param, struct mx_mysql_bind *result, void *from, void **to, size_t size, char cleanup)
{
    struct mx_mysql_stmt *stmt = NULL;
    unsigned long long num_rows = 0;
    int res;
    char *tmpdata;

    assert(mysql);

    mx_mysql_ping_forever(mysql);

    stmt = mx_mysql_statement_prepare_with_bindings(mysql, query, param, result);
    if (!stmt) {
        if (cleanup) {
            mx_mysql_bind_cleanup(param);
            mx_mysql_bind_cleanup(result);
        }
        return -EINVAL;
    }

    res = mx_mysql_statement_execute(stmt, &num_rows);
    if (res < 0) {
        if (cleanup)
            mx_mysql_statement_close(&stmt);
        else
            mx_mysql_statement_close_no_bind_cleanup(&stmt);
        return res;
    }

    if (result && result->count && num_rows) {
        tmpdata = mx_calloc_forever(num_rows, size);

        for (unsigned long cnt = 0; cnt < num_rows; cnt++) {
            res = mx_mysql_statement_fetch(stmt);
            if (res < 0) {
                mx_free_null(tmpdata);
                if (cleanup)
                    mx_mysql_statement_close(&stmt);
                else
                    mx_mysql_statement_close_no_bind_cleanup(&stmt);
                return res;
            }
            memcpy(tmpdata+(cnt*size), from, size);
        }
        *to = tmpdata;
    }
    if (cleanup)
        mx_mysql_statement_close(&stmt);
    else
        mx_mysql_statement_close_no_bind_cleanup(&stmt);

    return num_rows;
}

int mx_mysql_do_statement(struct mx_mysql *mysql, char *query, struct mx_mysql_bind *param, struct mx_mysql_bind *result, void *from, void **to, size_t size)
{
    return _mx_mysql_do_statement(mysql, query, param, result, from, to, size, 1);
}

static int mx_mysql_do_statement_no_bind_cleanup(struct mx_mysql *mysql, char *query, struct mx_mysql_bind *param, struct mx_mysql_bind *result, void *from, void **to, size_t size)
{
    return _mx_mysql_do_statement(mysql, query, param, result, from, to, size, 0);
}

int mx_mysql_do_statement_retry_on_fail(struct mx_mysql *mysql, char *query, struct mx_mysql_bind *param, struct mx_mysql_bind *result, void *from, void **to, size_t size)
{
    int res;

    while (1) {
        res = mx_mysql_do_statement_no_bind_cleanup(mysql, query, param, result, from, to, size);

        if (res >= 0)
            break;

        mx_mysql_assert_usage_ok(res);

        mx_log_warning("mx_mysql_do_statement() failed: %s", strerror(-res));

        if (res != -EAGAIN)
            break;

        mx_mysql_ping_forever(mysql);
    }
    mx_mysql_bind_cleanup(param);
    mx_mysql_bind_cleanup(result);

    return res;
}


static struct mx_mysql_stmt *mx_mysql_statement_prepare_with_bindings(struct mx_mysql *mysql, char *statement, struct mx_mysql_bind *param, struct mx_mysql_bind *result)
{
    int res;
    struct mx_mysql_stmt *stmt = NULL;

    assert(mysql);
    assert(statement);
    assert(*statement);

    res = mx_mysql_statement_init(mysql, &stmt);
    if (res < 0)
        return NULL;

    while (1) {
        res = mx__mysql_stmt_prepare(stmt, statement);
        if (res < 0)
            break;

        res = mx_mysql_stmt_param_count_set(stmt);
        if (res < 0)
            break;

        res = mx_mysql_stmt_field_count_set(stmt);
        if (res < 0)
            break;

        mx_mysql_bind_init_from(&stmt->param, stmt->param_count, MX_MYSQL_BIND_TYPE_PARAM, param);
        mx_mysql_bind_init_from(&stmt->result, stmt->field_count, MX_MYSQL_BIND_TYPE_RESULT, result);

        return stmt;
    };

    mx_mysql_statement_close_no_bind_cleanup(&stmt);

    return NULL;
}

struct mx_mysql_stmt *mx_mysql_statement_prepare(struct mx_mysql *mysql, char *statement)
{
    assert(mysql);
    assert(statement);
    assert(*statement);

    return mx_mysql_statement_prepare_with_bindings(mysql, statement, NULL, NULL);
}

int mx_mysql_statement_close(struct mx_mysql_stmt **stmt)
{
    if (*stmt == NULL)
        return 0;

    mx__mysql_stmt_free_result(*stmt);
    mx__mysql_stmt_close(*stmt);

    mx_mysql_bind_cleanup(&(*stmt)->param);
    mx_mysql_bind_cleanup(&(*stmt)->result);
    mx_free_null(*stmt);

    return 0;
}

int mx_mysql_statement_close_no_bind_cleanup(struct mx_mysql_stmt **stmt)
{
    if (*stmt == NULL)
        return 0;

    mx__mysql_stmt_free_result(*stmt);
    mx__mysql_stmt_close(*stmt);

    mx_free_null(*stmt);

    return 0;
}

static int mx_mysql_bind_integer(struct mx_mysql_bind *b, unsigned int index, void *value, int type, int is_unsigned)
{
    int res;

    res = _mx_mysql_bind_integer(b, index, value, type, is_unsigned);
    if (res == 0)
        return 0;

    mx_log_debug("Failed to set index %d: %s", index, strerror(-res));
    return res;
}

void mx_mysql_bind_int8(struct mx_mysql_bind *b, unsigned int index, int8_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_TINY, 0);
}

void mx_mysql_bind_uint8(struct mx_mysql_bind *b, unsigned int index, uint8_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_TINY, 1);
}

void mx_mysql_bind_int16(struct mx_mysql_bind *b, unsigned int index, int16_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_SHORT, 0);
}

void mx_mysql_bind_uint16(struct mx_mysql_bind *b, unsigned int index, uint16_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_SHORT, 1);
}

void mx_mysql_bind_int32(struct mx_mysql_bind *b, unsigned int index, int32_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_LONG, 0);
}

void mx_mysql_bind_uint32(struct mx_mysql_bind *b, unsigned int index, uint32_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_LONG, 1);
}

void mx_mysql_bind_int64(struct mx_mysql_bind *b, unsigned int index, int64_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_LONGLONG, 0);
}

void mx_mysql_bind_uint64(struct mx_mysql_bind *b, unsigned int index, uint64_t *value)
{
    mx_mysql_bind_integer(b, index, (void *)value, MYSQL_TYPE_LONGLONG, 1);
}