#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include "mqtt_session.h"
#include "dy_utils/dy_common.h"
#include "cloud_mqtt_common.h"
#include "dy_utils/cJSON.h"
#include "dy_utils/protocol.h"

#define HEART_BEAT_INTERVAL_SET  "heartbeat/interval/set"
#define HEART_BEAT_INTERVAL_GET  "heartbeat/interval/get"

typedef struct _service_info_s
{
    char identifier[IDENTIFIER_LEN];
    unsigned int mi;
    char devSn[SN_MAX_LEN];
    unsigned int timestamp;
} service_info_s;



static int login_prase(login_info_t *info, const char *str)
{
    if (!info || !str)
    {
        return -1;
    }
    cJSON *root = cJSON_Parse(str);
    if (!root)
    {
        return -1;
    }
    GET_JSON_VALUE_INT(root, "mi", info->serial_num);
    GET_JSON_VALUE_INT(root, "timestamp", info->timestamp);
    GET_JSON_VALUE_INT(root, "error_code", info->error_code);
    cJSON_Delete(root);
    return 0;
}

static void msg_login_result(channel_t *channel, mqtt_message_t *mqtt_msg)
{
    login_info_t info = {0};
    int ret = login_prase(&info, mqtt_msg->payload);

    if (!ret)
    {
        if (info.error_code == LOGIN_PASS)
        {
            channel->comm_status = COMM_STATUS_LOGIN;
            dbg_syslog(LOG_INFO, "gateway login success");
        }
        else if (info.error_code == LOGIN_REFUSE)
        {
            dbg_syslog(LOG_ERR, "gateway login failed");
        }
    }
}

static void msg_confirm_connect(channel_t *channel)
{
    channel->heart_beat_lost_cnt = 0;
}

char *get_customer_id(char *topic)      // topic: "/sys/cERZSo10/[GW_SN]/heartbeat/interval/get"
{
    char* str = strtok(topic, "/");     // get "sys"
    if(str != NULL) {
        return strtok(NULL, "/");       // get "cERZSo10"
    }
    return NULL;
}

static void send_interval_get_reply(channel_t *channel, mqtt_message_t *mqtt_msg)
{
    unsigned int interval_val = 0;
    service_info_s interval_get_reply;

    char *customer_id = get_customer_id(mqtt_msg->topic);
    if (NULL == customer_id) {
        dbg_syslog(LOG_WARNING, "get customer_id failed!!!");
        return;
    }
    printf("customer_id: %s\n", customer_id);
    dbg_syslog(LOG_INFO, "customer_id: %s\n", customer_id);

    cJSON *root_recv = cJSON_Parse(mqtt_msg->payload);
    if (!root_recv) {
        dbg_syslog(LOG_WARNING, "parse payload failed!!!");
        return;
    }

    GET_JSON_VALUE_INT(root_recv, "mi", interval_get_reply.mi);
    cJSON_Delete(root_recv);

    get_board_sn(interval_get_reply.devSn);
    interval_get_reply.timestamp = time(NULL);
    strncpy(interval_get_reply.identifier, "heartBeatIntervalGet", IDENTIFIER_LEN);

    printf("interval: %d\n", interval_val);
    dbg_syslog(LOG_INFO, "interval: %d\n", interval_val);

    cJSON *root_send, *sub;
    char *out;
    char topic[TOPIC_MAX_LEN] = {0};
    root_send = cJSON_CreateObject();
    if (root_send != NULL)
    {
        cJSON_AddStringToObject(root_send, "identifier", interval_get_reply.identifier);
        cJSON_AddStringToObject(root_send, "sn", interval_get_reply.devSn);
        cJSON_AddNumberToObject(root_send, "mi", interval_get_reply.mi);
        cJSON_AddNumberToObject(root_send, "timestamp", interval_get_reply.timestamp);
        sub = cJSON_CreateObject();
        int interval = get_heart_beat_interval(CLOUD_MQTT_CFG);
        cJSON_AddNumberToObject(sub, "interval", interval);
        cJSON_AddItemToObject(root_send, "tags", sub);
        out = cJSON_Print(root_send);
        cJSON_Delete(root_send);
        snprintf(topic, TOPIC_MAX_LEN, "/sys/%s/%s/heartbeat/interval/get_reply", customer_id, interval_get_reply.devSn);
        mqtt_session_publish(channel->session, topic, out, strlen(out));
        free(out);
    }
}

