/**
 * @file main.cc
 *
 * @brief this is the main file implements 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 "spdlog/sinks/rotating_file_sink.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"
#include <maque_watchdog.h>
#include "qr.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);
atomic<int32_t> gQRCodeDuration(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 configFilePath = "/etc/evcamera.json";
json gJsonConfig;

//
static int gPusherFailsFlag = 0;
json gRecordStat;

//
const string sysCfgJsonFilePath = "/etc/sys.json";
const string sysWpaCfgFilePath = "/etc/wpa.conf";

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/";

// rotating logger
// create a thread safe sink which will keep its file size to a maximum of 5MB and a maximum of 3 rotated files.
shared_ptr<spdlog::logger> rlogger = spdlog::rotating_logger_mt("evc", "/mnt/sd/logs.txt", 1048576 * 1, 3);

/// 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;

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

    if(sig == SIGINT) {
        LibXmMaQue_Watchdog_enable(0);
    }

    rlogger->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()) {
        rlogger->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)) {
        rlogger->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;
    }
    rlogger->info("main: fps={}", oldCfg.nFps);
    res = LibXmMaQue_VideoEnc_setCfg(0, eStreamChn, &oldCfg);
    if (!res) {
        memcpy(cfg, &oldCfg, sizeof(oldCfg));
    }

    return res;
}

//
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;

    if(frame->eEncodeType == MAQUE_ENCODE_TYPE_H264 && frame->eFrameType == MAQUE_FRAME_TYPE_VIDEO && frame->nDataLen > 0) {
        zmq_msg_init_size(&msg, sizeof(evpacket_t) + frame->nDataLen );
        evpacket_t* pkt = (evpacket_t*)zmq_msg_data(&msg);
        memcpy((char*)pkt + sizeof(evpacket_t), frame->pData, frame->nDataLen);
        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++;
        }

        zmq_msg_send(&msg, pPub, 0);
        zmq_msg_close(&msg);
        packetId++;
    }

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

void frame_send_entry(void *args)
{
    int ret = 0;
    gPusherFailsFlag = 0; // 1 - sub, 2 - dealer , 3 - both, 4 - disabled
    evpacket_t *pkt = nullptr;
    zmq_msg_t msg;
    string res;
    chrono::system_clock::time_point ts_start, ts_now;
    // should never exit
    while(1) {
        rlogger->info("perpare for pusher");
        if(!gConfigSystem.module.sys.push) {
            gPusherFailsFlag |= 1 >> 2;
            this_thread::sleep_for(4s);
            continue;
        }
        gPusherFailsFlag &= ~4;

        // setup dealer
        string vgwUrl = gJsonConfig[consts::kMsgConfigVgw].as<string>();
        if(vgwUrl.substr(0, 6) != "tcp://") {
            vgwUrl = "tcp://" + vgwUrl;
        }
        auto uri = httplib::Uri::Parse(vgwUrl);
        int s = 0;
        auto r = raw_connect(uri.Host, uri.Port, &s);
        if(r < 0 || s == 0) {
            auto res = fmt::format("failed to connect cloud vgw: {}", vgwUrl);
            rlogger->error(res);
            if(gConfigSystem.pClient) {
                MqttMgr::report_response_args(gConfigSystem.pClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, res, "pusher", "", json());
            }
            this_thread::sleep_for(4s);
            continue;
        }

        ret = setupDealer(&pDealerCtx, &pDealer, string(dev_sn), vgwUrl, 10 * 1000, 0);
        if(ret < 0) {
            auto res = fmt::format("failed to connect cloud vgw: {}", vgwUrl);
            rlogger->error(res);
            if(gConfigSystem.pClient) {
                MqttMgr::report_response_args(gConfigSystem.pClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, res, "pusher", "", json());
            }
            gPusherFailsFlag |= 2;
        }

        if(!gPusherFailsFlag) {
            rlogger->info("success connect to vgw: {}", vgwUrl);
            gPusherFailsFlag &= ~2;

            // setup sub
            pSub = zmq_socket(pPubCtx, ZMQ_SUB);
            if(pSub == nullptr) {
                rlogger->error("failed to create frame receiving socket");
                continue;
            }

            // int onlyOne = 1;
            // zmq_setsockopt(pSub, ZMQ_CONFLATE, &onlyOne, sizeof(onlyOne));
            // if(ret < 0) {
            //     rlogger->error("failed to set ZMQ_CONFLATE on pusher {}", zmq_strerror(ret));
            // }
            // ret = 10;
            // ret = zmq_setsockopt(pPub, ZMQ_RCVHWM, &ret, sizeof(ret));
            // if(ret < 0) {
            //     rlogger->error("failed to set ZMQ_RCVHWM on pusher sub {}", zmq_strerror(zmq_errno()));
            // }
            ret = zmq_connect(pSub, consts::strPubUrl.c_str());
            if(ret != 0) {
                rlogger->error("failed connect to frame pub: {}", consts::strPubUrl);
                continue;
            }
            ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
            rlogger->info("zmq sub successed (for frame sending)");
            gPusherFailsFlag &= ~1;

        }

        if(!gPusherFailsFlag) {
            // force I Frame
            LibXmMaQue_VideoEnc_forceIFrame(0, (MaQueStreamChannel_e) 0);
            rlogger->info("start pushing stream");
            while (1) {
                zmq_msg_init(&msg);
                ret = zmq_msg_recv(&msg, pSub, 0);
                if(ret < 0) {
                    rlogger->error("failed to recv zmq msg: {}", zmq_strerror(zmq_errno()));
                    gPusherFailsFlag |= 1 << 0;
                    break;
                }
                // ssize_t sz = zmq_msg_size(&msg);
                // char *ptr = (char*)zmq_msg_data(&msg);
                if(!gConfigSystem.module.sys.push) {
                    gPusherFailsFlag  |= 1 << 2;
                    zmq_msg_close(&msg);
                    continue;
                }
                else {
                    ret = zmq_msg_send(&msg, pDealer, 0);
                    zmq_msg_close(&msg);
                }
                if(ret < 0) {
                    auto res = fmt::format("failed to send vgw {}: {}", gJsonConfig[consts::kMsgConfigVgw].as<string>(), zmq_strerror(zmq_errno()));
                    rlogger->error(res);
                    if(gConfigSystem.pClient) {
                        MqttMgr::report_response_args(gConfigSystem.pClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, res, "pusher", "", json());
                    }
                    gPusherFailsFlag |= 1 << 1;
                    break;
                }
            }
        }

        if(gPusherFailsFlag & 1) {
            if(pSub) {
                zmq_close(pSub);
            }
            pSub = nullptr;
        }
        if(gPusherFailsFlag & 2) {
            if(pDealer) {
                zmq_close(pDealer);
            }
            if(pDealerCtx) {
                zmq_ctx_destroy(pDealerCtx);
            }
            pDealer = nullptr;
            pDealerCtx = nullptr;
        }

        this_thread::sleep_for(4s);
        gPusherFailsFlag = 0;
    }
}

/// 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()) {
        rlogger->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_GOOD;
        gConfigSystem.module.record.interval = 60 * 2; // 2 minutes interval
        gConfigSystem.module.sys.bitrate_kb = 1024; // 1Mbps
        gConfigSystem.module.sys.bitrate_type = MAQUE_BITRATE_CTRL_VBR;
        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>());
                rlogger->set_level(logLevel);
            }

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

            // record
            if(cfgFeatures.contains("recordLen")) {
                gConfigSystem.module.record.interval = cfgFeatures["recordLen"].as<int>();
            }
            else {
                rlogger->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 {
                rlogger->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>();
                if(gConfigSystem.module.sys.bitrate_kb > 4096 || gConfigSystem.module.sys.bitrate_kb < 200) {
                    gConfigSystem.module.sys.bitrate_kb = 1024;
                }
            }
            else {
                rlogger->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 {
                rlogger->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 {
                rlogger->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(gConfigSystem.module.motion.level > 6 || gConfigSystem.module.motion.level < 1) {
                        gConfigSystem.module.motion.level = 3;
                    }
                }
                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>()};
                    if(gConfigSystem.module.motion.region.minx >= 0 && gConfigSystem.module.motion.region.maxx > gConfigSystem.module.motion.region.minx && gConfigSystem.module.motion.region.maxx <=1 &&
                            gConfigSystem.module.motion.region.miny >= 0 && gConfigSystem.module.motion.region.maxy > gConfigSystem.module.motion.region.miny && gConfigSystem.module.ai.region.maxy <=1
                      ) {
                        //
                    }
                    else {
                        rlogger->error("invalid region for ai: {}", motion.to_string());
                        gConfigSystem.module.motion.region = {0,0,1,1};
                    }
                }
            }
            rlogger->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>()};
                    if(gConfigSystem.module.ai.region.minx >= 0 && gConfigSystem.module.ai.region.maxx > gConfigSystem.module.ai.region.minx && gConfigSystem.module.ai.region.maxx <=1 &&
                            gConfigSystem.module.ai.region.miny >= 0 && gConfigSystem.module.ai.region.maxy > gConfigSystem.module.ai.region.miny && gConfigSystem.module.ai.region.maxy <=1
                      ) {
                        //
                    }
                    else {
                        rlogger->error("invalid region for ai: {}", ai.to_string());
                        gConfigSystem.module.ai.region = {0,0,1,1};
                    }
                }
            }
            rlogger->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());
        rlogger->error(rc);
    }
    return rc;
}

bool verify_qr_target(string target)
{
    if(target == "WIFI" || target == "SN" || target == "MQ") {
        return true;
    }
    else {
        rlogger->error("invalid qr target: {}", target);
        return false;
    }
}

/// parse QR configure string
/// "WIFI:EVCAMERA1,1234;SN:012345678901"
void process_qr_event(string strQR)
{
    bool reboot = false;
    /// TODO: refinement of string literals into namespace consts::X
    /// parse QR string
    json result; // key -> array<string>
    auto items = evutils::split(strQR, ';');
    bool invalid = false;
    for(auto &item: items) {
        auto secs = evutils::split(item, ':');
        if(secs.size() >= 2) {
            if(!verify_qr_target(secs[0])) {
                return;
            }
            auto values = evutils::split(secs[1], ',');
            if(values.size() > 0) {
                result[secs[0]] = json(json_array_arg);
                for(auto &v: values) {
                    result[secs[0]].push_back(v);
                }
            }
        }
    }

    for(const auto &member: result.object_range()) {
        auto v = member.value().array_value();
        /// wifi
        if(member.key() == "WIFI") {
            if(v.size() != 2) {
                rlogger->error("handle_qr: invalid WIFI config data: {}", v.size());
            }
            else {
                string buffer = fmt::format("network={{\nssid=\"{}\"\npsk=\"{}\"\n}}", v[0].as<string>(), v[1].as<string>());
                ofstream wpa(sysWpaCfgFilePath, std::ios::trunc);
                if(!wpa.is_open()) {
                    rlogger->error("handle_qr: failed to open wpa.conf");
                }
                else {
                    wpa.write(buffer.c_str(), buffer.size());
                }
            }
            reboot = true;
        }
        else if(member.key() == "SN") {
            if(v.size() != 1) {
                rlogger->error("handle_qr: invalid SN config data: {}", v.size());
            }
            else {
                json sysCfg;
                string sn_ = v[0].as<string>();
                if(sn_.size() != 12) {
                    rlogger->error("handle_qr: sn size is not 12 bytes but {}", sn_.size());
                }
                else {
                    evutils::load_json(sysCfg, sysCfgJsonFilePath);
                    sysCfg["sn"] = v[0].as<string>();
                    evutils::save_json(sysCfg, sysCfgJsonFilePath);
                    reboot = true;
                }
            }
        }
        else if(member.key() == "MQ") {
            if(v.size() != 1) {
                rlogger->error("handle_qr: invalid MQ config data: {}", v.size());
            }
            else {
                json sysCfg;
                string mq = v[0].as<string>();
                // std::replace(mq.begin(), mq.end(), '/', ':');
                auto uri = httplib::Uri::Parse(mq);
                if(uri.Host.empty() || uri.Port.empty()) {
                    rlogger->error("handle_qr: invalid MQ config data: {}", mq);
                }
                else {
                    gJsonConfig["mqtt"] = mq;
                    save_configuration(gJsonConfig);
                    reboot = true;
                }
            }
        }
        else {
            rlogger->error("handle_qr: unkown content: {}", member.key(), member.value().to_string());
            // revoke any preceding reboot request
            // reboot = false;
            break;
        }
    }

    if(reboot) {
        rlogger->warn("handler_qr: system rebooting");
        /// caveate: make sure it will reboot
        LibXmMaQue_System_reboot();
        // net goes here
        this_thread::sleep_for(chrono::seconds(2));
        exit(0);
    }
}

/// 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);
    rlogger->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()) {
            rlogger->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>();
        /// implementation of cmds
        if(cmd == consts::kMsgCmdConfig) {
            bool needReboot = false;
            // verify configuration
            if(data.size() == 0) {
                MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", cmd, rid, gJsonConfig);
                rlogger->info("local config sent");
                return;
            }
            auto rv = verify_config(data);
            if(!rv.empty()) {
                rlogger->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
                    rlogger->info("json diff: {}", patch.to_string());
                    auto strP = patch.to_string();
                    if(strP.find("mqtt") != string::npos || strP.find("upload") != string::npos || strP.find("vgw") != string::npos || strP.find("region") != string::npos) {
                        needReboot = true;
                    }
                    auto rc = verify_config(data);
                    if(!rc.empty()) {
                    }
                    else {
                        rc = apply_config(data);
                    }
                    if(!rc.empty()) {
                        rlogger->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);
                        if(needReboot) {
                            rlogger->info("rebooting to apply new configuration");
                            LibXmMaQue_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()));
            }

            //
            stat["pusher"] = json();
            stat["pusher"]["val"] = gPusherFailsFlag;
            stat["pusher"]["msg"] = json(json_array_arg);
            if(gPusherFailsFlag &2) {
                stat["pusher"]["msg"].push_back(fmt::format("failed to connect vgw: {}", gJsonConfig[consts::kMsgConfigVgw].as<string>()));
            }
            if(gPusherFailsFlag &4) {
                stat["pusher"]["msg"].push_back("pusher disabled in configuration");
            }

            /// 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
            rlogger->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) {
                    rlogger->info("mqtt begin move");
                    string rc = ptz_move(data["action"].as<string>(), data["degree"].as<float>());
                    rlogger->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 == consts::kMsgCmdQR) {
            auto duration = data["duration"].as<int>();
            gQRCodeDuration.store(duration, memory_order_relaxed);
        }
        else if(cmd == "list_videos") {
            if(gRecFilesList != nullptr)
                print_ts_files(*gRecFilesList);
            else {
                rlogger->warn("no local video files");
            }
        }
        else {
            auto str = fmt::format("unsupported cmd {}: {}", cmd, js.to_string());
            rlogger->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;
        rlogger->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) {
        rlogger->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()) {
        rlogger->info("create path: {}", k);
        system((string("mkdir -p ") + k).c_str());
    }
}

///
int main(int argc, char *argv[])
{
    int ret = XM_SUCCESS;
    auto envlog = getenv("CONSOLE");
    if(envlog != nullptr) {
        if(!strncmp(envlog, "true", 4)){
            rlogger = spdlog::default_logger();
        }
    }
    
    // disable watchdog first
    LibXmMaQue_Watchdog_enable(0);
    

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

    memset(dev_sn, 0, sizeof(dev_sn));
    json sysCfg;
    auto res = evutils::load_json(sysCfg, sysCfgJsonFilePath);
    bool snReady = false;
    if(res.empty()) {
        if(sysCfg.count("sn") != 0) {
            auto s = sysCfg["sn"].as<string>();
            if(s.size() == 12) {
                memcpy(dev_sn, s.data(), 12);
                snReady = true;
            }
            else {
                fs::remove(sysCfgJsonFilePath);
            }
        }
    }
    if(!snReady) {
        get_mac_addr(dev_sn);
    }
    rlogger->info("hi3518ev300, sn: {}, version: {}, BE: {}", dev_sn, (char*)evutils::consts::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()) {
        rlogger->error("failed to apply configuration: {}", gJsonConfig.to_string());
        rlogger->error("defaults is used instead");
    }

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

    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);
    rlogger->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);
    rlogger->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);
    rlogger->info("abt ret: {}", ret);
    if (XM_SUCCESS == ret) {
        rlogger->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 = 5;
    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;
    cfg.eCapSize = MAQUE_CAPTURE_SIZE_1080P;
    cfg.bUseUserQp = 1;
    cfg.iQpMax = 48;
    cfg.iQpMin = 24;
   
    ret = configure_stream((MaQueStreamChannel_e)0, &cfg);
    rlogger->info("cfg stream ret: {}", ret);

    ret = LibXmMaQue_VideoEnc_startStream(0, (MaQueStreamChannel_e)0, cb_frame_proc, (void *)&args);
    if (XM_SUCCESS == ret) {
        rlogger->info("LibXmMaQue_VideoEnc_startStream success");
    }

    //
    thread thQR = thread([] {
        while(true)
        {
            auto dura = gQRCodeDuration.load(memory_order_relaxed);
            bool always = dura == -1? true:false;
            if(dura > 0 || dura < 0) {
                auto start = chrono::system_clock::now();
                auto end = start;
                int64_t delta = 0;
                while(delta < dura||always) {
                    string result;
                    rlogger->info("got qr job");
                    string res = qr_analysis(result);
                    if(res.empty()) {
                        /// TODO: handle result
                        rlogger->info("processing qr event");
                        process_qr_event(result);
                        if(!always) {
                            // set to stop
                            gQRCodeDuration.store(0, memory_order_relaxed);
                        }
                        break;
                    }
                    if(always) {
                        break;
                    }
                    // scan every 3s unless we got a code
                    this_thread::sleep_for(chrono::milliseconds(3000));
                    end = chrono::system_clock::now();
                    delta = chrono::duration_cast<chrono::seconds>(end - start).count();
                }
                /// no code got
                if(delta >= dura && dura > 0) {
                    /// TODO:  handle failed task
                }
            }
            else {
                // no task
            }
            /// check for task every 3s
            // rlogger->info("debug qr task");
            this_thread::sleep_for(chrono::seconds(4));
        }
    });

    rlogger->info("created QR task");

    // blocking update time, since time is critical to every other component
    bool bGotTime = false;
    time_t stm;
    LibXmMaQue_Watchdog_enable(1);
    LibXmMaQue_Watchdog_setTimeOutSecs(30);
    while(!bGotTime) {
        if(getNtpTime(&stm) >=0) {
            rlogger->info("ntp got time");
            ::stime(&stm);
            bGotTime = true;
        }
        else {
            rlogger->warn("failed to get ntp time");
            // enable QR
            gQRCodeDuration.store(10, memory_order_relaxed);
        }
        //rlogger->info("debug ntp");
        LibXmMaQue_Watchdog_keepAlive();
        this_thread::sleep_for(chrono::seconds(3));
    }
    LibXmMaQue_Watchdog_enable(0);
    // disable qr
    gQRCodeDuration.store(0, memory_order_relaxed);

    /// subscribe to mqtt
    while(gMqttClient == nullptr) {
        gQRCodeDuration.store(-1, memory_order_relaxed);
        /// TODO: handle result
        gMqttClient = start_mqtt(nullptr);
        this_thread::sleep_for(2s);
    }
    gQRCodeDuration.store(0, memory_order_relaxed);

    /// only after we connected network, start ptz service
    ptz_service_start();

    // frame dispatching
    pPubCtx = zmq_ctx_new();
    pPub = zmq_socket(pPubCtx, ZMQ_PUB);
    // int onlyOne = 1;
    // zmq_setsockopt(pPub, ZMQ_CONFLATE, &onlyOne, sizeof(onlyOne));
    // if(ret < 0) {
    //     rlogger->error("failed to set ZMQ_CONFLATE on pub {}", zmq_strerror(zmq_errno()));
    // }
    // ret = 10;
    // ret = zmq_setsockopt(pPub, ZMQ_SNDHWM, &ret, sizeof(ret));
    // if(ret < 0) {
    //     rlogger->error("failed to set ZMQ_SNDHWM on pub {}", zmq_strerror(zmq_errno()));
    // }
    ret = zmq_bind(pPub, consts::strPubUrl.c_str());
    if(ret < 0) {
        rlogger->error("{} failed to create zmq pub topic: {}", consts::strPubUrl);
        exit(1);
    }
    /// TODO: configuration issues handling
    if(msg_field(gJsonConfig, consts::kMsgConfigVgw).empty()) {
        rlogger->error("missing vgw config");
    }

    thread thPush = thread(frame_send_entry, &args);
    rlogger->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) {
        rlogger->info("waiting for record svc to start");
        this_thread::sleep_for(chrono::seconds(3));
    }

    rlogger->info("recording svc started, starting upload svc");

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

    start_md_bd(&gConfigSystem);

    gQRCodeDuration.store(0);

    thread thSmart = thread(maq_smart_task_entry, &gConfigSystem);
    if(thSmart.joinable()) {
        thSmart.detach();
    }

    // thread for watchdog
    LibXmMaQue_Watchdog_enable(1);
    LibXmMaQue_Watchdog_setTimeOutSecs(33);
    while(1) {
        this_thread::sleep_for(20s);
        LibXmMaQue_Watchdog_keepAlive();
    }
}