#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 <signal.h>

#include "dy_utils/dy_mqtt.h"
#include "dy_utils/dy_common.h"
#include "cloud_daslink_common.h"
#include "dy_utils/cJSON.h"
#include "dy_utils/protocol.h"

#define daslink_STR		"daslink_"
#if 0
int recv_cloud_daslink_msg(channel_t *channel, mqtt_message_t *mqtt_msg)
{
    char topic[TOPIC_MAX_LEN] = {0};
    unsigned char matched;
    int j = 0, i = 0;
    int ret;
    cloud_mqtt_var_t *var = channel->var;

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

    for (j = 0; j < channel->topic_maps.downlink_cnt; j++)
    {
        ret = mqtt_topic_matches_sub(channel->topic_maps.downlink[j].external_topic, mqtt_msg->topic, &matched);
        if (ret == 0 && matched)
        {

            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)
                {
                    dy_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);
                dy_syslog(LOG_DEBUG, "forward to internal topic:%s", topic);
                dy_mqtt_session_publish(var->session, topic, mqtt_msg->payload, mqtt_msg->payloadLen);
                cJSON_Delete(node);
            }
            else
            {
                dy_syslog(LOG_DEBUG, "forward to internal topic:%s", channel->topic_maps.downlink[j].internal_topic);
                dy_mqtt_session_publish(var->session, channel->topic_maps.downlink[j].internal_topic,
                                        mqtt_msg->payload, mqtt_msg->payloadLen);
            }
        }
    }

out:
    return 0;
}
#else
int daslink_message_response(channel_t *channel, char *msgId, cJSON *data, int code)
{
	int ret;
	cJSON *payload = cJSON_CreateObject();
	
	cJSON_AddStringToObject(payload, "msgId", msgId);
	cJSON_AddNumberToObject(payload, "code", code);
	cJSON_AddItemToObject(payload, "data", data);
	
	char daslink_topic[TOPIC_MAX_LEN] = {0};
	snprintf(daslink_topic, TOPIC_MAX_LEN, "das/iot/%s/%s/sub/service/response", channel->thing_id,channel->device_id);
	char *daslink_package = cJSON_Print(payload);
	dy_syslog(LOG_DEBUG, "daslink_topic %s,len %d daslink_package:%s",daslink_topic,strlen(daslink_package),daslink_package);
	printf("daslink_topic %s,len %d daslink_package:%s \n",daslink_topic,strlen(daslink_package),daslink_package);
	ret = daslink_pub(daslink_topic, 0, daslink_package, strlen(daslink_package));
	//ret = daslink_report_property("daslink_json", daslink_json);
	dy_syslog(LOG_DEBUG, "daslink_pub ret:%d", ret);
	cJSON_Delete(payload);
	free(daslink_package);
	
	return ret;
}

