#include "utable.h"

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

#include "mpack/mpack.h"

#define YYJSON_HAS_STDINT_H 1
#define YYJSON_DISABLE_READER 1
#define YYJSON_HAS_STDBOOL_H 1
#include "yyjson/yyjson.h"

#include "uthash/uthash.h"
#include "base64/b64.h"

#include "nmx_pool/mempool.h"

#ifdef DEBUG_MODE
    #define DEBUG_PRINT(...) printf(__VA_ARGS__)
#else
    #define DEBUG_PRINT(...) 
#endif

#define ALLOC(number, type) ((type *)calloc(sizeof(type), number))
#define	FREE(p)	{free(p);p=NULL;}

#define MEMPOOL_TYPE mem_pool_t

#define MEMPOOL_CREATE(pool_size) create_mem_pool(pool_size)
#define MEMPOOL_DESTROY(pool) destroy_mem_pool(pool)

#define MEMPOOL_ALLOC(pool, number, type) ((type *)mem_alloc(pool, sizeof(type) * number))
#define MEMPOOL_FREE(pool, p) mem_free(pool, p)


#define MEM_ALLOC(pool, number, type) ((pool) ? MEMPOOL_ALLOC(pool, number, type) : ALLOC(number, type))
#define MEM_FREE(pool, p) do { \
    if (pool) { \
        MEMPOOL_FREE(pool, p); \
    } else { \
        FREE(p); \
    } \
} while (0)


static char*MEM_STRDUP(MEMPOOL_TYPE  *mempool, const char* s, size_t len) 
{
    char* new_str = MEM_ALLOC(mempool, len + 1, char);  
    memcpy(new_str, s, len + 1);
    return new_str;  
}

struct utable_item
{
    struct utable_kv *kv;
    UT_hash_handle hh;
};

struct utable 
{
    struct utable_item *items;
    struct utable_item *iter;
    MEMPOOL_TYPE *mempool;
    struct utable_stat stat;
};

struct utable *utable_new_with_size(size_t sz)
{
    struct utable *table = ALLOC(1, struct utable);
    table->mempool = MEMPOOL_CREATE(sz);
    return table; 
}

#define UTABLE_INITIAL_MEMPOOL_SIZE 2 * 1024

inline struct utable *utable_new(void)
{
    return utable_new_with_size(UTABLE_INITIAL_MEMPOOL_SIZE);
}

void utable_free(struct utable *table)
{
    if(table->items) HASH_CLEAR(hh, table->items);
    if(table->mempool) MEMPOOL_DESTROY(table->mempool);
    FREE(table);
}

inline void utable_stat(struct utable *table, struct utable_stat *stat)
{
    if(stat == NULL)
        return;
    memcpy(stat, &table->stat, sizeof(struct utable_stat));
}



struct utable_kv *utable_kv_new_with_cstring_from_mempool(MEMPOOL_TYPE *pool, const char *key, size_t key_sz, const char *value, size_t value_sz)
{
    struct utable_kv *kv = MEM_ALLOC(pool, 1, struct utable_kv);
    kv->key = MEM_STRDUP(pool, key, key_sz);
    kv->key_sz = key_sz;
    kv->value_type = utable_value_type_cstring;
    kv->cstring = MEM_STRDUP(pool, value, value_sz);
    kv->cstring_sz = value_sz;
    kv->value_sz = value_sz;
    return kv;
}

