/*
 * Created by jiaqiang.ye@lnxall.com
 *
 * 创业惠康 Protocol support
 *
 * 2022/01/10
 */

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

#include "cyhk_protocol.h"
#include "mqtt_session.h"
#include "ipc_session.h"
#include "cJSON.h"

#define CYHK_MSGLEN 32768

static void cyhk_msleep(unsigned int msec)
{
    struct timespec spec;
    spec.tv_sec = (time_t) (msec / 1000);
    spec.tv_nsec = (long) ((msec % 1000) * 1000000);
    nanosleep(&spec, NULL);
}

static int cyhk_handle_ipc_dummy(void * obj, ipc_msg_t * msg)
{
    (void) obj;
    (void) msg;
    /* dummy callback function for IPC session */
    return 0;
}

static int cyhk_handle_ipc(void * obj, ipc_msg_t * msg)
{
    char * upmsg;
    int plen, rval;
    char tmpstr[32];
    const char * pload;
    cyhk_proto_t * cyhk;
    cJSON * j_body, * j_first;
    cJSON * payload, * upload;
    cJSON * j_msg, * j_header;

    rval = 0;
    upmsg = NULL;
    j_body = j_msg = NULL;
    j_first = j_header = NULL;
    payload = upload = NULL;
    cyhk = (cyhk_proto_t *) obj;

    plen = msg ? msg->payloadLen : 0;
    pload = msg ? msg->payload : NULL;
    if (plen <= 0 || pload == NULL)
        return 0; /* sliently drop the invalid message */

    payload = cJSON_Parse(pload);
    if (payload == NULL) {
        fprintf(stderr, "invalid payload as JSON: %s\n", pload);
        fflush(stderr);
        return -1;
    }

    /* create JSON message for CYHK cloud */
    upload = cJSON_CreateObject();
    j_header = cJSON_CreateObject();
    if (upload == NULL || j_header == NULL) {
        rval = -2;
        goto err0;
    }

    snprintf(tmpstr, sizeof(tmpstr), "%u", cyhk_messageid(cyhk));
    cJSON_AddStringToObject(j_header, "id", tmpstr);
    cJSON_AddNumberToObject(j_header, "structId", CYHK_SID_DATA_REPORT);
    cJSON_AddItemToObject(upload, "header", j_header);

    /* create 'body' JSON object */
    j_body = cJSON_CreateArray();
    if (j_body == NULL) {
        rval = -4;
        goto err0;
    }

    /* create the first (and only) object in `messages */
    j_first = cJSON_CreateObject();
    if (j_first == NULL) {
        rval = -5;
        goto err0;
    }

    /* add `msg_type to json */
    cJSON_AddNumberToObject(j_first, "msg_type", 61);
    /* add `msg to json */
    j_msg = cJSON_CreateObject();
    if (j_msg == NULL) {
        rval = -6;
        goto err0;
    }

    /* add card_id field */
    cJSON_AddNumberToObject(j_msg, "card_id", 0);
    do { /* add dev_sn field */
        cJSON * psn;
        const char * dev_sn = "unknown_sn";
        psn = cJSON_GetObjectItemCaseSensitive(payload, "sn");
        if (psn && cJSON_IsString(psn) && psn->valuestring)
            dev_sn = psn->valuestring;
        cJSON_AddStringToObject(j_msg, "dev_sn", dev_sn);
    } while (0);
    /* add protocol filed */
    cJSON_AddNumberToObject(j_msg, "protocol", 0);
    do { /* add tagnode data fields */
        cJSON * tagn;
        tagn = cJSON_GetObjectItemCaseSensitive(payload, "tag_node");
        if (tagn && cJSON_IsString(tagn) && tagn->valuestring) {
            cJSON * tagnode;
            tagnode = cJSON_Parse(tagn->valuestring);
            if (tagnode && cJSON_IsObject(tagnode)) {
                cJSON * first, * next;
                first = tagnode->child;
                next = first;
                while (next != NULL) {
                    cJSON * jt;
                    jt = cJSON_Duplicate(next, 1);
                    if (jt != NULL)
                        cJSON_AddItemToObject(j_msg, next->string, jt);
                    next = next->next;
                    if (next == first)
                        break;
                }
            }
            if (tagnode != NULL)
                cJSON_Delete(tagnode);
        }
    } while (0);

    cJSON_AddItemToObject(j_first, "msg", j_msg);
    j_msg = NULL;

    cJSON_AddItemToArray(j_body, j_first);
    j_first = NULL;

    do {
        char * bodystr = cJSON_PrintUnformatted(j_body);
        if (bodystr != NULL) {
            cJSON_AddStringToObject(upload, "body", bodystr);
            free(bodystr);
        } else {
            fputs("Error, system out of memory!\n", stderr);
            fflush(stderr);
        }
    } while (0);

    upmsg = (char *) malloc(CYHK_HEAD_SIZE + CYHK_MSGLEN + 1);
    if (upmsg == NULL) {
        rval = -7;
        goto err0;
    }

    do {
        struct mosquitto * msess;
        struct cyhk_magic * cmp;

        cmp = (struct cyhk_magic *) upmsg;
        cmp->serial_id = CYHK_SERIAL_JSON;
        cmp->struct_class = 50; cmp->struct_id = 11582;

        if (cJSON_PrintPreallocated(upload, upmsg + CYHK_HEAD_SIZE, CYHK_MSGLEN, 0)) {
            int plen, msgid;
            msgid = 0;
            plen = CYHK_HEAD_SIZE;
            upmsg[CYHK_HEAD_SIZE + CYHK_MSGLEN] = '\0';
            plen += strlen(upmsg + CYHK_HEAD_SIZE);
            msess = (struct mosquitto *) cyhk->sess;
            /* publish the message: */
            if (mosquitto_publish(msess, &msgid,
                (const char *) cyhk->data_topic, plen, (const void *) upmsg, 1, 0) < 0) {
                fputs("Error, failed to publish message to cloud!\n", stderr);
                fflush(stderr);
            }
        } else {
            fputs("Error, failed to format JSON data!\n", stderr);
            fflush(stderr);
        }
    } while (0);

err0:
    if (upmsg != NULL)
        free(upmsg);
    if (upload != NULL)
        cJSON_Delete(upload);
    if (j_body != NULL)
        cJSON_Delete(j_body);
    if (j_msg != NULL)
        cJSON_Delete(j_msg);
    if (j_first != NULL)
        cJSON_Delete(j_first);
    if (payload != NULL)
        cJSON_Delete(payload);
    if (rval < 0) {
        /* make some noise */
        fprintf(stderr, "Error in [%s], return value: %d\n",
            __FUNCTION__, rval);
        fflush(stderr);
    }
    return rval;
}