void recv_cloud_daslink_msg(const void *user_args, daslink_topic_info_t *info)
{
	dy_syslog(LOG_DEBUG, "Daslink MQTT client received MQTT topic:%s payload:%s", info->topic_len,info->payload_len,info->ptopic,info->payload);

	char topic[TOPIC_MAX_LEN] = {0};
    int ret = -1;
	channel_t *channel = (channel_t *)user_args;
    cloud_mqtt_var_t *var = channel->var;
	//das/iot/<gateway_thing_id>/<gateway_device_id>/sub/service/request

	char *thingId = NULL;
	char *deviceId = NULL;
	char *identifier = NULL;
	char *msgId = NULL;
	time_t timeStamp;
	
    cJSON *node = cJSON_Parse(info->payload);
    if (!node)
    {
        dy_syslog(LOG_WARNING, "json payload %s error!!!", info->payload);
        goto _out;
    }

	GET_JSON_VALUE_DY_STRING(node, "msgId", msgId);
	GET_JSON_VALUE_INT(node, "timeStamp", timeStamp);

	if (cJSON_GetObjectItem(node, "payload") != NULL)
    {
    	cJSON *payload = cJSON_GetObjectItem(node, "payload");
		if (cJSON_GetObjectItem(payload, "subDevice") != NULL)
    	{
    		cJSON *subDevice = cJSON_GetObjectItem(payload, "subDevice");
			if (cJSON_GetObjectItem(subDevice, "identity") != NULL)
	    	{
	    		cJSON *identity = cJSON_GetObjectItem(subDevice, "identity");
				GET_JSON_VALUE_DY_STRING(identity, "thingId", thingId);
        		GET_JSON_VALUE_DY_STRING(identity, "deviceId", deviceId);
			}

			if (cJSON_GetObjectItem(subDevice, "service") != NULL)
	    	{
	    		cJSON *service = cJSON_GetObjectItem(subDevice, "service");
				cJSON *child = NULL;
			    for (child = service->child; child; child = child->next)
			    {
			    	if (child->string)
			    	{
			    		identifier = strdup(child->string);
						break;
			    	}
			    }
			}

			if(thingId == NULL || deviceId == NULL || identifier == NULL)
			{
				dy_syslog(LOG_WARNING, "thingId %p deviceId %p identifier %p error!!!", thingId, deviceId, identifier);
		        goto _out;
			}

			daslink_params_list_t* daslink_params = NULL;
			list_for_each_entry(daslink_params, &channel->var->node_list, list)
			{
				if(strcmp(daslink_params->daslink_params.device_id, deviceId) == 0)
				{
					cJSON *payload_json = cJSON_CreateObject();	
					cJSON_AddStringToObject(payload_json, "sn", deviceId);
					cJSON_AddStringToObject(payload_json, "identifier", identifier);
					cJSON_AddNumberToObject(payload_json, "mi", time(NULL));
					cJSON_AddNumberToObject(payload_json, "timestamp", timeStamp);
					snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/Set_Rglt", channel->var->sn_str,daslink_params->port,deviceId);
				
					ret = 200;
					char *payload_data = cJSON_Print(payload_json);
					dy_syslog(LOG_DEBUG, "forward to internal topic:%s", topic);
				    dy_mqtt_session_publish(var->session, topic, payload_data, strlen(payload_data));
					free(payload_data);
					break;
				}
			}

			daslink_message_response(channel, msgId, payload, ret);
		}
    }

_out:
	cJSON_Delete(node);
	if(thingId) free(thingId);
	if(deviceId) free(deviceId);
	if(identifier) free(identifier);
}
#endif

int daslink_message_produce(channel_t *channel, char *topic, char *payload, int payloadLen)
{
	dy_syslog(LOG_DEBUG, "topic:%s payloadLen %d payload %s", topic,payloadLen,payload);
	
	int ret = 0;
	cJSON *root = cJSON_Parse(payload);
	if(root)
	{
		char sn[64] = {0};
		time_t timeStamp = 0;

		GET_JSON_VALUE_STRING(root, "sn", sn);
		GET_JSON_VALUE_INT(root, "time", timeStamp);
		daslink_params_list_t* daslink_params = NULL;
		list_for_each_entry(daslink_params, &channel->var->node_list, list)
		{
			if(strcmp(daslink_params->daslink_params.device_id, sn) == 0)
			{
				cJSON *daslink_json = cJSON_CreateObject();
				cJSON_AddStringToObject(daslink_json, "msgId", "5a0BByAv05vCb4UpDq");
				cJSON_AddNumberToObject(daslink_json, "timeStamp", timeStamp);
				
				cJSON *payload_json = cJSON_CreateObject();
				
				cJSON *subDevices = cJSON_CreateArray();
				cJSON_AddItemToObject(payload_json, "subDevices", subDevices);
				
				cJSON *node = cJSON_CreateObject();
				cJSON_AddItemToArray(subDevices, node);
				
				cJSON *node_identity = cJSON_CreateObject();
				cJSON_AddItemToObject(node, "identity", node_identity);
				cJSON_AddStringToObject(node_identity, "thingId", daslink_params->daslink_params.thing_id);
				cJSON_AddStringToObject(node_identity, "deviceId", daslink_params->daslink_params.device_id);

				cJSON *node_properties = cJSON_CreateObject();
				cJSON_AddItemToObject(node, "properties", node_properties);

				cJSON *tags = cJSON_GetObjectItem(root, "tags");
		        if (tags)
		        {
		        	cJSON *child = NULL;
				    for (child = tags->child; child; child = child->next)
				    {
				    	if (child->string)
				        {
				        	cJSON *propertie = cJSON_CreateObject();
							if (child->valuestring)
                            {
								cJSON_AddStringToObject(propertie, "value", child->valuestring);
							}
							else
							{
								cJSON_AddNumberToObject(propertie, "value", child->valuedouble);
							}
							cJSON_AddNumberToObject(propertie, "time", timeStamp);
							cJSON_AddItemToObject(node_properties, child->string, propertie);
				        }
				    }
		        }
				
				cJSON_AddItemToObject(daslink_json, "payload", payload_json);		

				char daslink_topic[TOPIC_MAX_LEN] = {0};
				snprintf(daslink_topic, TOPIC_MAX_LEN, "das/iot/%s/%s/sub/batch/post", channel->thing_id,channel->device_id);
				char *daslink_package = cJSON_Print(daslink_json);
				dy_syslog(LOG_DEBUG, "daslink_topic %s,len %d daslink_package:%s",daslink_topic,strlen(daslink_package),daslink_package);
				printf("daslink_topic %s,len %d daslink_package:%s \n",daslink_topic,strlen(daslink_package),daslink_package);
				ret = daslink_pub(daslink_topic, 0, daslink_package, strlen(daslink_package));
				//ret = daslink_report_property("daslink_json", daslink_json);
				dy_syslog(LOG_DEBUG, "daslink_report_property ret:%d", ret);
				cJSON_Delete(daslink_json);
				free(daslink_package);
				break;
			}
		}
	}

_out:
	return ret;
}