struct utable_kv *utable_kv_new_with_blob_from_mempool(MEMPOOL_TYPE *pool,const char *key, size_t key_sz, const char *blob, size_t blob_sz)
{
    struct utable_kv *kv = MEM_ALLOC(pool,1, struct utable_kv);
    kv->key = MEM_STRDUP(pool, key, key_sz);
    kv->key_sz = key_sz;
    kv->value_type = utable_value_type_blob;
    kv->blob = MEM_ALLOC(pool, blob_sz, char);
    memcpy(kv->blob, blob, blob_sz);
    kv->blob_sz = blob_sz;
    kv->value_sz = blob_sz;
    return kv;
}
struct utable_kv *utable_kv_new_with_integer_from_mempool(MEMPOOL_TYPE *pool, const char *key, size_t key_sz, int64_t value)
{
    struct utable_kv *kv = MEM_ALLOC(pool,1, struct utable_kv);
    kv->key = MEM_STRDUP(pool, key, key_sz);
    kv->key_sz = key_sz;
    kv->value_type = utable_value_type_integer;
    kv->integer = value;
    kv->value_sz = sizeof(int64_t);
    return kv;
}
struct utable_kv *utable_kv_new_with_integer_array_from_mempool(MEMPOOL_TYPE *pool, const char *key, size_t key_sz, int64_t value[], size_t n_value)
{
    struct utable_kv *kv = MEM_ALLOC(pool,1, struct utable_kv);
    kv->key = MEM_STRDUP(pool, key, key_sz);
    kv->key_sz = key_sz;
    kv->value_type = utable_value_type_integer_array;
    kv->interger_array = MEM_ALLOC(pool, n_value, int64_t);
    memcpy(kv->interger_array, value, sizeof(int64_t) * n_value);
    kv->n_integer = n_value;
    kv->value_sz = sizeof(int64_t) * n_value;
    return kv;
}
struct utable_kv *utable_kv_new_with_cstring_array_from_mempool(MEMPOOL_TYPE *pool,const char *key, size_t key_sz, const char* value[], size_t value_sz[], size_t n_value)
{
    struct utable_kv *kv = MEM_ALLOC(pool,1, struct utable_kv);
    kv->key = MEM_STRDUP(pool, key, key_sz);
    kv->key_sz = key_sz;
    kv->value_type = utable_value_type_cstring_array;
    kv->cstring_array = MEM_ALLOC(pool, n_value, char *);
    kv->cstring_array_sz = MEM_ALLOC(pool, n_value, size_t);
    kv->value_sz = 0;
    for(size_t i =0; i < n_value; i++)
    {
        kv->cstring_array[i] = MEM_STRDUP(pool, value[i],  value_sz[i]);
        kv->cstring_array_sz[i] = value_sz[i];
        kv->value_sz += value_sz[i];
    }
    kv->n_cstring = n_value;
    return kv;
}

void utable_kv_free_from_pool(MEMPOOL_TYPE *pool, struct utable_kv *kv)
{
    if(kv->key) MEM_FREE(pool, kv->key);
    MEM_FREE(pool, kv);
}

struct utable_kv *utable_kv_new_with_cstring(const char *key, size_t key_sz, const char *value, size_t value_sz)
{
    return utable_kv_new_with_cstring_from_mempool(NULL, key, key_sz, value, value_sz);
}

struct utable_kv *utable_kv_new_with_blob(const char *key, size_t key_sz, const char *blob, size_t blob_sz)
{
    return utable_kv_new_with_blob_from_mempool(NULL, key, key_sz, blob, blob_sz);
}
struct utable_kv *utable_kv_new_with_integer(const char *key, size_t key_sz, int64_t value)
{
    return utable_kv_new_with_integer_from_mempool(NULL, key, key_sz, value);
}
struct utable_kv *utable_kv_new_with_integer_array(const char *key, size_t key_sz, int64_t value[], size_t n_value)
{
    return utable_kv_new_with_integer_array_from_mempool(NULL, key, key_sz, value, n_value);
}
struct utable_kv *utable_kv_new_with_cstring_array(const char *key, size_t key_sz, const char* value[], size_t value_sz[], size_t n_value)
{
    return utable_kv_new_with_cstring_array_from_mempool(NULL, key, key_sz, value, value_sz, n_value);
}

void utable_kv_free(struct utable_kv *kv)
{
    return utable_kv_free_from_pool(NULL, kv);
}

