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

#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <signal.h>
#include <sys/types.h>
#include <unistd.h>

#include "cyhk_protocol.h"
#include "mqtt_session.h"

#include "lua.h"
#include "lualib.h"
#include "lauxlib.h"

struct cyhk_msg {
    char *              topic;
    char *              payload;
    int                 plen;
#define CYHK_MSG_MAGIC  0x20220722
    unsigned int        magic; /* for memory error detection */
    struct cyhk_msg *   next;
};

/* global CYHK protocol private data */
static cyhk_proto_t * g_cyhk_proto;
#define DECLARE_CYHK_PROTO(cpn__) \
    cyhk_proto_t * cpn__ = g_cyhk_proto

static time_t cyhk_gettime(struct timespec * pts)
{
    int ret, error;
    struct timespec tspec;

    if (pts == NULL)
        pts = &tspec;
    pts->tv_sec = 0;
    pts->tv_nsec = 0;

    ret = clock_gettime(CLOCK_MONOTONIC, pts);
    if (ret == -1) {
        error = errno;
        pts->tv_sec = time(NULL);
        pts->tv_nsec = 0;
        fprintf(stderr, "Error, failed to get time for cyhk: %s\n",
            strerror(error));
        fflush(stderr);
    }
    return pts->tv_sec;
}

static int cyhk_lock(pthread_mutex_t * pmut,
    const char * errmsg)
{
    int ret;
    if (errmsg == NULL)
        errmsg = "unknown";
    ret = pthread_mutex_lock(pmut);
    if (ret != 0) {
        fprintf(stderr, "[%s]: failed to lock mutex: %d\n",
            errmsg, ret);
        fflush(stderr);
    }
    return ret;
}

static int cyhk_unlock(pthread_mutex_t * pmut,
    const char * errmsg)
{
    int ret;
    if (errmsg == NULL)
        errmsg = "unknown";
    ret = pthread_mutex_unlock(pmut);
    if (ret != 0) {
        fprintf(stderr, "[%s]: failed to release mutex: %d\n",
            errmsg, ret);
        fflush(stderr);
    }
    return ret;
}

static void * cyhk_thread_func(void * fdata)
{
    int ret;
    cyhk_proto_t * cyp;
    struct mosquitto * mosq;
    const char * fname = __FUNCTION__;

    cyp = (cyhk_proto_t *) fdata;

    /* acquire mutex lock between main thread */
    ret = cyhk_lock(&cyp->mlock, fname);
    if (ret)
        return NULL;

    /* create IPC/MQTT session */
    ret = cyhk_create_session(cyp);
    if (ret < 0)
        goto err0;

    mosq = (struct mosquitto *) cyp->sess;
    while (cyp->loopflag != 0) {
        struct timespec tspec;
        struct cyhk_msg * msgs;

        msgs = (struct cyhk_msg *) cyp->cyhk_msgs;
        while (msgs == NULL) {
            cyhk_gettime(&tspec);
            tspec.tv_sec += 15;
            tspec.tv_nsec = 0;
            ret = pthread_cond_timedwait(&cyp->condition, &cyp->mlock, &tspec);
            if (cyp->loopflag == 0)
                break;
            if (ret == ETIMEDOUT) {
                msgs = (struct cyhk_msg *) cyp->cyhk_msgs;
                continue;
            }

            if (ret) {
                const char * errstr;
                if (ret < 0)
                    ret = -ret;
                errstr = strerror(ret);
                if (!errstr) errstr = "unknown";
                fprintf(stderr, "Error, cond_timedwait(...) has failed with %d: %s\n",
                    ret, errstr);
                fflush(stderr);
                break;
            }
            msgs = (struct cyhk_msg *) cyp->cyhk_msgs;
        }

        if (msgs == NULL)
            break;
        cyp->cyhk_msgs = NULL;
        while (msgs != NULL) {
            int msgid = 0;
            struct cyhk_msg * next;

            if (msgs->magic != CYHK_MSG_MAGIC) {
                fprintf(stderr, "Error, invalid internal msg-magic: %p, %#x\n",
                    msgs, msgs->magic);
                fflush(stderr);
                _exit(2);
            }

            ret = mosquitto_publish(mosq, &msgid,
                msgs->topic, msgs->plen, msgs->payload, 1, 0);
            if (ret != MOSQ_ERR_SUCCESS) {
                fprintf(stderr, "Error, failed to send MQTT message: %d\n", ret);
                fflush(stderr);
            }

            next = msgs->next;
            msgs->topic = NULL;
            msgs->payload = NULL;
            msgs->plen = 0;
            msgs->magic = 0;
            msgs->next = NULL;
            free(msgs);
            msgs = next;
        }
    }

    fprintf(stderr, "CYHK working thread will exit now: %d\n", cyp->loopflag);
    fflush(stdout);
err0:
    cyhk_close_session(cyp);
    cyhk_unlock(&cyp->mlock, fname);
    if (cyp->loopflag)
        _exit(2);
    return NULL;
}

