/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
#include "postgres.h"

#ifndef USE_INTERNAL_FTS

#include <stdio.h>
#include <stdbool.h>
#include <string.h>
#include <curl/curl.h>
#include <jansson.h>
#include <pthread.h>
#include <stdint.h>

#include "utils/etcd.h"
#include "utils/etcdlib.h"
#include "common/base64.h"
#include "common/fe_memutils.h"

#define ETCD_JSON_NODE                  "node"
#define ETCD_JSON_PREVNODE              "prevNode"
#define ETCD_JSON_NODES                 "nodes"
#define ETCD_JSON_ACTION                "action"
#define ETCD_JSON_KEY                   "key"
#define ETCD_JSON_VALUE                 "value"
#define ETCD_JSON_DIR                   "dir"
#define ETCD_JSON_MODIFIEDINDEX         "modifiedIndex"
#define ETCD_JSON_INDEX                 "index"
#define ETCD_JSON_ERRORCODE             "errorCode"
#define ETCD_HEADER_INDEX               "X-Etcd-Index: "

#define ETCD_JSON_V3_HEADER             "header"
#define ETCD_JSON_V3_KVS                "kvs"
#define ETCD_JSON_V3_COUNT              "count"
#define ETCD_JSON_V3_DELETED            "deleted"
#define ETCD_JSON_V3_REVISION           "revision"
#define ETCD_JSON_V3_MOD_REVISION       "mod_revision"
#define ETCD_JSON_V3_LEASE_ID           "ID"
#define ETCD_JSON_V3_LEASE_TTL          "TTL"
#define ETCD_JSON_V3_MEMBER_ID           "member_id"
#define ETCD_JSON_V3_LEADER          "leader"

#define MAX_OVERHEAD_LENGTH           64
#define DEFAULT_CURL_TIMEOUT          10
#define DEFAULT_CURL_CONNECT_TIMEOUT  10

#define DEFAULT_ENABLE_ETCD_V3_API 1

#define ETCD_V32_URL_PREFIX "v3alpha"
#define ETCD_V33_URL_PREFIX "v3beta"
#define ETCD_V34_URL_PREFIX "v3"
#define DEFAULT_ETCD_URL_PREFIX ETCD_V33_URL_PREFIX

#define MAX_GLOBAL_HOSTNAME 128
#define MAX_ETCD_ENDPOINTS 32
#define MAX_ETCD_FAILOVER_RETRY 60
#define MAX_ETCD_FAILOVER_LOCK_TIMEOUT 10

#define NO_SUPPORT_IN_V2(ectd) \
    do { \
        if (!etcdlib_enable_v3(ectd)) return ETCDLIB_NOSUPPORT_IN_V2; \
    } \
    while(0);

#define NO_SUPPORT_IN_V3(ectd) \
    do { \
        if (etcdlib_enable_v3(ectd)) return ETCDLIB_NOSUPPORT_IN_V3; \
    } \
    while(0);

typedef enum {
    GET, PUT, DELETE, POST
} request_t;

static etcdlib_t *g_etcdlib = NULL;
static etcdlib_t *etcdlib_endpoints_handler[MAX_ETCD_ENDPOINTS] = {0};
static int etcdlib_endpoints_num = 0;
static char current_etcd_host[MAX_GLOBAL_HOSTNAME];
static pthread_mutex_t etcd_failover_lock = PTHREAD_MUTEX_INITIALIZER;
static bool etcd_failover_running = false;

static bool etcd_failover(etcdlib_t **etcdlib);
static etcdlib_t *etcdlib_create_internal(const char* server, int port, int flags);
static void etcdlib_destroy_internal(etcdlib_t *etcdlib);
static int etcdlib_get_internal(etcdlib_t *etcdlib, const char *key, char **value, int *modifiedIndex);
static int etcdlib_get_directory_internal(etcdlib_t *etcdlib, const char *directory, etcdlib_key_value_callback callback, void *arg, long long *modifiedIndex);
static int etcdlib_set_internal(etcdlib_t *etcdlib, const char *key, const char *value, long long ttl, bool prevExist);
static int etcdlib_refresh_internal(etcdlib_t *etcdlib, const char *key, int ttl);
static int etcdlib_watch_internal(etcdlib_t *etcdlib, const char *key, long long index, char **action, char **prevValue, char **value, char **rkey, long long *modifiedIndex);
static int etcdlib_del_internal(etcdlib_t *etcdlib, const char *key);
static int etcdlib_grant_lease_internal(etcdlib_t *etcdlib, long long *lease, int ttl);
static int etcdlib_renew_lease_internal(etcdlib_t *etcdlib, long long lease);
static int etcdlib_lock_internal(etcdlib_t *etcdlib, const char *name, long long lease, char **lock);
static int etcdlib_unlock_internal(etcdlib_t *etcdlib, const char *lock);

struct MemoryStruct {
    char *memory;
    char *header;
    size_t memorySize;
    size_t headerSize;
};

static
void memory_struct_init(struct MemoryStruct * memory) {
    memory->memory = palloc(1); /* will be grown as needed by the realloc above */
    memory->memorySize = 0; /* no data at this point */
    memory->header = NULL; /* will be grown as needed by the realloc above */
    memory->headerSize = 0; /* no data at this point */
}

static 
char *base64(const char *data,
                    int input_length,
                    int *output_length,
                    bool encode) {
    int b64_out_len;
    char * output;

    b64_out_len = encode ? pg_b64_enc_len(input_length) : pg_b64_dec_len(input_length);
    output = palloc0(b64_out_len + 1);
    
    b64_out_len = encode ? pg_b64_encode(data, input_length, output, b64_out_len):
        pg_b64_decode(data, input_length, output, b64_out_len);
    if (b64_out_len < 0) {
        if (output)
            pfree(output);
        if (output_length) {
            *output_length = -1;
        }
        return NULL;
    }

    output[b64_out_len] = '\0';

    if (output_length) {
        *output_length = b64_out_len;
    }

    return output;
}