struct utable_kv *utable_kv_duplicate_from_mempool(MEMPOOL_TYPE *pool, const struct utable_kv *kv)
{
    struct utable_kv *new_kv = NULL;
    switch (kv->value_type) 
    {
        case utable_value_type_cstring:
            new_kv=utable_kv_new_with_cstring_from_mempool(pool, kv->key, kv->key_sz, kv->cstring, kv->cstring_sz);
            break;
        case utable_value_type_blob:
            new_kv=utable_kv_new_with_blob_from_mempool(pool, kv->key, kv->key_sz, kv->blob, kv->blob_sz);
            break;
        case utable_value_type_integer:
            new_kv=utable_kv_new_with_integer_from_mempool(pool, kv->key, kv->key_sz, kv->integer);
            break;
        case utable_value_type_integer_array:
            new_kv=utable_kv_new_with_integer_array_from_mempool(pool, kv->key, kv->key_sz, kv->interger_array, kv->n_integer);
            break;
        case utable_value_type_cstring_array:
            new_kv=utable_kv_new_with_cstring_array_from_mempool(pool, kv->key, kv->key_sz, (const char **)kv->cstring_array, kv->cstring_array_sz, kv->n_cstring);
            break;
        default:
            break;
    }
    return new_kv;
}

struct utable_kv *utable_kv_duplicate(const struct utable_kv *kv)
{
    return utable_kv_duplicate_from_mempool(NULL, kv);
}

void utable_add_kv(struct utable *table, struct utable_kv *kv)
{
    struct utable_item *item;
    HASH_FIND(hh, table->items, kv->key, kv->key_sz, item);
    if (item) {
        DEBUG_PRINT("ERR: key %s already exists\n", kv->key);
        return;
    }
    item = MEM_ALLOC(table->mempool, 1, struct utable_item);
    item->kv = utable_kv_duplicate_from_mempool(table->mempool, kv);
    HASH_ADD_KEYPTR(hh, table->items, item->kv->key, item->kv->key_sz, item);
}

void utable_add_kv_array(struct utable *table, struct utable_kv *kv_array, size_t n_kv)
{
    for (size_t i = 0; i < n_kv; i++) {
        utable_add_kv(table, &kv_array[i]);
    }
}

struct utable_kv *utable_get0_kv(struct utable *table, const char *key, size_t key_sz)
{
    struct utable_item *item;
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        return item->kv;
    }
    return NULL;
}

struct utable_kv *utable_next_kv(struct utable *table)
{
    if (table->iter == NULL) {
        table->iter = table->items;
    } else {
        table->iter = (struct utable_item *)table->iter->hh.next;
    }
    if (table->iter == NULL) {
        return NULL;
    }
    return table->iter->kv;
}


static void utable_item_stat_add(struct utable_stat *stat, struct utable_kv *item)
{
    if(stat==NULL || item == NULL)
        return;
    stat->n_item++;
    stat->n_item_size += item->key_sz;
    stat->n_item_size += item->value_sz;
    switch (item->value_type) 
    {
        case utable_value_type_cstring:
            stat->n_cstring++;
            stat->n_cstring_size += item->value_sz;
            break;
        case utable_value_type_blob:
            stat->n_blob++;
            stat->n_blob_size += item->value_sz;
            break;
        case utable_value_type_integer:
            stat->n_integer++;
            stat->n_integer_size += item->value_sz;
            break;
        case utable_value_type_integer_array:
            stat->n_integer_array++;
            stat->n_integer_array_size += item->value_sz;
            break;
        case utable_value_type_cstring_array:
            stat->n_cstring_array++;
            stat->n_cstring_array_size += item->value_sz;
            break;            
        default:
            break;
    }
}

static void utable_item_stat_sub(struct utable_stat *stat, struct utable_kv *item)
{
    if(stat==NULL || item == NULL)
        return;
    stat->n_item--;
    stat->n_item_size -= item->key_sz;
    stat->n_item_size -= item->value_sz;
    switch (item->value_type) 
    {
        case utable_value_type_cstring:
            stat->n_cstring--;
            stat->n_cstring_size -= item->cstring_sz;
            break;
        case utable_value_type_blob:
            stat->n_blob--;
            stat->n_blob_size -= item->blob_sz;
            break;
        case utable_value_type_integer:
            stat->n_integer--;
            stat->n_integer_size -= sizeof(item->integer);
            break;
        case utable_value_type_integer_array:
            stat->n_integer_array--;
            stat->n_integer_array_size -= sizeof(int64_t) * item->n_integer;
            break;
        case utable_value_type_cstring_array:
            stat->n_cstring_array--;
            stat->n_cstring_array_size -= item->value_sz;
            break;               
        default:
            break;
    }
}

