#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include "cloud_mqtt_common.h"

static int load_uplink_cfg(cloud_mqtt_var_t *var, char *data)
{
    cJSON *root = NULL;
    cJSON *objs = NULL;
    cJSON *obj = NULL;
    int cnt = 0;
    int i = 0, j = 0;
    int ret = 0;

    root = cJSON_Parse(data);
    if (!root)
    {
        dbg_syslog(LOG_ERR, "parse cfg file error");
        ret = -1;
        goto out;
    }

    // MQTT
    objs = cJSON_GetObjectItem(root, "MQTT");
    if (!objs)
    {
        dbg_syslog(LOG_ERR, "get MQTT failed");
        ret = -1;
        goto KAFKA;
    }

    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        dbg_syslog(LOG_INFO, "No MQTT UPLINK Channel");
        goto KAFKA;
    }

    if (cnt > 8)
    {
        dbg_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
        cnt = MAX_UPLINK_CHANNEL_CNT;
    }
    var->channel_cnt = cnt;

    for (i = 0; i < cnt; i++)
    {
        char clientId[128] = {0}; ///< clientId
        channel_t *channel = NULL;

        channel = &var->channel[i];
        channel->type = LINK_TYPE_MQTT;
        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "client_id") != NULL)
        {
            GET_JSON_VALUE_STRING(obj, "client_id", clientId);
        }
        if (strlen(clientId) == 0)
        {
            snprintf(clientId, sizeof(clientId), "%s_UP_%d", var->sn_str, i);
        }

        mqtt_session_t *session = mqtt_session_new(clientId, (void*)channel);
        if (session == NULL)
        {
            dbg_syslog(LOG_ERR, "malloc error");
            goto KAFKA;
        }
        channel->session = session;

        GET_JSON_VALUE_STRING(obj, "host", channel->addr);
        GET_JSON_VALUE_INT(obj, "port", channel->port);
        GET_JSON_VALUE_STRING(obj, "user", channel->user);
        GET_JSON_VALUE_STRING(obj, "pass", channel->pass);
        if (cJSON_GetObjectItem(obj, "resume") != NULL)
        {
            GET_JSON_VALUE_INT(obj, "resume", channel->resume);
        }
        else
        {
            channel->resume = 0;
        }

        char cafile_url[128] = {0}, certfile_url[128] = {0}, keyfile_url[128] = {0};
        GET_JSON_VALUE_STRING(obj, "cafile", cafile_url);
        GET_JSON_VALUE_STRING(obj, "certfile", certfile_url);
        GET_JSON_VALUE_STRING(obj, "keyfile", keyfile_url);
        if (strlen(cafile_url) || strlen(certfile_url) || strlen(keyfile_url))
        {
            char cafile[128] = {0};
            char certfile[128] = {0};
            char keyfile[128] = {0};
            char path[128] = {0};

            sprintf(path, "%s/%d", CLOUD_MQTT_TLS_FILE_PATH, i);
            if (strlen(cafile_url))
            {
                mqtt_tls_file_overlow(cafile_url, path, cafile, 128);
            }
            if (strlen(certfile_url))
            {
                mqtt_tls_file_overlow(certfile_url, path, certfile, 128);
            }
            if (strlen(keyfile_url))
            {
                mqtt_tls_file_overlow(keyfile_url, path, keyfile, 128);
            }
            dbg_syslog(LOG_INFO, "cafile:%s,certfile:%s,keyfile:%s", cafile, certfile, keyfile);
            mqtt_session_set_tls(session, cafile, certfile, keyfile);
        }
        mqtt_session_set_address(session, channel->addr, channel->port, channel->user, channel->pass);
        mqtt_session_set_opts(session, DEFAULT_QOS, KEEPALIVEINTERVAL);

        if (cJSON_GetObjectItem(obj, "heart_beat") != NULL)
        {
            cJSON *hb = cJSON_GetObjectItem(obj, "heart_beat");
            GET_JSON_VALUE_INT(hb, "interval", channel->heart_beat.interval);
            GET_JSON_VALUE_DY_STRING(hb, "pub_topic", channel->heart_beat.pub_topic);
            GET_JSON_VALUE_DY_STRING(hb, "sub_topic", channel->heart_beat.sub_topic);
            if (cJSON_GetObjectItem(hb, "pub_payload") != NULL)
            {
                GET_JSON_VALUE_DY_STRING(hb, "pub_payload", channel->heart_beat.payload);
            }
        }
        if (cJSON_GetObjectItem(obj, "login") != NULL)
        {
            cJSON *login = cJSON_GetObjectItem(obj, "login");
            GET_JSON_VALUE_INT(login, "interval", channel->login.interval);
            GET_JSON_VALUE_DY_STRING(login, "pub_topic", channel->login.pub_topic);
            GET_JSON_VALUE_DY_STRING(login, "sub_topic", channel->login.sub_topic);
            if (cJSON_GetObjectItem(login, "pub_payload") != NULL)
            {
                GET_JSON_VALUE_DY_STRING(login, "pub_payload", channel->login.payload);
            }
        }

        if (cJSON_GetObjectItem(obj, "topic_maps") != NULL)
        {
            cJSON *maps = cJSON_GetObjectItem(obj, "topic_maps");
            topic_maps_t *topic_maps = &channel->topic_maps;
            if (cJSON_GetObjectItem(maps, "uplink") != NULL)
            {
                cJSON *uplinks = cJSON_GetObjectItem(maps, "uplink");
                int cnt = cJSON_GetArraySize(uplinks);
                if (cnt > 0)
                {
                    topic_maps->uplink_cnt = cnt;
                    topic_maps->uplink = calloc(cnt, sizeof(topic_pair_t));
                    if (topic_maps->uplink == NULL)
                    {
                        dbg_syslog(LOG_WARNING, "calloc failed, cnt:%d", cnt);
                        continue;
                    }
                    for (j = 0; j < cnt; j++)
                    {
                        cJSON *uplink = cJSON_GetArrayItem(uplinks, j);
                        GET_JSON_VALUE_DY_STRING(uplink, "internal", topic_maps->uplink[j].internal_topic);
                        GET_JSON_VALUE_DY_STRING(uplink, "external", topic_maps->uplink[j].external_topic);
                    }
                }
            }
            if (cJSON_GetObjectItem(maps, "downlink") != NULL)
            {
                cJSON *downlinks = cJSON_GetObjectItem(maps, "downlink");
                int cnt = cJSON_GetArraySize(downlinks);
                if (cnt > 0)
                {
                    topic_maps->downlink_cnt = cnt;
                    topic_maps->downlink = calloc(cnt, sizeof(topic_pair_t));
                    if (topic_maps->downlink == NULL)
                    {
                        dbg_syslog(LOG_WARNING, "calloc failed, cnt:%d", cnt);
                        continue;
                    }
                    for (j = 0; j < cnt; j++)
                    {
                        cJSON *downlink = cJSON_GetArrayItem(downlinks, j);
                        GET_JSON_VALUE_DY_STRING(downlink, "internal", topic_maps->downlink[j].internal_topic);
                        GET_JSON_VALUE_DY_STRING(downlink, "external", topic_maps->downlink[j].external_topic);
                    }
                }
            }
        }
    }