static int cyhk_thread_productkey(lua_State * L)
{
    int ntop;
    const char * pkey;
    const char * macaddr;
    DECLARE_CYHK_PROTO(cyp);

    ntop = lua_gettop(L);
    if (ntop < 0x2 ||
        lua_isstring(L, 1) == 0 ||
        lua_isstring(L, 2) == 0) {
err0:
        lua_pushnil(L);
        lua_pushstring(L, "invalid argument(s) for CYHK start");
        return 2;
    }

    pkey = lua_tolstring(L, 1, NULL);
    macaddr = lua_tolstring(L, 2, NULL);
    if (pkey == NULL || pkey[0] == '\0' ||
        macaddr == NULL || macaddr[0] == '\0')
        goto err0;

    /* free existing productkey and MAC address */
    if (cyp->productKey != NULL)
        free(cyp->productKey);
    if (cyp->macAddr == NULL)
        free(cyp->macAddr);

    /* duplicate productKey and MAC address */
    cyp->productKey = strdup(pkey);
    cyp->macAddr = strdup(macaddr);
    if (cyp->productKey == NULL ||
        cyp->macAddr == NULL) {
        lua_pushnil(L);
        lua_pushstring(L, "failed to allocate productKey or MAC address");
        return 2;
    }

    lua_pushboolean(L, 1);
    return 1;
}

static int cyhk_thread_start(lua_State * L)
{
    int ret;
    const char * mstr;
    unsigned long mptr;
    DECLARE_CYHK_PROTO(cyp);

    mptr = 0;
    mstr = NULL;
    if (lua_checkstack(L, 2) == 0)
        return 0;

    if (lua_gettop(L) >= 1 && lua_type(L, 1) == LUA_TSTRING) {
        int error;
        mstr = lua_tolstring(L, 1, NULL);
        if (mstr && mstr[0]) {
            errno = 0;
            mptr = strtoul(mstr, NULL, 0);
            error = errno;
            if (error || mptr == ULONG_MAX)
                mptr = 0;
        }
    }

    if (mptr == 0) {
        lua_pushnil(L);
        lua_pushfstring(L, "Error, invalid MQTT handle %s", mstr ? : "nil");
        return 2;
    }

    if (cyp->tidwork) {
        cyp->loopflag = 0;
        ret = pthread_kill(cyp->tidwork, 0);
        if (ret == 0)
            goto next;
    }

    cyp->sess = (void *) mptr;
    cyp->tidwork = 0;
    cyp->loopflag = -1;
    ret = pthread_create(&cyp->tidwork,
        NULL, cyhk_thread_func, (void *) cyp);
    if (ret) {
        lua_pushnil(L);
        lua_pushfstring(L, "failed to create thread: %d", ret);
        return 2;
    }

next:
    lua_pushboolean(L, 1);
    return 1;
}

static int cyhk_thread_stop(lua_State * L)
{
    int ret;
    void * rval = NULL;
    DECLARE_CYHK_PROTO(cyp);

    if (lua_checkstack(L, 2) == 0)
        return 0;

    cyp->loopflag = 0; /* set loop flag to zero */

    /* check the thread-id of working thread */
    if (cyp->tidwork == 0) {
        lua_pushboolean(L, 1);
        return 1;
    }

    /* stop the working thread: */
    ret = pthread_cancel(cyp->tidwork);
    if (ret) {
        fprintf(stderr, "Error, failed to cancel thread: %d\n", ret);
        fflush(stderr);
    }

    /* wait for working thread to exit */
    ret = pthread_join(cyp->tidwork, &rval);
    if (ret) {
        fprintf(stderr, "Error, failed to join thread(%lx): %d\n",
            (unsigned long) cyp->tidwork, ret);
        fflush(stderr);
        lua_pushnil(L);
        lua_pushfstring(L, "failed to join thread: %d", ret);
        return 2;
    }

    cyp->tidwork = 0;
    lua_pushboolean(L, 1);
    return 1;
}