static 
size_t WriteMemoryCallback(void *contents, size_t size, size_t nmemb, void *userp) {
    size_t realsize = size * nmemb;
    struct MemoryStruct *mem = (struct MemoryStruct *) userp;

    mem->memory = repalloc(mem->memory, mem->memorySize + realsize + 1);
    if (mem->memory == NULL) {
        /* out of memory! */
        fprintf(stderr, "[ETCDLIB] Error: not enough memory (realloc returned NULL)\n");
        return 0;
    }

    memcpy(&(mem->memory[mem->memorySize]), contents, realsize);
    mem->memorySize += realsize;
    mem->memory[mem->memorySize] = 0;

    return realsize;
}

static size_t
WriteHeaderCallback(void *contents, size_t size, size_t nmemb, void *userp) {
    size_t realsize = size * nmemb;
    struct MemoryStruct *mem = (struct MemoryStruct *) userp;

    mem->header = repalloc(mem->header, mem->headerSize + realsize + 1);
    if (mem->header == NULL) {
        /* out of memory! */
        fprintf(stderr, "[ETCDLIB] Error: not enough header-memory (realloc returned NULL)\n");
        return 0;
    }

    memcpy(&(mem->header[mem->headerSize]), contents, realsize);
    mem->headerSize += realsize;
    mem->header[mem->headerSize] = 0;

    return realsize;
}

static int
performRequest(CURL **curl, char *url, request_t request, void *reqData, void *repData) {
    CURLcode res = 0;
    
    if (*curl == NULL) {
        *curl = curl_easy_init();
    } else {
        curl_easy_reset(*curl);
    }

    curl_easy_setopt(*curl, CURLOPT_NOSIGNAL, 1);
    curl_easy_setopt(*curl, CURLOPT_TIMEOUT, DEFAULT_CURL_TIMEOUT);
    curl_easy_setopt(*curl, CURLOPT_CONNECTTIMEOUT, DEFAULT_CURL_CONNECT_TIMEOUT);
    curl_easy_setopt(*curl, CURLOPT_FOLLOWLOCATION, 1L);
    /*curl_easy_setopt(*curl, CURLOPT_VERBOSE, 1L); */
    curl_easy_setopt(*curl, CURLOPT_URL, url);
    curl_easy_setopt(*curl, CURLOPT_WRITEFUNCTION, WriteMemoryCallback);
    curl_easy_setopt(*curl, CURLOPT_WRITEDATA, repData);
    if (((struct MemoryStruct *) repData)->header) {
        curl_easy_setopt(*curl, CURLOPT_HEADERDATA, repData);
        curl_easy_setopt(*curl, CURLOPT_HEADERFUNCTION, WriteHeaderCallback);
    }

    if (request == PUT) {
        curl_easy_setopt(*curl, CURLOPT_CUSTOMREQUEST, "PUT");
        curl_easy_setopt(*curl, CURLOPT_POST, 1L);
        curl_easy_setopt(*curl, CURLOPT_POSTFIELDS, reqData);
    } else if (request == DELETE) {
        curl_easy_setopt(*curl, CURLOPT_CUSTOMREQUEST, "DELETE");
    } else if (request == GET) {
        curl_easy_setopt(*curl, CURLOPT_CUSTOMREQUEST, "GET");
    } else if (request == POST) {
        curl_easy_setopt(*curl, CURLOPT_CUSTOMREQUEST, "POST");
        curl_easy_setopt(*curl, CURLOPT_POST, 1L);
        curl_easy_setopt(*curl, CURLOPT_POSTFIELDS, reqData);
    }

    res = curl_easy_perform(*curl);

    if (res != CURLE_OK && res != CURLE_OPERATION_TIMEDOUT) {
        const char *m = request == GET ? "GET" : request == PUT ? "PUT" : request == DELETE ? "DELETE" :
            request == POST ? "POST" :"?";
        fprintf(stderr, "[etcdlib] Curl error for %s @ %s: %s\n", url, m, curl_easy_strerror(res));

        curl_easy_cleanup(*curl);
        *curl = NULL;
    }

    return res;
}

/**
 * Static function declarations
 */
static int
performRequest(CURL **curl, char *url, request_t request, void *reqData, void *repData);
static size_t WriteMemoryCallback(void *contents, size_t size, size_t nmemb, void *userp);
/**
 * External function definition
 */

static bool
etcd_failover(etcdlib_t **etcdlib) {
    bool isEletedLeader = false;
    bool res = false;
    int retry_count = 0;
    struct timespec time_out;
    clock_gettime(CLOCK_REALTIME, &time_out);
    time_out.tv_sec += MAX_ETCD_FAILOVER_LOCK_TIMEOUT;

    /* etcd_failover_running flag ensures that there is only one failover request running at the same time.*/
    if (etcd_failover_running) {
        return true;
    }

    while (isEletedLeader == false && retry_count <= MAX_ETCD_FAILOVER_RETRY) {
        for (int i = 0; i < etcdlib_endpoints_num; i++) {
            if (strcmp(etcdlib_endpoints_handler[i]->host, current_etcd_host) == 0)
                continue;
            pthread_mutex_timedlock(&etcd_failover_lock, &time_out);
            etcd_failover_running = true;
            if (etcdlib_get_leader(etcdlib_endpoints_handler[i], &isEletedLeader) != ETCDLIB_RC_OK)
                continue;
            etcd_failover_running = false;
            pthread_mutex_unlock(&etcd_failover_lock);
            if (isEletedLeader) {
                printf("failoverHostLeaderFromETCD failover etcd connection to new leader: %s:%d\n", 
                      etcdlib_endpoints_handler[i]->host, etcdlib_endpoints_handler[i]->port);
                *etcdlib = etcdlib_endpoints_handler[i];
                res = true;
                break;
            }
        }
        sleep(1);
        retry_count ++;
    }
    return res;
}

/**
 * etcd_init
 */
int
etcd_init(etcdlib_endpoint_t *etcd_endpoints, int etcd_endpoints_num, int flags) {
    int status = 0;
    g_etcdlib = etcdlib_create(etcd_endpoints, etcd_endpoints_num, flags);
    if (!g_etcdlib)
        status = -1;
    return status;
}

static etcdlib_t *
etcdlib_create_internal(const char* server, int port, int flags) {
    
    curl_global_init(CURL_GLOBAL_ALL);

    etcdlib_t *lib = palloc(sizeof(*lib));
    lib->host = pstrdup(server);
    lib->port = port;
    lib->curl = NULL;
    lib->v3_api = DEFAULT_ENABLE_ETCD_V3_API;

    return lib;
}

