/*
 * Copyright (c) 2018, Oracle and/or its affiliates.  All rights reserved.
 *
 * This software is dual-licensed to you under the MIT License (MIT) and
 * the Universal Permissive License (UPL).  See the LICENSE file in the root
 * directory for license terms.  You may choose either license, or both.
 */

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "sqlite3.h"
#include "message_persistence.h"
#include "iotcs.h"
#include "advanced/iotcs_message.h"
#include "util/util.h"
#include "util/util_memory.h"
#include "iotcs_port_system.h"
#include "log/log.h"
#define IOTCSP_MODULE_LOG_CHANNEL LOG_CHANNEL_MSG
#include "log/log_template.h"

static const char *DB_NAME = "msg_storage.sqlite";
static const char *GUARANTED_DELIVERY_SAVE1_TEMPLATE = "INSERT INTO MESSAGE_PERSISTENT_STORE VALUES (%ld, %d, '";
static const char *GUARANTED_DELIVERY_SAVE2_TEMPLATE = "', '";
static const char *GUARANTED_DELIVERY_SAVE3_TEMPLATE = "', 5)";
static const char *GUARANTED_DELIVERY_DELETE_TEMPLATE = "DELETE FROM MESSAGE_PERSISTENT_STORE WHERE uid = %d;";
static const char *GUARANTED_DELIVERY_LOAD_TEMPLATE = "SELECT * FROM MESSAGE_PERSISTENT_STORE WHERE ENDPOINT_ID = '%s' ORDER BY timestamp";
static const char *GUARANTED_DELIVERY_INDEX_TEMPLATE = "CREATE INDEX endpoint_id ON MESSAGE_PERSISTENT_STORE(ENDPOINT_ID)";
static const char *GUARANTED_DELIVERY_UPDATE_TEMPLATE = "UPDATE SET RETRIES = %d WHERE UID = %d";


static const char *BATCH_BY_SAVE1_TEMPLATE = "INSERT INTO BATCH_BY_MESSAGE_STORE VALUES (%ld, %d, '";
static const char *BATCH_BY_SAVE2_TEMPLATE = "')";
static const char *BATCH_BY_DELETE_UID_TEMPLATE = "DELETE FROM BATCH_BY_MESSAGE_STORE WHERE uid = %d;";
static const char *BATCH_BY_DELETE_TEMPLATE = "DELETE FROM BATCH_BY_MESSAGE_STORE;";
static const char *BATCH_BY_LOAD_TEMPLATE = "SELECT * FROM BATCH_BY_MESSAGE_STORE WHERE uid > %d ORDER BY uid";

static sqlite3 *db = NULL;
static iotcs_bool guaranted_delivery_store_ok = IOTCS_FALSE;
static iotcs_bool batch_by_store_ok = IOTCS_FALSE;
#ifndef IOTCS_SQL_BUFFER_SIZE
#define IOTCS_SQL_BUFFER_SIZE 1000
#endif
char *sql_exec_buffer = NULL;

/**
 * Creates the message persistent storage table if it doesn't exist.
 */
iotcs_result iotcs_message_persistence_init() {
    int rc = sqlite3_open(DB_NAME, &db);
    char *err_msg = 0;
    sql_exec_buffer = (char*)util_malloc(sizeof(char) * IOTCS_SQL_BUFFER_SIZE);
    char *mp_sql = "CREATE TABLE IF NOT EXISTS MESSAGE_PERSISTENT_STORE"
                "(TIMESTAMP BIGINT NOT NULL, "
                "UID INT, "
                "ENDPOINT_ID VARCHAR(100) NOT NULL, "
                "MESSAGE TEXT NOT NULL, "
                "RETRIES INT, "
                "PRIMARY KEY (UID))";

    char *bb_sql = "CREATE TABLE IF NOT EXISTS BATCH_BY_MESSAGE_STORE"
                "(TIMESTAMP BIGINT NOT NULL, "
                "UID INT, "
                "MESSAGE TEXT NOT NULL, "
                "PRIMARY KEY (UID))";

    if (rc != SQLITE_OK) {
        LOG_ERR("Cannot open database: %s\n", sqlite3_errmsg(db));
        sqlite3_close(db);

        return IOTCS_RESULT_FAIL;
    }
    rc = sqlite3_exec(db, mp_sql, 0, 0, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("SQL error can not create table %s: %s\n", "MESSAGE_PERSISTENT_STORE", err_msg);
        sqlite3_free(err_msg);
        guaranted_delivery_store_ok = IOTCS_FALSE;
    } else {
        guaranted_delivery_store_ok = IOTCS_TRUE;
    }
    rc = sqlite3_exec(db, bb_sql, 0, 0, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("SQL error can not create table: %s\n", err_msg);
        sqlite3_free(err_msg);
        batch_by_store_ok = IOTCS_FALSE;
    } else {
        batch_by_store_ok = IOTCS_TRUE;
    }
    if (guaranted_delivery_store_ok && batch_by_store_ok) {
        return IOTCS_RESULT_OK;
    } else {
        if (!guaranted_delivery_store_ok && !batch_by_store_ok) {
            iotcs_message_persistence_close();
        }
        return IOTCS_RESULT_FAIL;
    }
}

