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

#include "ipc_session.h"
#include "mqtt_session.h"
#include "dy_utils/dy_common.h"
#include "common.h"
#include "dy_utils/dy_ipc.h"


static int get_rtd_value(lanzhuo_var_t *var, mqtt_message_t *mqtt_msg)
{
    int ret = 0;
    char *tag_node = NULL;
    tag_table_t tag = {0};
    long long now = get_time_stamp_ms();

    dy_syslog(LOG_DEBUG, "=====time stamp ms %lld", now);

    cJSON *root = cJSON_Parse(mqtt_msg->payload);
    if (root)
    {
        GET_JSON_VALUE_DY_STRING(root, "tag_node", tag_node);
        cJSON *root_single = cJSON_Parse(tag_node);
        if (root_single)
        {
            int cnt = cJSON_GetArraySize(root_single);
            ValueSequence value_seq = VALUE_SEQUENCE__INIT;
            int i = 0;
            int index = 0;

            value_seq.n_values = cnt;
            NamedValue **values = calloc(cnt, sizeof(NamedValue *));
            value_seq.values = values;

            for (i = 0; i < cnt; i++)
            {
                values[i] = calloc(1, sizeof(NamedValue));
                named_value__init(values[i]);
                values[i]->value = malloc(sizeof(RtdValue));
                rtd_value__init(values[i]->value);
            }

            if (root_single->child)
            {
                cJSON *child = root_single->child;
                while (child)
                {
                    char *name = child->string;
                    RtdValue__ValueCase value_case = RTD_VALUE__VALUE__NOT_SET;

                    values[index]->name = strdup(name);
                    values[index]->value->timestamp = now;
                    values[index]->value->quality = 0;

                    for (i = 0; i < var->meta_tag_seq.n_tags; i++)
                    {
                        if (strcmp(name, var->meta_tag_seq.tags[i]->name) == 0)
                        {
                            switch (var->meta_tag_seq.tags[i]->type)
                            {
                                case VALUE_TYPE__Double:
                                    values[index]->value->value_case = RTD_VALUE__VALUE_DBL_VAL;
                                    values[index]->value->dblval = child->valuedouble;
                                    break;
                                case VALUE_TYPE__Integer:
                                    values[index]->value->value_case = RTD_VALUE__VALUE_INT_VAL;
                                    values[index]->value->intval = child->valueint;
                                    break;
                                case VALUE_TYPE__String:
                                    values[index]->value->value_case = RTD_VALUE__VALUE_STR_VAL;
                                    values[index]->value->strval = strdup(child->valuestring);
                                    break;
                                case VALUE_TYPE__Bytes:
                                    values[index]->value->value_case = RTD_VALUE__VALUE_BYT_VAL;
                                    int bin_len = strlen(child->valuestring) * 2;
                                    char *bin_buff = malloc(bin_len);
                                    int buf_len = b64_decode(child->valuestring, bin_buff, bin_len);
                                    values[index]->value->bytval.len = buf_len;
                                    values[index]->value->bytval.data = bin_buff;

                                    break;
                                case VALUE_TYPE__Boolean:
                                    values[index]->value->value_case = RTD_VALUE__VALUE_BOOL_VAL;
                                    values[index]->value->boolval = child->valueint;
                                    break;

                                default:
                                    break;
                            }
                            break;
                        }
                    }
                    index++;
                    child = child->next;
                }
            }

            int len = value_sequence__get_packed_size(&value_seq);
            char *buf = malloc(len);
            value_sequence__pack(&value_seq, buf);
            char topic[TOPIC_MAX_LEN] = {0};

            snprintf(topic, TOPIC_MAX_LEN, "/%s/%s/%s/%s/%s", var->token, var->ep_id, var->ep_name, "rtdvalue", "report");
            ret = mqtt_session_publish_with_flag(var->session_ext, topic, buf, len, 1, false);
            if (1)
            {
                tag.tag_node = buf;
                GET_JSON_VALUE_STRING(root, "identifier", tag.identifier);
                tag.time = now;
                if (ret == 0)
                {
                    tag.report = REPORTED;
                }
                else
                {
                    tag.report = NOT_REPORT;
                }
                tag.tag_len = len;
                dy_db_session_insert_tag(var->db_session, &tag);
            }
            dy_syslog(LOG_DEBUG, "pub topic:%s", topic);

            for (i = 0; i < cnt; i++)
            {
                if (values[i]->value->value_case == RTD_VALUE__VALUE_STR_VAL)
                {
                    free(values[index]->value->strval);
                }
                else if (values[i]->value->value_case == RTD_VALUE__VALUE_BYT_VAL)
                {
                    free(values[i]->value->bytval.data);
                }
                free(values[i]->value);
                free(values[i]);
            }
            free(values);
        }
        else
        {
            dy_syslog(LOG_WARNING, "tag_node:%s parse error", tag_node);
        }

        free(tag_node);
    }
    else
    {
        dy_syslog(LOG_ERR, "json parse error");
        ret = -1;
    }

out:
    cJSON_Delete(root);
    return ret;
}

static int internal_mqtt_handle_recv_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    lanzhuo_var_t *var = (lanzhuo_var_t *)obj;
    dy_syslog(LOG_DEBUG, "received MQTT topic:%s payload length:%d",
              mqtt_msg->topic, mqtt_msg->payloadLen);

    get_rtd_value(var, mqtt_msg);

    return 0;
}

static void internal_mqtt_subscribe_all(lanzhuo_var_t *var)
{
    ipc_session_t *session = var->session_int;
    char topic[TOPIC_MAX_LEN] = {0};

    snprintf(topic, TOPIC_MAX_LEN, "ipc/+/+/device/+/data/property/+");
    ipc_session_subscribe(session, topic);
}

// 建立与内部broker之间的MQTT连接
int internal_client_init(lanzhuo_var_t *var)
{
    ipc_session_set_callbacks(var->session_int, internal_mqtt_handle_recv_msg, NULL);
    internal_mqtt_subscribe_all(var);
    ipc_session_start(var->session_int);
}