etcdlib_t *
etcdlib_create(etcdlib_endpoint_t *etcd_endpoints, int etcd_endpoints_num, int flags) {
    etcdlib_endpoints_num = etcd_endpoints_num;
    etcdlib_endpoint_t *petcd_endpoints = etcd_endpoints;
    strncpy(current_etcd_host, petcd_endpoints->etcd_host, MAX_GLOBAL_HOSTNAME);
    for (int i = 0; i < etcd_endpoints_num; i++, petcd_endpoints++) {
        etcdlib_endpoints_handler[i] = etcdlib_create_internal(petcd_endpoints->etcd_host, petcd_endpoints->etcd_port, flags);
    }
    return etcdlib_endpoints_handler[0];
}

static void
etcdlib_destroy_internal(etcdlib_t *etcdlib) {
    if (etcdlib != NULL) {
        if (etcdlib->host) {
            pfree(etcdlib->host);
            etcdlib->host = NULL;
        }
    }
}

void
etcdlib_destroy(etcdlib_t *etcdlib) {
    for (int i = 0; i < etcdlib_endpoints_num; i++) {
       etcdlib_destroy_internal(etcdlib_endpoints_handler[i]);
    }
}

const char *
etcdlib_host(etcdlib_t *etcdlib) {
    return etcdlib->host;
}

int
etcdlib_port(etcdlib_t *etcdlib) {
    return etcdlib->port;
}

int
etcdlib_enable_v3(etcdlib_t *etcdlib) {
    return etcdlib->v3_api;
}

int
etcd_get(const char *key, char **value, int *modifiedIndex) {
    return etcdlib_get(g_etcdlib, key, value, modifiedIndex);
}

static int
etcdlib_get_request(etcdlib_t *etcdlib, const char *key, struct MemoryStruct *reply) {
    char *url = NULL;
    int res = -1;

    if (etcdlib_enable_v3(etcdlib)) {
        int key_encode_len;
        char *key_encode = NULL;
        char *request = NULL;

        key_encode = base64(key,strlen(key),&key_encode_len, true);
        if (!key_encode || key_encode_len == -1) {
            return res;
        }

        request = psprintf("{\"key\":\"%s\"}", key_encode);

        url = psprintf("http://%s:%d/%s/kv/range", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);
        res = performRequest(&etcdlib->curl, url, POST, request, (void *) reply);

        if (request)
            pfree(request);
        if (key_encode)
            pfree(key_encode);
    } else {
        url = psprintf("http://%s:%d/v2/keys/%s", etcdlib->host, etcdlib->port, key);
        res = performRequest(&etcdlib->curl, url, GET, NULL, (void *) reply);
    }
    if (url)
        pfree(url);
    return res;
}

static int
etcdlib_get_decode_v3(etcdlib_t *etcdlib, struct MemoryStruct *reply, const char *key, char **value, int *modifiedIndex) {
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    json_t *js_key = NULL;
    json_t *js_value = NULL;
    json_t *js_counts = NULL;
    json_t *js_modifiedIndex = NULL;
    json_error_t error;

    int retVal = ETCDLIB_RC_ERROR;
    int counts = 0;
    char * counts_str = NULL;
    char * modified_index_str = NULL;
    char * key_encode = NULL;
    char * value_str = NULL;

    key_encode = base64(key,strlen(key), NULL, true);
    if (!key_encode) {
        return ETCDLIB_RC_ERROR;
    }

    js_root = json_loads(reply->memory, 0, &error);
    if (!js_root) {
        pfree(key_encode);
        return ETCDLIB_JSON_DECODE_ERROR;
    }

    js_counts = json_object_get(js_root, ETCD_JSON_V3_COUNT);
    if (!js_counts) {
        retVal = ETCDLIB_JSON_STRUCT_ERROR;
        goto finish;
    }

    js_node = json_object_get(js_root, ETCD_JSON_V3_KVS);
    counts_str = pstrdup(json_string_value(js_counts));
    counts = atoi(counts_str);

    if (counts != 1 || js_node == NULL || json_array_size(js_node) != 1) {
        retVal = ETCDLIB_JSON_STRUCT_ERROR;
        goto finish;
    }

    js_node = json_array_get(js_node, 0);
    if (!js_node){
        retVal = ETCDLIB_JSON_STRUCT_ERROR;
        goto finish;
    }

    js_key = json_object_get(js_node, ETCD_JSON_KEY);
    js_value = json_object_get(js_node, ETCD_JSON_VALUE);
    
    if (!js_key || !js_value || \
        !json_is_string(js_key) || !json_is_string(js_value)  \
        || strlen(json_string_value(js_key)) != strlen(key_encode) || strcmp(json_string_value(js_key), key_encode) != 0) 
    {
        retVal = ETCDLIB_JSON_STRUCT_ERROR;
        goto finish;
    }

    if (modifiedIndex) {
        js_modifiedIndex = json_object_get(js_node, ETCD_JSON_V3_MOD_REVISION);
        if (!js_modifiedIndex) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto finish;
        }
        modified_index_str = pstrdup(json_string_value(js_modifiedIndex));
        *modifiedIndex = atoi(modified_index_str);
    }

    value_str = pstrdup(json_string_value(js_value));
    *value = base64(value_str, strlen(value_str), NULL, false);
    retVal = ETCDLIB_RC_OK;

finish:
    if (key_encode)
        pfree(key_encode);
    if (counts_str)
        pfree(counts_str);
    if (modified_index_str)
        pfree(modified_index_str);
    if (value_str)
        pfree(value_str);
    json_decref(js_root);

    return retVal;
}

static int
etcdlib_get_decode_v2(struct MemoryStruct *reply, char **value, int *modifiedIndex) {
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    json_t *js_index = NULL;
    json_t *js_value = NULL;
    json_t *js_modifiedIndex = NULL;
    json_error_t error;

    int retVal = ETCDLIB_RC_ERROR;

    js_root = json_loads(reply->memory, 0, &error);
    if (!js_root) {
        return ETCDLIB_JSON_DECODE_ERROR;
    }

    js_node = json_object_get(js_root, ETCD_JSON_NODE);
    js_index = json_object_get(js_root, ETCD_JSON_INDEX);

    if (js_node != NULL) {
        js_value = json_object_get(js_node, ETCD_JSON_VALUE);
        js_modifiedIndex = json_object_get(js_node, ETCD_JSON_MODIFIEDINDEX);

        if (js_modifiedIndex != NULL && js_value != NULL) {
            if (modifiedIndex) {
                *modifiedIndex = json_integer_value(js_modifiedIndex);
            }
            *value = pstrdup(json_string_value(js_value));
            retVal = ETCDLIB_RC_OK;
        }
    } else if ((modifiedIndex != NULL) && (js_index != NULL)) {
        /* Error occurred, retrieve the index of ETCD from the error code */
        *modifiedIndex = json_integer_value(js_index);
        retVal = ETCDLIB_RC_OK;
    }

    json_decref(js_root);
    return retVal;
}