void iotcs_message_persistence_close() {
    util_free(sql_exec_buffer);
    sql_exec_buffer = NULL;
    sqlite3_close(db);
    db = NULL;
}

iotcs_result iotcs_message_persistence_delete_messages(iotcs_message *message, int count, int additional_uid) {
    iotcs_result rv;
    char *err_msg = 0;
    int ctx = 0;
    int rc, i;

    if (!db && !guaranted_delivery_store_ok) {
        LOG_WARNS("Can not delete message. Database is not initialized.");
        return IOTCS_RESULT_FAIL;
    }

    if (additional_uid) {
        rv = util_safe_snprintf(sql_exec_buffer + ctx, IOTCS_SQL_BUFFER_SIZE - ctx, GUARANTED_DELIVERY_DELETE_TEMPLATE, additional_uid);
        if (rv > 0) {
            ctx += rv;
        }
    }
    for (i = 0; i < count; i++) {
        if (message[i].uid <= 0) {
            LOG_DBGS("Can not delete message. Message UID is zero.");
            continue;
        }
        rv = util_safe_snprintf(sql_exec_buffer + ctx, IOTCS_SQL_BUFFER_SIZE - ctx, GUARANTED_DELIVERY_DELETE_TEMPLATE, message[i].uid);
        if (rv > 0) {
            ctx += rv;
        } else {
            rc = sqlite3_exec(db, sql_exec_buffer, 0, 0, &err_msg);
            if (rc != SQLITE_OK ) {
                LOG_ERR("Can not delete message. SQL error: %s", err_msg);
                sqlite3_free(err_msg);
            }
            ctx = 0;
        }
    }
    if (rc != SQLITE_OK ) {
        return IOTCS_RESULT_FAIL;
    }
    return IOTCS_RESULT_OK;
}

iotcs_result iotcs_batch_by_delete_message(int uid) {
    iotcs_result rv;
    char *err_msg = 0;
    int ctx = 0;
    int rc, i;

    if (!db && !batch_by_store_ok) {
        LOG_WARNS("Can not delete message. Database is not initialized.");
        return IOTCS_RESULT_FAIL;
    }

    rv = util_safe_snprintf(sql_exec_buffer + ctx, IOTCS_SQL_BUFFER_SIZE - ctx, BATCH_BY_DELETE_UID_TEMPLATE, uid);
    if (rv > 0) {
        ctx += rv;
    }
    rc = sqlite3_exec(db, sql_exec_buffer, 0, 0, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("Can not delete message. SQL error: %s", err_msg);
        sqlite3_free(err_msg);
    }
    if (rc != SQLITE_OK ) {
        return IOTCS_RESULT_FAIL;
    }
    return IOTCS_RESULT_OK;
}

iotcs_result iotcs_batch_by_clear_messages() {
    char *err_msg = 0;
    int rc;

    if (!db && !batch_by_store_ok) {
        LOG_WARNS("Can not delete message. Database is not initialized.");
        return IOTCS_RESULT_FAIL;
    }
    rc = sqlite3_exec(db, BATCH_BY_DELETE_TEMPLATE, 0, 0, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("Can not delete message. SQL error: %s", err_msg);
        sqlite3_free(err_msg);
        return IOTCS_RESULT_FAIL;
    }
    return IOTCS_RESULT_OK;
}

struct buffer_information {
    char *ptr;
    int *leng;
    int *uid;
    int retries;
    int size;
};