KAFKA:
    // KAFKA
    objs = cJSON_GetObjectItem(root, "KAFKA");
    if (!objs)
    {
        dbg_syslog(LOG_ERR, "get KAFKA failed");
        ret = -1;
        goto out;
    }
    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        dbg_syslog(LOG_INFO, "No KAFKA UPLINK Channel");
        goto out;
    }
    if (cnt > 8)
    {
        dbg_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
        cnt = MAX_UPLINK_CHANNEL_CNT;
    }
    var->kafka_channel_cnt = cnt;
    for (i = 0; i < cnt; i++)
    {
        char clientId[128] = {0}; ///< clientId
        channel_t *channel = NULL;

        channel = &var->kafka_channel[i];
        channel->type = LINK_TYPE_KAFKA;

        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "client_id") != NULL)
        {
            GET_JSON_VALUE_STRING(obj, "client_id", clientId);
        }
        if (strlen(clientId) == 0)
        {
            snprintf(clientId, sizeof(clientId), "%s_UP_%d", var->sn_str, i);
        }

        GET_JSON_VALUE_STRING(obj, "host", channel->addr);
        GET_JSON_VALUE_INT(obj, "port", channel->port);
        GET_JSON_VALUE_STRING(obj, "user", channel->user);
        GET_JSON_VALUE_STRING(obj, "pass", channel->pass);
        if (cJSON_GetObjectItem(obj, "resume") != NULL)
        {
            GET_JSON_VALUE_INT(obj, "resume", channel->resume);
        }
        else
        {
            channel->resume = 0;
        }
        if (cJSON_GetObjectItem(obj, "properties") != NULL)
        {
            cJSON *props = cJSON_GetObjectItem(obj, "properties");
            int t = cJSON_GetArraySize(props);
            if (t > 0)
            {
                channel->property_cnt = t;
                channel->properties = calloc(t, sizeof(property_t));
                if (channel->properties == NULL)
                {
                    dbg_syslog(LOG_WARNING, "calloc failed, cnt:%d", t);
                    continue;
                }
                for (j = 0; j < t; j++)
                {
                    cJSON *prop = cJSON_GetArrayItem(props, j);
                    GET_JSON_VALUE_DY_STRING(prop, "name", channel->properties[j].property);
                    GET_JSON_VALUE_DY_STRING(prop, "value", channel->properties[j].value);
                }
            }
        }

        if (cJSON_GetObjectItem(obj, "heart_beat") != NULL)
        {
            cJSON *hb = cJSON_GetObjectItem(obj, "heart_beat");
            GET_JSON_VALUE_INT(hb, "interval", channel->heart_beat.interval);
            GET_JSON_VALUE_DY_STRING(hb, "pub_topic", channel->heart_beat.pub_topic);
            GET_JSON_VALUE_DY_STRING(hb, "sub_topic", channel->heart_beat.sub_topic);
            if (cJSON_GetObjectItem(hb, "pub_payload") != NULL)
            {
                GET_JSON_VALUE_DY_STRING(hb, "pub_payload", channel->heart_beat.payload);
            }
        }
        if (cJSON_GetObjectItem(obj, "login") != NULL)
        {
            cJSON *login = cJSON_GetObjectItem(obj, "login");
            GET_JSON_VALUE_INT(login, "interval", channel->login.interval);
            GET_JSON_VALUE_DY_STRING(login, "pub_topic", channel->login.pub_topic);
            GET_JSON_VALUE_DY_STRING(login, "sub_topic", channel->login.sub_topic);
            if (cJSON_GetObjectItem(login, "pub_payload") != NULL)
            {
                GET_JSON_VALUE_DY_STRING(login, "pub_payload", channel->login.payload);
            }
        }

        if (cJSON_GetObjectItem(obj, "topic_maps") != NULL)
        {
            cJSON *maps = cJSON_GetObjectItem(obj, "topic_maps");
            topic_maps_t *topic_maps = &channel->topic_maps;
            if (cJSON_GetObjectItem(maps, "uplink") != NULL)
            {
                cJSON *uplinks = cJSON_GetObjectItem(maps, "uplink");
                int cnt = cJSON_GetArraySize(uplinks);
                if (cnt > 0)
                {
                    topic_maps->uplink_cnt = cnt;
                    topic_maps->uplink = calloc(cnt, sizeof(topic_pair_t));
                    if (topic_maps->uplink == NULL)
                    {
                        dbg_syslog(LOG_WARNING, "calloc failed, cnt:%d", cnt);
                        continue;
                    }
                    for (j = 0; j < cnt; j++)
                    {
                        cJSON *uplink = cJSON_GetArrayItem(uplinks, j);
                        GET_JSON_VALUE_DY_STRING(uplink, "internal", topic_maps->uplink[j].internal_topic);
                        GET_JSON_VALUE_DY_STRING(uplink, "external", topic_maps->uplink[j].external_topic);
                    }
                }
            }
            if (cJSON_GetObjectItem(maps, "downlink") != NULL)
            {
                cJSON *downlinks = cJSON_GetObjectItem(maps, "downlink");
                int cnt = cJSON_GetArraySize(downlinks);
                if (cnt > 0)
                {
                    topic_maps->downlink_cnt = cnt;
                    topic_maps->downlink = calloc(cnt, sizeof(topic_pair_t));
                    if (topic_maps->downlink == NULL)
                    {
                        dbg_syslog(LOG_WARNING, "calloc failed, cnt:%d", cnt);
                        continue;
                    }
                    for (j = 0; j < cnt; j++)
                    {
                        cJSON *downlink = cJSON_GetArrayItem(downlinks, j);
                        GET_JSON_VALUE_DY_STRING(downlink, "internal", topic_maps->downlink[j].internal_topic);
                        GET_JSON_VALUE_DY_STRING(downlink, "external", topic_maps->downlink[j].external_topic);
                    }
                }
            }
        }
    }

