/**
 * @file main.cc
 *
 * @brief this is the main file impelents evcamera
 *
 * @ingroup evcamera
 *
 * @author Bruce Lu
 * Contact: lzbgt@icloud.com
 *
 */

#include <iostream>
#include <string>
#include <chrono>
#include <sys/stat.h>
#include <fmt/format.h>
#include <spdlog/spdlog.h>
#include "ref-memory.hpp"
#include <maque_system.h>
#include <maque_video_enc.h>
#include <maque_time.h>
#include <evpacket.h>
#include "common.h"
#include "ntp.h"
#include <time.h>
#include "raw_tcp.h"
#include "smart.h"
#include "motion.h"
#include <thread>
#include "evutils.hpp"
#include <ghc/filesystem.hpp>
#include <mqtt_helper.hpp>
#include <signal.h>
#include <jsoncons_ext/jsonpatch/jsonpatch.hpp>
#include "internal_types.h"
#include "ptz.h"
#include <queue>
#include "record_upload.h"

using namespace std;
using namespace jsoncons;
using namespace chrono_literals;
namespace fs = ghc::filesystem;
using namespace evutils;

///
#define NUM_IFRAME_PICK 2
#define NUM_MAX_QUEQUE_SIZE 60
#define NUM_MAX_PACKET_BYTES 1200
#define INTERVAL_LOG_QFULL 3    // its about 60/15 *3 secs

/// device info
char dev_sn[TERMINAL_SN_SIZE] = {0};

/// default video gateway info
int videoQuality = MAQUE_IMG_QUALITY_BETTER;
int gFrameFPS = 15;
atomic<uint64_t> frameCntTotal(0);
atomic<uint64_t> frameCntIframe(0);
atomic<uint64_t> frameCntPframe(0);
atomic<uint64_t> packetId(0);
spdlog::level::level_enum logLevel = spdlog::level::info;
bool bPFrameAvail = false;
bool bNeedRetry = false;
int readFaildCnt = 0;
bool enablePush = false;
bool enableRecord = false;
bool bCanRecord = false;
bool bConnected = false;
queue<DataItem> _frameQueue;
mutex _mutFrame;
condition_variable _condFrame;
Notifier _notiFrame = {&_mutFrame, &_condFrame};
CallBackArg args = {nullptr, &_notiFrame, &_frameQueue};
bool bGotTime = false;

// mqtt
MqttHelper *gMqttClient = nullptr;

// zmq
void *pPubCtx = nullptr;
void *pPub = nullptr;
void *pSub = nullptr;
void *pDealerCtx = nullptr;
void *pDealer = nullptr;
string vgwUrl = "tcp://192.168.55.104:7123";

//
string configFilePath = "/etc/evcamera.json";
json gJsonConfig;

char hostArr[] = "192.168.55.104";
char portArr[] = "7123";
char *host = hostArr;
char *port = portArr;

//
OrderedList<int64_t> *gRecFilesList = nullptr;

namespace _upvariable_ {
mutex mut;
condition_variable cv;
queue<upload_item_t> que;
};

thread_upload_args_t gUploadArgs{&_upvariable_::mut, &_upvariable_::cv, &_upvariable_::que, &gJsonConfig, dev_sn, &gRecFilesList};

//
string strMaQuePath = "/mnt/sd/Config/";

/// why using get function because global vriable initialization order between files is undefined!
/// this avoids getting empty result
vector<string> get_sdcard_resered_dirs()
{
    return {consts::recFilePath, strMaQuePath};
}

ev_module_config_t gConfigSystem;

enum EV_MSG_ERROR_CODE {
    EV_MSG_ERROR_NONE,
    EV_MSG_ERROR_INVALID_PARAM,
    EV_MSG_ERROR_EXCEPTION,
    EV_MSG_ERROR_UNSUPPORTED_CMD,
    EV_MSG_ERROR_CONTENT_SYNTAX,
    EV_MSG_ERROR_NUM
};

void clean_up(int sig)
{
    if(sig == SIGPIPE) {
        // vgw connection error
        bConnected = false;
        bNeedRetry = true;
        return;
    }

    spdlog::warn("clearn up called, with sig: {}", sig);
    if (args.recFD) {
        fclose(args.recFD);
    }

    //stop_md_bd();
    LibXmMaQue_System_destroy();
    exit(1);
}

