/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 *    Copyright 2018 (c) basysKom GmbH <opensource@basyskom.com> (Author: Peter Rustler)
 */

#include <open62541/plugin/historydata/history_data_backend_tdengine.h>

#include <limits.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <tdengine/client/taos.h>
#include <tdengine/util/taoserror.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>

#define MAX_SQL_LINES 256
#define MAX_SQL_LEN 512
typedef struct UA_TDengineValue_struct {
    char values[MAX_SQL_LEN][MAX_SQL_LINES];
    int lines;
    uint64_t last_flust;
} UA_TDengineValue;

typedef struct UA_TDengineContext_struct {
    TAOS * taos_p;
    char * dbname;
    char * ipaddr;
    char * username;
    char * password;
    pthread_mutex_t llock;
    UA_TDengineValue *values;
    pthread_t thread; // flush thread
} UA_TDengineContext;

static inline uint64_t get_time_ms() {
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return (uint64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
}

int ua_history_tdengine_values_init(UA_TDengineContext *ctx) {
    ctx->values = (UA_TDengineValue *)calloc(1, sizeof(UA_TDengineValue));
    ctx->values->lines = 0;
    ctx->values->last_flust = 0;
    return 0;
}
static int UA_TDengineContext_connect(UA_TDengineContext *ctx);

int ua_history_tdengine_values_flush(UA_TDengineContext *ctx) {
    if(ctx->values->lines == 0) {
        return 0;
    }

    if(ctx->taos_p == NULL) {
        int ret = UA_TDengineContext_connect(ctx);
        if(ret < 0) {
            fprintf(stderr, "Error, failed to connect to TDengine\n");
            return -1;
        }
    }

    char *sql = (char *)malloc(MAX_SQL_LEN * MAX_SQL_LINES);
    int len = sprintf(sql, "INSERT INTO ");
    for(int i = 0; i < ctx->values->lines; i++) {
        len += sprintf(sql + len, "%s ", ctx->values->values[i]);
    }

    TAOS_RES *res = taos_query(ctx->taos_p, sql);
    int code = taos_errno(res);
    if(code != 0) {
        fprintf(stderr, "Error code: %d; Message: %s\n", code, taos_errstr(res));
    }
    taos_free_result(res);
    free(sql);
    ctx->values->lines = 0;
    ctx->values->last_flust = get_time_ms();
    memset(ctx->values->values, 0, MAX_SQL_LEN * MAX_SQL_LINES);
    return 0;
}




int ua_history_tdengine_values_add(UA_TDengineContext *ctx, const char* tagstr, int taglen, uint64_t ts, uint8_t datatype, const char* result) {
    if(tagstr == NULL || tagstr[0] == '\0' || taglen <= 0) {
        return -1;
    }
    char tbname[64] = {0};
    char tagname[64] = {0};
    char *p = NULL, *tag_temp = NULL;
    uint32_t tbname_len = 0, tag_len = 0;

    p = (char*)memrchr(tagstr, '.', taglen);
    if(p == NULL){
        fprintf(stderr, "Error, invalid tag name: %s\n", tagstr);
        fflush(stderr);
        return -1;
    }
    tag_temp = p + 1;
    tag_len = (uint32_t)(tagstr + taglen - p - 1);
    tbname_len = (uint32_t)(p - tagstr);

    tbname_len = tbname_len > (sizeof(tbname) -1) ? (sizeof(tbname) -1) : tbname_len;
    tag_len = tag_len > (sizeof(tagname) - 1) ? (sizeof(tagname) - 1) : tag_len;

    if (tag_temp == NULL || tbname_len <= 0 || tag_len <= 0 ) {
        fprintf(stderr, "Error, invalid tag name: %s\n", tagstr);
        fflush(stderr);
        return -1;
    }
    strncpy(tbname, tagstr, tbname_len);
    tbname[tbname_len] = '\0';

    strncpy(tagname, tag_temp, tag_len);
    tagname[tag_len] = '\0';

    str_replace_chr(tbname);
    str_replace_chr(tagname);
    int ret = pthread_mutex_lock(&ctx->llock);
    if(ret != 0) {
        fprintf(stderr, "Error, failed to lock TDengine statement: %s\n", strerror(ret));
        return -1;
    }
    snprintf(ctx->values->values[ctx->values->lines++], MAX_SQL_LEN,
             "%s_%s USING %s TAGS('%s') VALUES(%llu, %d, '%s')", tbname, tagname, tbname,
             tagname, ts, datatype, result);
    if(ctx->values->lines >= MAX_SQL_LINES) {
        ua_history_tdengine_values_flush(ctx);
    }
    pthread_mutex_unlock(&ctx->llock);
    return 0;
}

void *
ua_history_tdengine_flush_task(void *arg) {
    UA_TDengineContext *ctx = (UA_TDengineContext *)arg;
    while(ctx->taos_p != NULL) {
        sleep(3);
        int ret = pthread_mutex_trylock(&ctx->llock);
        if(ret != 0) {
            if(ret != EBUSY)
                fprintf(stderr, "Error, failed to lock TDengine statement: %s\n", strerror(ret));
            continue;
        }
        if(ctx->values->last_flust + 2000 > get_time_ms() || ctx->values->lines == 0) {
            pthread_mutex_unlock(&ctx->llock);
            continue;
        }
        ua_history_tdengine_values_flush(ctx);
        pthread_mutex_unlock(&ctx->llock);

    }
    return  NULL;
}


static void UA_TDengineContext_free(UA_TDengineContext * ctx)
{
    if (ctx == NULL)
        return;

    if (ctx->taos_p != NULL) {
        taos_close(ctx->taos_p);
        ctx->taos_p = NULL;
    }

    if (ctx->dbname != NULL) {
        free(ctx->dbname);
        ctx->dbname = NULL;
    }

    if (ctx->ipaddr != NULL) {
        free(ctx->ipaddr);
        ctx->ipaddr = NULL;
    }

    if (ctx->username != NULL) {
        free(ctx->username);
        ctx->username = NULL;
    }

    if (ctx->password != NULL) {
        free(ctx->password);
        ctx->password = NULL;
    }

    if(ctx->values != NULL) {
        free(ctx->values);
        ctx->values = NULL;
    }

    free(ctx);
}

static int UA_TDengineContext_connect(UA_TDengineContext * ctx)
{
    int err, ret;
    TAOS * t;
    TAOS_RES * tres;
    const char * ip, * errstr;
    char buff[192];

    ip = ctx->ipaddr ? ctx->ipaddr : "127.0.0.1";
    t = taos_connect(ip, ctx->username ? ctx->username : "root",
        ctx->password ? ctx->password : "taosdata", NULL, 0);
    if (t == NULL) {
        fprintf(stderr, "Error, failed to connect to TAOSD/%s\n", ip);
        fflush(stderr);
        return -1;
    }

    ret = snprintf(buff, sizeof(buff), "CREATE DATABASE IF NOT EXISTS %s PRECISION 'ms' BUFFER 8 CACHEMODEL 'last_value';", ctx->dbname);
    if (ret >= (int) sizeof(buff))
        ret = (int) (sizeof(buff) - 1);
    buff[ret] = '\0';

    /* create database if not exists */
    tres = taos_query(t, buff);
    err = tres ? taos_errno(tres) : EINVAL;
    if (err != 0) {
        errstr = taos_errstr(tres);
        if (errstr == NULL)
            errstr = "unknown";
        fprintf(stderr, "Error, failed to create database '%s': %d, %s\n",
            ctx->dbname, err, errstr);
        fflush(stderr);
        taos_free_result(tres);
        taos_close(t);
        return -1;
    }
    taos_free_result(tres);

    /* switch to database */
    ret = snprintf(buff, sizeof(buff), "USE %s;", ctx->dbname);
    if (ret >= (int) sizeof(buff))
        ret = (int) (sizeof(buff) - 1);
    buff[ret] = '\0';

    tres = taos_query(t, buff);
    err = tres ? taos_errno(tres) : EINVAL;
    if (err != 0) {
        errstr = taos_errstr(tres);
        if (errstr == NULL)
            errstr = "unknown";
        fprintf(stderr, "Error, failed to switch to database '%s': %d, %s\n",
            ctx->dbname, err, errstr);
        fflush(stderr);
        taos_free_result(tres);
        taos_close(t);
        return -1;
    }
    taos_free_result(tres);
    ctx->taos_p = t;
    return 0;
}

static void todo_func(const char * fn, unsigned int * supval)
{
    unsigned int val = 0;
    if (supval != NULL) {
        val = *supval;
        *supval = val + 1;
    }
    if ((val & 0x1F) == 0) {
        fprintf(stderr,
            "************************************************\n"
            "Error, function not yet implemented: %s\n"
            "************************************************\n", fn);
        fflush(stderr);
    }
}

static size_t
resultSize_backend_tdengine(UA_Server *server,
                          void *context,
                          const UA_NodeId *sessionId,
                          void *sessionContext,
                          const UA_NodeId * nodeId,
                          size_t startIndex,
                          size_t endIndex) {
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return 0; // UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static size_t
getDateTimeMatch_backend_tdengine(UA_Server *server,
                                void *context,
                                const UA_NodeId *sessionId,
                                void *sessionContext,
                                const UA_NodeId * nodeId,
                                const UA_DateTime timestamp,
                                const MatchStrategy strategy) {
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return 0; // UA_STATUSCODE_BADNOTIMPLEMENTED;
}



static inline long long UA_Datetime_2ms(UA_DateTime d2ms)
{
    long long ret;
    ret = (long long) ((d2ms - UA_DATETIME_UNIX_EPOCH) / UA_DATETIME_MSEC);
    return ret;
}

static inline UA_DateTime ms2_UA_DateTime(long long ms2d)
{
    UA_DateTime ret;
    ret = (UA_DateTime) ms2d;
    ret = ret * UA_DATETIME_MSEC + UA_DATETIME_UNIX_EPOCH;
    return ret;
}

static UA_StatusCode
serverSetHistoryData_backend_tdengine(UA_Server *server,
                                    void *context,
                                    const UA_NodeId *sessionId,
                                    void *sessionContext,
                                    const UA_NodeId * nodeId,
                                    UA_Boolean historizing,
                                    const UA_DataValue *value)
{
    UA_TDengineContext * ctx = (UA_TDengineContext *) context;
    UA_DateTime timestamp = 0;
    int taglen = 0, ret = 0;
    char *result = NULL;
    const char * tagstr = NULL;

    if (ctx == NULL) {
        fprintf(stderr, "Invalid NULL context in [%s]\n", __func__);
        fflush(stderr);
        return UA_STATUSCODE_GOOD;
    }

    if(nodeId->identifierType == UA_NODEIDTYPE_NUMERIC)
        return UA_STATUSCODE_GOOD;

    tagstr = (const char *)nodeId->identifier.string.data;
    taglen = (int)nodeId->identifier.string.length;
    
    if (value->hasSourceTimestamp) {
        timestamp = value->sourceTimestamp;
    } else if (value->hasServerTimestamp) {
        timestamp = value->serverTimestamp;
    } else {
        timestamp = UA_DateTime_now();
    }

	result = GetDataValueToStr(&value->value);
    if (result == NULL || result[0] == '\0') {
        free(result);
        return UA_STATUSCODE_GOOD;
    }
    
    ret = ua_history_tdengine_values_add(ctx, tagstr, taglen, UA_Datetime_2ms(timestamp),  GetDataTypeToInt(&value->value) & 0x7f, result);
    if(ret != 0) {
        fprintf(stderr, "Error, failed to add data to tdengine: %s\n", result);
        fflush(stderr);
    } 
    free(result);
    return UA_STATUSCODE_GOOD;
}

static size_t
getEnd_backend_tdengine(UA_Server *server,
                      void *context,
                      const UA_NodeId *sessionId,
                      void *sessionContext,
                      const UA_NodeId * nodeId) {
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return 0; // UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static size_t
lastIndex_backend_tdengine(UA_Server *server,
                         void *context,
                         const UA_NodeId *sessionId,
                         void *sessionContext,
                         const UA_NodeId * nodeId) {
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return 0; // UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static size_t
firstIndex_backend_tdengine(UA_Server *server,
                          void *context,
                          const UA_NodeId *sessionId,
                          void *sessionContext,
                          const UA_NodeId * nodeId) {
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return 0; // UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static UA_Boolean
boundSupported_backend_tdengine(UA_Server *server,
                              void *context,
                              const UA_NodeId *sessionId,
                              void *sessionContext,
                              const UA_NodeId * nodeId) {
    return true;
}

static UA_Boolean
timestampsToReturnSupported_backend_tdengine(UA_Server *server,
                                           void *context,
                                           const UA_NodeId *sessionId,
                                           void *sessionContext,
                                           const UA_NodeId *nodeId,
                                           const UA_TimestampsToReturn timestampsToReturn) {
    return true;
}

static const UA_DataValue*
getDataValue_backend_tdengine(UA_Server *server,
                            void *context,
                            const UA_NodeId *sessionId,
                            void *sessionContext,
                            const UA_NodeId * nodeId, size_t index) {
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return NULL;
    // return UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static UA_StatusCode
copyDataValues_backend_tdengine(UA_Server *server,
                              void *context,
                              const UA_NodeId *sessionId,
                              void *sessionContext,
                              const UA_NodeId * nodeId,
                              size_t startIndex,
                              size_t endIndex,
                              UA_Boolean reverse,
                              size_t maxValues,
                              UA_NumericRange range,
                              UA_Boolean releaseContinuationPoints,
                              const UA_ByteString *continuationPoint,
                              UA_ByteString *outContinuationPoint,
                              size_t * providedValues,
                              UA_DataValue * values)
{
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    // return 0;
    return UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static UA_StatusCode
insertDataValue_backend_tdengine(UA_Server *server,
                   void *hdbContext,
                   const UA_NodeId *sessionId,
                   void *sessionContext,
                   const UA_NodeId *nodeId,
                   const UA_DataValue *value)
{
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static UA_StatusCode
replaceDataValue_backend_tdengine(UA_Server *server,
                    void *hdbContext,
                    const UA_NodeId *sessionId,
                    void *sessionContext,
                    const UA_NodeId *nodeId,
                    const UA_DataValue *value)
{
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static UA_StatusCode
updateDataValue_backend_tdengine(UA_Server *server,
                   void *hdbContext,
                   const UA_NodeId *sessionId,
                   void *sessionContext,
                   const UA_NodeId *nodeId,
                   const UA_DataValue *value)
{
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static UA_StatusCode
removeDataValue_backend_tdengine(UA_Server *server,
                               void *hdbContext,
                               const UA_NodeId *sessionId,
                               void *sessionContext,
                               const UA_NodeId *nodeId,
                               UA_DateTime startTimestamp,
                               UA_DateTime endTimestamp)
{
    static unsigned int msgcnt = 0;
    // TODO: implement the method.
    todo_func(__func__, &msgcnt);

    return UA_STATUSCODE_BADNOTIMPLEMENTED;
}

static void
deleteMembers_backend_tdengine(UA_HistoryDataBackend *backend)
{
    if (backend == NULL || backend->context == NULL)
        return;

    UA_TDengineContext_free((UA_TDengineContext *) backend->context);
    backend->context = NULL;
}

static int tdengine_get_count(UA_TDengineContext * ctx, const char * sql, int * countp)
{
    int ret, rval, found, count;
    TAOS_RES * tres;
    TAOS_FIELD * fps;

    rval = 0;
    found = count = 0;
    tres = taos_query(ctx->taos_p, sql);
    if (tres == NULL) {
        fprintf(stderr, "Error, failed to execute TAOS SQL: %s\n", sql);
        fflush(stderr);
        rval = -1;
        goto err0;
    }

    ret = taos_field_count(tres);
    fps = taos_fetch_fields(tres);
    if (ret <= 0 || fps == NULL) {
        fprintf(stderr, "Error, field count for SQL: %s => %d\n", sql, ret);
        fflush(stderr);
        rval = -1;
        goto err0;
    }

    for (;;) {
        TAOS_ROW row_p;
        row_p = taos_fetch_row(tres);
        if (row_p == NULL)
            break;
        if (row_p[0] == NULL)
            continue;

        switch (fps->type) {
        case TSDB_DATA_TYPE_INT:
            found = 1;
            count = *((int *) row_p[0]);
            break;

        case TSDB_DATA_TYPE_BIGINT:
            found = 1;
            count = (int) *((int64_t *) row_p[0]);
            break;

        default:
            break;
        }

        if (found != 0)
            break;
    }

    if (found == 0) {
        rval = -1;
        fprintf(stderr, "Error, failed to find query count in [%s]\n", __func__);
        fflush(stderr);
    } else {
        rval = 0;
        *countp = count;
    }

err0:
    if (tres != NULL)
        taos_free_result(tres);
    return rval;
}

static UA_StatusCode
getHistoryData_backend_tdengine(UA_Server *server,
    const UA_NodeId *sessionId,
    void *sessionContext,
    const UA_HistoryDataBackend *backend,
    const UA_DateTime start,
    const UA_DateTime end,
    const UA_NodeId *nodeId,
    size_t maxSizePerResponse,
    UA_UInt32 numValuesPerNode,
    UA_Boolean returnBounds,
    UA_TimestampsToReturn timestampsToReturn,
    UA_NumericRange range,
    UA_Boolean releaseContinuationPoints,
    const UA_ByteString *continuationPoint,
    UA_ByteString *outContinuationPoint,
    UA_HistoryData *result)
{
    int totalCount, ret, skip;
	int i, nrow, maxSize, numfields;
	char sql[384], buff[128], sqlCount[384];
    UA_TDengineContext * ctx;
    UA_DataValue * outResult;
    TAOS_RES * tres;
    TAOS_FIELD * fps;
    long long udts, udte;

    fps = NULL;
    tres = NULL;
    outResult = NULL;
    numfields = 0;
    udts = udte = 0;
    skip = totalCount = nrow = 0;
	maxSize = (int) maxSizePerResponse;
        ctx = (UA_TDengineContext *)backend->context;

    if (ctx == NULL) {
        fprintf(stderr, "Error, invalid NULL tdengine contex!!!\n");
        fflush(stderr);
        return UA_STATUSCODE_GOOD;
    }

    if (continuationPoint->length > 0) {
        if (continuationPoint->length >= sizeof(size_t)) {
            size_t size_skip;
            size_skip = *((size_t *) continuationPoint->data);
            skip = (int) size_skip;
        } else {
            return UA_STATUSCODE_BADCONTINUATIONPOINTINVALID;
        }
    }

    if (releaseContinuationPoints) {
        fprintf(stdout, "Release continuationPoint...\n");
        fflush(stdout);
        UA_ByteString_deleteMembers((UA_ByteString *) continuationPoint);
    }

    if (nodeId->identifierType == UA_NODEIDTYPE_NUMERIC) {
        fprintf(stderr, "Warning, invalid NUMERIC nodeID...\n");
        fflush(stdout);
        return UA_STATUSCODE_GOOD;
    }

    udts = UA_Datetime_2ms(start);
    udte = UA_Datetime_2ms(end);

    memset(buff, 0, sizeof(buff));
    memcpy(buff, nodeId->identifier.string.data, nodeId->identifier.string.length);
    str_replace_chr(buff);
    if (returnBounds) {
        if (start < end) {
            snprintf(sql, sizeof(sql), "SELECT time,datatype,result FROM %s WHERE time >= %lld AND time <= %lld ORDER BY time LIMIT %d OFFSET %d;",
                buff, udts, udte, maxSize, skip);
            snprintf(sqlCount, sizeof(sqlCount), "SELECT count(*) FROM %s WHERE time >= %lld AND time <= %lld;",
                buff, udts, udte);
        } else if (end == 0) {
            snprintf(sql, sizeof(sql), "SELECT time,datatype,result FROM %s WHERE time >= %lld ORDER BY time LIMIT %d OFFSET %d;",
                buff, udts, maxSize, skip);
            snprintf(sqlCount, sizeof(sqlCount), "SELECT count(*) FROM %s WHERE time >= %lld;",
                buff, udts);
        } else {
            snprintf(sql, sizeof(sql), "SELECT time,datatype,result FROM %s WHERE time >= %lld AND time <= %lld ORDER BY time LIMIT %d OFFSET %d;",
                buff, udte, udts, maxSize, skip);
            snprintf(sqlCount, sizeof(sqlCount), "SELECT count(*) FROM %s WHERE time >= %lld AND time <= %lld;",
                buff, udte, udts);
                }
    } else {
        if (start < end) {
            snprintf(sql, sizeof(sql), "SELECT time,datatype,result FROM %s WHERE time > %lld AND time < %lld ORDER BY time LIMIT %d OFFSET %d;",
                buff, udts, udte, maxSize, skip);
            snprintf(sqlCount, sizeof(sqlCount), "SELECT count(*) FROM %s WHERE time > %lld AND time < %lld;",
                buff, udts, udte);
        } else if(end == 0) {
            snprintf(sql, sizeof(sql), "SELECT time,datatype,result FROM %s WHERE time > %lld ORDER BY time LIMIT %d OFFSET %d;",
                buff, udts, maxSize, skip);
            snprintf(sqlCount, sizeof(sqlCount), "SELECT count(*) FROM %s WHERE time > %lld;",
                buff, udts);
        } else {
            snprintf(sql, sizeof(sql), "SELECT time,datatype,result FROM %s WHERE time > %lld AND time < %lld ORDER BY time LIMIT %d OFFSET %d;",
                buff, udte, udts, maxSize, skip);
            snprintf(sqlCount, sizeof(sqlCount), "SELECT count(*) FROM %s WHERE time > %lld AND time < %lld;",
                buff, udte, udts);
        }
    }
    ret = pthread_mutex_lock(&ctx->llock);
    if (ret != 0) {
        fprintf(stderr, "Error, failed to lock TDengine context: %d\n", ret);
        fflush(stderr);
        return UA_STATUSCODE_GOOD;
    }

    if (ctx->taos_p == NULL) {
        UA_TDengineContext_connect(ctx);
        if (ctx->taos_p == NULL) {
            pthread_mutex_unlock(&ctx->llock);
            fprintf(stderr, "Error, failed to connect to TAOSD.\n");
            fflush(stderr);
            return UA_STATUSCODE_GOOD;
        }
    }

    tdengine_get_count(ctx, sqlCount, &totalCount);
    if (totalCount <= 0)
        goto next;

    outResult = (UA_DataValue *) UA_Array_new((size_t) totalCount, &UA_TYPES[UA_TYPES_DATAVALUE]);
    if (outResult == NULL) {
        pthread_mutex_unlock(&ctx->llock);
        return UA_STATUSCODE_BADOUTOFMEMORY;
    }

    /* execute `sql statement */
    tres = taos_query(ctx->taos_p, sql);
    i = tres ? taos_errno(tres) : EFAULT;
    if (i != 0) {
        const char * errmsg;
        errmsg = taos_errstr(tres);
        if (errmsg == NULL)
            errmsg = "unknown";
        fprintf(stderr, "Error, failed to execute TAOS SQL statement: %d, %s\n",
            i, errmsg);
        fflush(stderr);
        goto next;
    }

    numfields = taos_num_fields(tres);
    fps = taos_fetch_fields(tres);
    if (numfields < 3 || fps == NULL) {
        fprintf(stderr, "Error, invalid number of fields: %d\n", numfields);
        fflush(stderr);
        goto next;
    }

    if (fps[0].type != TSDB_DATA_TYPE_TIMESTAMP ||
        fps[1].type != TSDB_DATA_TYPE_TINYINT ||
        fps[2].type != TSDB_DATA_TYPE_VARCHAR) {
        fprintf(stderr, "Error, invalid field types: %d, %d, %d\n",
            fps[0].type, fps[1].type, fps[2].type);
        fflush(stderr);
        goto next;
    }

    for (i = 0; i < totalCount; ++i) {
        // int64_t ts_v;
        int vtype;
        TAOS_ROW row_p;
        UA_DataValue resultValue;
        UA_DateTime timestamp;
        long long dts;
        const char * restr;
        char restr_buf[128];
        int *lengths;

        row_p = taos_fetch_row(tres);
        if (row_p == NULL)
            break;
        
        lengths = taos_fetch_lengths(tres);
        if (lengths == NULL) {
            fprintf(stderr, "Error, taos_fetch_lengths failed\n");
            fflush(stderr);
            continue;
        }

        if (row_p[0] == NULL || row_p[1] == NULL || row_p[2] == NULL) {
            fprintf(stderr, "Error, invalid ROW pointers: %p, %p, %p\n",
                row_p[0], row_p[1], row_p[2]);
            fflush(stderr);
            continue;
        }

        UA_DataValue_init(&resultValue);
        resultValue.hasValue = true;
        UA_Variant_init(&resultValue.value);

        dts = *((long long *) row_p[0]);
        timestamp = ms2_UA_DateTime(dts);
        vtype = (int) *((signed char *) row_p[1]);
        restr = (const char *) row_p[2];

        int restr_len = lengths[2];
        if ((size_t)restr_len < sizeof(restr_buf)) {
            memcpy(restr_buf, restr, restr_len);
            restr_buf[restr_len] = '\0';
            restr = restr_buf;
        } else {
            // Handle case where string is too long for the buffer, maybe log an error
            fprintf(stderr, "Warning, result string too long: %d\n", restr_len);
            fflush(stderr);
            continue;
        }
        
        switch (vtype) {
        case UA_TYPES_BOOLEAN: {
            UA_Boolean temp1 = (UA_Boolean) strtol(restr, NULL, 0);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_BOOLEAN]);
            break;
        }

        case UA_TYPES_INT16: {
            int16_t temp1 = (int16_t) strtol(restr, NULL, 0);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_INT16]);
            break;
        }

        case UA_TYPES_INT32: {
            int32_t temp1 = (int32_t) strtol(restr, NULL, 0);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_INT32]);
            break;
        }

        case UA_TYPES_INT64: {
            int64_t temp1 = (int64_t) strtol(restr, NULL, 0);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_INT64]);
            break;
        }

        case UA_TYPES_FLOAT: {
            float temp1 = (float) strtod(restr, NULL);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_FLOAT]);
            break;
        }

        case UA_TYPES_DOUBLE: {
            double temp1 = strtod(restr, NULL);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_DOUBLE]);
            break;
        }

        case UA_TYPES_UINT16: {
            uint16_t temp1 = (uint16_t) strtoul(restr, NULL, 10);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_UINT16]);
            break;
        }

        case UA_TYPES_UINT32: {
            uint32_t temp1 = (uint32_t) strtoul(restr, NULL, 10);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_UINT32]);
            break;
        }

        case UA_TYPES_UINT64: {
            uint64_t temp1 = strtoul(restr, NULL, 10);
            UA_Variant_setScalar(&resultValue.value, &temp1, &UA_TYPES[UA_TYPES_UINT64]);
            break;
        }

        case UA_TYPES_STRING: {
            UA_String ua_result;
            ua_result.length = strlen(restr) + 1;
            ua_result.data = (UA_Byte *) malloc(ua_result.length + 1);
            strcpy((char*) ua_result.data, restr);
            UA_Variant_setScalarCopy(&resultValue.value, &ua_result, &UA_TYPES[UA_TYPES_STRING]);
            UA_String_clear(&ua_result);
            free(ua_result.data);
            break;
        }
        }

        resultValue.sourceTimestamp = timestamp;
        resultValue.hasSourceTimestamp = true;
        resultValue.serverTimestamp = timestamp;
        resultValue.hasServerTimestamp = true;
        resultValue.hasSourcePicoseconds = false;
        resultValue.hasServerPicoseconds = false;
        resultValue.hasStatus = true;
        resultValue.status = UA_STATUSCODE_GOOD;

        UA_DataValue_copy(&resultValue, &outResult[i]);
        nrow++;
        }