out:
    cJSON_Delete(root);

    return ret;
}

int cloud_mqtt_load_cfg(cloud_mqtt_var_t *var, const char *cfg_file)
{
    char *data = NULL;
    char *new_data = NULL;
    int ret = 0;

    if (var == NULL || cfg_file == NULL)
    {
        dbg_syslog(LOG_ERR, "p_nodes_cfg:%p cfg_file:%p", var, cfg_file);
        return -1;
    }

    data = read_file_data(cfg_file);
    if (data == NULL)
    {
        dbg_syslog(LOG_ERR, "read cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }
    printf("raw_data:\n%s\n", data);
    new_data = calloc(strlen(var->sn_str) / strlen("[GW_SN]") + 1, strlen(data));
    replace_sub_str(data, "[GW_SN]", var->sn_str, new_data);
    free(data);
    data = new_data;
    printf("replaced data:\n%s\n", data);

    ret = load_uplink_cfg(var, data);
    if (ret != 0)
    {
        dbg_syslog(LOG_WARNING, "uplink config load fail");
        ret = 0;
    }

out:
    if (data != NULL)
    {
        free(data);
    }

    return ret;
}

int get_heart_beat_interval(const char *cfg_file)
{
    char *data = NULL;
    cJSON *root = NULL;
    cJSON *objs = NULL;
    cJSON *obj = NULL;
    cJSON *hb = NULL;
    int cnt = 0;
    int i = 0;
    int ret = 0;
    int interval = 0;

    if (cfg_file == NULL)
    {
        dbg_syslog(LOG_ERR, "cfg_file:%p", cfg_file);
        return -1;
    }

    data = read_file_data(cfg_file);
    if (data == NULL)
    {
        dbg_syslog(LOG_ERR, "read cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }

    root = cJSON_Parse(data);
    if (!root)
    {
        dbg_syslog(LOG_ERR, "parse cfg file error");
        ret = -1;
        goto out;
    }

    // MQTT
    objs = cJSON_GetObjectItem(root, "MQTT");
    if (!objs)
    {
        dbg_syslog(LOG_ERR, "get MQTT failed");
        ret = -1;
        goto KAFKA;
    }

    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        dbg_syslog(LOG_INFO, "No MQTT UPLINK Channel");
        goto KAFKA;
    }

    if (cnt > 8)
    {
        dbg_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
        cnt = MAX_UPLINK_CHANNEL_CNT;
    }

    for (i = 0; i < cnt; i++)
    {
        obj = cJSON_GetArrayItem(objs, i);

        if (cJSON_GetObjectItem(obj, "heart_beat") != NULL)
        {
            hb = cJSON_GetObjectItem(obj, "heart_beat");
            GET_JSON_VALUE_INT(hb, "interval", interval);
            return interval;
        }
    }

KAFKA:
    // KAFKA
    objs = cJSON_GetObjectItem(root, "KAFKA");
    if (!objs)
    {
        dbg_syslog(LOG_ERR, "get KAFKA failed");
        ret = -1;
        goto out;
    }
    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        dbg_syslog(LOG_INFO, "No KAFKA UPLINK Channel");
        goto out;
    }
    if (cnt > 8)
    {
        dbg_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
        cnt = MAX_UPLINK_CHANNEL_CNT;
    }

    for (i = 0; i < cnt; i++)
    {
        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "heart_beat") != NULL)
        {
            hb = cJSON_GetObjectItem(obj, "heart_beat");
            GET_JSON_VALUE_INT(hb, "interval", interval);
            return interval;
        }
    }

out:
    cJSON_Delete(root);

    return ret;
}