bool is_big_endian(void)
{
    union {
        uint32_t i;
        char c[4];
    } bint = {0x01020304};

    return bint.c[0] == 1;
}

string save_configuration(json &data)
{
    auto rc = save_json(data, configFilePath, false, nullptr);
    if(!rc.empty()) {
        spdlog::error(rc);
    }

    return rc;
}

string load_configuration(json &config)
{
    auto rc = load_json(config, configFilePath, evutils::make_default_cloud_config);
    return rc;
}

MaQueVideoEncodeCfg_s *init_stream_cfg(MaQueVideoEncodeCfg_s *cfg)
{
    cfg->eVidComp = (MaQueVideoCompress_e)-1;
    cfg->eCapSize = (MaQueCaptureSize_e)-1;
    cfg->eBitrateCtrl = (MaQueBitrateCtrl_e)-1;
    cfg->eImageQuality = (MaQueImageQuality_e)-1;
    cfg->nFps = -1;
    cfg->nBitRate = -1;
    cfg->nIFrameInterval = -1;
    cfg->iQpMin = -1;
    cfg->iQpMax = -1;

    return cfg;
}

int32_t configure_stream(MaQueStreamChannel_e eStreamChn, MaQueVideoEncodeCfg_s *cfg)
{
    int32_t res;
    MaQueVideoEncodeCfg_s oldCfg;

    if (LibXmMaQue_VideoEnc_getCfg(0, eStreamChn, &oldCfg)) {
        spdlog::error("LibXmMaQue_VideoEncode_getCfg error");
        return -1;
    }

    if (cfg->eVidComp != -1) {
        oldCfg.eVidComp = cfg->eVidComp;
    }
    if (cfg->eCapSize != -1) {
        oldCfg.eCapSize = cfg->eCapSize;
    }
    if (cfg->eBitrateCtrl != -1) {
        oldCfg.eBitrateCtrl = cfg->eBitrateCtrl;
    }

    if (cfg->eImageQuality != -1) {
        oldCfg.eImageQuality = cfg->eImageQuality;
    }

    if (cfg->nFps != -1) {
        oldCfg.nFps = cfg->nFps;
    }
    if (cfg->nBitRate != -1) {
        oldCfg.nBitRate = cfg->nBitRate;
    }

    if (cfg->nIFrameInterval != -1) {
        oldCfg.nIFrameInterval = cfg->nIFrameInterval;
    }

    if (cfg->iQpMin > 0 && cfg->iQpMin <= 51 && cfg->iQpMax > 0 && cfg->iQpMax <= 51) {
        oldCfg.iQpMin = cfg->iQpMin;
        oldCfg.iQpMax = cfg->iQpMax;
        oldCfg.bUseUserQp = 1;
    }

    spdlog::info("main: fps={}", oldCfg.nFps);
    res = LibXmMaQue_VideoEnc_setCfg(0, eStreamChn, &oldCfg);
    if (!res) {
        memcpy(cfg, &oldCfg, sizeof(oldCfg));
    }

    return res;
}

//
void frame_free(void *data, void*hint)
{
    MaQueVideoEncFrameInfo_s *frame = (MaQueVideoEncFrameInfo_s *)hint;
    MaQue_Demo_Mem_release(frame->handleMem);
}

//
void my_free(void *data, void* hint)
{
    free(data);
}