static int
etcdlib_get_decode(etcdlib_t *etcdlib, struct MemoryStruct *reply, const char *key, char **value, int *modifiedIndex) {
    
    return etcdlib_enable_v3(etcdlib) ? etcdlib_get_decode_v3(etcdlib, reply, key, value, modifiedIndex):
        etcdlib_get_decode_v2(reply, value, modifiedIndex);
}

int
etcdlib_get(etcdlib_t *etcdlib, const char *key, char **value, int *modifiedIndex) {
    int ret = etcdlib_get_internal(etcdlib, key, value, modifiedIndex);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_get_internal(etcdlib, key, value, modifiedIndex);
    }
    return ret;
}

static int
etcdlib_get_internal(etcdlib_t *etcdlib, const char *key, char **value, int *modifiedIndex) {
    int res = -1;
    struct MemoryStruct reply;

    memory_struct_init(&reply);

    int retVal = ETCDLIB_RC_ERROR;
    res = etcdlib_get_request(etcdlib, key, &reply);

    if (res == CURLE_OK) {
        retVal = etcdlib_get_decode(etcdlib, &reply, key, value, modifiedIndex);
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        /* process curl timeout case */
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        retVal = ETCDLIB_RC_ERROR;
        fprintf(stderr, "Error getting etcd value, curl error: '%s'\n", curl_easy_strerror(res));
    }

    if (reply.memory)
        pfree(reply.memory);

    if (retVal != ETCDLIB_RC_OK) {
        *value = NULL;
    }

    return retVal;
}

static int
etcd_get_recursive_values(json_t *js_root, etcdlib_key_value_callback callback, void *arg, json_int_t *mod_index) {
    json_t *js_nodes;
    if ((js_nodes = json_object_get(js_root, ETCD_JSON_NODES)) != NULL) {
        if (json_is_array(js_nodes)) {
            int len = json_array_size(js_nodes);
            for (int i = 0; i < len; i++) {
                json_t *js_object = json_array_get(js_nodes, i);
                json_t *js_mod_index = json_object_get(js_object, ETCD_JSON_MODIFIEDINDEX);

                if (js_mod_index != NULL) {
                    json_int_t index = json_integer_value(js_mod_index);
                    if (*mod_index < index) {
                        *mod_index = index;
                    }
                } else {
                    printf("[ETCDLIB] Error: No INDEX found for key!\n");
                }

                if (json_object_get(js_object, ETCD_JSON_NODES)) {
                    /* node contains nodes */
                    etcd_get_recursive_values(js_object, callback, arg, mod_index);
                } else {
                    /*else empty etcd directory, not an error.*/
                    json_t *js_key = json_object_get(js_object, ETCD_JSON_KEY);
                    json_t *js_value = json_object_get(js_object, ETCD_JSON_VALUE);

                    if (js_key && js_value) {
                        if (!json_object_get(js_object, ETCD_JSON_DIR)) {
                            callback(json_string_value(js_key), json_string_value(js_value), arg);
                        }
                    }
                }
            }
        } else {
            fprintf(stderr, "[ETCDLIB] Error: misformatted JSON: nodes element is not an array !!\n");
        }
    } else {
        fprintf(stderr, "[ETCDLIB] Error: nodes element not found!!\n");
    }

    return (*mod_index > 0 ? 0 : 1);
}

static long long
etcd_get_current_index(const char *headerData) {
    long long index = -1;
    char *indexStr = strstr(headerData, ETCD_HEADER_INDEX);
    indexStr += strlen(ETCD_HEADER_INDEX);

    if (sscanf(indexStr, "%lld\n", &index) == 1) {
    } else {
        index = -1;
    }
    return index;
}

int
etcd_get_directory(const char *directory, etcdlib_key_value_callback callback, void *arg, long long *modifiedIndex) {
    return etcdlib_get_directory(g_etcdlib, directory, callback, arg, modifiedIndex);
}

int
etcdlib_get_directory(etcdlib_t *etcdlib, const char *directory, etcdlib_key_value_callback callback, void *arg,
                          long long *modifiedIndex) {
    int ret = etcdlib_get_directory_internal(etcdlib, directory, callback, arg, modifiedIndex);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_get_directory_internal(etcdlib, directory, callback, arg, modifiedIndex);
    }
    return ret;
}

static int
etcdlib_get_directory_internal(etcdlib_t *etcdlib, const char *directory, etcdlib_key_value_callback callback, void *arg,
                          long long *modifiedIndex) {

    json_t *js_root = NULL;
    json_t *js_rootnode = NULL;
    json_t *js_index = NULL;

    json_error_t error;
    int res;
    struct MemoryStruct reply;

    NO_SUPPORT_IN_V3(etcdlib);
    memory_struct_init(&reply);

    int retVal = ETCDLIB_RC_OK;
    char *url = NULL;

    url = psprintf("http://%s:%d/v2/keys/%s?recursive=true", etcdlib->host, etcdlib->port, directory);
    res = performRequest(&etcdlib->curl, url, GET, NULL, (void *) &reply);

    if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);
        if (js_root != NULL) {
            js_rootnode = json_object_get(js_root, ETCD_JSON_NODE);
            js_index = json_object_get(js_root, ETCD_JSON_INDEX);
        } else {
            retVal = ETCDLIB_RC_ERROR;
            fprintf(stderr, "[ETCDLIB] Error: %s in js_root not found\n", ETCD_JSON_NODE);
        }
        if (js_rootnode != NULL) {
            long long modIndex = 0;
            long long *ptrModIndex = NULL;
            if (modifiedIndex != NULL) {
                *modifiedIndex = 0;
                ptrModIndex = modifiedIndex;
            } else {
                ptrModIndex = &modIndex;
            }
            retVal = etcd_get_recursive_values(js_rootnode, callback, arg, (json_int_t *) ptrModIndex);
            long long indexFromHeader = etcd_get_current_index(reply.header);
            if (indexFromHeader > *ptrModIndex) {
                *ptrModIndex = indexFromHeader;
            }
        } else if ((modifiedIndex != NULL) && (js_index != NULL)) {
            // Error occurred, retrieve the index of ETCD from the error code
            *modifiedIndex = json_integer_value(js_index);;
        }
        if (js_root != NULL) {
            json_decref(js_root);
        }
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        retVal = ETCDLIB_RC_ERROR;
    }

    if (url)
        pfree(url);
    if (reply.memory)
        pfree(reply.memory);
    if (reply.header)
        pfree(reply.header);

    return retVal;
}