void utable_add_cstring(struct utable *table, const char *key, const char *value, size_t value_sz)
{
    // check if key already exists
    struct utable_item *item;
    size_t key_sz = strlen(key);
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        DEBUG_PRINT("ERR: key %s already exists\n", key);
        return;
    }
    item = MEMPOOL_ALLOC(table->mempool ,1,struct utable_item);
    item->kv= utable_kv_new_with_cstring(key, key_sz, value, value_sz);
    HASH_ADD_KEYPTR(hh, table->items, item->kv->key, item->kv->key_sz, item);
    utable_item_stat_add(&table->stat, item->kv);
}

void utable_add_blob(struct utable *table, const char *key, const char *blob, size_t blob_sz)
{
    // check if key already exists
    struct utable_item *item;
    size_t key_sz = strlen(key);
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        DEBUG_PRINT("ERR: key %s already exists\n", key);
        return;
    }
    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
    item->kv = utable_kv_new_with_blob(key, key_sz, blob, blob_sz);

    HASH_ADD_KEYPTR(hh, table->items, item->kv->key, item->kv->key_sz, item);
    utable_item_stat_add(&table->stat, item->kv);
}

void utable_add_integer(struct utable *table, const char *key, int64_t value)
{
    // check if key already exists
    struct utable_item *item;
    size_t key_sz = strlen(key);
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        DEBUG_PRINT("ERR: key %s already exists\n", key);
        return;
    }
    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
    item->kv = utable_kv_new_with_integer(key, key_sz, value);

    HASH_ADD_KEYPTR(hh, table->items, item->kv->key, item->kv->key_sz, item);
    utable_item_stat_add(&table->stat, item->kv);
}

void utable_add_integer_array(struct utable *table, const char *key, int64_t value_array[], size_t n_value)
{
    // check if key already exists
    struct utable_item *item;
    size_t key_sz = strlen(key);
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        DEBUG_PRINT("ERR: key %s already exists\n", key);
        return;
    }
    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
    item->kv = utable_kv_new_with_integer_array(key, key_sz, value_array, n_value);

    HASH_ADD_KEYPTR(hh, table->items, item->kv->key, item->kv->key_sz, item);
    utable_item_stat_add(&table->stat, item->kv); 
}

void utable_add_cstring_array(struct utable *table, const char *key, const char* value_array[], size_t value_sz[], size_t n_value)
{
    // check if key already exists
    struct utable_item *item;
    size_t key_sz = strlen(key);
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        DEBUG_PRINT("ERR: key %s already exists\n", key);
        return;
    }
    item = MEMPOOL_ALLOC(table->mempool ,1, struct utable_item);
    item->kv = utable_kv_new_with_cstring_array(key, key_sz, value_array, value_sz, n_value);

    HASH_ADD_KEYPTR(hh, table->items, item->kv->key, item->kv->key_sz, item);
    utable_item_stat_add(&table->stat, item->kv);   
}

void utable_delete(struct utable *table, const char *key, size_t key_sz)
{
    struct utable_item *item;
    HASH_FIND(hh, table->items, key, key_sz, item);
    if (item) {
        HASH_DEL(table->items, item);
        utable_item_stat_sub(&table->stat, item->kv);
        utable_kv_free(item->kv);
        MEMPOOL_FREE(table->mempool,item);
    }
}

void utable_reset_iter(struct utable *table)
{
    table->iter = NULL;
}

const char *utable_next_key(struct utable *table)
{
    if (table->iter == NULL) {
        table->iter = table->items;
    } else {
        table->iter = (struct utable_item *)table->iter->hh.next;
    }
    if (table->iter == NULL) {
        return NULL;
    }
    return table->iter->kv->key;
}

enum utable_value_type utable_get_value_type(const struct utable *table, const char *key)
{
    struct utable_item *item;
    HASH_FIND_STR(table->items, key, item);
    if (item) {
        return item->kv->value_type;
    }
    return utable_value_type_undefined;
}

int utable_get0_blob_value(const struct utable *table, const char *key, char **value, size_t *value_len)
{
    struct utable_item *item;
    HASH_FIND_STR(table->items, key, item);
    if (item) {
        if (item->kv->value_type == utable_value_type_blob) {
            *value = item->kv->blob;
            *value_len = item->kv->blob_sz;
            return 0;
        }
    }
    return -1;
}

