#include "common.h"
#include "dy_utils/dy_common.h"

static int update_sn_socket_session(sock_sess_node_t *new_sess)
{
    sock_sess_node_t *sess = NULL;

    pthread_rwlock_wrlock(&socket_mqtt_var.sock_sess_lock);
    sess = find_session_by_sn_nolock(new_sess->sn);
    if (sess == NULL)
    {
        char topic[TOPIC_MAX_LEN] = {0};
        sess = malloc(sizeof(sock_sess_node_t));
        sess->socket = new_sess->socket;
        strcpy(sess->sn, new_sess->sn);
        sess->mqtt = new_sess->mqtt;
        add_session_nolock(sess);
        dy_syslog(LOG_INFO, "add session, sn:%s socket:%d", new_sess->sn, sess->socket);
        sprintf(topic, "GW/%s/transparent", new_sess->sn);
        mqtt_session_subscribe(sess->mqtt, topic);
    }
    else
    {
        del_session_nolock(sess);
        sess->socket = new_sess->socket;
        sess->mqtt = new_sess->mqtt;
        add_session_nolock(sess);
        dy_syslog(LOG_INFO, "update session, sn:%s, socket:%d", sess->sn, sess->socket);
    }
    pthread_rwlock_unlock(&socket_mqtt_var.sock_sess_lock);

    return 0;
}

int mqtt_handle_sync_sn(mqtt_session_t *mqtt, mqtt_message_t *mqtt_msg)
{
    int socket = -1;
    char sn[SN_MAX_LEN] = {0};
    char ipAddr[INET_ADDRSTRLEN];
    int port;
    sock_sess_node_t sess;
    cJSON *root = cJSON_Parse(mqtt_msg->payload);

    if (root)
    {
        GET_JSON_VALUE_INT(root, "socket", socket);
        GET_JSON_VALUE_STRING(root, "sn", sn);
        GET_JSON_VALUE_STRING(root, "ip", ipAddr);
        GET_JSON_VALUE_INT(root, "port", port);
    }
    cJSON_Delete(root);

    strcpy(sess.sn, sn);
    sess.socket = socket;
    sess.mqtt = mqtt;
    update_sn_socket_session(&sess);
    return 0;
}

static int send_data_to_socket(int socket, char *payload, int payload_len)
{
    int len;

    len = socket_send(socket, payload, payload_len);
    if (len < 0)
    {
        dy_syslog(LOG_INFO, "send error: errno:%d, reason:%s", errno, strerror(errno));
    }
    else
    {
        dy_syslog(LOG_INFO, "send buf:%s to socket:%d", payload, socket);
    }

    return 0;
}

static int mqtt_handle_data(mqtt_message_t *mqtt_msg)
{
    char sn[SN_MAX_LEN] = {0};
    char *buf = NULL;
    sock_sess_node_t *sess;
    cJSON *root = cJSON_Parse(mqtt_msg->payload);
    char s[2][32] = {0};
    int len;

    if (sscanf(mqtt_msg->topic, "%32[^/]/%32[^/]/%32s", s[0], sn, s[2]) != 3)
    {
        dy_syslog(LOG_ERR, "sscanf error, topic:%s", mqtt_msg->topic);
        return -1;
    }

    if (root)
    {
        char *payload = NULL;
        GET_JSON_VALUE_DY_STRING(root, "payload", payload);
        buf = (char *)malloc(MAXBUF);
        len = b64_decode(payload, buf, MAXBUF);
        free(payload);
    }
    else
    {
        dy_syslog(LOG_ERR, "json parse error");
        return -1;
    }
    cJSON_Delete(root);
    sess = find_session_by_sn(sn);
    if (sess != NULL)
    {
        send_data_to_socket(sess->socket, buf, len);
    }
    free(buf);

    return 0;
}

static int recv_mqtt_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    char topic[TOPIC_MAX_LEN] = {0};
    unsigned char matched;
    int j = 0, i = 0;
    int ret;

    mqtt_session_t *session = (mqtt_session_t *)obj;

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

    if (strstr(mqtt_msg->topic, "RspSearchSN"))
    {
        mqtt_handle_sync_sn(session, mqtt_msg);
    }
    else if (strstr(mqtt_msg->topic, "transparent"))
    {
        // data
        mqtt_handle_data(mqtt_msg);
    }

out:
    return 0;
}

int mqtt_init(socket_mqtt_var_t *var)
{
    int i = 0;
    char topic[TOPIC_MAX_LEN] = {0};

    for (i = 0; i < var->mqtt_cnt; i++)
    {
        mqtt_session_t *session = var->mqtt[i];
        if (session == NULL)
        {
            dy_syslog(LOG_ERR, "malloc error");
            continue;
        }

        mqtt_session_set_opts(session, DEFAULT_QOS, KEEPALIVEINTERVAL);
        mqtt_session_set_callbacks(session, recv_mqtt_msg, NULL);
        mqtt_session_start(session);

        sprintf(topic, "GW/%s/RspSearchSN", session->client.clientId);
        mqtt_session_subscribe(session, topic);
    }
}