int
etcd_set(const char *key, const char *value, long long ttl, bool prevExist) {
    return etcdlib_set(g_etcdlib, key, value, ttl, prevExist);
}

static int
etcdlib_set_decode(etcdlib_t *etcdlib, const char * value, struct MemoryStruct *reply) {
    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    json_t *js_value = NULL;
    int retVal = ETCDLIB_RC_ERROR;

    js_root = json_loads(reply->memory, 0, &error);

    if (!js_root) {
        return ETCDLIB_JSON_DECODE_ERROR;
    }
    
    if (etcdlib_enable_v3(etcdlib)) {
        js_node = json_object_get(js_root, ETCD_JSON_V3_HEADER);
        if (!js_node) {
            return ETCDLIB_JSON_STRUCT_ERROR;
        }
        js_value = json_object_get(js_node, ETCD_JSON_V3_REVISION);
        if (js_value != NULL){
            retVal = ETCDLIB_RC_OK;
        } 
    } else {
        js_node = json_object_get(js_root, ETCD_JSON_NODE);
        if (!js_node) {
            return ETCDLIB_JSON_STRUCT_ERROR;
        }

        js_value = json_object_get(js_node, ETCD_JSON_VALUE);
        if (js_value != NULL \
            && json_is_string(js_value)) {
            retVal = ETCDLIB_RC_OK;
        } else {
            return ETCDLIB_JSON_STRUCT_ERROR;
        }
    }

    json_decref(js_root);
    return retVal;
}

static int
etcdlib_set_request(etcdlib_t *etcdlib, const char *key, const char *value, long long ttl, bool prevExist, struct MemoryStruct * reply) {
    char *url = NULL;
    int res = -1;
    char *request = NULL;

    if (etcdlib_enable_v3(etcdlib)) {
        int key_encode_len, value_encode_len;
        char *key_encode = NULL, *value_encode = NULL; 

        key_encode = base64(key, strlen(key), &key_encode_len, true);
        value_encode = base64(value, strlen(value), &value_encode_len, true);

        if (!key_encode || !value_encode || key_encode_len == -1 || value_encode_len == -1) {
            goto done_v3;
        }

        request = psprintf("{\"key\":\"%s\", \"value\": \"%s\", \"lease\": \"%lld\"}", key_encode, value_encode, ttl);
        url = psprintf("http://%s:%d/%s/kv/put", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);

        res = performRequest(&etcdlib->curl, url, POST, request, (void *) reply);

done_v3:
        if (request)
            pfree(request);
        if (key_encode)
            pfree(key_encode);
        if (value_encode)
            pfree(value_encode);
        
    } else {
        size_t req_len = strlen(value) + MAX_OVERHEAD_LENGTH;
        char *requestPtr;

        request = malloc(req_len);
        requestPtr = request;

        url = psprintf("http://%s:%d/v2/keys/%s", etcdlib->host, etcdlib->port, key);

        requestPtr += snprintf(requestPtr, req_len, "value=%s", value);
        if (ttl > 0) {
            requestPtr += snprintf(requestPtr, req_len - (requestPtr - request), ";ttl=%lld", ttl);
        }

        if (prevExist) {
            requestPtr += snprintf(requestPtr, req_len - (requestPtr - request), ";prevExist=true");
        }

        res = performRequest(&etcdlib->curl, url, PUT, request, (void *) reply);
        if (request)
            free(request);
    }

    if (url)
        pfree(url);
    return res;
}

int
etcdlib_set(etcdlib_t *etcdlib, const char *key, const char *value, long long ttl, bool prevExist) {
    int ret = etcdlib_set_internal(etcdlib, key, value, ttl, prevExist);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_set_internal(etcdlib, key, value, ttl, prevExist);
    }
    return ret;
}

static int
etcdlib_set_internal(etcdlib_t *etcdlib, const char *key, const char *value, long long ttl, bool prevExist) {
    int retVal = ETCDLIB_RC_ERROR;
    int res;
    struct MemoryStruct reply;

    if(!etcdlib_enable_v3(etcdlib)) {
        /* Skip leading '/', etcd v2 cannot handle this. */
        while (*key == '/') {
            key++;
        }
    }

    memory_struct_init(&reply);

    res = etcdlib_set_request(etcdlib, key, value, ttl, prevExist, &reply);

    if (res == CURLE_OK) {
        retVal = etcdlib_set_decode(etcdlib, value, &reply);
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        //timeout
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        retVal = ETCDLIB_RC_ERROR;
        fprintf(stderr, "Error getting etcd value, curl error: '%s'\n", curl_easy_strerror(res));
    }

    if (reply.memory)
        pfree(reply.memory);
    return retVal;
}

int
etcd_refresh(const char *key, int ttl) {
    return etcdlib_refresh(g_etcdlib, key, ttl);
}

int
etcdlib_refresh(etcdlib_t *etcdlib, const char *key, int ttl) {
    int ret = etcdlib_refresh_internal(etcdlib, key, ttl);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_refresh_internal(etcdlib, key, ttl);
    }
    return ret;
}