int utable_get0_cstring_value(const struct utable *table, const char *key, char **value, size_t *value_len)
{
    struct utable_item *item;
    HASH_FIND_STR(table->items, key, item);
    if (item) {
        if (item->kv->value_type == utable_value_type_cstring) {
            *value = item->kv->cstring;
            *value_len = item->kv->cstring_sz;
            return 0;
        }
    }
    return -1;
}

int utable_get0_integer_value(const struct utable *table, const char *key, int64_t *value)
{
    struct utable_item *item;
    HASH_FIND_STR(table->items, key, item);
    if (item) {
        if (item->kv->value_type == utable_value_type_integer) {
            *value = item->kv->integer;
            return 0;
        }
    }
    return -1;
}

int utable_get0_integer_value_array(const struct utable *table, const char *key, int64_t **value_array, size_t *n_value)
{
    struct utable_item *item;
    HASH_FIND_STR(table->items, key, item);
    if (item) {
        if (item->kv->value_type == utable_value_type_integer_array) {
            *value_array = item->kv->interger_array;
            *n_value = item->kv->n_integer;
            return 0;
        }
    }
    return -1;
}

int utable_get0_cstring_value_array(const struct utable *table, const char *key, char ***value_array, size_t **value_len, size_t *n_value)
{
    struct utable_item *item;
    HASH_FIND_STR(table->items, key, item);
    if (item) {
        if (item->kv->value_type == utable_value_type_cstring_array) {
            *value_array = item->kv->cstring_array;
            *value_len = item->kv->cstring_array_sz;
            *n_value = item->kv->n_cstring;
            return 0;
        }
    }
    return -1;
}

struct utable_item *utable_item_dup(MEMPOOL_TYPE *mempool, const struct utable_item *item)
{
    struct utable_item *new_item = MEMPOOL_ALLOC(mempool, 1, struct utable_item);
    new_item->kv = utable_kv_duplicate_from_mempool(mempool, item->kv);
    return new_item;
}

struct utable *utable_duplicate(const struct utable *table)
{
    struct utable *new_table = utable_new();
    struct utable_item *item, *tmp;
    HASH_ITER(hh, table->items, item, tmp) {
        struct utable_item *new_item = utable_item_dup(new_table->mempool, item);
        HASH_ADD_KEYPTR(hh, new_table->items, new_item->kv->key, new_item->kv->key_sz, new_item);
    }
    new_table->stat = table->stat;
    return new_table;
}

int utable_union(struct utable *dst, const struct utable *src)
{
    struct utable_item *item, *tmp;
    HASH_ITER(hh, src->items, item, tmp) {
        struct utable_item *dst_item;
        HASH_FIND(hh, dst->items, item->kv->key, item->kv->key_sz, dst_item);
        if (dst_item) {
            continue;
        }
        struct utable_item *new_item = utable_item_dup(dst->mempool, item);
        HASH_ADD_KEYPTR(hh, dst->items, new_item->kv->key, new_item->kv->key_sz, new_item);
        utable_item_stat_add(&dst->stat, item->kv);
    }
    return 0;
}

