/*
 * Copyright (C) 2018 Andreas Granig (sipwise.com)
 *
 * This file is part of Kamailio, a free SIP server.
 *
 * Kamailio is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version
 *
 * Kamailio is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 */

#include <stdlib.h>
#include <time.h>

#include "db_redis_mod.h"
#include "redis_connection.h"
#include "redis_dbase.h"
#include "redis_table.h"

#define TIMESTAMP_STR_LENGTH 19

static void db_redis_dump_reply(redisReply *reply) {
    int i;
    if (reply->type == REDIS_REPLY_STRING) {
        LM_DBG("%s\n", reply->str);
    } else if (reply->type == REDIS_REPLY_INTEGER) {
        LM_DBG("%lld\n", reply->integer);
    } else if (reply->type == REDIS_REPLY_NIL) {
        LM_DBG("<null>\n");
    } else if (reply->type == REDIS_REPLY_ARRAY) {
        LM_DBG("printing %lu elements in array reply\n", (unsigned long) reply->elements);
        for(i = 0; i < reply->elements; ++i) {
            db_redis_dump_reply(reply->element[i]);
        }
    } else {
        LM_DBG("not printing invalid reply type\n");
    }
}

// TODO: utilize auto-expiry? on insert/update, also update expire value
// of mappings

/*
 * Initialize database module
 * No function should be called before this
 */
db1_con_t* db_redis_init(const str* _url) {
    return db_do_init(_url, (void *)db_redis_new_connection);
}

/*
 * Shut down database module
 * No function should be called after this
 */
void db_redis_close(db1_con_t* _h) {
    LM_DBG("closing redis db connection\n");
    db_do_close(_h, db_redis_free_connection);
}


static db1_res_t* db_redis_new_result(void) {
    db1_res_t* obj;

    obj = db_new_result();
    if (!obj)
        return NULL;
    return obj;
}