next:
    result->dataValuesSize = nrow;
    if (nrow > 0 && nrow < totalCount) {
        UA_StatusCode r;
        size_t olds = (size_t) totalCount;
        r = UA_Array_resize((void **) &outResult, &olds, (size_t) nrow, &UA_TYPES[UA_TYPES_DATAVALUE]);
        if (r != UA_STATUSCODE_GOOD) {
            fprintf(stderr, "Warning, UA_Array_resize has failed: %d => %d\n", totalCount, nrow);
            fflush(stderr);
        }
    } else if (nrow <= 0 && outResult) {
        UA_Array_delete(outResult, (size_t) totalCount, &UA_TYPES[UA_TYPES_DATAVALUE]);
        outResult = NULL;
    }
    result->dataValues = outResult;

    if (tres != NULL) {
        taos_free_result(tres);
        tres = NULL;
    }

    // there are more values
    if (skip + nrow < totalCount && skip + nrow < maxSize) {
        if (UA_ByteString_allocBuffer(outContinuationPoint, sizeof(size_t) + sizeof(size_t)) != UA_STATUSCODE_GOOD) {
            pthread_mutex_unlock(&ctx->llock);
            return UA_STATUSCODE_BADOUTOFMEMORY;
        }

        outContinuationPoint->length = sizeof(size_t);
        outContinuationPoint->data = (UA_Byte*) UA_malloc(sizeof(size_t));
        *((size_t*) outContinuationPoint->data) = (size_t) (skip + nrow);
    }
    pthread_mutex_unlock(&ctx->llock);
    return UA_STATUSCODE_GOOD;
}