// 建立与broker之间的MQTT连接
int cloud_daslink_channel_create(cloud_mqtt_var_t *var)
{
    int i = 0, j = 0;
    char tmp[128] = {0};
    const char *board_name = NULL;
    board_name = get_board_name();

	if(var->daslink_channel_cnt <= 0)
	{
		dy_syslog(LOG_DEBUG, "daslink_channel_cnt:%d", var->daslink_channel_cnt);
		return 0;
	}
	
    for (i = 0; i < var->daslink_channel_cnt; i++)
    {
        channel_t *channel = &var->daslink_channel[i];
        char db_name[32] = {0};

		dy_syslog(LOG_DEBUG, "channel:%p board_name:%s", channel,board_name);
		dy_syslog(LOG_DEBUG, "resume:%d", channel->resume);

        channel->var = var;

        if (channel->resume)
        {
        	dy_syslog(LOG_DEBUG, "resume:%d", channel->resume);
            if (strcmp(board_name, BOARD_WOOLINK_MT7628) == 0)
            {
                sprintf(db_name, "/tmp/daslink_REPORT_%d.db", i);
            }
            else
            {
                sprintf(db_name, "/app/daslink_REPORT_%d.db", i);
            }
			dy_syslog(LOG_DEBUG, "db_name:%s", db_name);
            dy_db_session_init(&channel->db_session, db_name, (void*)channel);
        }
		dy_syslog(LOG_DEBUG, "=== resume:%d addr:%s port:%d ===", channel->resume, channel->addr, channel->port);
		channel->params.thing_id = channel->thing_id;
		channel->params.device_id = channel->device_id;
		channel->params.device_secret = channel->device_secret;
		channel->params.host = channel->addr;
		channel->params.port = channel->port;
		strcpy(channel->params.firmware_ver, "v1.1.0");
		channel->params.extra_obj = NULL;

		int retCode = daslink_init(&channel->params, NULL, NULL);
		dy_syslog(LOG_DEBUG, "=== daslink_init retCode %d (%s %d)(%s %s %s)===", retCode,channel->params.host,channel->params.port,channel->params.thing_id,channel->params.device_id,channel->params.device_secret);
	    if (retCode != RET_SUCCESS) {
	        dy_syslog(LOG_DEBUG, "das link init failed");
	        return 0;
	    }
		dy_syslog(LOG_DEBUG, "=== resume:%d addr:%s port:%d end===", channel->resume, channel->addr, channel->port);

		char daslink_topic[TOPIC_MAX_LEN] = {0};
		snprintf(daslink_topic, TOPIC_MAX_LEN, "das/iot/%s/%s/sub/service/request", channel->thing_id,channel->device_id);
		retCode = daslink_sub(daslink_topic, 0, recv_cloud_daslink_msg, channel, 5000);
		dy_syslog(LOG_DEBUG, "=== daslink_sub retCode %d daslink_topic %s===", retCode,daslink_topic);
    }
}