int cyhk_create_session(cyhk_proto_t * cyhk)
{
    int ret;
    char tmpstr[256];
    struct mosquitto * sess = NULL;
    ipc_session_t * ipcsess = NULL;

    if (cyhk == NULL)
        return -1;

    ret = snprintf(tmpstr, sizeof(tmpstr), "CYHK_%s", cyhk->macAddr);
    if (ret >= sizeof(tmpstr)) {
        ret = sizeof(tmpstr) - 1;
        tmpstr[ret] = '\0';
    }

	/* remote MQTT session already created, cloud service of 创业惠康 */
	sess = (struct mosquitto *) cyhk->sess;
	if (sess == NULL) {
        fprintf(stderr, "Error, failed to create remote MQTT session for [%s]\n",
            tmpstr);
        fflush(stderr);
        goto err0;
    }

    /* create local IPC session */
    ipcsess = ipc_session_new(tmpstr, cyhk, IPC_MQTT);
    if (ipcsess == NULL) {
        fprintf(stderr, "Error, failed to create local IPC session for [%s]\n",
            tmpstr);
        fflush(stderr);
        goto err0;
    }

    /* handle local IPC messages */
    ret = ipc_session_set_callbacks(ipcsess, cyhk_handle_ipc, NULL);
    if (ret < 0) {
        fputs("Error, cannot register local message callback!\n", stderr);
        fflush(stderr);
        goto err0;
    }

    /* subscribe to local message topic */
    ret = ipc_session_subscribe(ipcsess, "ipc/+/+/device/+/data/service/#");
    if (ret < 0) {
        fputs("Error, failed to subscribe local topic!\n", stderr);
        fflush(stderr);
        goto err0;
    }

    /* save the sessions before starting them */
    cyhk->sess_local = (void *) ipcsess;
    asprintf(&cyhk->data_topic, "%s/%s/out/%d",
        cyhk->productKey, cyhk->macAddr, CYHK_CMD_DATA_REPORT);
    if (cyhk->data_topic == NULL) {
        fputs("Error, failed to allocate data topic!\n", stderr);
        fflush(stderr);
        goto err0;
    }

    /* start local session */
    ret = ipc_session_start(ipcsess);
    if (ret < 0) {
        fputs("Error, failed to start local session!\n", stderr);
        fflush(stderr);
        goto err0;
    }
    return 0;

err0:
    if (ipcsess != NULL) {
        ipc_session_destroy_and_free(ipcsess);
        ipcsess = NULL;
    }
    return -1;
}

void cyhk_close_session(cyhk_proto_t * cyhk)
{
    ipc_session_t * isess;
    if (cyhk == NULL)
        return;

    isess = (ipc_session_t *) cyhk->sess_local;
    if (isess != NULL) {
        ipc_session_set_callbacks(isess,
            cyhk_handle_ipc_dummy, NULL);
        cyhk_msleep(250); /* delay 0.25 second */
        ipc_session_destroy_and_free(isess);
        cyhk->sess_local = NULL;
    }
    cyhk->sess = NULL;
}

/*
 * multiple thread will frequently call this function,
 * atomic operation is needed to avoid pthread_mutex_lockING.
 */
unsigned int cyhk_messageid(cyhk_proto_t * cyhk)
{
    unsigned int rval;
    unsigned int * pmsgid;
    pmsgid = &cyhk->msgid;
    rval = __atomic_fetch_add(pmsgid, 0x1u, __ATOMIC_SEQ_CST);
    rval &= 0x7FFFFFFFu;
    return rval;
}