//
XM_S32 cb_frame_proc(XM_VOID *pUserArg, MaQueVideoEncFrameInfo_s *frame)
{
    MaQueSystemTime_s *pstSysTime = &frame->stTimeStamp.stSysTime;
    CallBackArg *args = (CallBackArg *)pUserArg;
    struct timeval stTimeVal;
    int ret;
    frameCntTotal++;
    zmq_msg_t msg;
    void *hint = NULL;
    if(frame->eEncodeType == MAQUE_ENCODE_TYPE_H264 && frame->eFrameType == MAQUE_FRAME_TYPE_VIDEO && frame->nDataLen > 0) {
#ifdef RAW_FORMAT
        evpacket_t* pkt = (evpacket_t*)malloc(sizeof(evpacket_t) + frame->nDataLen);
#else
        evpacket_t* pkt = (evpacket_t*)((char*)frame->pData - sizeof(evpacket_t));
#endif
        pkt->meta.magic[0] = 0xBE;
        pkt->meta.magic[1] = 0xEF;
        memcpy(pkt->meta.sn, dev_sn, sizeof(dev_sn));
        /// NOTES: since both ends use little-endian, we don't use htonl, htons to save clocks
        pkt->vpara.frame_type = frame->eSubType + 1;
        pkt->vpara.packet_type = frame->eEncodeType;
        pkt->vpara.res.width = frame->nWidth;
        pkt->vpara.res.height = frame->nHeight;
        /// TODO:
        pkt->vpara.fps = 15;
        //pkt->vpara.ts = frame->stTimeStamp;
        pkt->length = frame->nDataLen;
        pkt->meta.packet_id = packetId.load(memory_order_relaxed);
        pkt->meta.crc = 0;
        pkt->meta.crc = crc16((unsigned char*)pkt, sizeof(evpacket_t));
        if(frame->eSubType == MAQUE_FRAME_SUBTYPE_I) {
            frameCntIframe++;
        }
        else if(frame->eSubType == MAQUE_FRAME_SUBTYPE_P) {
            frameCntPframe++;
        }
#ifdef RAW_FORMAT
        memcpy((char*)pkt + sizeof(evpacket_t), frame->pData, frame->nDataLen);
#endif
        zmq_msg_init_data(&msg, (char*)pkt, frame->nDataLen + sizeof(evpacket_t), my_free, nullptr);
        zmq_msg_send(&msg, pPub, 0);
        packetId++;
    }

    MaQue_Demo_Mem_release(frame->handleMem);
    return XM_SUCCESS;
}

void frame_send_entry(void *args)
{
    int ret = 0;
    int flagFail = 3; // 1 - sub, 2 - dealer , 3 - both
    // should never exit
    while(1) {
        /// TODO: check enable status
        this_thread::sleep_for(4s);
        if(false) {
            continue;
        }
        // setup dealer
        if(flagFail & 2) {
            ret = setupDealer(&pDealerCtx, &pDealer, string(dev_sn), vgwUrl, 10);
            if(ret < 0) {
                spdlog::error("failed to connect cloud vgw: {}", vgwUrl);
                continue;
            }
            spdlog::info("success connect to vgw: {}", vgwUrl);

        }
        // setup sub
        if(flagFail &1) {
            pSub = zmq_socket(pPubCtx, ZMQ_SUB);
            if(pSub == nullptr) {
                spdlog::error("failed to create frame sending socket");
                continue;
            }

            ret = zmq_connect(pSub, consts::strPubUrl.c_str());
            if(ret != 0) {
                spdlog::error("failed connect frame pub: {}", consts::strPubUrl);
                continue;
            }
            ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
            spdlog::info("zmq sub successed (for frame sending)");
        }

        zmq_msg_t msg;
        while (1) {
            // TODO: reconnect
            zmq_msg_init(&msg);
            ret = zmq_msg_recv(&msg, pSub, 0);
            if(ret < 0) {
                spdlog::error("failed to recv zmq msg: {}", zmq_strerror(zmq_errno()));
                flagFail = 1;
                break;
            }
            // ssize_t sz = zmq_msg_size(&msg);
            // char *ptr = (char*)zmq_msg_data(&msg);
            ret = zmq_msg_send(&msg, pDealer, 0);
            zmq_msg_close(&msg);
            if(ret < 0) {
                spdlog::error("failed to send vgw: {}, {}", zmq_strerror(zmq_errno()));
                flagFail = 2;
                break;
            }
        }
        if(flagFail & 1) {
            zmq_close(pSub);
            pSub = nullptr;
        }
        if(flagFail &2) {
            zmq_close(pDealer);
            zmq_ctx_destroy(pDealerCtx);
            pDealer = nullptr;
            pDealerCtx = nullptr;
        }
    }
}

/// verify request message
string verify_request(json &js)
{
    string ret;
    string check[] = {consts::kMsgCmd, consts::kMsgData, consts::kMsgRid, consts::kMsgTime};
    for(auto &k: check) {
        if(!js.contains(k)) {
            ret += k + ",";
        }
    }
    if(!ret.empty()) {
        ret = "missing fields in request: " + ret;
    }
    return ret;
}