static int db_redis_val2str(const db_val_t *v, str *_str) {
    const char *s;
    const str *tmpstr;
    int vtype = VAL_TYPE(v);
    struct tm _time;
    _str->s = NULL;
    _str->len = 32; // default for numbers

    if (VAL_NULL(v)) {
        LM_DBG("converting <null> value to str\n");
        _str->len = 0;
        return 0;
    }

    switch (vtype) {
        case DB1_INT:
            LM_DBG("converting int value %d to str\n", VAL_INT(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%d", VAL_INT(v));
            _str->len = strlen(_str->s);
            break;
       case DB1_UINT:
            LM_DBG("converting uint value %u to str\n", VAL_UINT(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%u", VAL_UINT(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_BIGINT:
            LM_DBG("converting bigint value %lld to str\n", VAL_BIGINT(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%010lld", VAL_BIGINT(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_UBIGINT:
            LM_DBG("converting ubigint value %llu to str\n", VAL_UBIGINT(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%010llu", VAL_UBIGINT(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_STRING:
            s = VAL_STRING(v);
            _str->len = strlen(s);
            LM_DBG("converting string value '%s' with len %d to str\n", s, _str->len);
            _str->s = (char*)pkg_malloc(_str->len + 1);
            if (!_str->s) goto memerr;
            //memcpy(_str->s, s, _str->len);
            //_str->s[_str->len] = '\0';
            memset(_str->s, 0, _str->len + 1);
            strncpy(_str->s, s, _str->len);
            break;
        case DB1_STR:
            tmpstr = &(VAL_STR(v));
            LM_DBG("converting str value '%.*s' with len %d to str\n", tmpstr->len, tmpstr->s, tmpstr->len);
            // copy manually to add 0 termination
            _str->s = (char*)pkg_malloc(tmpstr->len + 1);
            if (!_str->s) goto memerr;
            _str->len = tmpstr->len;
            memcpy(_str->s, tmpstr->s, _str->len);
            _str->s[_str->len] = '\0';
            break;
        case DB1_DATETIME:
            LM_DBG("converting datetime value %ld to str\n", VAL_TIME(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            localtime_r(&(VAL_TIME(v)), &_time);
            strftime(_str->s, _str->len, "%Y-%m-%d %H:%M:%S", &_time);
            _str->len = strlen(_str->s);
            break;
        case DB1_DOUBLE:
            LM_DBG("converting double value %f to str\n", VAL_DOUBLE(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%.6f", VAL_DOUBLE(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_BITMAP:
            LM_DBG("converting bitmap value %u to str\n", VAL_BITMAP(v));
            _str->s = (char*)pkg_malloc(_str->len);
            if (!_str->s) goto memerr;
            snprintf(_str->s, _str->len, "%u", VAL_BITMAP(v));
            _str->len = strlen(_str->s);
            break;
        case DB1_BLOB:
        default:
            LM_ERR("Unsupported val type %d\n", vtype);
            goto err;
    }

    return 0;

memerr:
    LM_ERR("Failed to allocate memory to convert value to string\n");
err:
    return -1;
}

static int db_redis_return_version(const db1_con_t* _h, km_redis_con_t *con, const str *table_name,
    db1_res_t **_r) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    db_val_t* dval;
    db_row_t* drow;

    LM_DBG("get table version\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define it in schema file!\n",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    *_r = db_redis_new_result();
    if (!*_r) {
        LM_ERR("Failed to allocate memory for result");
        return -1;
    }
    RES_NUM_ROWS(*_r) = 1;
    RES_COL_N(*_r) = 1;
    RES_ROW_N(*_r) = 1;
    if (db_allocate_rows(*_r) != 0) {
        LM_ERR("Failed to allocate memory for rows\n");
        goto err;
    }
    if (db_allocate_columns(*_r, 1) != 0) {
        LM_ERR("Failed to allocate memory for result columns");
        goto err;
    }

    drow = &(RES_ROWS(*_r)[0]);

    if (db_allocate_row(*_r, drow) != 0) {
        LM_ERR("Failed to allocate row %d\n", RES_NUM_ROWS(*_r));
        goto err;
    }

    dval = &(ROW_VALUES(drow)[0]);

    VAL_TYPE(dval) = DB1_INT;
    VAL_NULL(dval) = 0;
    VAL_INT(dval) = table->version;

    LM_DBG("returning short-cut table version %d for table '%.*s'",
        table->version, table_name->len, table_name->s);

    return 0;

err:
    if (*_r) db_redis_free_result((db1_con_t*)_h, *_r);
    return -1;
}

static int db_redis_build_entry_manual_keys(redis_table_t *table, const db_key_t *_k,
        const db_val_t *_v, const int _n, int **manual_keys, int *manual_key_count) {

    // TODO: we also put keys here which are already part of type mapping!
    // there must be removed for performance reasons

    redis_key_t *key = NULL;

    *manual_keys = (int*)pkg_malloc(_n * sizeof(int));
    if (! *manual_keys) {
        LM_ERR("Failed to allocate memory for manual key indices\n");
        goto err;
    }
    memset(*manual_keys, 0, _n * sizeof(int));
    *manual_key_count = 0;

    for (key = table->entry_keys; key; key = key->next) {
        int subkey_found = 0;
        int i;
        *manual_key_count = 0;
        LM_DBG("checking for existence of entry key '%.*s' in query to get manual key\n",
                key->key.len, key->key.s);
        for (i = 0; i < _n; ++i) {
            const db_key_t k = _k[i];
            if (!str_strcmp(&key->key, (str*)k)) {
                LM_DBG("found key in entry key\n");
                subkey_found = 1;
                break;
            } else {
                (*manual_keys)[*manual_key_count] = i;
                (*manual_key_count)++;
            }
        }
        if (!subkey_found) {
            break;
        }
    }
    return 0;

err:
    if (*manual_keys) {
        pkg_free(*manual_keys);
        *manual_keys = NULL;
    }
    return -1;
}

static int db_redis_find_query_key(redis_key_t *key, const str *table_name,
        redis_table_t *table,
        str *type_name, const db_key_t *_k, const db_val_t *_v, const db_op_t *_op, const int _n,
        str *key_name, int *key_found, uint64_t *ts_scan_start) {

    unsigned int len;
    str val = {NULL, 0};

    *key_found = 1;
    key_name->len = 0;
    key_name->s = NULL;

    for (; key; key = key->next) {
        int subkey_found = 0;
        int i;
        LM_DBG("checking for existence of entry key '%.*s' in query\n",
                key->key.len, key->key.s);
        for (i = 0; i < _n; ++i) {
            const db_key_t k = _k[i];
            const db_val_t v = _v[i];
            const db_op_t op = _op ? _op[i] : NULL;

            if (VAL_NULL(&v)) {
                LM_DBG("Skipping null value for given key '%.*s'\n",
                        k->len, k->s);
                break;
            } else if (op && strcmp(op, OP_EQ)
                    && !((VAL_TYPE(&v) == DB1_DATETIME || VAL_TYPE(&v) == DB1_BIGINT
                            || VAL_TYPE(&v) == DB1_UBIGINT)
                        && (!strcmp(op, OP_LT) || !strcmp(op, OP_GT)))) {
                LM_DBG("Skipping non-EQ op (%s) for given key '%.*s'\n",
                        op, k->len, k->s);
                break;
            } else if (!str_strcmp(&key->key, (str*)k)) {
                LM_DBG("found key in entry key\n");
                if (db_redis_val2str(&v, &val) != 0) goto err;
                if (val.s == NULL) {
                    LM_DBG("key value in entry key is null, skip key\n");
                    subkey_found = 0;
                    break;
                }
                if (!key_name->len) {
                    // <version>:<table_name>:<type>::<val>
                    len = table->version_code.len + table_name->len + 1 + type_name->len + 2 + val.len + 1; //snprintf writes term 0 char
                    key_name->s = (char*)pkg_malloc(len);
                    if (!key_name->s) {
                        LM_ERR("Failed to allocate key memory\n");
                        goto err;
                    }
                    snprintf(key_name->s, len, "%.*s%.*s:%.*s::%.*s",
                            table->version_code.len, table->version_code.s,
                            table_name->len, table_name->s,
                            type_name->len, type_name->s,
                            val.len, val.s);
                    key_name->len = len-1; // subtract the term 0 char

                } else {
                    // :<val>
                    key_name->s = (char*)pkg_realloc(key_name->s, key_name->len + val.len + 2);
                    if (!key_name->s) {
                        LM_ERR("Failed to allocate key memory\n");
                        goto err;
                    }
                    snprintf(key_name->s + key_name->len, 1 + val.len + 1, ":%.*s",
                            val.len, val.s);
                    key_name->len += (1 + val.len);
                }
                if (op && (VAL_TYPE(&v) == DB1_DATETIME || VAL_TYPE(&v) == DB1_BIGINT
                            || VAL_TYPE(&v) == DB1_UBIGINT)
                        && (!strcmp(op, OP_LT) || !strcmp(op, OP_GT))) {
                    // Special case: we support matching < or > against timestamps and ints using a special
                    // key scanning method. We do this only for a single timestamp/int occurance, and we
                    // still do a table scan, just not a full table scan.
                    if (!ts_scan_start) {
                        LM_DBG("key '%.*s' for type '%.*s' found as timestamp or int, but table scans "
                                "not supported, unable to use this type\n",
                                key->key.len, key->key.s, type_name->len, type_name->s);
                        break;
                    }
                    // ts_scan_start is: 31 bits of current full key length, 31 bits of this value length,
                    // one bit of directionality, one bit of length variable indicator
                    if (VAL_TYPE(&v) == DB1_DATETIME && *ts_scan_start == 0 && val.len == TIMESTAMP_STR_LENGTH) {
                        *ts_scan_start = key_name->len | ((uint64_t) TIMESTAMP_STR_LENGTH << 31);
                        if (!strcmp(op, OP_LT))
                            *ts_scan_start |= 0x8000000000000000ULL;
                        LM_DBG("preparing for timestamp range scan at key offset %llx\n",
                                    (unsigned long long) *ts_scan_start);
                        *key_found = 0; // this forces a table scan using the new match key
                    }
                    else if ((VAL_TYPE(&v) == DB1_BIGINT
                                || VAL_TYPE(&v) == DB1_UBIGINT) && *ts_scan_start == 0) {
                        *ts_scan_start = key_name->len | ((uint64_t) val.len << 31);
                        *ts_scan_start |= 0x4000000000000000ULL; // length is variable
                        if (!strcmp(op, OP_LT))
                            *ts_scan_start |= 0x8000000000000000ULL;
                        LM_DBG("preparing for int range scan at key offset %llx\n",
                                    (unsigned long long) *ts_scan_start);
                        *key_found = 0; // this forces a table scan using the new match key
                    }
                }
                LM_DBG("entry key so far is '%.*s'\n", key_name->len, key_name->s);
                subkey_found = 1;
                pkg_free(val.s);
                val.s = NULL;
                break;
            }
        }
        if (!subkey_found) {
            LM_DBG("key '%.*s' for type '%.*s' not found, unable to use this type\n",
                    key->key.len, key->key.s, type_name->len, type_name->s);
            if (key_name->s) {
                pkg_free(key_name->s);
                key_name->s = NULL;
                key_name->len = 0;
            }
            *key_found = 0;
            break;
        }
    }

    // for value-less master keys
    if (!key_name->len) {
        // <version>:<table_name>:<type>
        len = table->version_code.len + table_name->len + 1 + type_name->len + 1;
        key_name->s = (char*)pkg_malloc(len);
        if (!key_name->s) {
            LM_ERR("Failed to allocate key memory\n");
            goto err;
        }
        snprintf(key_name->s, len, "%.*s%.*s:%.*s",
                table->version_code.len, table->version_code.s,
                table_name->len, table_name->s,
                type_name->len, type_name->s);
        key_name->len = len-1;
    }

    return 0;

err:
    if (val.s)
        pkg_free(val.s);
    if(key_name->s) {
        pkg_free(key_name->s);
        key_name->s = NULL;
        key_name->len = 0;
    }
    return -1;
}

static int db_redis_build_entry_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const db_val_t *_v, const int _n,
        redis_key_t **keys, int *keys_count) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_key_t *key;
    int key_found;
    str type_name = str_init("entry");
    str keyname = {NULL, 0};

    LM_DBG("build entry keys\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;
    key = table->entry_keys;
    if (db_redis_find_query_key(key, table_name, table, &type_name, _k, _v, NULL, _n, &keyname, &key_found, NULL) != 0) {
        goto err;
    }
    if (key_found) {
        if (db_redis_key_add_str(keys, &keyname) != 0) {
            LM_ERR("Failed to add key string\n");
            goto err;
        }
        LM_DBG("found suitable entry key '%.*s' for query\n",
                (*keys)->key.len, (*keys)->key.s);
        *keys_count = 1;
    } else {
        LM_ERR("Failed to create direct entry key, no matching key definition\n");
        goto err;
    }
    if (keyname.s)
        pkg_free(keyname.s);

    return 0;

err:
    db_redis_key_free(keys);
    if (keyname.s)
        pkg_free(keyname.s);
    return -1;
}

static int db_redis_get_keys_for_all_types(km_redis_con_t *con, const str *table_name,
        redis_key_t **keys, int *keys_count) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_type_t *type;
    redis_key_t *key;

    *keys = NULL;
    *keys_count = 0;

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    for (type = table->types; type; type = type->next) {
        for (key = type->keys; key; key = key->next) {
            if (db_redis_key_add_str(keys, &key->key) != 0) {
                LM_ERR("Failed to add key string\n");
                goto err;
            }
            (*keys_count)++;
        }
    }

    return 0;

err:
    db_redis_key_free(keys);
    return -1;
}

static int db_redis_build_type_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const db_val_t *_v, const int _n,
        redis_key_t **keys, redis_key_t **set_keys, int *keys_count) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_type_t *type;
    redis_key_t *key;

    *keys = NULL;
    *keys_count = 0;

    LM_DBG("build type keys\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    for (type = table->types; type; type = type->next) {
        str *type_name = &(type->type);
        int key_found = 0;
        str keyname = {NULL, 0};
        key = type->keys;

        if (db_redis_find_query_key(key, table_name, table, &type->type, _k, _v, NULL, _n, &keyname, &key_found, NULL) != 0) {
            goto err;
        }
        if (key_found) {
            if (db_redis_key_add_str(keys, &keyname) != 0) {
                LM_ERR("Failed to add query key to key list\n");
                goto err;
            }
            (*keys_count)++;
            LM_DBG("found key '%.*s' for type '%.*s'\n",
                    keyname.len, keyname.s,
                    type_name->len, type_name->s);

            if (set_keys) {
                // add key for parent set
                // <version>:<table>::index::<type>
                pkg_free(keyname.s);
                keyname.len = table->version_code.len + table_name->len + 9 + type->type.len;
                keyname.s = pkg_malloc(keyname.len + 1);
                if (!keyname.s) {
                    LM_ERR("Failed to allocate memory for parent set key\n");
                    goto err;
                }
                sprintf(keyname.s, "%.*s%.*s::index::%.*s",
                        table->version_code.len, table->version_code.s,
                        table_name->len, table_name->s,
                        type->type.len, type->type.s);
                if (db_redis_key_add_str(set_keys, &keyname) != 0) {
                    LM_ERR("Failed to add query key to set key list\n");
                    goto err;
                }
            }
        }
        if (keyname.s)
            pkg_free(keyname.s);
    }

    return 0;

err:
    LM_ERR("Failed to get type key\n");
    db_redis_key_free(keys);
    return -1;
}

static int db_redis_build_query_keys(km_redis_con_t *con, const str *table_name,
        const db_key_t *_k, const db_val_t *_v, const db_op_t *_op, const int _n,
        redis_key_t **query_keys, int *query_keys_count, int **manual_keys, int *manual_keys_count,
        int *do_table_scan, uint64_t *ts_scan_start, str *ts_scan_key) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    redis_type_t *type;
    redis_key_t *key;
    str keyname;
    int key_found;
    redisReply *reply = NULL;
    str typename = str_init(REDIS_DIRECT_PREFIX);

    *query_keys = NULL;
    *query_keys_count = 0;
    *do_table_scan = 1;

    LM_DBG("build query keys\n");

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define in db_redis keys parameter!",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    // check if given keys directly match entry key
    keyname.s = NULL;
    keyname.len = 0;
    key = table->entry_keys;

    if (db_redis_find_query_key(key, table_name, table, &typename, _k, _v, _op, _n, &keyname, &key_found, NULL) != 0) {
        goto err;
    }
    if (key_found) {
        LM_DBG("found suitable entry key '%.*s' for query\n",
                keyname.len, keyname.s);
        if (db_redis_key_add_str(query_keys, &keyname) != 0) {
            LM_ERR("Failed to add key name to query keys\n");
            goto err;
        }
        *query_keys_count = 1;
        pkg_free(keyname.s);
        keyname.s = NULL;
    } else {
        if (keyname.s)
            pkg_free(keyname.s);
        keyname.s = NULL;
        LM_DBG("no direct entry key found, checking type keys\n");
        for (type = table->types; type; type = type->next) {
            key = type->keys;
            LM_DBG("checking type '%.*s'\n", type->type.len, type->type.s);
            if (db_redis_find_query_key(key, table_name, table, &type->type, _k, _v, _op, _n, &keyname,
                        &key_found, ts_scan_start) != 0) {
                goto err;
            }
            if (key_found) {
                redis_key_t *query_v = NULL;
                char *prefix = "SMEMBERS";

                if (db_redis_key_add_string(&query_v, prefix, strlen(prefix)) != 0) {
                    LM_ERR("Failed to add smembers command to query\n");
                    db_redis_key_free(&query_v);
                    goto err;
                }
                if (db_redis_key_add_str(&query_v, &keyname) != 0) {
                    LM_ERR("Failed to add key name to smembers query\n");
                    db_redis_key_free(&query_v);
                    goto err;
                }

                reply = db_redis_command_argv(con, query_v);
                pkg_free(keyname.s);
                keyname.s = NULL;
                db_redis_key_free(&query_v);
                db_redis_check_reply(con, reply, err);
                if (reply->type == REDIS_REPLY_ARRAY) {
                    if (reply->elements == 0) {
                        LM_DBG("type query returned empty list\n");
                        *query_keys_count = 0;
                        *do_table_scan = 0;
                        db_redis_free_reply(&reply);
                        break;
                    } else {
                        int i;
                        LM_DBG("populating query keys list with result of type query\n");
                        *query_keys_count = reply->elements;
                        for (i = 0; i < reply->elements; ++i) {
                            redisReply *subreply = reply->element[i];
                            if (subreply->type == REDIS_REPLY_STRING) {
                                LM_DBG("adding resulting entry key '%s' from type query\n", subreply->str);
                                if (db_redis_key_prepend_string(query_keys, subreply->str, strlen(subreply->str)) != 0) {
                                    LM_ERR("Failed to add query key\n");
                                    goto err;
                                }
                            } else {
                                LM_ERR("Unexpected entry key type in type query, expecting a string\n");
                                goto err;
                            }
                        }
                    }
                } else {
                    LM_ERR("Unexpected reply for type query, expecting an array\n");
                    goto err;
                }

                db_redis_free_reply(&reply);
                break;
            }
            else if (keyname.s && *ts_scan_start) {
                LM_DBG("will use key '%.*s' at offset %llx for timestamp/int range scan\n",
                        keyname.len, keyname.s, (unsigned long long) *ts_scan_start);
                *ts_scan_key = keyname;
                keyname.s = NULL;
            }
            else if (keyname.s) {
                pkg_free(keyname.s);
                keyname.s = NULL;
            }
        }
    }

    if (*query_keys_count > 0) {
        LM_DBG("building manual keys\n");
        if (db_redis_build_entry_manual_keys(table, _k, _v, _n, manual_keys, manual_keys_count) != 0) {
            LM_ERR("Failed to build manual entry key list\n");
            goto err;

        }
    }

    return 0;
err:
    if (keyname.s) {
        pkg_free(keyname.s);
        keyname.s = NULL;
    }
    if (reply) {
        db_redis_free_reply(&reply);
    }
    db_redis_key_free(query_keys);
    if (*manual_keys) {
        pkg_free(*manual_keys);
        *manual_keys = NULL;
    }
    return -1;
}

static int db_redis_scan_query_keys_pattern(km_redis_con_t *con, const str *match_pattern,
        const int _n,
        redis_key_t **query_keys, int *query_keys_count,
        int **manual_keys, int *manual_keys_count, unsigned int match_count_start_val) {

    size_t i = 0;
    redis_key_t *query_v = NULL;
    redisReply *reply = NULL;
    redisReply *keys_list = NULL;
    size_t j;
    int l;


#undef USE_SCAN

#ifdef USE_SCAN

    char cursor_str[32] = "";
    unsigned long cursor = 0;
    unsigned int match_count = match_count_start_val;
    char match_count_str[16];

    do {
        snprintf(cursor_str, sizeof(cursor_str), "%lu", cursor);

        if (db_redis_key_add_string(&query_v, "SCAN", 4) != 0) {
            LM_ERR("Failed to add scan command to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, cursor_str, strlen(cursor_str)) != 0) {
            LM_ERR("Failed to add cursor to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, "MATCH", 5) != 0) {
            LM_ERR("Failed to add match command to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, match_pattern->s, match_pattern->len) != 0) {
            LM_ERR("Failed to add match pattern to scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, "COUNT", 5) != 0) {
            LM_ERR("Failed to add count command to scan query\n");
            goto err;
        }
        l = snprintf(match_count_str, sizeof(match_count_str), "%u", match_count);
        if (l <= 0) {
            LM_ERR("Failed to print integer for scan query\n");
            goto err;
        }
        if (db_redis_key_add_string(&query_v, match_count_str, l) != 0) {
            LM_ERR("Failed to add count value to scan query\n");
            goto err;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, err);
        if (reply->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Invalid reply type for scan on table '%.*s', expected array\n",
                    match_pattern->len, match_pattern->s);
            goto err;
        }
        if (reply->elements != 2) {
            LM_ERR("Invalid number of reply elements for scan on table '%.*s', expected 2, got %lu\n",
                    match_pattern->len, match_pattern->s, reply->elements);
            goto err;
        }

        if (reply->element[0]->type == REDIS_REPLY_STRING) {
            cursor = atol(reply->element[0]->str);
        } else if (reply->element[0]->type == REDIS_REPLY_INTEGER) {
            // should not happen, but play it safe
            cursor = reply->element[0]->integer;
        } else {
            LM_ERR("Invalid cursor type for scan on table '%.*s', expected string or integer\n",
                    match_pattern->len, match_pattern->s);
            goto err;
        }
        LM_DBG("cursor is %lu\n", cursor);

        keys_list = reply->element[1];

#else // use KEYS

    if (db_redis_key_add_string(&query_v, "KEYS", 4) != 0) {
        LM_ERR("Failed to add scan command to scan query\n");
        goto err;
    }
    if (db_redis_key_add_string(&query_v, match_pattern->s, match_pattern->len) != 0) {
        LM_ERR("Failed to add match pattern to scan query\n");
        goto err;
    }

    reply = db_redis_command_argv(con, query_v);
    db_redis_key_free(&query_v);
    db_redis_check_reply(con, reply, err);

    keys_list = reply;

#endif

        if (keys_list->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Invalid content type for scan on table '%.*s', expected array\n",
                    match_pattern->len, match_pattern->s);
            goto err;
        }

        *query_keys_count += keys_list->elements;

        for (j = 0; j < keys_list->elements; ++i, ++j) {
            redisReply *key = keys_list->element[j];
            if (!key) {
                LM_ERR("Invalid null key at cursor result index %lu while scanning table '%.*s'\n",
                        (unsigned long) j, match_pattern->len, match_pattern->s);
                goto err;
            }
            if (key->type != REDIS_REPLY_STRING) {
                LM_ERR("Invalid key type at cursor result index %lu while scanning table '%.*s', expected string\n",
                        (unsigned long) j, match_pattern->len, match_pattern->s);
                goto err;
            }
            if (db_redis_key_prepend_string(query_keys, key->str, strlen(key->str)) != 0) {
                LM_ERR("Failed to prepend redis key\n");
                goto err;
            }
        }

#ifdef USE_SCAN
        // exponential increase and falloff, hovering around 1000 results
        if (keys_list->elements > 1300 && match_count > 500)
            match_count /= 2;
        else if (keys_list->elements < 700 && match_count < 500000)
            match_count *= 2;
#endif

        db_redis_free_reply(&reply);

#ifdef USE_SCAN
    } while (cursor > 0);
#endif

    // for full table scans, we have to manually match all given keys
    // but only do this once for repeated invocations
    if (!*manual_keys) {
        *manual_keys_count = _n;
        *manual_keys = (int*)pkg_malloc(*manual_keys_count * sizeof(int));
        if (! *manual_keys) {
            LM_ERR("Failed to allocate memory for manual keys\n");
            goto err;
        }
        memset(*manual_keys, 0, *manual_keys_count * sizeof(int));
        for (l = 0; l < _n; ++l) {
            (*manual_keys)[l] = l;
        }
    }

    if (reply) {
        db_redis_free_reply(&reply);
    }

    LM_DBG("got %lu entries by scan\n", (unsigned long) i);
    return 0;

err:
    if (reply)
        db_redis_free_reply(&reply);
    db_redis_key_free(&query_v);
    db_redis_key_free(query_keys);
    *query_keys_count = 0;
    if (*manual_keys) {
        pkg_free(*manual_keys);
        *manual_keys = NULL;
    }
    return -1;
}

static int db_redis_scan_query_keys(km_redis_con_t *con, const str *table_name,
        const int _n,
        redis_key_t **query_keys, int *query_keys_count,
        int **manual_keys, int *manual_keys_count, uint64_t ts_scan_start, const str *ts_scan_key) {

    struct str_hash_entry *table_e;
    redis_table_t *table;
    char *match = NULL;
    int ret;
    redisReply *reply = NULL;
    redis_key_t *set_key = NULL;
    int i, j;

    *query_keys = NULL;
    *query_keys_count = 0;
    *manual_keys = NULL;
    *manual_keys_count = 0;
    redis_key_t *set_keys = NULL;
    int set_keys_count = 0;

    table_e = str_hash_get(&con->tables, table_name->s, table_name->len);
    if (!table_e) {
        LM_ERR("query to undefined table '%.*s', define it in schema file!\n",
                table_name->len, table_name->s);
        return -1;
    }
    table = (redis_table_t*)table_e->u.p;

    if (!ts_scan_start) {
        // full table scan
        match = (char*)pkg_malloc(table->version_code.len
                + table_name->len + 10); // length of ':entry::*' plus \0
        if (!match) {
            LM_ERR("Failed to allocate memory for match pattern\n");
            return -1;
        }
        int len = sprintf(match, "%.*s%.*s:entry::*",
                table->version_code.len, table->version_code.s,
                table_name->len, table_name->s);
        str match_pattern = {match, len};
        ret = db_redis_scan_query_keys_pattern(con, &match_pattern, _n, query_keys, query_keys_count,
                manual_keys, manual_keys_count, 1000);
        pkg_free(match);
        return ret;
    }

    // timestamp range scan
    // ex: 2019-07-17 17:33:16
    // if >, we match: [3-9]???-??-?? ??:??:??, 2[1-9]??-??-?? ??:??:??, 20[2-9]?-??-?? ??:??:??, etc
    // if <, we match: [0-1]???-??-?? ??:??:??, 200..., 201[0-8]..., etc
    // the maximum match string length is ts_scan_key->len with one character replaced by 5 ('[a-b]')
    //
    // int range scan
    // ex: 12345
    // if >, we match: 2????, 1[3-9]???, ..., plus ?????*
    // if <. we match: ?, ??, ???, ????, 1[0-1]???, 12[0-2]??, etc
    //    ... however we expect a minimum length of 10 digits as per BIGINT printf format

    match = pkg_malloc(ts_scan_key->len + 6);
    if (!match) {
        LM_ERR("Failed to allocate memory for match pattern\n");
        return -1;
    }

    int scan_lt = (ts_scan_start & 0x8000000000000000ULL) ? 1 : 0;
    int scan_len_variable = (ts_scan_start & 0x4000000000000000ULL) ? 1 : 0;
    unsigned int scan_offset = ts_scan_start & 0x7fffffffULL;
    unsigned int scan_length = (ts_scan_start >> 31) & 0x7fffffffULL;
    scan_offset -= scan_length;
    const char *suffix = ts_scan_key->s + scan_offset + scan_length;

    LM_DBG("running timestamp/int range matching: lt %i, lv %i, off %u, len %u\n",
            scan_lt, scan_len_variable, scan_offset, scan_length);

    if (scan_lt && scan_len_variable) {
        // match shorter strings

        // copy unchanged prefix
        memcpy(match, ts_scan_key->s, scan_offset);

        // append a number of ?. minimum string length is 10 digits
        for (i = 0; i < scan_length - 1; i++) {
            int len = scan_offset + i;
            char match_char = ts_scan_key->s[len];
            // skip non-numbers
            if (match_char < '0' || match_char > '9') {
                match[len] = match_char;
                continue;
            }
            // append a single ?
            match[len] = '?';
            // append unchanged suffix
            strcpy(match + len + 1, suffix);
            len = strlen(match);

            // minimum bigint printf string length
            if (i < 10)
                continue;

            str match_pattern = {match, len};
            LM_DBG("running timestamp/int range matching using pattern '%.*s'\n", len, match);

            ret = db_redis_scan_query_keys_pattern(con, &match_pattern, _n, &set_keys, &set_keys_count,
                    manual_keys, manual_keys_count, 5000);
            if (ret)
                goto out;
        }
    }

    for (i = 0; i < scan_length; i++) {
        int len = scan_offset + i;
        char match_char = ts_scan_key->s[len];
        // skip non-numbers
        if (match_char < '0' || match_char > '9')
            continue;
        // skip numbers that are at the edge of their match range
        if (match_char == '0' && scan_lt)
            continue;
        if (match_char == '1' && scan_lt && i == 0) // no leading 0
            continue;
        if (match_char == '9' && !scan_lt)
            continue;

        // copy unchanged prefix
        memcpy(match, ts_scan_key->s, len);
        // append range matcher
        if (scan_lt)
            len += sprintf(match + len, "[0-%c]", match_char - 1);
        else
            len += sprintf(match + len, "[%c-9]", match_char + 1);
        // finish with trailing ?s
	for (j = i + 1; j < scan_length; j++) {
            match_char = ts_scan_key->s[scan_offset + j];
            // skip non-numbers
            if (match_char < '0' || match_char > '9') {
                match[len++] = match_char;
                continue;
            }
            match[len++] = '?';
        }
        // append unchanged suffix
        strcpy(match + len, suffix);
        len = strlen(match);

        str match_pattern = {match, len};
        LM_DBG("running timestamp/int range matching using pattern '%.*s'\n", len, match);

        ret = db_redis_scan_query_keys_pattern(con, &match_pattern, _n, &set_keys, &set_keys_count,
                manual_keys, manual_keys_count, 5000);
        if (ret)
            goto out;
    }

    if (!scan_lt && scan_len_variable) {
        // match longer strings
        int len = sprintf(match, "%.*s*%s", scan_offset + scan_length, ts_scan_key->s, suffix);

        str match_pattern = {match, len};
        LM_DBG("running timestamp/int range matching using pattern '%.*s'\n", len, match);

        ret = db_redis_scan_query_keys_pattern(con, &match_pattern, _n, &set_keys, &set_keys_count,
                manual_keys, manual_keys_count, 5000);
        if (ret)
            goto out;
    }

    // we not have a list of matching type keys in set_keys. now we have to iterate through them
    // and retrieve the set members, and finally build our actual key list

    ret = -1;

    for (set_key = set_keys; set_key; set_key = set_key->next) {
        LM_DBG("pulling set members from key '%.*s'\n", set_key->key.len, set_key->key.s);

        redis_key_t *query_v = NULL;
        if (db_redis_key_add_string(&query_v, "SMEMBERS", 8) != 0) {
            LM_ERR("Failed to add smembers command to query\n");
            db_redis_key_free(&query_v);
            goto out;
        }
        if (db_redis_key_add_str(&query_v, &set_key->key) != 0) {
            LM_ERR("Failed to add key name to smembers query\n");
            db_redis_key_free(&query_v);
            goto out;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, out);

        if (reply->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Unexpected reply for type query, expecting an array\n");
            goto out;
        }

        LM_DBG("adding %i keys returned from set", (int) reply->elements);

        for (i = 0; i < reply->elements; i++) {
            if (reply->element[i]->type != REDIS_REPLY_STRING) {
                LM_ERR("Unexpected entry key type in type query, expecting a string\n");
                goto out;
            }
            if (db_redis_key_prepend_string(query_keys, reply->element[i]->str, strlen(reply->element[i]->str))
                    != 0) {
                LM_ERR("Failed to prepend redis key\n");
                goto out;
            }
            LM_DBG("adding key '%s'\n", reply->element[i]->str);
        }
        *query_keys_count += reply->elements;

        db_redis_free_reply(&reply);
    }

    ret = 0;

out:
    pkg_free(match);
    db_redis_key_free(&set_keys);
    db_redis_free_reply(&reply);
    if (ret) {
        db_redis_key_free(query_keys);
        *query_keys_count = 0;
        if (*manual_keys) {
            pkg_free(*manual_keys);
            *manual_keys = NULL;
        }
    }
    return ret;
}

static int db_redis_compare_column(db_key_t k, db_val_t *v, db_op_t op, redisReply *reply) {
    int i_value;
    long long ll_value;
    double d_value;
    str *tmpstr;
    char tmp[32] = "";
    struct tm _time;

    int vtype = VAL_TYPE(v);

    if (VAL_NULL(v) && reply->type == REDIS_REPLY_NIL) {
        LM_DBG("comparing matching NULL values\n");
        return 0;
    } else if (VAL_NULL(v) || reply->type == REDIS_REPLY_NIL) {
        LM_DBG("comparing non-matching NULL values\n");
        return -1;
    }

    switch (vtype) {
        case DB1_INT:
            i_value = atoi(reply->str);
            LM_DBG("comparing INT %d %s %d\n", i_value, op, VAL_INT(v));
            if (!strcmp(op, OP_EQ)) {
                if (i_value == VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_LT)) {
                if (i_value < VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_GT)) {
                if (i_value > VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_LEQ)) {
                if (i_value <= VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_GEQ)) {
                if (i_value >= VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_NEQ)) {
                if (i_value != VAL_INT(v))
                    return 0;
            } else if (!strcmp(op, OP_BITWISE_AND)) {
                if (i_value & VAL_INT(v))
                    return 0;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_BIGINT:
            ll_value = atoll(reply->str);
            LM_DBG("comparing BIGINT %lld %s %lld\n", ll_value, op, VAL_BIGINT(v));
            if (!strcmp(op, OP_EQ)) {
                if (ll_value == VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_LT)) {
                if (ll_value < VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_GT)) {
                if (ll_value > VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_LEQ)) {
                if (ll_value <= VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_GEQ)) {
                if (ll_value >= VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_NEQ)) {
                if (ll_value != VAL_BIGINT(v))
                    return 0;
            } else if (!strcmp(op, OP_BITWISE_AND)) {
                if (ll_value & VAL_BIGINT(v))
                    return 0;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_STRING:
            LM_DBG("comparing STRING %s %s %s\n", reply->str, op, VAL_STRING(v));
            if (!strcmp(op, OP_EQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) == 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (strcmp(reply->str, VAL_STRING(v)) < 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (strcmp(reply->str, VAL_STRING(v)) > 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) <= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) >= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (strcmp(reply->str, VAL_STRING(v)) != 0) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_STR:
            tmpstr = (struct _str*) &(VAL_STR(v));
            LM_DBG("comparing STR %s %s %.*s\n", reply->str, op, tmpstr->len, tmpstr->s);
            if (!strcmp(op, OP_EQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) == 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) < 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) > 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) <= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) >= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (strncmp(reply->str, tmpstr->s, tmpstr->len) != 0) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_DOUBLE:
            d_value = atof(reply->str);
            LM_DBG("comparing DOUBLE %f %s %f\n", d_value, op, VAL_DOUBLE(v));
            if (!strcmp(op, OP_EQ)) {
                return (d_value == VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (d_value < VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (d_value > VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (d_value <= VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (d_value >= VAL_DOUBLE(v)) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (d_value != VAL_DOUBLE(v)) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_DATETIME:
            // TODO: insert int value to db for faster comparison!
            localtime_r(&(VAL_TIME(v)), &_time);
            strftime(tmp, sizeof(tmp), "%Y-%m-%d %H:%M:%S", &_time);
            LM_DBG("comparing DATETIME %s %s %s\n", reply->str, op, tmp);
            if (!strcmp(op, OP_EQ)) {
                return (strcmp(reply->str, tmp) == 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LT)) {
                return (strcmp(reply->str, tmp) < 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GT)) {
                return (strcmp(reply->str, tmp) > 0) ? 0 : -1;
            } else if (!strcmp(op, OP_LEQ)) {
                return (strcmp(reply->str, tmp) <= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_GEQ)) {
                return (strcmp(reply->str, tmp) >= 0) ? 0 : -1;
            } else if (!strcmp(op, OP_NEQ)) {
                return (strcmp(reply->str, tmp) != 0) ? 0 : -1;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_BITMAP:
            i_value = atoi(reply->str);
            LM_DBG("comparing BITMAP %d %s %d\n", i_value, op, VAL_BITMAP(v));
            if (!strcmp(op, OP_EQ)) {
                if (i_value == VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_LT)) {
                if (i_value < VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_GT)) {
                if (i_value > VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_LEQ)) {
                if (i_value <= VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_GEQ)) {
                if (i_value >= VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_NEQ)) {
                if (i_value != VAL_BITMAP(v))
                    return 0;
            } else if (!strcmp(op, OP_BITWISE_AND)) {
                if (i_value & VAL_BITMAP(v))
                    return 0;
            } else {
                LM_ERR("Unsupported op type '%s'\n", op);
                return -1;
            }
            return -1;
        case DB1_BLOB:
        default:
            LM_ERR("Unsupported val type %d\n", vtype);
            return -1;
    }
}

static int db_redis_convert_row(km_redis_con_t *con, db1_res_t* _r, const db_key_t *_k,
        const db_val_t *_v, const db_op_t *_op,
        redisReply *reply, const str *table_name, const db_key_t* _c, int _nc,
        int *manual_keys, int manual_keys_count) {
    db_val_t* dval;
    db_row_t* drow;
    size_t col;

    if (reply->type != REDIS_REPLY_ARRAY) {
        LM_ERR("Unexpected redis reply type, expecting array\n");
        return -1;
    }

    if (!reply->elements) {
        LM_DBG("skip empty row");
        return 0;
    }

    // manually filter non-matching replies
    for (col = 0; col < reply->elements; ++col) {
        if (col < manual_keys_count) {
            int idx = manual_keys[col];
            db_key_t k = _k[idx];
            db_val_t v = _v[idx];
            db_op_t o = _op[idx];
            LM_DBG("manually filtering key '%.*s'\n",
                    k->len, k->s);
            if (db_redis_compare_column(k, &v, o, reply->element[col]) != 0) {
                LM_DBG("column %lu does not match, ignore row\n", (unsigned long) col);
                return 0;
            }
        }
    }

    RES_NUM_ROWS(_r) = RES_ROW_N(_r) = RES_NUM_ROWS(_r) + 1;
    drow = &(RES_ROWS(_r)[RES_NUM_ROWS(_r)-1]);

    if (db_allocate_row(_r, drow) != 0) {
        LM_ERR("Failed to allocate row %d\n", RES_NUM_ROWS(_r));
        return -1;
    }

    if (reply->elements - manual_keys_count > RES_COL_N(_r)) {
        LM_ERR("Invalid number of columns at row %d/%d, expecting %d, got %lu\n",
                RES_NUM_ROWS(_r), RES_ROW_N(_r), RES_COL_N(_r), (unsigned long) reply->elements - manual_keys_count);
        return -1;
    }
    for (col = manual_keys_count; col < reply->elements; ++col) {
        size_t colidx = col - manual_keys_count;
        size_t redisidx = col;
        int coltype;
        redisReply *col_val = reply->element[redisidx];
        str *col_name = _c[colidx];

        LM_DBG("converting column #%lu of row #%d", (unsigned long) colidx, RES_ROW_N(_r));

        if (col_val->type != REDIS_REPLY_STRING &&
            col_val->type != REDIS_REPLY_NIL) {

            LM_ERR("Invalid column value type in column '%.*s' of row %d, expecting string or null\n",
                    col_name->len, col_name->s, RES_NUM_ROWS(_r));
            return -1;
        }

        if (RES_NUM_ROWS(_r) == 1) {
            coltype = db_redis_schema_get_column_type(con, table_name, col_name);
            RES_TYPES(_r)[colidx] = coltype;
        } else {
            coltype = RES_TYPES(_r)[colidx];
        }

        dval = &(ROW_VALUES(drow)[colidx]);
        VAL_TYPE(dval) = coltype;

        if (col_val->type == REDIS_REPLY_NIL) {
            VAL_NULL(dval) = 1;
        } else {
            if (db_str2val(coltype, dval, col_val->str, strlen(col_val->str), 1) != 0) {
                LM_ERR("Failed to convert redis column '%.*s' to db value\n",
                        col_name->len, col_name->s);
                return -1;
            }
        }
    }

    return 0;
}

static int db_redis_perform_query(const db1_con_t* _h, km_redis_con_t *con, const db_key_t* _k,
        const db_val_t* _v, const db_op_t *_op, const db_key_t* _c,
        const int _n, const int _nc, db1_res_t** _r,
        redis_key_t **keys, int *keys_count,
        int **manual_keys, int *manual_keys_count, int do_table_scan, uint64_t ts_scan_start,
        const str *ts_scan_key) {

    redisReply *reply = NULL;
    redis_key_t *query_v = NULL;
    int num_rows = 0;
    redis_key_t *key;
    int i, j, max;

    *_r = db_redis_new_result();
    if (!*_r) {
        LM_ERR("Failed to allocate memory for result");
        goto error;
    }

    if (db_allocate_columns(*_r, _nc) != 0) {
        LM_ERR("Failed to allocate memory for result columns");
        goto error;
    }
    RES_NUM_ROWS(*_r) = RES_ROW_N(*_r) = 0;
    RES_COL_N(*_r) = _nc;

    if (!(*keys_count) && do_table_scan) {
        if(_n > 0) {
            LM_WARN("performing full table scan on table '%.*s' while doing the query\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
            for(i = 0; i < _n; ++i) {
                LM_WARN("  scan key %d is '%.*s'\n",
                        i, _k[i]->len, _k[i]->s);
            }
        } else {
            LM_DBG("loading full table: '%.*s\n", CON_TABLE(_h)->len, CON_TABLE(_h)->s);
        }
        if (db_redis_scan_query_keys(con, CON_TABLE(_h), _n,
                    keys, keys_count,
                    manual_keys, manual_keys_count, ts_scan_start, ts_scan_key) != 0) {
            LM_ERR("failed to scan query keys\n");
            goto error;
        }
    }

    // we allocate best case scenario (all rows match)
    RES_NUM_ROWS(*_r) = RES_ROW_N(*_r) = *keys_count;
    if (db_allocate_rows(*_r) != 0) {
        LM_ERR("Failed to allocate memory for rows\n");
        return -1;
    }
    RES_COL_N(*_r) = _nc;
    // reset and increment in convert_row
    RES_NUM_ROWS(*_r) = RES_ROW_N(*_r) = 0;

    for (key = *keys; key; key = key->next) {
        redis_key_t *tmp = NULL;
        str *keyname = &(key->key);

        num_rows++;

        LM_DBG("checking key '%s' in redis\n", keyname->s);

        if (db_redis_key_add_string(&query_v, "EXISTS", 6) != 0) {
            LM_ERR("Failed to add exists query to list\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, keyname) != 0) {
            LM_ERR("Failed to add key name to list\n");
            goto error;
        }
        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }
        tmp = db_redis_key_shift(&query_v);
        if (tmp)
            db_redis_key_free(&tmp);

        // construct HMGET query
        if (_nc + (*manual_keys_count) == 0) {
            if (db_redis_key_prepend_string(&query_v, "HGETALL", 7) != 0) {
                LM_ERR("Failed to add hgetall query to list\n");
                goto error;
            }
        } else {
            if (db_redis_key_prepend_string(&query_v, "HMGET", 5) != 0) {
                LM_ERR("Failed to add hmget query to list\n");
                goto error;
            }
        }

        // we put the manual comparison columns first, so we can skip them
        // easily in result, for the cost of potential duplicate column returns
        for (j = 0; j < *manual_keys_count; ++j) {
            int idx = (*manual_keys)[j];
            str *k_name = _k[idx];
            if (db_redis_key_add_str(&query_v, k_name) != 0) {
                LM_ERR("Failed to add manual key to query list\n");
                goto error;
            }
        }
        for (j = 0; j < _nc; ++j) {
            str *k_name = _c[j];
            if (db_redis_key_add_str(&query_v, k_name) != 0) {
                LM_ERR("Failed to add manual key to query list\n");
                goto error;
            }
        }

        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }

        db_redis_key_free(&query_v);
        query_v = NULL;

        max = 0;
        if (*keys_count == num_rows)
            max = (*keys_count) % 1000;
        else if (num_rows % 1000 == 0)
            max = 1000;

        if (max) {
            LM_DBG("fetching next %d results\n", max);
            for (i = 0; i < max; ++i) {
                // get reply for EXISTS query
                if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
                    LM_ERR("Failed to get reply for query: %s\n",
                            db_redis_get_error(con));
                    goto error;
                }
                db_redis_check_reply(con, reply, error);
                if (reply->integer == 0) {
                    LM_DBG("key does not exist, returning no row for query\n");
                    db_redis_free_reply(&reply);
                    // also free next reply, as this is a null row for the HMGET
                    if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
                        LM_ERR("Failed to get reply for query: %s\n",
                                db_redis_get_error(con));
                        goto error;
                    }
                    db_redis_check_reply(con, reply, error);
                    db_redis_free_reply(&reply);
                    continue;
                }
                db_redis_free_reply(&reply);

                // get reply for actual HMGET query
                if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
                    LM_ERR("Failed to get reply for query: %s\n",
                            db_redis_get_error(con));
                    goto error;
                }
                db_redis_check_reply(con, reply, error);
                if (reply->type != REDIS_REPLY_ARRAY) {
                    LM_ERR("Unexpected reply, expected array\n");
                    goto error;
                }
                LM_DBG("dumping full query reply for row\n");
                db_redis_dump_reply(reply);

                if (db_redis_convert_row(con, *_r, _k, _v, _op, reply, CON_TABLE(_h), _c, _nc, *manual_keys, *manual_keys_count)) {
                    LM_ERR("Failed to convert redis reply for row\n");
                    goto error;
                }
                db_redis_free_reply(&reply);
            }
        }
    }

    LM_DBG("done performing query\n");
    return 0;

error:
    LM_ERR("failed to perform the query\n");
    db_redis_key_free(&query_v);
    if(reply)
        db_redis_free_reply(&reply);
    if(*_r) {
        db_redis_free_result((db1_con_t*)_h, *_r); *_r = NULL;
    }
    return -1;
}

static int db_redis_perform_delete(const db1_con_t* _h, km_redis_con_t *con, const db_key_t* _k,
        const db_val_t* _v, const db_op_t *_op, const int _n,
        redis_key_t **keys, int *keys_count,
        int **manual_keys, int *manual_keys_count, int do_table_scan, uint64_t ts_scan_start,
        const str *ts_scan_key) {

    int i = 0, j = 0;
    redis_key_t *k = NULL;
    int type_keys_count = 0;
    int all_type_keys_count = 0;
    size_t col;

    redisReply *reply = NULL;
    redis_key_t *query_v = NULL;
    redis_key_t *type_keys = NULL;
    redis_key_t *set_keys = NULL;
    redis_key_t *all_type_keys = NULL;
    db_val_t *db_vals = NULL;
    db_key_t *db_keys = NULL;
    redis_key_t *type_key;
    redis_key_t *set_key;

    if (!*keys_count && do_table_scan) {
        if (!ts_scan_start)
            LM_WARN("performing full table scan on table '%.*s' while performing delete\n",
                    CON_TABLE(_h)->len, CON_TABLE(_h)->s);
        else
            LM_WARN("performing table scan on table '%.*s' while performing delete using match key "
                    "'%.*s' at offset %llx\n",
                    CON_TABLE(_h)->len, CON_TABLE(_h)->s,
                    ts_scan_key->len, ts_scan_key->s, (unsigned long long) ts_scan_start);
        for(i = 0; i < _n; ++i) {
            LM_WARN("  scan key %d is '%.*s'\n",
                    i, _k[i]->len, _k[i]->s);
        }
        if (db_redis_scan_query_keys(con, CON_TABLE(_h), _n,
                    keys, keys_count,
                    manual_keys, manual_keys_count, ts_scan_start, ts_scan_key) != 0) {
            LM_ERR("failed to scan query keys\n");
            goto error;
        }
    }

    // TODO: this should be moved to redis_connection structure
    // and be parsed at startup:
    //
    // fetch list of keys in all types
    if (db_redis_get_keys_for_all_types(con, CON_TABLE(_h),
                &all_type_keys, &all_type_keys_count) != 0) {
            LM_ERR("failed to get full list of type keys\n");
            goto error;
    }

    LM_DBG("delete all keys\n");
    for (k = *keys; k; k = k->next) {
        redis_key_t *all_type_key;
        str *key = &k->key;
        redis_key_t *tmp = NULL;
        int row_match;
        LM_DBG("delete key '%.*s'\n", key->len, key->s);

        if (db_redis_key_add_string(&query_v, "EXISTS", 6) != 0) {
            LM_ERR("Failed to add exists command to pre-delete query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, key) != 0) {
            LM_ERR("Failed to add key name to pre-delete query\n");
            goto error;
        }

        // TODO: pipeline commands!
        reply = db_redis_command_argv(con, query_v);
        db_redis_check_reply(con, reply, error);
        if (reply->integer == 0) {
            LM_DBG("key does not exist in redis, skip deleting\n");
            db_redis_free_reply(&reply);
            db_redis_key_free(&query_v);
            continue;
        }
        db_redis_free_reply(&reply);
        tmp = db_redis_key_shift(&query_v);
        if (tmp)
            db_redis_key_free(&tmp);

        if (db_redis_key_prepend_string(&query_v, "HMGET", 5) != 0) {
            LM_ERR("Failed to set hmget command to pre-delete query\n");
            goto error;
        }

        // add all manual keys to query
        for (j = 0; j < *manual_keys_count; ++j) {
            int idx = (*manual_keys)[j];
            str *col = _k[idx];

            if (db_redis_key_add_str(&query_v, col) != 0) {
                LM_ERR("Failed to add manual key to pre-delete query\n");
                goto error;
            }
        }
        // add all type keys to query
        for (all_type_key = all_type_keys; all_type_key; all_type_key = all_type_key->next) {
            if (db_redis_key_add_str(&query_v, &all_type_key->key) != 0) {
                LM_ERR("Failed to add type key to pre-delete query\n");
                goto error;
            }
        }
        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);

        LM_DBG("dumping full query reply\n");
        db_redis_dump_reply(reply);

        // manually filter non-matching replies
        row_match = 1;
        for (col = 0; col < reply->elements; ++col) {
            if (col < *manual_keys_count) {
                int idx = (*manual_keys)[col];
                db_key_t k = _k[idx];
                db_val_t v = _v[idx];
                db_op_t o = _op[idx];
                LM_DBG("manually filtering key '%.*s'\n",
                        k->len, k->s);
                if (db_redis_compare_column(k, &v, o, reply->element[col]) != 0) {
                    LM_DBG("column %lu does not match, ignore row\n", (unsigned long) col);
                    row_match = 0;
                    break;
                }
            }
        }
        if (!row_match) {
            db_redis_free_reply(&reply);
            continue;
        } else {
            LM_DBG("row matches manual filtering, proceed with deletion\n");
        }

        db_keys = (db_key_t*) pkg_malloc(all_type_keys_count * sizeof(db_key_t));
        if (!db_keys) {
            LM_ERR("Failed to allocate memory for db type keys\n");
            goto error;
        }
        for (j = 0, tmp = all_type_keys; tmp; ++j, tmp = tmp->next) {
            db_keys[j] = &tmp->key;
        }

        db_vals = (db_val_t*) pkg_malloc(all_type_keys_count * sizeof(db_val_t));
        if (!db_vals) {
            LM_ERR("Failed to allocate memory for manual db vals\n");
            goto error;
        }

        for (j = 0, all_type_key = all_type_keys; all_type_key; ++j, all_type_key = all_type_key->next) {
            db_val_t *v = &(db_vals[j]);
            str *key = &all_type_key->key;
            char *value = reply->element[*manual_keys_count + j]->str;
            int coltype = db_redis_schema_get_column_type(con, CON_TABLE(_h), key);
            if (value == NULL) {
                VAL_NULL(v) = 1;
            } else if (db_str2val(coltype, v, value, strlen(value), 0) != 0) {
                LM_ERR("Failed to convert redis reply column to db value\n");
                goto error;
            }
        }
        if (db_redis_build_type_keys(con, CON_TABLE(_h), db_keys, db_vals, all_type_keys_count,
                    &type_keys, &set_keys, &type_keys_count) != 0) {
            LM_ERR("failed to build type keys\n");
            goto error;
        }
        pkg_free(db_keys);
        db_keys = NULL;
        pkg_free(db_vals);
        db_vals = NULL;
        db_redis_free_reply(&reply);

        if (db_redis_key_add_string(&query_v, "DEL", 3) != 0) {
            LM_ERR("Failed to add del command to delete query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, key) != 0) {
            LM_ERR("Failed to add key to delete query\n");
            goto error;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);

        for (type_key = type_keys, set_key = set_keys; type_key;
                type_key = type_key->next, set_key = set_key->next) {

            if (db_redis_key_add_string(&query_v, "EVALSHA", 7) != 0) {
                LM_ERR("Failed to add srem command to post-delete query\n");
                goto error;
            }
            if (db_redis_key_add_string(&query_v, con->srem_key_lua, strlen(con->srem_key_lua)) != 0) {
                LM_ERR("Failed to add srem command to post-delete query\n");
                goto error;
            }
            if (db_redis_key_add_string(&query_v, "3", 1) != 0) {
                LM_ERR("Failed to add srem command to post-delete query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, &type_key->key) != 0) {
                LM_ERR("Failed to add key to delete query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, &set_key->key) != 0) {
                LM_ERR("Failed to add key to delete query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, key) != 0) {
                LM_ERR("Failed to add key to delete query\n");
                goto error;
            }
            reply = db_redis_command_argv(con, query_v);
            db_redis_key_free(&query_v);
            db_redis_check_reply(con, reply, error);
            db_redis_free_reply(&reply);
        }
        LM_DBG("done with loop '%.*s'\n", k->key.len, k->key.s);
        db_redis_key_free(&type_keys);
        db_redis_key_free(&set_keys);
    }
    db_redis_key_free(&all_type_keys);
    db_redis_key_free(&query_v);

    return 0;

error:
    LM_ERR("failed to perform the delete\n");
    if(reply)
        db_redis_free_reply(&reply);
    if (db_keys)
        pkg_free(db_keys);
    if (db_vals)
        pkg_free(db_vals);
    db_redis_key_free(&query_v);
    db_redis_key_free(&type_keys);
    db_redis_key_free(&set_keys);
    db_redis_key_free(&all_type_keys);
    return -1;
}

static int db_redis_perform_update(const db1_con_t* _h, km_redis_con_t *con, const db_key_t* _k,
        const db_val_t* _v, const db_op_t *_op, const db_key_t* _uk, const db_val_t *_uv,
        const int _n, const int _nu,
        redis_key_t **keys, int *keys_count,
        int **manual_keys, int *manual_keys_count, int do_table_scan, uint64_t ts_scan_start,
        const str *ts_scan_key) {

    redisReply *reply = NULL;
    redis_key_t *query_v = NULL;
    int update_queries = 0;
    redis_key_t *key;
    int i;
    int j;
    size_t col;
    redis_key_t *all_type_keys = NULL;
    int all_type_keys_count = 0;
    db_val_t *db_vals = NULL;
    db_key_t *db_keys = NULL;
    redis_key_t *type_keys = NULL;
    redis_key_t *set_keys = NULL;
    int type_keys_count = 0;
    redis_key_t *new_type_keys = NULL;
    int new_type_keys_count = 0;
    redis_key_t *all_type_key;

    if (!(*keys_count) && do_table_scan) {
        LM_WARN("performing full table scan on table '%.*s' while performing update\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
        for(i = 0; i < _n; ++i) {
            LM_WARN("  scan key %d is '%.*s'\n",
                    i, _k[i]->len, _k[i]->s);
        }
        if (db_redis_scan_query_keys(con, CON_TABLE(_h), _n,
                    keys, keys_count,
                    manual_keys, manual_keys_count, ts_scan_start, ts_scan_key) != 0) {
            LM_ERR("failed to scan query keys\n");
            goto error;
        }
    }

    // TODO: this should be moved to redis_connection structure
    // and be parsed at startup:
    //
    // fetch list of keys in all types
    if (db_redis_get_keys_for_all_types(con, CON_TABLE(_h),
                &all_type_keys, &all_type_keys_count) != 0) {
            LM_ERR("failed to get full list of type keys\n");
            goto error;
    }

    if (db_redis_build_type_keys(con, CON_TABLE(_h), _uk, _uv, _nu,
                &new_type_keys, NULL, &new_type_keys_count) != 0) {
        LM_ERR("failed to build type keys\n");
        goto error;
    }
    LM_DBG("%i new type keys\n", new_type_keys_count);

    for (key = *keys; key; key = key->next) {
        str *keyname = &key->key;

        LM_DBG("fetching row for '%.*s' from redis for update\n", keyname->len, keyname->s);


        if (db_redis_key_add_string(&query_v, "EXISTS", 6) != 0) {
            LM_ERR("Failed to set exists command to pre-update exists query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, keyname) != 0) {
            LM_ERR("Failed to add key name to pre-update exists query\n");
            goto error;
        }
        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }
        db_redis_key_free(&query_v);

        // construct HMGET query
        if (db_redis_key_add_string(&query_v, "HMGET", 5) != 0) {
            LM_ERR("Failed to set hgetall command to pre-update hget query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, keyname) != 0) {
            LM_ERR("Failed to add key name to pre-update exists query\n");
            goto error;
        }

        for (j = 0; j < *manual_keys_count; ++j) {
            int idx = (*manual_keys)[j];
            str *k_name = _k[idx];
            if (db_redis_key_add_str(&query_v, k_name) != 0) {
                LM_ERR("Failed to add manual key name to hget query\n");
                goto error;
            }
        }
        // add all type keys to query
        for (all_type_key = all_type_keys; all_type_key; all_type_key = all_type_key->next) {
            if (db_redis_key_add_str(&query_v, &all_type_key->key) != 0) {
                LM_ERR("Failed to add type key to pre-update query\n");
                goto error;
            }
        }

        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }
        db_redis_key_free(&query_v);
    }

/*
    key_value = (str*)pkg_malloc(_nu * sizeof(str));
    if (!key_value) {
        LM_ERR("Failed to allocate memory for key buffer\n");
        goto error;
    }
    memset(key_value, 0, _nu * sizeof(str));

    col_value = (str*)pkg_malloc(_nu * sizeof(str));
    if (!col_value) {
        LM_ERR("Failed to allocate memory for column buffer\n");
        goto error;
    }
    memset(col_value, 0, _nu * sizeof(str));
    */


    for (key = *keys; key; key = key->next) {
        redis_key_t *tmp = NULL;
        redis_key_t *type_key;
        redis_key_t *set_key;
        redis_key_t *new_type_key;
        int row_match;

        LM_DBG("fetching replies for '%.*s' from redis for update\n", key->key.len, key->key.s);

        // get reply for EXISTS query
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    db_redis_get_error(con));
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        if (reply->integer == 0) {
            LM_DBG("key does not exist, returning no row for query\n");
            db_redis_free_reply(&reply);
            // also free next reply, as this is a null row for the HMGET
            LM_DBG("also fetch hmget reply after non-existent key\n");
            if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
                LM_ERR("Failed to get reply for query: %s\n",
                        db_redis_get_error(con));
                goto error;
            }
            db_redis_check_reply(con, reply, error);
            db_redis_free_reply(&reply);
            LM_DBG("continue fetch reply loop\n");
            continue;
        }
        db_redis_free_reply(&reply);

        // get reply for actual HMGET query
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    db_redis_get_error(con));
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        if (reply->type != REDIS_REPLY_ARRAY) {
            LM_ERR("Unexpected reply, expected array\n");
            goto error;
        }
        LM_DBG("dumping full query reply for row\n");
        db_redis_dump_reply(reply);

        // manually filter non-matching replies
        row_match = 1;
        for (col = 0; col < reply->elements; ++col) {
            if (col < *manual_keys_count) {
                int idx = (*manual_keys)[col];
                db_key_t k = _k[idx];
                db_val_t v = _v[idx];
                db_op_t o = _op[idx];
                LM_DBG("manually filtering key '%.*s'\n",
                        k->len, k->s);
                if (db_redis_compare_column(k, &v, o, reply->element[col]) != 0) {
                    LM_DBG("column %lu does not match, ignore row\n", (unsigned long) col);
                    row_match = 0;
                    break;
                }
            }
        }
        if (!row_match) {
            continue;
        } else {
            LM_DBG("row matches manual filtering, proceed with update\n");
        }

        db_keys = (db_key_t*) pkg_malloc(all_type_keys_count * sizeof(db_key_t));
        if (!db_keys) {
            LM_ERR("Failed to allocate memory for db type keys\n");
            goto error;
        }
        for (j = 0, tmp = all_type_keys; tmp; ++j, tmp = tmp->next) {
            db_keys[j] = &tmp->key;
        }

        db_vals = (db_val_t*) pkg_malloc(all_type_keys_count * sizeof(db_val_t));
        if (!db_vals) {
            LM_ERR("Failed to allocate memory for manual db vals\n");
            goto error;
        }

        for (j = 0, all_type_key = all_type_keys; all_type_key; ++j, all_type_key = all_type_key->next) {
            db_val_t *v = &(db_vals[j]);
            str *key = &all_type_key->key;
            char *value = reply->element[*manual_keys_count + j]->str;
            int coltype = db_redis_schema_get_column_type(con, CON_TABLE(_h), key);
            if (value == NULL) {
                VAL_NULL(v) = 1;
            } else if (db_str2val(coltype, v, value, strlen(value), 0) != 0) {
                LM_ERR("Failed to convert redis reply column to db value\n");
                goto error;
            }
        }
        if (db_redis_build_type_keys(con, CON_TABLE(_h), db_keys, db_vals, all_type_keys_count,
                    &type_keys, &set_keys, &type_keys_count) != 0) {
            LM_ERR("failed to build type keys\n");
            goto error;
        }
        pkg_free(db_keys);
        db_keys = NULL;
        pkg_free(db_vals);
        db_vals = NULL;
        db_redis_free_reply(&reply);

        if (db_redis_key_add_string(&query_v, "HMSET", 5) != 0) {
            LM_ERR("Failed to add hmset command to update query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &key->key) != 0) {
            LM_ERR("Failed to add key to update query\n");
            goto error;
        }

        for (i = 0; i < _nu; ++i) {
            str *k = _uk[i];
            str v = {NULL, 0};

            if (db_redis_val2str(&(_uv[i]), &v) != 0) {
                LM_ERR("Failed to convert update value to string\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, k) != 0) {
                LM_ERR("Failed to add key to update query\n");
                goto error;
            }
            if (db_redis_key_add_str(&query_v, &v) != 0) {
                LM_ERR("Failed to add key to update query\n");
                goto error;
            }
            if (v.s)
                pkg_free(v.s);
        }
        update_queries++;
        if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
            LM_ERR("Failed to append redis command\n");
            goto error;
        }

        db_redis_key_free(&query_v);

        for (type_key = type_keys, set_key = set_keys; type_key;
                type_key = type_key->next, set_key = set_key->next) {

            LM_DBG("checking for update of type key '%.*s'\n",
                    type_key->key.len, type_key->key.s);
            char *prefix = ser_memmem(type_key->key.s, "::", type_key->key.len, 2);
            if (!prefix || prefix == type_key->key.s) {
                LM_DBG("Invalid key without :: '%.*s'\n",
                        type_key->key.len, type_key->key.s);
                continue;
            }
            for (new_type_key = new_type_keys; new_type_key; new_type_key = new_type_key->next) {
                // compare prefix to see if this is the same key
                if (memcmp(new_type_key->key.s, type_key->key.s, prefix - type_key->key.s))
                    continue;
                LM_DBG("checking for update of type key against '%.*s'\n",
                        new_type_key->key.len, new_type_key->key.s);
                if (!str_strcmp(&new_type_key->key, &type_key->key))
                    continue;

                // add to new set key and delete from old

                if (db_redis_key_add_string(&query_v, "SADD", 4) != 0) {
                    LM_ERR("Failed to set sadd command to post-update query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &new_type_key->key) != 0) {
                    LM_ERR("Failed to add map key to post-update query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &key->key) != 0) {
                    LM_ERR("Failed to set entry key to post-update query\n");
                    goto error;
                }

                update_queries++;
                if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
                    LM_ERR("Failed to append redis command\n");
                    goto error;
                }

                db_redis_key_free(&query_v);

                if (db_redis_key_add_string(&query_v, "SADD", 4) != 0) {
                    LM_ERR("Failed to set sadd command to post-update query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &set_key->key) != 0) {
                    LM_ERR("Failed to add map key to post-update query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &new_type_key->key) != 0) {
                    LM_ERR("Failed to set entry key to post-update query\n");
                    goto error;
                }

                update_queries++;
                if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
                    LM_ERR("Failed to append redis command\n");
                    goto error;
                }

                db_redis_key_free(&query_v);

                if (db_redis_key_add_string(&query_v, "EVAL", 4) != 0) {
                    LM_ERR("Failed to add srem command to post-delete query\n");
                    goto error;
                }
                if (db_redis_key_add_string(&query_v, SREM_KEY_LUA, strlen(SREM_KEY_LUA)) != 0) {
                    LM_ERR("Failed to add srem command to post-delete query\n");
                    goto error;
                }
                if (db_redis_key_add_string(&query_v, "3", 1) != 0) {
                    LM_ERR("Failed to add srem command to post-delete query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &type_key->key) != 0) {
                    LM_ERR("Failed to add key to delete query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &set_key->key) != 0) {
                    LM_ERR("Failed to add key to delete query\n");
                    goto error;
                }
                if (db_redis_key_add_str(&query_v, &key->key) != 0) {
                    LM_ERR("Failed to add key to delete query\n");
                    goto error;
                }

                update_queries++;
                if (db_redis_append_command_argv(con, query_v, 1) != REDIS_OK) {
                    LM_ERR("Failed to append redis command\n");
                    goto error;
                }

                db_redis_key_free(&query_v);
            }
        }

        db_redis_key_free(&type_keys);
        db_redis_key_free(&set_keys);
    }

    LM_DBG("getting replies for %d queries\n", update_queries);

    for (i = 0; i < update_queries; ++i) {
        if (db_redis_get_reply(con, (void**)&reply) != REDIS_OK) {
            LM_ERR("Failed to get reply for query: %s\n",
                    db_redis_get_error(con));
            goto error;
        }
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);
    }

    LM_DBG("done performing update\n");

    db_redis_key_free(&all_type_keys);
    db_redis_key_free(&new_type_keys);
    return 0;

error:
    LM_ERR("failed to perform the update\n");
    if (reply)
        db_redis_free_reply(&reply);
    db_redis_key_free(&query_v);
    db_redis_key_free(&all_type_keys);
    db_redis_key_free(&type_keys);
    db_redis_key_free(&set_keys);
    db_redis_key_free(&new_type_keys);
    return -1;
}


/*
 * Query table for specified rows
 * _h: structure representing database connection
 * _k: key names
 * _op: operators
 * _v: values of the keys that must match
 * _c: column names to return
 * _n: number of key=values pairs to compare
 * _nc: number of columns to return
 * _o: order by the specified column
 */
int db_redis_query(const db1_con_t* _h, const db_key_t* _k, const db_op_t* _op,
        const db_val_t* _v, const db_key_t* _c, const int _n, const int _nc,
        const db_key_t _o, db1_res_t** _r)
{
    km_redis_con_t *con = NULL;
    int free_op = 0;
    int do_table_scan = 0;
    uint64_t ts_scan_start = 0;
    str ts_scan_key = {0,};

    redis_key_t *keys = NULL;
    int keys_count = 0;
    int *manual_keys = NULL;
    int manual_keys_count = 0;
    db_op_t *query_ops = NULL;
    int i;

    // TODO: implement order-by
    // TODO: optimize mapping-based manual post-check (remove check for keys already
    // in type query key)

    if (!_r) {
        LM_ERR("db result is null\n");
        return -1;
    }

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("querying prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    *_r = NULL;

    // check if we have a version query, and return version directly from
    // schema instead of loading it from redis
    if (_nc == 1 && _n >= 1 && VAL_TYPE(&_v[0]) == DB1_STR &&
            CON_TABLE(_h)->len == strlen("version") && strncmp(CON_TABLE(_h)->s, "version", CON_TABLE(_h)->len) == 0 &&
            _k[0]->len == strlen("table_name") && strncmp(_k[0]->s, "table_name", _k[0]->len) == 0 &&
            _c[0]->len == strlen("table_version") && strncmp(_c[0]->s, "table_version", _c[0]->len) == 0) {

        if (db_redis_return_version(_h, con, &VAL_STR(&_v[0]), _r) == 0) {
            return 0;
        }
        // if we fail to return a version from the schema, go query the table, just in case
    }

    free_op = 0;

    if (_op == NULL) {
        char *op = "=";
        free_op = 1;
        query_ops = (db_op_t*)pkg_malloc(_n * sizeof(db_op_t));
        if (!query_ops) {
            LM_ERR("Failed to allocate memory for query op list\n");
            goto error;
        }
        for (i = 0; i < _n; ++i) {
            query_ops[i] = op;
        }
    } else {
        query_ops = (db_op_t*)_op;
    }

    if (_n > 0) {
        if (db_redis_build_query_keys(con, CON_TABLE(_h), _k, _v, query_ops, _n,
                    &keys, &keys_count, &manual_keys, &manual_keys_count, &do_table_scan, &ts_scan_start,
                    &ts_scan_key) != 0) {
            LM_ERR("failed to build query keys\n");
            goto error;
        }
        if (!keys_count) {
            if (do_table_scan) {
                LM_DBG("unable to build query keys, falling back to full table scan\n");
            } else {
                LM_DBG("query keys not member of suitable mapping, skip full table scan\n");
            }
        }
    } else {
        LM_DBG("no columns given to build query keys, falling back to full table scan\n");
        keys_count = 0;
        do_table_scan = 1;
    }

    if (db_redis_perform_query(_h, con, _k, _v, query_ops, _c, _n, _nc, _r,
        &keys, &keys_count, &manual_keys, &manual_keys_count, do_table_scan, ts_scan_start, &ts_scan_key) != 0) {
        goto error;
    }

    LM_DBG("done performing query\n");

    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);

    if (manual_keys) {
        pkg_free(manual_keys);
    }
    if (ts_scan_key.s)
        pkg_free(ts_scan_key.s);

    db_redis_consume_replies(con);
    return 0;

error:
    LM_ERR("failed to do the query\n");
    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys) {
        pkg_free(manual_keys);
    }
    if (ts_scan_key.s)
        pkg_free(ts_scan_key.s);
    db_redis_consume_replies(con);


    return -1;
}

/*
 * Execute a raw SQL query
 */
int db_redis_raw_query(const db1_con_t* _h, const str* _s, db1_res_t** _r)
{
    LM_DBG("perform redis raw query\n");
    return -1;
}

/*
 * Insert a row into specified table
 * _h: structure representing database connection
 * _k: key names
 * _v: values of the keys
 * _n: number of key=value pairs
 */
int db_redis_insert(const db1_con_t* _h, const db_key_t* _k, const db_val_t* _v, const int _n)
{
    km_redis_con_t *con = NULL;
    redis_key_t *key = NULL;
    int keys_count = 0;
    redis_key_t *type_keys = NULL;
    redis_key_t *set_keys = NULL;
    int type_keys_count = 0;
    redis_key_t *query_v = NULL;
    redisReply *reply = NULL;
    int i;
    redis_key_t *k;
    redis_key_t *set_key;

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("inserting to prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    if (db_redis_build_entry_keys(con, CON_TABLE(_h), _k, _v, _n,
                &key, &keys_count) != 0) {
        LM_ERR("failed to build entry keys\n");
        goto error;
    }
    if (db_redis_build_type_keys(con, CON_TABLE(_h), _k, _v, _n,
                &type_keys, &set_keys, &type_keys_count) != 0) {
        LM_ERR("failed to build type keys\n");
        goto error;
    }

    if (db_redis_key_add_string(&query_v, "HMSET", 5) != 0) {
        LM_ERR("Failed to add hmset command to insert query\n");
        goto error;
    }
    if (db_redis_key_add_str(&query_v, &key->key) != 0) {
        LM_ERR("Failed to add key to insert query\n");
        goto error;
    }

    for (i = 0; i < _n; ++i) {
        str *k = _k[i];
        str v;

        if (db_redis_key_add_str(&query_v, k) != 0) {
            LM_ERR("Failed to add column name to insert query\n");
            goto error;
        }
        if (db_redis_val2str(&(_v[i]), &v) != 0) {
            LM_ERR("Failed to allocate memory for query value\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &v) != 0) {
            LM_ERR("Failed to add column value to insert query\n");
            goto error;
        }
        if (v.s)
            pkg_free(v.s);
    }

    reply = db_redis_command_argv(con, query_v);
    db_redis_key_free(&query_v);
    db_redis_check_reply(con, reply, error);
    db_redis_free_reply(&reply);

    for (k = type_keys, set_key = set_keys; k; k = k->next, set_key = set_key->next) {
        str *type_key = &k->key;

        LM_DBG("inserting entry key '%.*s' to type map '%.*s'\n",
            key->key.len, key->key.s, type_key->len, type_key->s);

        if (db_redis_key_add_string(&query_v, "SADD", 4) != 0) {
            LM_ERR("Failed to set sadd command to post-insert query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, type_key) != 0) {
            LM_ERR("Failed to add map key to post-insert query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &key->key) != 0) {
            LM_ERR("Failed to set entry key to post-insert query\n");
            goto error;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);

        if (db_redis_key_add_string(&query_v, "SADD", 4) != 0) {
            LM_ERR("Failed to set sadd command to post-insert query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, &set_key->key) != 0) {
            LM_ERR("Failed to add map key to post-insert query\n");
            goto error;
        }
        if (db_redis_key_add_str(&query_v, type_key) != 0) {
            LM_ERR("Failed to set entry key to post-insert query\n");
            goto error;
        }

        reply = db_redis_command_argv(con, query_v);
        db_redis_key_free(&query_v);
        db_redis_check_reply(con, reply, error);
        db_redis_free_reply(&reply);
    }

    db_redis_key_free(&key);
    db_redis_key_free(&type_keys);
    db_redis_key_free(&set_keys);
    db_redis_consume_replies(con);

    return 0;

error:
    db_redis_key_free(&key);
    db_redis_key_free(&type_keys);
    db_redis_key_free(&set_keys);
    db_redis_key_free(&query_v);

    if (reply)
        db_redis_free_reply(&reply);

    db_redis_consume_replies(con);

    LM_ERR("failed to do the insert\n");
    return -1;
}

/*
 * Delete a row from the specified table
 * _h: structure representing database connection
 * _k: key names
 * _o: operators
 * _v: values of the keys that must match
 * _n: number of key=value pairs
 */
int db_redis_delete(const db1_con_t* _h, const db_key_t* _k,
        const db_op_t* _op, const db_val_t* _v, const int _n)
{
    km_redis_con_t *con = NULL;
    redis_key_t *keys = NULL;
    int keys_count = 0;
    int *manual_keys = NULL;
    int manual_keys_count = 0;
    int free_op = 0;
    int do_table_scan = 0;
    uint64_t ts_scan_start = 0;
    str ts_scan_key = {0,};
    db_op_t *query_ops = NULL;
    int i;

    // TODO: optimize mapping-based manual post-check (remove check for keys already
    // in type query key)

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("deleting from prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    free_op = 0;

    if (_op == NULL) {
        char *op = "=";
        free_op = 1;
        query_ops = (db_op_t*)pkg_malloc(_n * sizeof(db_op_t));
        if (!query_ops) {
            LM_ERR("Failed to allocate memory for query op list\n");
            goto error;
        }
        for (i = 0; i < _n; ++i) {
            query_ops[i] = op;
        }
    } else {
        query_ops = (db_op_t*)_op;
    }

    if (_n > 0) {
        if (db_redis_build_query_keys(con, CON_TABLE(_h), _k, _v, query_ops, _n,
                    &keys, &keys_count, &manual_keys, &manual_keys_count, &do_table_scan, &ts_scan_start,
                    &ts_scan_key) != 0) {
            LM_ERR("failed to build query keys\n");
            goto error;
        }
        if (!keys_count) {
            if (do_table_scan) {
                LM_DBG("unable to build query keys, falling back to full table scan\n");
            } else {
                LM_DBG("query keys not member of suitable mapping, skip full table scan\n");
            }
        }
    } else {
        LM_DBG("no columns given to build query keys, falling back to full table scan\n");
        keys_count = 0;
        do_table_scan = 1;
    }

    if (db_redis_perform_delete(_h, con, _k, _v, query_ops, _n,
        &keys, &keys_count, &manual_keys, &manual_keys_count, do_table_scan, ts_scan_start, &ts_scan_key) != 0) {
        goto error;
    }

    LM_DBG("done performing delete\n");

    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys)
        pkg_free(manual_keys);
    if (ts_scan_key.s)
        pkg_free(ts_scan_key.s);
    db_redis_consume_replies(con);

    return 0;

error:
    LM_ERR("failed to do the query\n");
    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys)
        pkg_free(manual_keys);
    if (ts_scan_key.s)
        pkg_free(ts_scan_key.s);
    db_redis_consume_replies(con);
    return -1;
}

/*
 * Update some rows in the specified table
 * _h: structure representing database connection
 * _k: key names
 * _op: operators
 * _v: values of the keys that must match
 * _uk: updated columns
 * _uv: updated values of the columns
 * _n: number of key=value pairs
 * _un: number of columns to update
 */
int db_redis_update(const db1_con_t* _h, const db_key_t* _k,
        const db_op_t* _op, const db_val_t* _v, const db_key_t* _uk,
        const db_val_t* _uv, const int _n, const int _nu)
{
    km_redis_con_t *con = NULL;
    int free_op = 0;
    int do_table_scan = 0;
    uint64_t ts_scan_start = 0;
    str ts_scan_key = {0,};

    redis_key_t *keys = NULL;
    int keys_count = 0;
    int *manual_keys = NULL;
    int manual_keys_count = 0;
    db_op_t *query_ops = NULL;
    int i;

    // TODO: optimize mapping-based manual post-check (remove check for keys already
    // in type query key)

    con = REDIS_CON(_h);
    if (con && con->con == NULL) {
        if (db_redis_connect(con) != 0) {
            LM_ERR("Failed to reconnect to server\n");
            return -1;
        }
    }
    if(con == NULL || con->id == NULL || con->con == NULL) {
        LM_ERR("connection to server is null\n");
        return -1;
    }
    if(CON_TABLE(_h)->s == NULL) {
        LM_ERR("prefix (table) name not set\n");
        return -1;
    } else {
        LM_DBG("updating prefix (table) '%.*s'\n",
                CON_TABLE(_h)->len, CON_TABLE(_h)->s);
    }

    free_op = 0;

    if (_op == NULL) {
        char *op = "=";
        free_op = 1;
        query_ops = (db_op_t*)pkg_malloc(_n * sizeof(db_op_t));
        if (!query_ops) {
            LM_ERR("Failed to allocate memory for query op list\n");
            goto error;
        }
        for (i = 0; i < _n; ++i) {
            query_ops[i] = op;
        }
    } else {
        query_ops = (db_op_t*)_op;
    }

    if (_n > 0) {
        if (db_redis_build_query_keys(con, CON_TABLE(_h), _k, _v, query_ops, _n,
                    &keys, &keys_count, &manual_keys, &manual_keys_count, &do_table_scan, &ts_scan_start,
                    &ts_scan_key) != 0) {
            LM_ERR("failed to build query keys\n");
            goto error;
        }
        if (!keys_count) {
            if (do_table_scan) {
                LM_DBG("unable to build query keys, falling back to full table scan\n");
            } else {
                LM_DBG("query keys not member of suitable mapping, skip full table scan\n");
            }
        }
    } else {
        LM_DBG("no columns given to build query keys, falling back to full table scan\n");
        keys_count = 0;
    }

    if (db_redis_perform_update(_h, con, _k, _v, query_ops, _uk, _uv, _n, _nu,
        &keys, &keys_count, &manual_keys, &manual_keys_count, do_table_scan, ts_scan_start, &ts_scan_key) != 0) {
        goto error;
    }

    LM_DBG("done performing update\n");

    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);

    if (manual_keys) {
        pkg_free(manual_keys);
    }
    if (ts_scan_key.s)
        pkg_free(ts_scan_key.s);
    db_redis_consume_replies(con);
    return 0;

error:
    LM_ERR("failed to do the query\n");
    if (free_op && query_ops) {
        pkg_free(query_ops);
    }
    db_redis_key_free(&keys);
    if (manual_keys) {
        pkg_free(manual_keys);
    }
    if (ts_scan_key.s)
        pkg_free(ts_scan_key.s);
    db_redis_consume_replies(con);
    return -1;
}

/*
 * Just like insert, but replace the row if it exists
 */
int db_redis_replace(const db1_con_t* _h, const db_key_t* _k,
        const db_val_t* _v, const int _n, const int _un, const int _m) {
    LM_DBG("perform redis replace\n");
    return -1;
}

/*
 * Store name of table that will be used by
 * subsequent database functions
 */
int db_redis_use_table(db1_con_t* _h, const str* _t) {
    return db_use_table(_h, _t);
}

int db_redis_free_result(db1_con_t* _h, db1_res_t* _r) {
    LM_DBG("perform redis free result\n");

    if(!_r)
        return -1;
    db_free_result(_r);
    return 0;
}