int replace_heart_beat_interval(const char *cfg_file, unsigned int interval)
{
    char *data = NULL;
    cJSON *root = NULL;
    cJSON *objs = NULL;
    cJSON *obj = NULL;
    cJSON *hb = NULL;
    char * pbuf = NULL;
    int cnt = 0;
    int i = 0;
    int ret = 0;
    int pfd = 0;

    if (cfg_file == NULL)
    {
        dbg_syslog(LOG_ERR, "cfg_file:%p", cfg_file);
        return -1;
    }

    data = read_file_data(cfg_file);
    if (data == NULL)
    {
        dbg_syslog(LOG_ERR, "read cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }

    root = cJSON_Parse(data);
    if (!root)
    {
        dbg_syslog(LOG_ERR, "parse cfg file error");
        ret = -1;
        goto out;
    }

    // MQTT
    objs = cJSON_GetObjectItem(root, "MQTT");
    if (!objs)
    {
        dbg_syslog(LOG_ERR, "get MQTT failed");
        ret = -1;
        goto KAFKA;
    }

    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        dbg_syslog(LOG_INFO, "No MQTT UPLINK Channel");
        goto KAFKA;
    }

    if (cnt > 8)
    {
        dbg_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
        cnt = MAX_UPLINK_CHANNEL_CNT;
    }

    for (i = 0; i < cnt; i++)
    {
        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "heart_beat") != NULL)
        {
            hb = cJSON_GetObjectItem(obj, "heart_beat");
            cJSON_ReplaceItemInObject(hb, "interval", cJSON_CreateNumber(interval));
            pbuf = cJSON_Print(root);
            pfd = open(cfg_file, O_TRUNC | O_CREAT | O_WRONLY, 0644);
            if (pfd != -1) {
                if (pbuf != NULL)
                    write(pfd, pbuf, strlen(pbuf));
                close(pfd);
            }
            free(pbuf);
        }
    }

KAFKA:
    // KAFKA
    objs = cJSON_GetObjectItem(root, "KAFKA");
    if (!objs)
    {
        dbg_syslog(LOG_ERR, "get KAFKA failed");
        ret = -1;
        goto out;
    }
    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        dbg_syslog(LOG_INFO, "No KAFKA UPLINK Channel");
        goto out;
    }
    if (cnt > 8)
    {
        dbg_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
        cnt = MAX_UPLINK_CHANNEL_CNT;
    }

    for (i = 0; i < cnt; i++)
    {
        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "heart_beat") != NULL)
        {
            hb = cJSON_GetObjectItem(obj, "heart_beat");
            cJSON_ReplaceItemInObject(hb, "interval", cJSON_CreateNumber(interval));
            pbuf = cJSON_Print(root);
            pfd = open(cfg_file, O_TRUNC | O_CREAT | O_WRONLY, 0644);
            if (pfd != -1) {
                if (pbuf != NULL)
                    write(pfd, pbuf, strlen(pbuf));
                close(pfd);
            }
            free(pbuf);
        }
    }

out:
    cJSON_Delete(root);

    return ret;
}