static int cyhk_thread_address(lua_State * L)
{
    int ntop;
    const char * mtaddr;
    unsigned short portno;
    DECLARE_CYHK_PROTO(cyp);

    if (lua_checkstack(L, 2) == 0)
        return 0;

    ntop = lua_gettop(L);
    if (ntop < 2 || lua_isstring(L, 1) == 0 ||
        lua_isinteger(L, 2) == 0) {
err0:
        lua_pushnil(L);
        lua_pushstring(L, "invalid argument(s) for IPC/MQTT address");
        return 2;
    }

    mtaddr = lua_tolstring(L, 1, NULL);
    if (mtaddr == NULL || mtaddr[0] == '\0')
        goto err0;

    /* free old IPC/MQTT server address */
    if (cyp->svrAddr != NULL) {
        free(cyp->svrAddr);
        cyp->svrAddr = NULL;
    }
    /* save the new address */
    cyp->svrAddr = strdup(mtaddr);
    if (cyp->svrAddr == NULL) {
        lua_pushnil(L);
        lua_pushstring(L, "system out of memory");
        return 2;
    }

    /* update IPC/MQTT server port */
    portno = (unsigned short) lua_tointeger(L, 2);
    if (portno != 0)
        cyp->svrPort = (unsigned long) portno;

    if (ntop > 2 && lua_isstring(L, 3)) {
        const char * mtuser;
        mtuser = lua_tolstring(L, 3, NULL);
        if (cyp->userName != NULL) {
            free(cyp->userName);
            cyp->userName = NULL;
        }
        if (mtuser && mtuser[0] != '\0') {
            /* TODO: check the return value of `strdup(...) */
            cyp->userName = strdup(mtuser);
        }
    }

    if (ntop > 3 && lua_isstring(L, 4)) {
        const char * mtpasswd;
        mtpasswd = lua_tolstring(L, 4, NULL);
        if(cyp->userPass != NULL) {
            free(cyp->userPass);
            cyp->userPass = NULL;
        }
        if (mtpasswd && mtpasswd[0] != '\0') {
            /* TODO: check the return value of `strdup(...) */
            cyp->userPass = strdup(mtpasswd);
        }
    }

    lua_pushboolean(L, 1);
    return 1;
}

static int cyhk_thread_alive(lua_State * L)
{
    pthread_t ptid;
    int ret, alive = 0;
    DECLARE_CYHK_PROTO(cyp);

    if (lua_checkstack(L, 2) == 0)
        return 0;
    ptid = cyp->tidwork;
    if (ptid == 0)
        goto err0;
    ret = pthread_kill(ptid, 0);
    if (ret == 0)
        alive = 1;
err0:
    lua_pushboolean(L, alive);
    return 1;
}

static int cyhk_thread_publish(lua_State * L)
{
    int ret, ntop;
    const char * topout;
    const char * payload;
    unsigned char * pmsg;
    size_t plen, tlen, mlen;
    DECLARE_CYHK_PROTO(cyp);
    struct cyhk_msg * cmsg, * dmsg;
    const char * fname = __FUNCTION__;

    cmsg = NULL;
    plen = tlen = 0;
    if (lua_checkstack(L, 2) == 0)
        return 0;
    ntop = lua_gettop(L);
    if (ntop < 2 || lua_isstring(L, 1) == 0 ||
        lua_isstring(L, 2) == 0) {
err0:
        lua_pushnil(L);
        lua_pushstring(L, "invalid argument(s) for publish");
        return 2;
    }

    /* get IPC/MQTT topic */
    topout = lua_tolstring(L, 1, &tlen);
    if (topout == NULL || topout[0] == '\0')
        goto err0;
    /* get the payload */
    payload = lua_tolstring(L, 2, &plen);
    if (payload == NULL || payload[0] == '\0' || plen == 0)
        goto err0;

    mlen = sizeof(struct cyhk_msg) + plen + 1 + tlen + 1 + 0x8;
    cmsg = (struct cyhk_msg *) malloc(mlen);
    if (cmsg == NULL) {
        fprintf(stderr, "Error, system out of memory: %lu\n", (unsigned long) mlen);
        fflush(stderr);
        lua_pushnil(L);
        lua_pushstring(L, "System out of memory");
        return 2;
    }

    pmsg = (unsigned char *) cmsg;
    cmsg->topic = (char *) (pmsg + sizeof(*cmsg));
    cmsg->payload = (char *) (pmsg + sizeof(*cmsg) + tlen + 1);
    cmsg->plen = (int) plen;
    cmsg->magic = CYHK_MSG_MAGIC;
    cmsg->next = NULL;
    if (tlen > 0)
        memcpy(cmsg->topic, topout, tlen);
    cmsg->topic[tlen] = '\0';
    if (plen > 0)
        memcpy(cmsg->payload, payload, plen);
    cmsg->payload[plen] = '\0';

    ret = cyhk_lock(&cyp->mlock, fname);
    if (ret) {
        free(cmsg);
        lua_pushnil(L);
        lua_pushfstring(L, "Failed to acquire mutex lock: %d", ret);
        return 2;
    }

    dmsg = (struct cyhk_msg *) cyp->cyhk_msgs;
    cmsg->next = dmsg;
    cyp->cyhk_msgs = (void *) cmsg;
    ret = cyhk_unlock(&cyp->mlock, fname);
    if (ret) {
        lua_pushnil(L);
        lua_pushfstring(L, "Failed to release mutex lock: %d\n", ret);
        return 2;
    }

    ret = pthread_cond_signal(&cyp->condition);
    if (ret) {
        fprintf(stderr, "Error, failed to wakeup thread: %d\n", ret);
        fflush(stderr);
        lua_pushboolean(L, 0);
    } else
        lua_pushboolean(L, 1);
    return 1;
}