UA_HistoryDataBackend
UA_HistoryDataBackend_TDengine(const char * ipaddr, const char * user,
    const char * pass, const char * dbname)
{
    int error, ret;
    UA_HistoryDataBackend result;
    UA_TDengineContext * ctx;
    pthread_mutexattr_t attr;
    error = 0;
    ctx = NULL;

    memset(&result, 0, sizeof(UA_HistoryDataBackend));
    if (dbname == NULL || dbname[0] == '\0') {
        fputs("Error, database name not specified!\n", stderr);
        fflush(stderr);
        return result;
    }

    ctx = (UA_TDengineContext *) UA_calloc(1, sizeof(*ctx));
    if (ctx == NULL)
        return result;

    ctx->dbname = strdup(dbname);
    if (ctx->dbname == NULL) {
        fprintf(stderr, "Error, failed to duplicate dbname: %s\n", dbname);
        fflush(stderr);
        return result;
    }

    memset(&attr, 0, sizeof(attr));
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setrobust(&attr, PTHREAD_MUTEX_ROBUST);
    pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
    ret = pthread_mutex_init(&ctx->llock, &attr);
    pthread_mutexattr_destroy(&attr);
    if (ret != 0) {
        fprintf(stderr, "Error, failed to initialize mutex lock: %d\n", ret);
        fflush(stderr);
    }

    if (ipaddr && ipaddr[0]) {
        ctx->ipaddr = strdup(ipaddr);
        if (ctx->ipaddr == NULL) {
            error = ENOMEM;
            goto err0;
        }
    }

    if (user && user[0]) {
        ctx->username = strdup(user);
        if (ctx->username == NULL) {
            error = ENOMEM;
            goto err0;
        }
    }

    if (pass && pass[0]) {
        ctx->password = strdup(pass);
        if (ctx->password == NULL) {
            error = ENOMEM;
            goto err0;
        }
    }

    ret = UA_TDengineContext_connect(ctx);
    if (ret != 0){
        fprintf(stderr, "Error, failed to connect to TDengine: %d\n", ret);
        goto err0;
    }
    ua_history_tdengine_values_init(ctx);
    
    if((ret = pthread_create(&ctx->thread, NULL, ua_history_tdengine_flush_task, ctx)) != 0) {
        fprintf(stderr, "Error, failed to create thread: %d\n", ret);
        goto err0;
    } 

    result.serverSetHistoryData = &serverSetHistoryData_backend_tdengine;
    result.resultSize = &resultSize_backend_tdengine;
    result.getEnd = &getEnd_backend_tdengine;
    result.lastIndex = &lastIndex_backend_tdengine;
    result.firstIndex = &firstIndex_backend_tdengine;
    result.getDateTimeMatch = &getDateTimeMatch_backend_tdengine;
    result.copyDataValues = &copyDataValues_backend_tdengine;
    result.getDataValue = &getDataValue_backend_tdengine;
    result.boundSupported = &boundSupported_backend_tdengine;
    result.timestampsToReturnSupported = &timestampsToReturnSupported_backend_tdengine;
    result.insertDataValue =  &insertDataValue_backend_tdengine;
    result.updateDataValue =  &updateDataValue_backend_tdengine;
    result.replaceDataValue =  &replaceDataValue_backend_tdengine;
    result.removeDataValue =  &removeDataValue_backend_tdengine;
    result.deleteMembers = &deleteMembers_backend_tdengine;
    result.getHistoryData = getHistoryData_backend_tdengine;
    result.context = ctx;
    return result;

err0:
    if (error != 0) {
        fprintf(stderr, "Error, %s has failed: %s\n", __func__,
            strerror(error));
        fflush(stderr);
    }
    UA_TDengineContext_free(ctx);
    return result;
}

void
UA_HistoryDataBackend_TDengine_clear(UA_HistoryDataBackend *backend)
{
    UA_TDengineContext *ctx = (UA_TDengineContext*) backend->context;
    UA_TDengineContext_free(ctx);
    memset(backend, 0, sizeof(UA_HistoryDataBackend));
}