static int batch_by_load_callback(void *data, int argc, char **argv, char **azColName) {
    struct buffer_information *buffer = (struct buffer_information *)data;
    int i, tmp_uid;

    for(i = 0; i < argc; i++) {
        if (*buffer->leng == 0) {
            if (strcmp(azColName[i], "UID") == 0) {
                   *buffer->uid = atoi(argv[i]);
            } else if (strcmp(azColName[i], "MESSAGE") == 0) {
                if ((int)strlen(argv[i]) < buffer->size) {
                    *buffer->leng = *buffer->leng + sprintf(buffer->ptr, "%s", argv[i]);
                } else {
                    *buffer->leng = -1;
                    *buffer->uid = 0;
                    return 0;
                }
            }
        }
    }
    return 0;
}

int iotcs_batch_by_load_messages(char *buffer, int *length, int *uid, int max_length) {
    iotcs_result rv;
    char *err_msg = 0;
    *length = 0;
    int rc;
    struct buffer_information data = {
        .ptr = buffer,
        .leng = length,
        .uid = uid,
        .retries = 0,
        .size = max_length
    };

    if (!db && !batch_by_store_ok) {
        LOG_WARNS("Can not load message. Database is not initialized.");
        return 0;
    }
    rv = util_safe_snprintf(sql_exec_buffer, IOTCS_SQL_BUFFER_SIZE, BATCH_BY_LOAD_TEMPLATE, *uid);
    if (rv <= 0) {
        LOG_WARNS("Can not load message. SQL Command Buffer overflow.");
        return 0;
    }
    rc = sqlite3_exec(db, sql_exec_buffer, batch_by_load_callback, &data, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("Can not load message. SQL error: %s", err_msg);
        sqlite3_free(err_msg);
        return 0;
    }

    return *data.leng;
}

static int gd_load_callback(void *data, int argc, char **argv, char **azColName) {
    struct buffer_information *buffer = (struct buffer_information *)data;
    int i;

    if (*buffer->leng == 0) {
        for(i = argc - 1; i >= 0; i--) {
            if (strcmp(azColName[i], "RETRIES") == 0) {
               buffer->retries = atoi(argv[i]) - 1;
            }
        }
    }
    for(i = 0; i < argc; i++) {
        if (*buffer->leng == 0) {
            if (strcmp(azColName[i], "UID") == 0) {
                   *buffer->uid = atoi(argv[i]);
            } else if (strcmp(azColName[i], "MESSAGE") == 0) {
                if ((int)strlen(argv[i]) < buffer->size) {
                    *buffer->leng = *buffer->leng + sprintf(buffer->ptr, "%s,", argv[i]);
                } else {
                    *buffer->uid = 0;
                    return 0;
                }
            }
        }
    }
    return 0;
}

int iotcs_message_persistence_load_messages(char *endpoint_id, char *buffer, int *length, int *uid, int max_length) {
    iotcs_result rv;
    char *err_msg = 0;
    *length = 0;
    int rc;
    struct buffer_information data = {
        .ptr = buffer,
        .leng = length,
        .uid = uid,
        .retries = 5,
        .size = max_length
    };

    if (!db && !guaranted_delivery_store_ok) {
        LOG_WARNS("Can not load message. Database is not initialized.");
        return 0;
    }
    rv = util_safe_snprintf(sql_exec_buffer, IOTCS_SQL_BUFFER_SIZE, GUARANTED_DELIVERY_LOAD_TEMPLATE, endpoint_id);
    if (rv <= 0) {
        LOG_WARNS("Can not load message. SQL Command Buffer overflow.");
        return 0;
    }
    rc = sqlite3_exec(db, sql_exec_buffer, gd_load_callback, &data, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("Can not load message. SQL error: %s", err_msg);
        sqlite3_free(err_msg);
        return 0;
    }

    if (*length) {
        if (data.retries <= 0) {
            iotcs_message_persistence_delete_messages(NULL, 0, *data.uid);
        } else {
            if (util_safe_snprintf(sql_exec_buffer, IOTCS_SQL_BUFFER_SIZE, GUARANTED_DELIVERY_UPDATE_TEMPLATE, data.retries, *data.uid) <= 0) {
                LOG_WARNS("Can not load message. SQL Command Buffer overflow.");
                return 0;
            }
            rc = sqlite3_exec(db, sql_exec_buffer, 0, 0, &err_msg);
            if (rc != SQLITE_OK ) {
                LOG_ERR("Can not load message. SQL error can not update messages: %s", err_msg);
                sqlite3_free(err_msg);
                return 0;
            }
        }
    }

    return *data.leng;
}