/*
 * publish MQTT message to internal broker
 */
static int cyhk_thread_pubint(lua_State * L)
{
    size_t plen;
    int ret, ntop;
    const char * topout;
    const char * payload;
    ipc_session_t * isess;
    DECLARE_CYHK_PROTO(cyp);

    plen = 0;
    if (lua_checkstack(L, 2) == 0)
        return 0;
    isess = (ipc_session_t *) cyp->sess_local;
    if (isess == NULL) {
        fputs("Error, internal MQTT session not initialized!\n", stderr);
        fflush(stderr);
        lua_pushnil(L);
        lua_pushstring(L, "No internal MQTT session found.");
        return 2;
    }

    ntop = lua_gettop(L);
    if (ntop < 2 || lua_isstring(L, 1) == 0 ||
        lua_isstring(L, 2) == 0) {
err0:
        lua_pushnil(L);
        lua_pushstring(L, "invalid argument(s) for publish");
        return 2;
    }

    /* get IPC/MQTT topic */
    topout = lua_tolstring(L, 1, NULL);
    if (topout == NULL || topout[0] == '\0')
        goto err0;
    /* get the payload */
    payload = lua_tolstring(L, 2, &plen);
    if (payload == NULL || payload[0] == '\0' || plen == 0)
        goto err0;

    ret = ipc_session_publish(isess, (char *) topout,
        (unsigned char *) payload, (int) plen);
    if (ret) {
        fprintf(stderr, "Error, failed publish message to internal broker: %d\n", ret);
        fflush(stderr);
    }
    lua_pushboolean(L, ret == 0);
    return 1;
}

static int cyhk_thread_msgid(lua_State * L)
{
    unsigned int mid;
    DECLARE_CYHK_PROTO(cyp);
    if (lua_checkstack(L, 2) == 0)
        return 0;
    mid = cyhk_messageid(cyp);
    lua_pushinteger(L, (lua_Integer) mid);
    return 1;
}

static const struct luaL_Reg cyhk_proto_funcs[] = {
    { "setpkey",             cyhk_thread_productkey    },
    { "setaddr",             cyhk_thread_address       },
    { "start",               cyhk_thread_start         },
    { "alive",               cyhk_thread_alive         },
    { "stop",                cyhk_thread_stop          },
    { "publish",             cyhk_thread_publish       },
    { "pubint",              cyhk_thread_pubint        },
    { "msgid",               cyhk_thread_msgid         },
    { NULL,                  NULL                      },
};

int luaopen_cyhkthread(lua_State * L)
{
    int ret;
    pthread_condattr_t conda;
    cyhk_proto_t * proto = NULL;

    proto = (cyhk_proto_t *) calloc(0x1, sizeof(*proto));
    if (proto == NULL) {
        fputs("Error, system out of memory!\n", stderr);
        fflush(stderr);
        return 0; /* Lua module failed to load */
    }

    /* initialize mutex lock */
    ret = pthread_mutex_init(&proto->mlock, NULL);
    if (ret != 0) {
        fprintf(stderr, "Error, failed to initialize mutex: %d\n", ret);
        fflush(stderr);
        goto err0;
    }

    memset(&conda, 0, sizeof(conda));
    ret = pthread_condattr_init(&conda);
    if (ret == 0)
        ret = pthread_condattr_setclock(&conda, CLOCK_MONOTONIC);
    if (ret == 0)
        ret = pthread_cond_init(&proto->condition, &conda);

    if (ret) {
        fprintf(stderr, "Error, failed to initialize condition: %d\n", ret);
        fflush(stderr);
        goto err0;
    }

    g_cyhk_proto = proto;
    luaL_register(L, "cyhkthread", cyhk_proto_funcs);
    return 1;

err0:
    pthread_mutex_destroy(&proto->mlock);
    free(proto);
    return 0;
}