#include <stdio.h>
#include <stdlib.h>

#include "ipc_session.h"
#include "dy_utils/dy_common.h"
#include "dy_utils/dy_ipc.h"
#include "cloud_mqtt_common.h"
#include "dy_utils/protocol.h"
#include "dy_utils/dy_pp.h"

static int cloud_mqtt_internal_recv_msg_handle(cloud_mqtt_var_t *var, char *sn, char *raw_topic, char *payload, int payloadLen)
{
    char topic[TOPIC_MAX_LEN] = {0};
    int i = 0, j = 0;
    int ret = 0;
    time_t now = time(NULL);
    int channel_cnt = var->channel_cnt;

    for (i = 0; i < channel_cnt; i++)
    {
        channel_t *channel = &var->channel[i];
        unsigned char matched;

        for (j = 0; j < channel->topic_maps.uplink_cnt; j++)
        {
            ret = mqtt_topic_matches_sub(channel->topic_maps.uplink[j].internal_topic, raw_topic, &matched);
            if (ret == 0 && matched)
            {
                int report = 0;

                if (strstr(channel->topic_maps.uplink[j].external_topic, "[DEV_SN]"))
                {
                    replace_sub_str(channel->topic_maps.uplink[j].external_topic, "[DEV_SN]", strlen(sn) == 0 ? "UNKNOWN_SN" : sn, topic);
                    if (channel->resume)
                        report = mqtt_session_publish_sync(channel->session, topic, (unsigned char *) payload, payloadLen);
                    else
                        mqtt_session_publish(channel->session, topic, (unsigned char *) payload, payloadLen);
                }
                else
                {
                    strcpy(topic, channel->topic_maps.uplink[j].external_topic);
                    if (channel->resume)
                        report = mqtt_session_publish_sync(channel->session, topic, (unsigned char *) payload, payloadLen);
                    else
                        mqtt_session_publish(channel->session, topic, (unsigned char *) payload, payloadLen);
                }
                if (channel->resume)
                {
                    if (report != 0 || access(CLOUD_MQTT_FORCE_WRITEDB, F_OK) == 0)
                    {
                        // 上报失败，插入db中
                        tag_table_t tag = {0};

                        tag.report = NOT_REPORT;
                        tag.tag_node = malloc(payloadLen + 1);
                        memcpy(tag.tag_node, payload, payloadLen);
                        tag.tag_node[payloadLen] = '\0';
                        tag.tag_len = payloadLen;
                        tag.time = now;
                        tag.ext = strdup(topic);
                        dy_db_session_insert_tag(channel->db_session, &tag);
                    }
                }
            }
        }
    }

    channel_cnt = var->kafka_channel_cnt;

    for (i = 0; i < channel_cnt; i++)
    {
        channel_t *channel = &var->kafka_channel[i];
        bool matched;

        for (j = 0; j < channel->topic_maps.uplink_cnt; j++)
        {
            ret = mosquitto_topic_matches_sub(channel->topic_maps.uplink[j].internal_topic, raw_topic, &matched);
            if (ret == 0 && matched)
            {
                if (strstr(channel->topic_maps.uplink[j].external_topic, "[DEV_SN]"))
                {
                    replace_sub_str(channel->topic_maps.uplink[j].external_topic, "[DEV_SN]", strlen(sn) == 0 ? "UNKNOWN_SN" : sn, topic);
                }
                else
                {
                    strcpy(topic, channel->topic_maps.uplink[j].external_topic);
                }

                // 插入队列，在dr cb中处理
                p_msg_node_t *node = calloc(1, sizeof(p_msg_node_t) + payloadLen);
                strcpy(node->topic, topic);
                node->payloadLen = payloadLen;
                node->channel = channel;
                node->src = FROM_IPC;
                memcpy(node->payload, payload, payloadLen);

                pthread_mutex_lock(&channel->produce_lock);
                list_add_tail(&node->list, &channel->p_ready_list);
                channel->received++;
                pthread_mutex_unlock(&channel->produce_lock);
                pthread_cond_signal(&channel->produce_cond);
            }
        }
    }

    return 0;
}

static int cloud_mqtt_internal_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    cloud_mqtt_var_t *var = (cloud_mqtt_var_t *)obj;
    int i = 0;
    char requester[8] = {0};
    char sn[SN_MAX_LEN] = {0};
    cJSON *node = cJSON_Parse(mqtt_msg->payload);

    if (node == NULL)
    {
        return -1;
    }

    GET_JSON_VALUE_STRING(node, "requester", requester);
    GET_JSON_VALUE_STRING(node, "sn", sn);

    if (strlen(requester) > 0 && strcmp(requester, "local") == 0)
    {
        goto out;
    }

    // dbg_syslog(LOG_DEBUG, "Internal MQTT client received MQTT topic:%s payload length:%d",
    //            mqtt_msg->topic, mqtt_msg->payloadLen);

    for (i = 0; i < var->nodes_cfg_table->node_cnt; i++)
    {
        if (strcmp(sn, var->nodes_cfg_table->node[i].sn) == 0 && var->nodes_cfg_table->node[i].user_defined != NULL)
        {
            cJSON *user_defined = cJSON_Parse(var->nodes_cfg_table->node[i].user_defined);
            if (user_defined)
            {
                cJSON *child = NULL;
                for (child = user_defined->child; child; child = child->next)
                {
                    if (child->string)
                    {
                        if (child->valuestring)
                        {
                            cJSON_AddStringToObject(node, child->string, child->valuestring);
                        }
                        else
                        {
                            cJSON_AddNumberToObject(node, child->string, child->valuedouble);
                        }
                    }
                }
            }
            break;
        }
    }

    cJSON_AddStringToObject(node, "gw_sn", var->sn_str);
    if (strstr(mqtt_msg->topic, "/data_filtered/event/"))
        cJSON_AddStringToObject(node, "data_type", "event");
    else if (strstr(mqtt_msg->topic, "/data_filtered/period_property/"))
        cJSON_AddStringToObject(node, "data_type", "property");
    else
        cJSON_AddStringToObject(node, "data_type", "service");

    char *payload = cJSON_PrintUnformatted(node);
    cloud_mqtt_internal_recv_msg_handle(var, sn, mqtt_msg->topic, payload, strlen(payload));
    free(payload);
out:
    cJSON_Delete(node);

    return 0;
}

// 建立与内部broker之间的MQTT连接
int cloud_mqtt_internal_mqtt_init(cloud_mqtt_var_t *var)
{
    int i = 0, j = 0;
    char clientId[128] = {0};

    snprintf(clientId, sizeof(clientId), "MQTT_CLD_%s", var->sn_str);
    var->session = ipc_session_new(clientId, (void *)var, IPC_DEFAULT);
    if (var->session == NULL)
        return -1;

    ipc_session_set_callbacks(var->session, cloud_mqtt_internal_recv_msg, NULL);
    for (i = 0; i < var->channel_cnt; i++)
    {
        channel_t *channel = &var->channel[i];
        for (j = 0; j < channel->topic_maps.uplink_cnt; j++)
        {
            dbg_syslog(LOG_INFO, "subscribe topic:%s", channel->topic_maps.uplink[j].internal_topic);
            ipc_session_subscribe(var->session, channel->topic_maps.uplink[j].internal_topic);
        }
    }
    ipc_session_start(var->session);
	return 0;
}