iotcs_result iotcs_batch_by_save_message(iotcs_message *message, char *json_str) {
    iotcs_result rv;
    char *err_msg = 0;
    int rc, ctx;
    int64_t event_time;

    if (!db && !batch_by_store_ok) {
        LOG_WARNS("Can not persist message. Database is not initialized.");
        return IOTCS_RESULT_FAIL;
    }

    if (message->uid <= 0) {
        LOG_DBGS("Can not persist message. Message UID is zero.");
        return IOTCS_RESULT_OK;
    }
    event_time = message->event_time;
    if (event_time == 0) {
        event_time = iotcs_port_get_current_time_millis();
    }

    /* TODO investigate this. If the are only one snprintf then it cause Segmentation fault on RPI */
    ctx = util_safe_snprintf(sql_exec_buffer, IOTCS_SQL_BUFFER_SIZE, BATCH_BY_SAVE1_TEMPLATE, (long)event_time, (int)message->uid);
    if ((ctx > 0) && (ctx += strlen(json_str) < IOTCS_SQL_BUFFER_SIZE)) {
        strcat(sql_exec_buffer, json_str);
    }
    if ((ctx > 0) && (ctx += strlen(BATCH_BY_SAVE2_TEMPLATE) < IOTCS_SQL_BUFFER_SIZE)) {
        strcat(sql_exec_buffer, BATCH_BY_SAVE2_TEMPLATE);
    }

    if (ctx <= 0 || ctx > IOTCS_SQL_BUFFER_SIZE) {
        LOG_WARNS("Can not persist message. SQL Command Buffer overflow.");
        return IOTCS_RESULT_FAIL;
    }

    rc = sqlite3_exec(db, sql_exec_buffer, 0, 0, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("Can not persist message. SQL error : %s", err_msg);
        sqlite3_free(err_msg);
        return IOTCS_RESULT_FAIL;
    }
    return IOTCS_RESULT_OK;
}

iotcs_result iotcs_message_persistence_save_message(iotcs_message *message, char *json_str, char *endpoint_id) {
    iotcs_result rv;
    char *err_msg = 0;
    int rc, ctx;
    int64_t event_time;

    if (!db && !guaranted_delivery_store_ok) {
        LOG_WARNS("Can not persist message. Database is not initialized.");
        return IOTCS_RESULT_FAIL;
    }

    if (message->uid <= 0) {
        LOG_DBGS("Can not persist message. Message UID is zero.");
        return IOTCS_RESULT_OK;
    }
    event_time = message->event_time;
    if (event_time == 0) {
        event_time = iotcs_port_get_current_time_millis();
    }

    /* TODO investigate this. If the are only one snprintf then it cause Segmentation fault on RPI */
    ctx = util_safe_snprintf(sql_exec_buffer, IOTCS_SQL_BUFFER_SIZE, GUARANTED_DELIVERY_SAVE1_TEMPLATE, (long)event_time, (int)message->uid);
    if ((ctx += strlen(endpoint_id) < IOTCS_SQL_BUFFER_SIZE)) {
        strcat(sql_exec_buffer, endpoint_id);
    }
    if ((ctx > 0) && (ctx += strlen(GUARANTED_DELIVERY_SAVE2_TEMPLATE) < IOTCS_SQL_BUFFER_SIZE)) {
        strcat(sql_exec_buffer, GUARANTED_DELIVERY_SAVE2_TEMPLATE);
    }
    if ((ctx > 0) && (ctx += strlen(json_str) < IOTCS_SQL_BUFFER_SIZE)) {
        strcat(sql_exec_buffer, json_str);
    }
    if ((ctx > 0) && (ctx += strlen(GUARANTED_DELIVERY_SAVE3_TEMPLATE) <= IOTCS_SQL_BUFFER_SIZE)) {
        strcat(sql_exec_buffer, GUARANTED_DELIVERY_SAVE3_TEMPLATE);
    }

    if (ctx <= 0 || ctx > IOTCS_SQL_BUFFER_SIZE) {
        LOG_WARNS("Can not persist message. SQL Command Buffer overflow.");
        return IOTCS_RESULT_FAIL;
    }

    rc = sqlite3_exec(db, sql_exec_buffer, 0, 0, &err_msg);
    if (rc != SQLITE_OK ) {
        LOG_ERR("Can not persist message. SQL error : %s", err_msg);
        sqlite3_free(err_msg);
        return IOTCS_RESULT_FAIL;
    }
    return IOTCS_RESULT_OK;
}
