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

#include "dy_utils/dy_mqtt.h"
#include "dy_utils/dy_common.h"
#include "dy_utils/dy_ipc.h"
#include "cloud_daslink_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 *uplink_type)
{
    char topic[TOPIC_MAX_LEN] = {0};
    int i = 0, j = 0;
    int ret = 0;
    time_t now = time(NULL);
	int channel_cnt = var->daslink_channel_cnt;

    for (i = 0; i < channel_cnt; i++)
    {
        channel_t *channel = &var->daslink_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);
					report = daslink_message_produce(channel, topic, payload, payloadLen);
                }
                else
                {
                    strcpy(topic, channel->topic_maps.uplink[j].external_topic);
                    report = daslink_message_produce(channel, topic, payload, payloadLen);
                }
                if (channel->resume)
                {
                    if (report != 0)
                    {
                        // 上报失败，插入db中
                        tag_table_t tag = {0};

                        tag.report = NOT_REPORT;
                        tag.tag_node = malloc(payloadLen);
                        memcpy(tag.tag_node, payload, payloadLen);
                        tag.tag_len = payloadLen;
                        tag.time = now;
                        tag.ext = strdup(topic);
                        dy_db_session_insert_tag(channel->db_session, &tag);
                    }
                }
            }
        }
    }

    return 0;
}

static int cloud_mqtt_internal_recv_msg(void *obj, mqtt_message_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;
    }

    dy_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), "daslink");
	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[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "MQTT_CLD_daslink_%s", var->sn_str);
    var->session = dy_mqtt_session_new(clientId);
    if (var->session == NULL) return -1;

    dy_mqtt_session_set_address(var->session, INTERNAL_BROKER_ADDR, INTERNAL_BROKER_PORT, INTERNAL_BROKER_USER, INTERNAL_BROKER_PASS);
    dy_mqtt_session_set_opts(var->session, DEFAULT_IPC_QOS, KEEP_ALIVE_MAX);
    dy_mqtt_session_set_callbacks(var->session, cloud_mqtt_internal_recv_msg, NULL);
    dy_mqtt_session_init(var->session, (void*)var);

    for (i = 0; i < var->daslink_channel_cnt; i++)
    {
        channel_t *channel = &var->daslink_channel[i];
        for (j = 0; j < channel->topic_maps.uplink_cnt; j++)
        {
            dy_syslog(LOG_DEBUG, "subscribe topic:%s", channel->topic_maps.uplink[j].internal_topic);
            dy_mqtt_session_subscribe(var->session, channel->topic_maps.uplink[j].internal_topic);
        }
    }
}