static void send_interval_set_reply(channel_t *channel, mqtt_message_t *mqtt_msg)
{
    unsigned int interval_val = 0;
    service_info_s interval_set_reply;

    char *customer_id = get_customer_id(mqtt_msg->topic);
    if (NULL == customer_id) {
        dbg_syslog(LOG_WARNING, "get customer_id failed!!!");
        return;
    }
    printf("customer_id: %s\n", customer_id);
    dbg_syslog(LOG_INFO, "customer_id: %s\n", customer_id);

    cJSON *root_recv = cJSON_Parse(mqtt_msg->payload);
    if (!root_recv) {
        dbg_syslog(LOG_WARNING, "parse payload failed!!!");
        return;
    }
    GET_JSON_VALUE_INT(root_recv, "interval", interval_val);
    GET_JSON_VALUE_INT(root_recv, "mi", interval_set_reply.mi);
    cJSON_Delete(root_recv);
    get_board_sn(interval_set_reply.devSn);
    interval_set_reply.timestamp = time(NULL);
    strncpy(interval_set_reply.identifier, "heartBeatIntervalSet", IDENTIFIER_LEN);

    printf("interval: %d\n", interval_val);
    dbg_syslog(LOG_INFO, "interval: %d\n", interval_val);

    cJSON *root_send, *sub;
    char *out;
    char topic[TOPIC_MAX_LEN] = {0};
    root_send = cJSON_CreateObject();
    if (root_send != NULL)
    {
        cJSON_AddStringToObject(root_send, "identifier", interval_set_reply.identifier);
        cJSON_AddStringToObject(root_send, "sn", interval_set_reply.devSn);
        cJSON_AddNumberToObject(root_send, "mi", interval_set_reply.mi);
        cJSON_AddNumberToObject(root_send, "timestamp", interval_set_reply.timestamp);
        sub = cJSON_CreateObject();
        if(interval_val <= 0) {
            cJSON_AddNumberToObject(sub, "Result", 1);  // reply interval_set failed
        }
        else {
            cJSON_AddNumberToObject(sub, "Result", 0);  // reply interval_set success
        }
        cJSON_AddItemToObject(root_send, "tags", sub);
        out = cJSON_Print(root_send);
        cJSON_Delete(root_send);
        snprintf(topic, TOPIC_MAX_LEN, "/sys/%s/%s/heartbeat/interval/set_reply", customer_id, interval_set_reply.devSn);
        mqtt_session_publish(channel->session, topic, out, strlen(out));
        free(out);

        if (interval_val > 0) {
            replace_heart_beat_interval(CLOUD_MQTT_CFG, interval_val);
            system("/etc/init.d/cloud_mqtt reload");
        }
    }
}

int recv_cloud_mqtt_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    channel_t *channel = (channel_t *)obj;
    char topic[TOPIC_MAX_LEN] = {0};
    bool matched;
    int j = 0, i = 0;
    int ret;
    cloud_mqtt_var_t *var = channel->var;

    dbg_syslog(LOG_INFO, "External MQTT client received MQTT topic:%s payload:%s",
              mqtt_msg->topic, mqtt_msg->payload);

    if (channel->heart_beat.sub_topic != NULL && strlen(channel->heart_beat.sub_topic) > 0 && strcmp(mqtt_msg->topic, channel->heart_beat.sub_topic) == 0)
    {
        msg_confirm_connect(channel);
    }
    else if (channel->login.sub_topic != NULL && strlen(channel->login.sub_topic) > 0 && strcmp(mqtt_msg->topic, channel->login.sub_topic) == 0)
    {
        msg_login_result(channel, mqtt_msg);
    }
    else
    {
        for (j = 0; j < channel->topic_maps.downlink_cnt; j++)
        {
            ret = mosquitto_topic_matches_sub(channel->topic_maps.downlink[j].external_topic, mqtt_msg->topic, &matched);
            if (ret == 0 && matched)
            {
                if (NULL != strstr(mqtt_msg->topic , HEART_BEAT_INTERVAL_GET)) {
                    send_interval_get_reply(channel, mqtt_msg);
                    continue;
                }

                if (NULL != strstr(mqtt_msg->topic , HEART_BEAT_INTERVAL_SET)) {
                    send_interval_set_reply(channel, mqtt_msg);
                    continue;
                }

                if (strstr(channel->topic_maps.downlink[j].internal_topic, "[PORT]") != NULL ||
                    strstr(channel->topic_maps.downlink[j].internal_topic, "[DEV_SN]") != NULL ||
                    strstr(channel->topic_maps.downlink[j].internal_topic, "[SID]") != NULL)
                {
                    char port[64] = {0};
                    char sn[SN_MAX_LEN] = {0};
                    char identifier[64] = {0};
                    char topic_2[TOPIC_MAX_LEN] = {0};
                    cJSON *node = cJSON_Parse(mqtt_msg->payload);
                    if (!node)
                    {
                        dbg_syslog(LOG_WARNING, "json rsData %s error!!!", mqtt_msg->payload);
                        continue;
                    }
                    GET_JSON_VALUE_STRING(node, "sn", sn);
                    GET_JSON_VALUE_STRING(node, "identifier", identifier);
                    replace_sub_str(channel->topic_maps.downlink[j].internal_topic, "[DEV_SN]", strlen(sn) == 0 ? "UNKNOWN_SN" : sn, topic);

                    GET_JSON_VALUE_STRING(node, "port", port);
                    if (strlen(sn) != 0 && strlen(port) == 0)
                    {
                        // find port according SN
                        //遍历设备列表
                        for (i = 0; i < var->nodes_cfg_table->node_cnt; i++)
                        {
                            if (strcmp(sn, var->nodes_cfg_table->node[i].sn) == 0)
                            {
                                if (strlen(var->nodes_cfg_table->node[i].app_key) != 0)
                                {
                                    strcpy(port, var->nodes_cfg_table->node[i].app_key);
                                }
                                else
                                {
                                    strcpy(port, port_enum2char(var->nodes_cfg_table->node[i].port));
                                }
                                break;
                            }
                        }
                    }

                    replace_sub_str(topic, "[PORT]", strlen(port) == 0 ? "UNKNOWN_PORT" : port, topic_2);
                    replace_sub_str(topic_2, "[SID]", strlen(identifier) == 0 ? "UNKNOWN_SID" : identifier, topic);
                    dbg_syslog(LOG_INFO, "forward to internal topic:%s", topic);
                    ipc_session_publish(var->session, topic, mqtt_msg->payload, mqtt_msg->payloadLen);
                    cJSON_Delete(node);
                }
                else
                {
                    dbg_syslog(LOG_INFO, "forward to internal topic:%s", channel->topic_maps.downlink[j].internal_topic);
                    ipc_session_publish(var->session, channel->topic_maps.downlink[j].internal_topic,
                                            mqtt_msg->payload, mqtt_msg->payloadLen);
                }
            }
        }
    }

    return 0;
}