static int
etcdlib_refresh_internal(etcdlib_t *etcdlib, const char *key, int ttl) {
    int retVal = ETCDLIB_RC_ERROR;
    char *url = NULL;
    char *request;

    int res;
    struct MemoryStruct reply;

    NO_SUPPORT_IN_V3(etcdlib);

    /* Skip leading '/', etcd cannot handle this. */
    while (*key == '/') {
        key++;
    }

    memory_struct_init(&reply);

    url = psprintf("http://%s:%d/v2/keys/%s", etcdlib->host, etcdlib->port, key);
    request = psprintf("ttl=%d;prevExists=true;refresh=true", ttl);

    res = performRequest(&etcdlib->curl, url, PUT, request, (void *) &reply);
    if (res == CURLE_OK && reply.memory != NULL) {
        json_error_t error;
        json_t *root = json_loads(reply.memory, 0, &error);
        if (root != NULL) {
            json_t *errorCode = json_object_get(root, ETCD_JSON_ERRORCODE);
            if (errorCode == NULL) {
                //no curl error and no etcd errorcode reply -> OK
                retVal = ETCDLIB_RC_OK;
            } else {
                fprintf(stderr, "[ETCDLIB] errorcode %lli\n", json_integer_value(errorCode));
                retVal = ETCDLIB_RC_ERROR;
            }
            json_decref(root);
        } else {
            retVal = ETCDLIB_RC_ERROR;
            fprintf(stderr, "[ETCDLIB] Error: %s is not json\n", reply.memory);
        }
    }

    if (url)
        pfree(url);
    if (reply.memory)
        pfree(reply.memory);
    if (request)
        pfree(request);

    return retVal;
}

int
etcd_watch(const char *key, long long index, char **action, char **prevValue, char **value, char **rkey,
               long long *modifiedIndex) {
    return etcdlib_watch(g_etcdlib, key, index, action, prevValue, value, rkey, modifiedIndex);
}

int
etcdlib_watch(etcdlib_t *etcdlib, const char *key, long long index, char **action, char **prevValue, char **value, char **rkey,
               long long *modifiedIndex) {
    int ret = etcdlib_watch_internal(etcdlib, key, index, action, prevValue, value, rkey, modifiedIndex);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_watch_internal(etcdlib, key, index, action, prevValue, value, rkey, modifiedIndex);
    }
    return ret;
}

static int
etcdlib_watch_internal(etcdlib_t *etcdlib, const char *key, long long index, char **action, char **prevValue, char **value,
                  char **rkey, long long *modifiedIndex) {
    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    json_t *js_prevNode = NULL;
    json_t *js_action = NULL;
    json_t *js_value = NULL;
    json_t *js_rkey = NULL;
    json_t *js_prevValue = NULL;
    json_t *js_modIndex = NULL;
    json_t *js_index = NULL;
    int retVal = ETCDLIB_RC_ERROR;
    char *url = NULL;
    int res;
    struct MemoryStruct reply;

    // don't use shared curl/mutex for watch, that will lock everything.
    CURL *curl = NULL;

    NO_SUPPORT_IN_V3(etcdlib);
    memory_struct_init(&reply);

    if (index != 0)
        url = psprintf("http://%s:%d/v2/keys/%s?wait=true&recursive=true&waitIndex=%lld", etcdlib->host, etcdlib->port,
                 key, index);
    else
        url = psprintf("http://%s:%d/v2/keys/%s?wait=true&recursive=true", etcdlib->host, etcdlib->port, key);

    res = performRequest(&curl, url, GET, NULL, (void *) &reply);

    curl_easy_cleanup(curl);

    if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);

        if (js_root != NULL) {
            js_action = json_object_get(js_root, ETCD_JSON_ACTION);
            js_node = json_object_get(js_root, ETCD_JSON_NODE);
            js_prevNode = json_object_get(js_root, ETCD_JSON_PREVNODE);
            js_index = json_object_get(js_root, ETCD_JSON_INDEX);
            retVal = ETCDLIB_RC_OK;
        }
        if (js_prevNode != NULL) {
            js_prevValue = json_object_get(js_prevNode, ETCD_JSON_VALUE);
        }
        if (js_node != NULL) {
            js_rkey = json_object_get(js_node, ETCD_JSON_KEY);
            js_value = json_object_get(js_node, ETCD_JSON_VALUE);
            js_modIndex = json_object_get(js_node, ETCD_JSON_MODIFIEDINDEX);
        }
        if (js_prevNode != NULL) {
            js_prevValue = json_object_get(js_prevNode, ETCD_JSON_VALUE);
        }
        if ((prevValue != NULL) && (js_prevValue != NULL) && (json_is_string(js_prevValue))) {

            *prevValue = pstrdup(json_string_value(js_prevValue));
        }
        if (modifiedIndex != NULL) {
            if ((js_modIndex != NULL) && (json_is_integer(js_modIndex))) {
                *modifiedIndex = json_integer_value(js_modIndex);
            } else if ((js_index != NULL) && (json_is_integer(js_index))) {
                *modifiedIndex = json_integer_value(js_index);
            } else {
                *modifiedIndex = index;
            }
        }
        if ((rkey != NULL) && (js_rkey != NULL) && (json_is_string(js_rkey))) {
            *rkey = pstrdup(json_string_value(js_rkey));

        }
        if ((action != NULL) && (js_action != NULL) && (json_is_string(js_action))) {
            *action = pstrdup(json_string_value(js_action));
        }
        if ((value != NULL) && (js_value != NULL) && (json_is_string(js_value))) {
            *value = pstrdup(json_string_value(js_value));
        }
        if (js_root != NULL) {
            json_decref(js_root);
        }

    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        //ignore timeout
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

    if (url)
        pfree(url);
    if (reply.memory)
        pfree(reply.memory);

    return retVal;
}

static int
etcdlib_del_request(etcdlib_t *etcdlib, const char *key, struct MemoryStruct *reply) {
    char *url = NULL;
    int res = -1;

    if (etcdlib_enable_v3(etcdlib)) {
        int key_encode_len;
        char *key_encode = NULL;
        char *request;

        key_encode = base64(key,strlen(key),&key_encode_len, true);
        if (!key_encode || key_encode_len == -1) {
            return res;
        }

        request = psprintf("{\"key\":\"%s\"}", key_encode);

        url = psprintf("http://%s:%d/%s/kv/deleterange", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);
        res = performRequest(&etcdlib->curl, url, POST, request, (void *) reply);
        
        if (key_encode)
            pfree(key_encode);
        if (request)
            pfree(request);
    } else {
        url = psprintf("http://%s:%d/v2/keys/%s?recursive=true", etcdlib->host, etcdlib->port, key);
        res = performRequest(&etcdlib->curl, url, DELETE, NULL, (void *) &reply);
    }

    if (url)
        pfree(url);
    return res;
}