string verify_config(json &data)
{
    string str, strWarn;
    string veri[] = {consts::kMsgConfigVgw, consts::kMsgConfigMqtt, consts::kMsgConfigUpload};
    for(auto &k:veri) {
        auto value = msg_field(data, k);
        if(value.empty()) {
            str = fmt::format("no {} field in data: {}", k, data.to_string());
            break;
        }
        else {
            // parse host and port
            auto uri = httplib::Uri::Parse(value);
            if(uri.Host.empty()|| uri.Port.empty()) {
                str = fmt::format("host addr invalid {}:{}, {}", k, value, data.to_string());
                break;
            }
            // check network connections
            int socket = 0;
            auto rc = raw_connect_nonblock(uri.Host, uri.Port, &socket);
            if(rc < 0 || socket == 0) {
                str = fmt::format("failed to connect to {}:{}, {}", k, value, data.to_string());
                break;
            }
            else {
                // we must close the test socket
                ::close(socket);
            }
        }
    }

    if(!str.empty()) {
        spdlog::error(str);
    }

    return str;
}

/// TODO: we will figure out how to apply change without reboot.
/// this is a temprary implementaion for quich prototyping
string apply_config(json &data)
{
    string rc;
    try {
        /// make defaults
        gConfigSystem.module.sys.fps = 15; // secs
        gConfigSystem.module.sys.video_quality = MAQUE_IMG_QUALITY_BETTER;
        gConfigSystem.module.record.interval = 60 * 2; // 2 minutes interval
        gConfigSystem.module.sys.bitrate_kb = 1024; // 1Mbps
        gConfigSystem.module.sys.bitrate_type = MAQUE_BITRATE_CTRL_CBR;
        gConfigSystem.module.sys.push = 0;
        // motion
        gConfigSystem.module.motion.enabled = 1;
        gConfigSystem.module.motion.level = 3;
        gConfigSystem.module.motion.region = {0,0,1,1};
        // ai
        gConfigSystem.module.ai.enabled = 1;
        gConfigSystem.module.ai.region = {0,0,1,1};
        gConfigSystem.module.ai.face_thresh = 0.7;
        gConfigSystem.module.ai.human_thresh = 0.7;
        gConfigSystem.module.ai.image_report_interval = 10; // no two reports within 10 seconds

        // vgw
        if(gJsonConfig.contains(consts::kMsgConfigVgw)) {
            // gVgwAddr = gJsonConfig[consts::kMsgConfigVgw].as<string>();
        }
        else {
            gJsonConfig[consts::kMsgConfigVgw] = consts::vgw_addr;
        }

        // mqtt
        if(gJsonConfig.contains(consts::kMsgConfigMqtt)) {
            // gMqttAddr = gJsonConfig[consts::kMsgConfigMqtt].as<string>();
        }
        else {
            gJsonConfig[consts::kMsgConfigMqtt] = consts::mqtt_url;
        }

        // uploader
        if(gJsonConfig.contains(consts::kMsgConfigUpload)) {
            //gUploadAddr = gJsonConfig[consts::kMsgConfigUpload].as<string>();
        }
        else {
            gJsonConfig[consts::kMsgConfigUpload] = consts::upload_addr;
        }

        if(gJsonConfig.contains(consts::kMsgConfigFeatures)) {
            json &cfgFeatures = gJsonConfig[consts::kMsgConfigFeatures];
            if(cfgFeatures.contains("logLevel")) {
                logLevel = spdlog::level::from_str(cfgFeatures["logLevel"].as<string>());
                spdlog::set_level(logLevel);
            }

            // fps
            if(cfgFeatures.contains("fps")) {
                gConfigSystem.module.sys.fps = cfgFeatures["fps"].as<int>();
            }
            else {
                spdlog::warn("no fps configuration, using default: {}", gConfigSystem.module.sys.fps);
            }

            // record
            if(cfgFeatures.contains("recordLen")) {
                gConfigSystem.module.record.interval = cfgFeatures["recordLen"].as<int>();
            }
            else {
                spdlog::warn("no recordLen configuration, using default: {}", gConfigSystem.module.record.interval);
            }

            // quality
            if(cfgFeatures.contains("videoQuality")) {
                gConfigSystem.module.sys.video_quality = cfgFeatures["videoQuality"].as<int>();
            }
            else {
                spdlog::warn("no video quality configuration, using default: {}", gConfigSystem.module.sys.video_quality);
            }

            // bitrate
            if(cfgFeatures.contains("bitrate")) {
                gConfigSystem.module.sys.bitrate_kb = cfgFeatures["bitrate"].as<int>();
            }
            else {
                spdlog::warn("no video bitrate configuration, using default: {}", gConfigSystem.module.sys.bitrate_kb);
            }
            if(cfgFeatures.contains("bitrateType")) {
                gConfigSystem.module.sys.bitrate_type = cfgFeatures["bitrateType"].as<int>();
            }
            else {
                spdlog::warn("no video bitrate type configuration, using default: {}", gConfigSystem.module.sys.bitrate_type);
            }
            if(cfgFeatures.contains("push")) {
                gConfigSystem.module.sys.push = cfgFeatures["push"].as<int>();
            }
            else {
                spdlog::warn("no push configuration, using default: {}", gConfigSystem.module.sys.push);
            }

            // motion
            if(cfgFeatures.contains("motion")) {
                json &motion = cfgFeatures["motion"];
                if(motion.contains("enabled")) {
                    gConfigSystem.module.motion.enabled = motion["enabled"].as<int>();
                }
                if(motion.contains("level")) {
                    gConfigSystem.module.motion.level = motion["level"].as<int>();
                }
                if(motion.contains("region") && motion["region"].contains("minX") && motion["region"].contains("minY") && motion["region"].contains("maxX") && motion["region"].contains("maxY")) {
                    gConfigSystem.module.motion.region = {motion["region"]["minX"].as<float>(), motion["region"]["minY"].as<float>(), motion["region"]["maxX"].as<float>(), motion["region"]["maxY"].as<float>()};
                }
            }
            spdlog::info("motion configuration: enabled {}, level {}, region ({},{}), ({}, {})", gConfigSystem.module.motion.enabled, gConfigSystem.module.motion.level, gConfigSystem.module.motion.region.minx, gConfigSystem.module.motion.region.miny, gConfigSystem.module.motion.region.maxx, gConfigSystem.module.motion.region.maxy);

            // ai
            if(cfgFeatures.contains("ai")) {
                json &ai = cfgFeatures["ai"];
                if(ai.contains("enabled")) {
                    gConfigSystem.module.ai.enabled = ai["enabled"].as<int>();
                }
                if(ai.contains("faceThresh")) {
                    gConfigSystem.module.ai.face_thresh = ai["faceThresh"].as<float>();
                }
                if(ai.contains("humanThresh")) {
                    gConfigSystem.module.ai.human_thresh = ai["humanThresh"].as<float>();
                }
                if(ai.contains("region") && ai["region"].contains("minX") && ai["region"].contains("minY") && ai["region"].contains("maxX") && ai["region"].contains("maxY")) {
                    gConfigSystem.module.ai.region = {ai["region"]["minX"].as<float>(), ai["region"]["minY"].as<float>(), ai["region"]["maxX"].as<float>(), ai["region"]["maxY"].as<float>()};
                }
            }
            spdlog::info("ai configuration: enabled {}, faceThresh {}, humanThresh {}, region ({},{}), ({}, {})", gConfigSystem.module.ai.enabled, gConfigSystem.module.ai.face_thresh, gConfigSystem.module.ai.human_thresh, gConfigSystem.module.ai.region.minx, gConfigSystem.module.ai.region.miny, gConfigSystem.module.ai.region.maxx, gConfigSystem.module.ai.region.maxy);
        }
    }
    catch(exception &e) {
        /// TODO: save for reporting
        rc = fmt::format("failed to apply config: {}, {}", e.what(), data.to_string());
        spdlog::error(rc);
    }
    return rc;
}