int utable_json_export(const struct utable *table, char **blob, size_t *blob_len)
{
    yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL);
    yyjson_mut_val *root = yyjson_mut_obj(doc);
    yyjson_mut_doc_set_root(doc, root);
    char *encs[HASH_COUNT(table->items)];
    struct utable_item *item, *tmp;
    int enc_cnt = 0;
    HASH_ITER(hh, table->items, item, tmp) {
        switch (item->kv->value_type) {
            case utable_value_type_cstring: {
                    yyjson_mut_obj_add_str(doc, root, item->kv->key, item->kv->cstring); // key and cstring are shallow copied
                }
                break;
            case utable_value_type_blob: {
                    char *enc = b64_encode((unsigned char *)item->kv->blob, item->kv->blob_sz);
                    yyjson_mut_obj_add_str(doc, root, item->kv->key, enc);
                    // do not free enc now, it is shallow copied
                    encs[enc_cnt++] = enc;
                }
                break;
            case utable_value_type_integer: {
                    yyjson_mut_obj_add_int(doc, root, item->kv->key, item->kv->integer);
                }
                break;
            case utable_value_type_integer_array: {
                    yyjson_mut_val *arr = yyjson_mut_arr_with_sint64(doc, item->kv->interger_array, item->kv->n_integer);
                    yyjson_mut_obj_add_val(doc, root, item->kv->key, arr);
            }
                break;
            case utable_value_type_cstring_array: {
                    yyjson_mut_val *arr = yyjson_mut_arr_with_strn(doc, (const char **)item->kv->cstring_array, item->kv->cstring_array_sz, item->kv->n_cstring);
                    yyjson_mut_obj_add_val(doc, root, item->kv->key, arr);
            }
                break;                
            default:
                break;
        }
    }

    char *json_str = yyjson_mut_write(doc, YYJSON_WRITE_ALLOW_INVALID_UNICODE, blob_len);
    yyjson_mut_doc_free(doc);
    *blob = json_str;
    for (int j = 0; j < enc_cnt; j++) {
        free(encs[j]);
    }
    return 0;
}

/*
    [{
        "key":<tag key string>
        "type":"i_arr", "c_str", or "blob"
        "value":<
    }]
*/
int utable_msgpack_export(const struct utable *table, char **blob, size_t *blob_len)
{
    mpack_writer_t writer;
    mpack_writer_init_growable(&writer, blob, blob_len);
    mpack_start_array(&writer, HASH_COUNT(table->items));

    struct utable_item *item, *tmp;
    HASH_ITER(hh, table->items, item, tmp) {
        mpack_start_map(&writer, 3);
        mpack_write_cstr(&writer, "key");
        mpack_write_cstr(&writer, item->kv->key);

        mpack_write_cstr(&writer, "type");
        switch (item->kv->value_type) {
            case utable_value_type_cstring:
                mpack_write_cstr(&writer, "c_str");
                break;
            case utable_value_type_blob:
                mpack_write_cstr(&writer, "blob");
                break;
            case utable_value_type_integer:
                mpack_write_cstr(&writer, "i");
                break;
            case utable_value_type_integer_array:
                mpack_write_cstr(&writer, "i_arr");
                break;
            default:
                break;
        }

        mpack_write_cstr(&writer, "value");
        switch (item->kv->value_type) {
            case utable_value_type_cstring:
                mpack_write_bin(&writer, item->kv->cstring, item->kv->cstring_sz);
                break;
            case utable_value_type_blob:
                mpack_write_bin(&writer, item->kv->blob, item->kv->blob_sz);
                break;
            case utable_value_type_integer:
                mpack_write_i64(&writer, item->kv->integer);
                break;
            case utable_value_type_integer_array:
                mpack_start_array(&writer, item->kv->n_integer);
                for (size_t i = 0; i < item->kv->n_integer; i++) {
                    mpack_write_i64(&writer, item->kv->interger_array[i]);
                }
                mpack_finish_array(&writer);
                break;
            default:
                break;
        }

        mpack_finish_map(&writer);
    }

    mpack_finish_array(&writer);
    if (mpack_writer_destroy(&writer) != mpack_ok) {
        DEBUG_PRINT("ERR mpack writer fieldtag_list_serialize destroy failed\n");
        return -1;
    }
    return 0;
}

void utable_serialize(const struct utable *table, char **blob, size_t *blob_len) {
    utable_msgpack_export(table, blob, blob_len);
}

char *get_cstring_from_mpack_node(MEMPOOL_TYPE *mempool ,mpack_node_t node)
{
    size_t len = mpack_node_strlen(node);
    char *str = MEMPOOL_ALLOC(mempool, len + 1, char);
    memcpy(str, mpack_node_str(node), len);
    str[len] = '\0';
    return str;
}