static int
etcdlib_del_decode(etcdlib_t *etcdlib, struct MemoryStruct *reply) {
    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    int retVal = ETCDLIB_RC_ERROR;

    char * delete_str = NULL; 
    int delete_counts = 0; 

    js_root = json_loads(reply->memory, 0, &error);
    if (!js_root) {
        return ETCDLIB_JSON_DECODE_ERROR;
    }

    if (etcdlib_enable_v3(etcdlib)) {
        
        js_node = json_object_get(js_root, ETCD_JSON_V3_DELETED);
        if (!js_node) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }
        delete_str = pstrdup(json_string_value(js_node));
        delete_counts = atoi(delete_str);
        if (delete_counts != 1) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

    } else {
        js_root = json_loads(reply->memory, 0, &error);
        if (!js_root) {
            retVal = ETCDLIB_JSON_DECODE_ERROR;
            goto ret;
        }
        js_node = json_object_get(js_root, ETCD_JSON_NODE);

        if (!js_node) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }
    }
    retVal = ETCDLIB_RC_OK;

ret:
    if (delete_str)
        pfree(delete_str);
    if (js_root)
        json_decref(js_root);
    return retVal;
}

int
etcd_del(const char *key) {
    return etcdlib_del(g_etcdlib, key);
}

int
etcdlib_del(etcdlib_t *etcdlib, const char *key) {
    int ret = etcdlib_del_internal(etcdlib, key);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_del_internal(etcdlib, key);
    }
    return ret;
}

static int
etcdlib_del_internal(etcdlib_t *etcdlib, const char *key) {
    int retVal = ETCDLIB_RC_ERROR;
    int res;
    struct MemoryStruct reply;

    memory_struct_init(&reply);

    res = etcdlib_del_request(etcdlib, key, &reply);

    if (res == CURLE_OK) {
        retVal = etcdlib_del_decode(etcdlib, &reply);
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        /* ignore timeout in case delete operation to not being triggered failover operation.*/
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

    if (reply.memory)
        pfree(reply.memory);

    return retVal;
}

int
etcd_grant_lease(etcdlib_t *etcdlib, long long *lease, int ttl) {
    return etcdlib_grant_lease(g_etcdlib, lease, ttl);
}

int
etcdlib_grant_lease(etcdlib_t *etcdlib, long long *lease, int ttl) {
    int ret = etcdlib_grant_lease_internal(etcdlib, lease, ttl);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_grant_lease_internal(etcdlib, lease, ttl);
    }
    return ret;
}

static int
etcdlib_grant_lease_internal(etcdlib_t *etcdlib, long long *lease, int ttl) {
    struct MemoryStruct reply;
    int retVal = ETCDLIB_RC_ERROR;
    char *url = NULL;
    char *request = NULL;
    int res;

    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_id = NULL, *js_ttl = NULL;
    char *ttl_str = NULL;
    int origin_lease = *lease;

    NO_SUPPORT_IN_V2(etcdlib);
    memory_struct_init(&reply);

    if (*lease == 0) {
        request = psprintf("{ \"TTL\": \"%d\"}", ttl);
    } else {
        request = psprintf("{\"lease\": \"%lld\", \"TTL\": \"%d\"}", *lease, ttl);
    }
    url = psprintf("http://%s:%d/%s/lease/grant", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);

    res = performRequest(&etcdlib->curl, url, POST, request, (void *) &reply);

    if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);
        if (!js_root) {
            retVal = ETCDLIB_JSON_DECODE_ERROR;
            goto ret;
        }

        js_id = json_object_get(js_root, ETCD_JSON_V3_LEASE_ID);
        js_ttl = json_object_get(js_root, ETCD_JSON_V3_LEASE_TTL);
        if (!js_id || !js_ttl) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        *lease = atoll(pstrdup(json_string_value(js_id)));
        if (*lease == 0 || ((origin_lease) != 0 && (*lease != origin_lease))) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }
        
        ttl_str = pstrdup(json_string_value(js_ttl));
        if (atoi(ttl_str) == -1) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        retVal = ETCDLIB_RC_OK;
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

ret:
    if (ttl_str)
        pfree(ttl_str);
    if (url)
        pfree(url);
    if (reply.memory)
        pfree(reply.memory);
    if (request)
        pfree(request);
    if (js_root) {
        json_decref(js_root);
    }

    return retVal;
}

int
etcd_renew_lease(long long lease) {
    return etcdlib_renew_lease(g_etcdlib, lease);
}

int
etcdlib_renew_lease(etcdlib_t *etcdlib, long long lease) {
    int ret = etcdlib_renew_lease_internal(etcdlib, lease);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_renew_lease_internal(etcdlib, lease);
    }
    return ret;
}

static int
etcdlib_renew_lease_internal(etcdlib_t *etcdlib, long long lease) {
    struct MemoryStruct reply;
    int retVal = ETCDLIB_RC_ERROR;
    char * url = NULL;
    int res;

    json_error_t error;
    json_t *js_root = NULL;
    char *request = NULL;

    NO_SUPPORT_IN_V2(etcdlib);
    memory_struct_init(&reply);

    request = psprintf("{\"ID\":\"%lld\"}", lease);
    url = psprintf("http://%s:%d/%s/lease/keepalive", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);
    res = performRequest(&etcdlib->curl, url, POST, request, (void *) &reply);

    if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);

        if (!js_root) {
            retVal = ETCDLIB_JSON_DECODE_ERROR;
            goto ret;
        }
        
        retVal = ETCDLIB_RC_OK;
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

ret:
    if (url)
        pfree(url);
    if (request)
        pfree(request);
    if (reply.memory)
        pfree(reply.memory);
    if (js_root) {
        json_decref(js_root);
    }
    return retVal;
}

int
etcd_lock(const char *name, long long lease, char **lock) {
    return etcdlib_lock(g_etcdlib, name, lease, lock);
}