/// handles all incoming request from cloud services
void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string topic)
{
    char msg[len+1] = {0};
    memcpy(msg, data, len);
    spdlog::info("received request on {}:\n{}", topic, (char*)msg);
    auto now = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();

    try {
        auto js = json::parse(msg);
        auto str = verify_request(js);
        if(!str.empty()) {
            spdlog::error(str);
            MqttMgr::report_response_args(gMqttClient, consts::pub_topic_report, EV_MSG_ERROR_CONTENT_SYNTAX, str, js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), json());
            return;
        }

        /// dispatch msg
        auto cmd = js[consts::kMsgCmd].as<string>();
        auto rid = js[consts::kMsgRid].as<string>();
        auto data = js[consts::kMsgData].as<json>();
        /// TODO: implements cmds
        if(cmd == consts::kMsgCmdConfig) {
            // verify configuration
            auto rv = verify_config(data);
            if(!rv.empty()) {
                spdlog::error("failed to apply config: {}, {}", rv, data.to_string());
                MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid,EV_MSG_ERROR_INVALID_PARAM, rv, cmd, rid, data);
            }
            else {
                // calc diff
                auto patch = jsonpatch::from_diff(gJsonConfig, data);
                if(patch.empty()) {
                    MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "no diff to apply", cmd, rid, data);
                }
                else {
                    // apply diff to tmp and verify
                    spdlog::info("json diff: {}", patch.to_string());
                    auto rc = verify_config(data);
                    if(!rc.empty()) {
                    }
                    else {
                        rc = apply_config(data);
                    }
                    if(!rc.empty()) {
                        spdlog::error(rc);
                        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, rc, cmd, rid, data);
                    }
                    else {
                        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "rebooting to apply new config", cmd, rid, data);
                        /// TODO: using json diff to apply change without reboot
                        save_configuration(data);
                        spdlog::info("rebooting to apply new configuration");
                        // system("reboot");
                    }
                }
            }
        }
        else if(cmd == consts::kMsgCmdUploadVideo) {
            // upload video
            if(gRecFilesList == nullptr) {
                MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, -1, "no sd card", cmd, rid, data);
            }
            else {
                auto tss = data["start"].as<int64_t>();
                auto tse = data["end"].as<int64_t>();

                bool isTsValid = false;
                if(tss <= tse && tss > consts::TS_2020 && tse < consts::TS_MAX) {
                    // in ms
                    isTsValid = true;
                }
                else {
                    // in seconds
                    tss = tss*1000;
                    tse = tse*1000;
                    if(tss <= tse && tss > consts::TS_2020 && tse < consts::TS_MAX ) {
                        // in seconds
                        isTsValid = true;
                    }
                }
                if(!isTsValid) {
                    MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, "time stamps error. start <= end && start > 1577836800(000) and both in millisecon or second format ", cmd, rid, data);
                }
                else {
                    int64_t offsetE = 0, offsetS = 0;
                    int status = -2;
                    auto res = gRecFilesList->findByRange(tss, tse, offsetS, offsetE, gConfigSystem.module.record.interval, &status);
                    if(res.empty() && status < 0) {
                        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, "no video found: timestamp too far in the past", cmd, rid, data);
                    }
                    else {
                        //print_ts_files(*gRecFilesList);
                        printf("matched (%lld, %lld) with offsets %lld, %lld:\n\t", tss, tse, offsetS, offsetE);
                        for(auto &k: res) {
                            printf("%lld, ", k);
                        }
                        printf("\n");
                        /// TODO: upload videos
                        upload_item_t item{tss, tse, 6, -1, 0};
                        {
                            lock_guard<mutex> lg(*gUploadArgs.mut);
                            gUploadArgs.que->push(item);
                            gUploadArgs.cv->notify_one();
                        }
                        string extra;
                        if(status > 0) {
                            extra = ", but it is scheduled in the future";
                        }
                        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, string("request accepted") + extra, cmd, rid, data);
                    }
                }
            }
        }
        else if(cmd == consts::kMsgCmdGetConfig) {
            // response config
            MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", cmd, rid, gJsonConfig);
        }
        else if(cmd == consts::kMsgCmdStatus) {
            // response status
            json stat;
            stat["sd"] = json();
            if(is_sdcard_avail()) {
                uint64_t totalMB = 0, freeMB = 0;
                if(get_sdcard_megabytes(totalMB, freeMB)) {
                    stat["sd"]["MBfree"] = freeMB;
                    stat["sd"]["MBTotal"] = totalMB;
                }
            }
            // get first video ts
            stat["record"] = json();
            auto rec = gRecFilesList->items();
            stat["record"]["capacity"] = gRecFilesList->capacity();
            if(rec.size() > 0) {
                stat["record"]["num"] = rec.size();
                stat["record"]["first"] = *(rec.begin());
                stat["record"]["last"] = *((rec.rbegin()));
            }

            /// TODO: subsystems running status
            MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", cmd, rid, stat);
        }
        else if(cmd == consts::kMsgCmdPtz) {
            // response Ptz control
            spdlog::info("mqtt got ptz cmd: {}", data.to_string());
            if(!data.contains("action") || !data.contains("degree")) {
                MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, "invalid param in data field", cmd, rid, data);
            }
            else {
                /// TODO: issue need to clear
                if(true) {
                    spdlog::info("mqtt begin move");
                    string rc = ptz_move(data["action"].as<string>(), data["degree"].as<float>());
                    spdlog::info("mqtt end move");
                    ptz_param_position pos;
                    ptz_get_params(pos);
                    json _js;
                    _js["horizontal"] = (int)(pos.current_x_deg);
                    _js["vertical"] = 0;
                    if(!str.empty()) {
                        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_INVALID_PARAM, rc, cmd, rid, data);
                    }
                    else {
                        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", cmd, rid, _js);
                    }
                }
                else {
                    MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, EV_MSG_ERROR_UNSUPPORTED_CMD, "there is hanging issue need to be fixed, before this command is available", cmd, rid, data);
                }
            }
        }
        else if(cmd == "list_videos") {
            if(gRecFilesList != nullptr)
                print_ts_files(*gRecFilesList);
            else {
                spdlog::warn("no local video files");
            }
        }
        else {
            auto str = fmt::format("unsupported cmd {}: {}", cmd, js.to_string());
            spdlog::error(str);
            MqttMgr::report_response_args(gMqttClient, consts::pub_topic_report, EV_MSG_ERROR_UNSUPPORTED_CMD, str, cmd, rid, json());
        }

    }
    catch(exception &e) {
        auto str = string("exception: ") + e.what() + string(", ") + msg;
        spdlog::error(str);
        MqttMgr::report_response_args(gMqttClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, str, string(""), string(""), json());
    }
    //(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}