static void utable_item_free(MEMPOOL_TYPE *mempool, struct utable_item *item)
{
    if(item == NULL)
        return;
    if(item->kv->key)MEMPOOL_FREE(mempool, item->kv->key);
    switch (item->kv->value_type) {
        case utable_value_type_cstring:
            MEMPOOL_FREE(mempool,item->kv->cstring);
            break;
        case utable_value_type_blob:
            MEMPOOL_FREE(mempool,item->kv->blob);
            break;
        case utable_value_type_integer_array:
            MEMPOOL_FREE(mempool,item->kv->interger_array);
            break;
        case utable_value_type_cstring_array:
            for(size_t i=0; i < item->kv->n_cstring; i++)
            {
                MEMPOOL_FREE(mempool,item->kv->cstring_array[i]);
            }
            MEMPOOL_FREE(mempool,item->kv->cstring_array_sz);
            break;            
        default:
            break;
    }
}

struct utable *utable_deserialize(const char *blob, size_t blob_len)
{
    mpack_tree_t tree;
    mpack_tree_init_data(&tree, blob, blob_len);
    mpack_tree_parse(&tree);
    mpack_node_t root = mpack_tree_root(&tree);
    if (mpack_node_type(root) != mpack_type_array) {
        DEBUG_PRINT("ERR mpack root type is not array\n");
        return NULL;
    }
    size_t n_item = mpack_node_array_length(root);

    struct utable *table = utable_new();
    for (size_t i = 0; i < n_item; i++) {
        mpack_node_t item = mpack_node_array_at(root, i);
        mpack_node_t key = mpack_node_map_cstr(item, "key");
        mpack_node_t type = mpack_node_map_cstr(item, "type");
        mpack_node_t value = mpack_node_map_cstr(item, "value");
        char *type_str = get_cstring_from_mpack_node(table->mempool, type);

        struct utable_item *new_item = MEMPOOL_ALLOC(table->mempool,1, struct utable_item);
        new_item->kv=MEMPOOL_ALLOC(table->mempool,1, struct utable_kv);
        new_item->kv->key = get_cstring_from_mpack_node(table->mempool, key);
        new_item->kv->key_sz = strlen(new_item->kv->key);
        new_item->kv->value_sz = 0;
    
        if (strcmp(type_str, "c_str") == 0) {
            new_item->kv->value_type = utable_value_type_cstring;
            new_item->kv->cstring = MEMPOOL_ALLOC(table->mempool,mpack_node_bin_size(value), char);
            new_item->kv->cstring_sz = mpack_node_bin_size(value);
            new_item->kv->value_sz = new_item->kv->cstring_sz;
            memcpy(new_item->kv->cstring, mpack_node_bin_data(value), mpack_node_bin_size(value));
        } else if (strcmp(type_str, "blob") == 0) {
            new_item->kv->value_type = utable_value_type_blob;
            new_item->kv->blob = MEMPOOL_ALLOC(table->mempool,mpack_node_bin_size(value), char);
            new_item->kv->blob_sz = mpack_node_bin_size(value);
            new_item->kv->value_sz = new_item->kv->blob_sz;
            memcpy(new_item->kv->blob, mpack_node_bin_data(value), mpack_node_bin_size(value));
        } else if (strcmp(type_str, "i") == 0) {
            new_item->kv->value_type = utable_value_type_integer;
            new_item->kv->integer = mpack_node_i64(value);
        } else if (strcmp(type_str, "i_arr") == 0) {
            new_item->kv->value_type = utable_value_type_integer_array;
            new_item->kv->n_integer = mpack_node_array_length(value);
            new_item->kv->interger_array = MEMPOOL_ALLOC(table->mempool,new_item->kv->n_integer, int64_t);
            for (size_t j = 0; j < new_item->kv->n_integer; j++) {
                new_item->kv->interger_array[j] = mpack_node_i64(mpack_node_array_at(value, j));
            }
            new_item->kv->value_sz = sizeof(int64_t) * new_item->kv->n_integer;
        } else {
            DEBUG_PRINT("ERR unknown type %s\n", type_str);
            mpack_tree_destroy(&tree);
            MEMPOOL_FREE(table->mempool, type_str);
            utable_item_free(table->mempool, new_item);
            return NULL;
        }

        HASH_ADD_KEYPTR(hh, table->items, new_item->kv->key, new_item->kv->key_sz, new_item);
        utable_item_stat_add(&table->stat, new_item->kv);
        MEMPOOL_FREE(table->mempool, type_str);
    }

    mpack_tree_destroy(&tree);
    return table;
}