int
etcdlib_lock(etcdlib_t *etcdlib, const char *name, long long lease, char **lock) {
    /*TODO: Since could not distinguish the following case with the same return error code, then not to implement failover function for lock API.
       1. Lock acquire failed caued by lock already occupied, it would return ETCDLIB_RC_TIMEOUT.
       2. Connection failed with etcd node, it would return ETCDLIB_RC_TIMEOUT.
    */
    int ret = etcdlib_lock_internal(etcdlib, name, lease, lock);
    if (ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_lock_internal(etcdlib, name, lease, lock);
    }
    return ret;
}

static int
etcdlib_lock_internal(etcdlib_t *etcdlib, const char *name, long long lease, char **lock) {
    struct MemoryStruct reply;
    int retVal = ETCDLIB_RC_ERROR;
    char * url = NULL;
    int res;
    int key_encode_len;
    char * key_encode = NULL;

    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    char * lock_key_str = NULL;

    char *request = NULL;

    NO_SUPPORT_IN_V2(etcdlib);
    memory_struct_init(&reply);

    key_encode = base64(name, strlen(name), &key_encode_len, true);

    request = psprintf("{\"name\":\"%s\", \"lease\": \"%lld\"}", key_encode, lease);
    url = psprintf("http://%s:%d/%s/lock/lock", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);
    res = performRequest(&etcdlib->curl, url, POST, request, (void *) &reply);

    if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);
        if (!js_root) {
            retVal = ETCDLIB_JSON_DECODE_ERROR;
            goto ret;
        }

        js_node = json_object_get(js_root, ETCD_JSON_KEY);
        if (!js_node) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        lock_key_str = pstrdup(json_string_value(js_node));
        if (strlen(lock_key_str) == 0) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }
        if (lock) {
            *lock = lock_key_str;
        }

        retVal = ETCDLIB_RC_OK;
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

ret:
    if (url)
        pfree(url);
    if (request)
        pfree(request);
    if (key_encode)
        pfree(key_encode);
    if (reply.memory)
        pfree(reply.memory);
    if (js_root) {
        json_decref(js_root);
    }
    return retVal;
}

int
etcd_unlock(const char *lock)  {
    return etcdlib_unlock(g_etcdlib, lock);
}

int
etcdlib_unlock(etcdlib_t *etcdlib, const char *lock) {
    int ret = etcdlib_unlock_internal(etcdlib, lock);
    if (ret == ETCDLIB_RC_TIMEOUT || ret == ETCDLIB_RC_ERROR) {
        if (etcdlib_endpoints_num > 1 && etcd_failover(&etcdlib))
            ret = etcdlib_unlock_internal(etcdlib, lock);
    }
    return ret;
}

static int
etcdlib_unlock_internal(etcdlib_t *etcdlib, const char *lock) {
    struct MemoryStruct reply;
    int retVal = ETCDLIB_RC_ERROR;
    char * url = NULL;
    int res;

    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    char *request = NULL;

    NO_SUPPORT_IN_V2(etcdlib);
    memory_struct_init(&reply);

    request = psprintf("{\"key\":\"%s\"}", lock);
    url = psprintf("http://%s:%d/%s/lock/unlock", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);
    res = performRequest(&etcdlib->curl, url, POST, request, (void *) &reply);

    if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);

        if (!js_root) {
            retVal = ETCDLIB_JSON_DECODE_ERROR;
            goto ret;
        }

        js_node = json_object_get(js_root, ETCD_JSON_V3_HEADER);
        if (!js_node) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        js_node = json_object_get(js_node, ETCD_JSON_V3_REVISION);
        if (!js_node){
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        } 
        
        retVal = ETCDLIB_RC_OK;
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

ret:
    if (url)
        pfree(url);
    if (request)
        pfree(request);
    if (reply.memory)
        pfree(reply.memory);
    if (js_root) {
        json_decref(js_root);
    }
    return retVal;
}

int
etcd_get_leader(bool *isLeader) {
    return etcdlib_get_leader(g_etcdlib, isLeader);
}

int
etcdlib_get_leader(etcdlib_t *etcdlib, bool *isLeader) {
    struct MemoryStruct reply;
    int retVal = ETCDLIB_RC_ERROR;
    char * url = NULL;
    long member = 0;
    long leader = 0;
    int res;

    json_error_t error;
    json_t *js_root = NULL;
    json_t *js_node = NULL;
    json_t *memberId = NULL, *leaderId = NULL;
    char *request = "{}";

    NO_SUPPORT_IN_V2(etcdlib);
    memory_struct_init(&reply);

    url = psprintf("http://%s:%d/%s/maintenance/status", etcdlib->host, etcdlib->port, DEFAULT_ETCD_URL_PREFIX);
    res = performRequest(&etcdlib->curl, url, POST, request, (void *) &reply);

   if (res == CURLE_OK) {
        js_root = json_loads(reply.memory, 0, &error);

        if (!js_root) {
            retVal = ETCDLIB_JSON_DECODE_ERROR;
            goto ret;
        }

        js_node = json_object_get(js_root, ETCD_JSON_V3_HEADER);
        if (!js_node) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        memberId= json_object_get(js_node, ETCD_JSON_V3_MEMBER_ID);
        leaderId = json_object_get(js_root, ETCD_JSON_V3_LEADER);
        if (!memberId || !leaderId) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        member = atoll(pstrdup(json_string_value(memberId)));
        leader = atoll(pstrdup(json_string_value(leaderId)));
        if (!member || !leader) {
            retVal = ETCDLIB_JSON_STRUCT_ERROR;
            goto ret;
        }

        if (member == leader) {
            *isLeader = true;
        }
        
        retVal = ETCDLIB_RC_OK;
    } else if (res == CURLE_OPERATION_TIMEDOUT) {
        retVal = ETCDLIB_RC_TIMEOUT;
    } else {
        fprintf(stderr, "Got curl error: %s\n", curl_easy_strerror(res));
        retVal = ETCDLIB_RC_ERROR;
    }

ret:
    if (url)
        pfree(url);
    if (reply.memory)
        pfree(reply.memory);
    if (js_root) {
        json_decref(js_root);
    }
    return retVal;
}

bool
test_etcd_failover(etcdlib_t **etcdlib) {
    return etcd_failover(etcdlib);
}

void
test_etcd_mock_handler(etcdlib_t **etcdlib, int etcd_endpoints) {
    int random_endpoints = rand()%etcd_endpoints;
    *etcdlib = etcdlib_endpoints_handler[random_endpoints];
}

#endif