///
MqttHelper* start_mqtt(void *arg)
{
    auto client = MqttMgr::get_instance(consts::mqtt_url, dev_sn);
    if(client == nullptr) {
        spdlog::error("failed to create mqtt instance to {}", consts::mqtt_url);
        return nullptr;
    }
    client->subscribe(consts::sub_topic + dev_sn, handle_mqtt_req, false);
    /// TODO:
    gConfigSystem.pClient = client;

    return client;
}

void create_sd_directories()
{
    // create record paths
    for(auto k:get_sdcard_resered_dirs()) {
        spdlog::info("create path: {}", k);
        system((string("mkdir -p ") + k).c_str());
    }
}

///
int main(int argc, char *argv[])
{
    int ret = XM_SUCCESS;

    get_mac_addr(dev_sn);
    spdlog::info("hi3518ev300, sn: {}, version: {}, BE: {}", dev_sn, evutils::version, is_big_endian());

    // get env for host and port
    auto host_ = getenv("VGW_HOST");
    enablePush = true;
    if(host_) {
        host = host_;
    }
    else {
        enablePush = false;
    }

    auto port_ = getenv("VGW_PORT");
    if(port_) {
        port_ = port;
    }
    else {
        enablePush = false;
    }

    /// TODO: check sd availability
    bool bSDAvail = is_sdcard_avail();
    if(bSDAvail) {
        create_sd_directories();
    }

    /// TODO: load configuration
    load_configuration(gJsonConfig);
    verify_config(gJsonConfig);
    auto rc = apply_config(gJsonConfig);
    if(!rc.empty()) {
        spdlog::error("failed to apply configuration: {}", gJsonConfig.to_string());
        spdlog::error("defaults is used instead");
    }

    auto logLevel_ = getenv("LOG_LEVEL");
    if(logLevel_) {
        logLevel = spdlog::level::from_str(logLevel_);
        spdlog::set_level(logLevel);
    }
    else {
        spdlog::info("log level configured to info");
        spdlog::set_level(logLevel);
    }

    // blocking update time, since time is critical to every other component
    bool bGotTime = false;
    time_t stm;
    while(!bGotTime) {
        if(getNtpTime(&stm) >=0) {
            spdlog::info("ntp got time");
            ::stime(&stm);
            bGotTime = true;
        }
        else {
            spdlog::warn("failed to get ntp time");
        }
        this_thread::sleep_for(chrono::seconds(3));
    }

    MaQue_Demo_Mem_Init();

    MaQueStartParam_s startParam = {MAQUE_VIDEO_STANDARD_PAL, {MAQUE_VIDEO_COMPRESS_H264, MAQUE_VIDEO_COMPRESS_H265}};
    ::memcpy(startParam.aWritableDir,  strMaQuePath.c_str(), sizeof(startParam.aWritableDir));
    ret = LibXmMaQue_System_startUp(&startParam);
    spdlog::info("ret: {}", ret);

    //
    MaQueMemoryApi_s memApi = {MaQue_Demo_Mem_alloc, MaQue_Demo_Mem_release, MaQue_Demo_Mem_addRef, MaQue_Demo_Mem_setLength};
    ret = LibXmMaQue_Mem_init(&memApi);
    spdlog::info("mem ret: {}", ret);

    MaQueSystemTime_s tm;
    LibXmMaQue_Time_getCurrentTime(&tm);
    LibXmMaQue_Time_print(&tm, nullptr);
    //
    MaQueCodeAbilities_s capb;
    ret = LibXmMaQue_VideoEnc_getAbilities(0, &capb);
    spdlog::info("abt ret: {}", ret);
    if (XM_SUCCESS == ret) {
        spdlog::info(fmt::format("capbilities: supported codecs {0:#b},"
                                 "max fhd/s {0:d}, max res: {0:d}, per ch mres: {0:d}, {0:d}, {0:d}, {0:d}",
                                 int(capb.videoEncTypeMask),
                                 int(capb.maxEncPowerX1080P), int(capb.eDecImageSizeMax), int(capb.astVidEncChnAbility[0].eCapSizeMax),
                                 int(capb.astVidEncChnAbility[1].eCapSizeMax), int(capb.astVidEncChnAbility[2].eCapSizeMax), int(capb.astVidEncChnAbility[3].eCapSizeMax)));
    }

    //
    gConfigSystem.isRecordRunning = false;
    MaQueVideoEncodeCfg_s cfg;
    init_stream_cfg(&cfg);
    gConfigSystem.pPubCtx = &pPubCtx;
    gConfigSystem.pClient = nullptr;
    cfg.nFps = gConfigSystem.module.sys.fps;
    cfg.eIFrmIntvType = IFRAME_INTV_TYPE_TIME;
    cfg.nIFrameInterval = 10;
    cfg.eImageQuality = (MaQueImageQuality_e)gConfigSystem.module.sys.video_quality;
    cfg.eVidComp = MAQUE_VIDEO_COMPRESS_H264;
    cfg.eBitrateCtrl = (MaQueBitrateCtrl_e)gConfigSystem.module.sys.bitrate_type;
    cfg.nBitRate = gConfigSystem.module.sys.bitrate_kb;
    ret = configure_stream((MaQueStreamChannel_e)0, &cfg);
    spdlog::info("cfg stream ret: {}", ret);

    ret = LibXmMaQue_VideoEnc_startStream(0, (MaQueStreamChannel_e)0, cb_frame_proc, (void *)&args);
    if (XM_SUCCESS == ret) {
        spdlog::info("created record task successfully");
    }

    /// start ptz service
    ptz_service_start();

    signal(SIGINT, clean_up);
    // signal(SIGTERM, clean_up);
    signal(SIGKILL, clean_up);
    signal(SIGPIPE, SIG_IGN);

    // frame dispatching
    pPubCtx = zmq_ctx_new();
    pPub = zmq_socket(pPubCtx, ZMQ_PUB);
    ret = zmq_bind(pPub, consts::strPubUrl.c_str());
    if(ret < 0) {
        spdlog::error("{} failed to create zmq pub topic: {}", consts::strPubUrl);
        exit(1);
    }
    /// TODO: configuration issues handling
    if(msg_field(gJsonConfig, consts::kMsgConfigVgw).empty()) {
        spdlog::error("missing vgw config");
    }

    thread thPush = thread(frame_send_entry, &args);
    spdlog::info("sizeof pkt header {}, sizeof tv {}", sizeof(evpacket_t), sizeof(timeval));
    if(thPush.joinable()) {
        thPush.detach();
    }

    thread thVideoRecord = thread(record_video_entry, &gConfigSystem);
    if(thVideoRecord.joinable()) {
        thVideoRecord.detach();
    }

    while(gConfigSystem.isRecordRunning == false) {
        spdlog::info("waiting for record svc to start");
        this_thread::sleep_for(chrono::seconds(3));
    }

    spdlog::info("recording svc started, starting upload svc");

    // start upload thread
    thread thUpload = thread(upload_svc_entry, &gUploadArgs);
    if(thUpload.joinable()) {
        thUpload.detach();
        spdlog::info("started video upload service");
    }

    start_md_bd(&gConfigSystem);
    /// subscribe to mqtt
    gMqttClient = start_mqtt(nullptr);

    thread thSmart = thread(maq_smart_task_entry, &gConfigSystem);
    thSmart.join();
}