#pragma GCC diagnostic ignored "-Wunused-private-field"
#pragma GCC diagnostic ignored "-Wunused-variable"

#include <stdlib.h>
#include <string>
#include <thread>
#include <iostream>
#include <chrono>
#include <future>
#include <queue>

#ifdef OS_LINUX
#include <filesystem>
namespace fs = std::filesystem;
#endif

#include "zmqhelper.hpp"
#include "tinythread.hpp"
#include "common.hpp"
#include "database.h"

using namespace std;
using namespace zmqhelper;

/**
 *  functions:
 *  app update
 *  control msg
 *
 **/

class EvMgr:public TinyThread {
private:
    void *pRouterCtx = NULL;
    void *pRouter = NULL;
    json config;
    string devSn;
    json peerStatus;
    json jmgr;
    unordered_map<string, queue<vector<vector<uint8_t> >> > cachedMsg;
    mutex cacheLock;
    queue<string> eventQue;
    mutex eventQLock;

    //
    void init()
    {
        int ret;
        bool inited = false;
        // TODO: load config from local db
        devSn = "ILSEVMGR1";
        while(!inited) {
            try {
                config = json::parse(cloudutils::config);
                spdlog::info("config dumps: \n{}", config.dump());
                // TODO: verify sn
                if(!config.count("data")||!config["data"].count(devSn)||!config["data"][devSn].count("ipcs")){
                    spdlog::error("evmgr {} invalid config. reload now...", devSn);
                    this_thread::sleep_for(chrono::seconds(3));
                    continue;
                }
                jmgr =  config["data"][devSn];
                string proto = jmgr["proto"];
                string addr;
                
                if(proto != "zmq"){
                    spdlog::error("evmgr {} unsupported protocol: {}, try fallback to zmq instead now...", devSn, proto);
                }
                addr = "tcp://" + jmgr["addr"].get<string>() + ":" + to_string(jmgr["port-router"]);
                // setup zmq
                // TODO: connect to cloud

                // router service
                pRouterCtx = zmq_ctx_new();
                pRouter = zmq_socket(pRouterCtx, ZMQ_ROUTER);
                int opt_notify = ZMQ_NOTIFY_DISCONNECT;
                zmq_setsockopt (pRouter, ZMQ_ROUTER_NOTIFY, &opt_notify, sizeof (opt_notify));
                ret = zmq_bind(pRouter, addr.c_str());
                if(ret < 0) {
                    spdlog::error("evmgr {} failed to bind zmq at {} for reason: {}, retrying load configuration...", devSn, addr, zmq_strerror(zmq_errno()));
                    this_thread::sleep_for(chrono::seconds(3));
                    continue;
                }
                inited = true;
            }
            catch(exception &e) {
                spdlog::error("evmgr {} exception on init() for: {}, retrying load configuration...", devSn, e.what());
                this_thread::sleep_for(chrono::seconds(3));
                continue;
            }
        }
        spdlog::info("evmgr {} successfuly inited", devSn);
    }

    int mqErrorMsg(string cls, string devSn, string extraInfo, int ret) {
        if(ret < 0) {
            spdlog::error("{} {} {}:{} ", cls, devSn, extraInfo, zmq_strerror(zmq_errno()));
        }
        return ret;
    }

    int handleMsg(vector<vector<uint8_t> > &body) {
        int ret = 0;
        zmq_msg_t msg;
        // ID_SENDER, ID_TARGET, meta ,MSG
        string peerId;
        if(body.size() == 2 && body[1].size() == 0) {
            peerId = body2str(body[0]);
            spdlog::warn("evmgr {} peer disconnected: {}", devSn, peerId);
            peerStatus[peerId] = 0;
            return 0;
        }else if(body.size() != 4) {
            spdlog::warn("evmgr {} dropped an invalid message, size: {}", devSn, body.size());
            return 0;
        }
        
        string meta = body2str(body[2]);
        string selfId = body2str(body[0]);
        peerId = body2str(body[1]);
        // update status;
        this->peerStatus[selfId] = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count();

        // msg to peer
        if(memcmp((void*)(body[1].data()), (devSn +":0:0").data(), body[1].size()) != 0) {
            // message to other peer
            // check peer status
            vector<vector<uint8_t> >v = {body[1], body[0], body[2], body[3]};
            if(peerStatus.count(peerId)!= 0) {
                auto t = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count() - peerStatus[peerId].get<long long>();
                if(t > EV_HEARTBEAT_SECONDS){
                    peerStatus[peerId] = 0;
                    // need cache
                }else{
                    spdlog::info("evmgr {} route msg from {} to {}", devSn, selfId, peerId);
                    ret = z_send_multiple(pRouter, v);
                    if(ret < 0) {
                        spdlog::error("evmgr {} failed to send multiple: {}", devSn, zmq_strerror(zmq_errno()));
                    }
                }
            }else{
                peerStatus[peerId] = 0;
                // need cache
            }

            if(peerStatus[peerId] == 0) {
                // cache
                spdlog::warn("evmgr {} cached msg from {} to {}", devSn, selfId, peerId);
                lock_guard<mutex> lock(cacheLock);
                cachedMsg[peerId].push(v);
                if(cachedMsg[peerId].size() > EV_NUM_CACHE_PERPEER) {
                    cachedMsg[peerId].pop();
                }
            }

            // check if event
            try{
                string metaType = json::parse(meta)["type"];
                if(metaType == EV_MSG_META_EVENT) {
                    eventQue.push(body2str(body[3]));
                    if(eventQue.size() > MAX_EVENT_QUEUE_SIZE) {
                        eventQue.pop();
                    }
                }
            }catch(exception &e) {
                spdlog::error("evmgr {} exception parse event msg from {} to {}: ", devSn, selfId, peerId, e.what());
            }
        }else{
            // message to mgr
            spdlog::info("evmgr {} subsystem report msg received: {}; {}; {}", devSn, zmqhelper::body2str(body[0]), zmqhelper::body2str(body[1]), zmqhelper::body2str(body[2]));
            if(meta == "pong"||meta == "ping") {
                // update status
                spdlog::info("evmgr {}, ping msg from {}", devSn, selfId);
                if(meta=="ping") {
                    if(cachedMsg.find(selfId) != cachedMsg.end()) {
                        while(!cachedMsg[selfId].empty()){
                            lock_guard<mutex> lock(cacheLock);
                            auto v = cachedMsg[selfId].front();
                            cachedMsg[selfId].pop();
                            ret = z_send_multiple(pRouter, v);
                            if(ret < 0) {
                                spdlog::error("evmgr {} failed to send multiple: {}", devSn, zmq_strerror(zmq_errno()));
                            }  
                        }
                    }
                }
            }else{
                // TODO:
                spdlog::warn("evmgr {} received unknown meta {} from {}", devSn, meta, selfId);
            }
        }

        return ret;
    }
    int  get_monitor_event (void *monitor, int *value, char **address)
           {
               //  First frame in message contains event number and value
               zmq_msg_t msg;
               zmq_msg_init (&msg);
               if (zmq_msg_recv (&msg, monitor, 0) == -1)
                   return -1;              //  Interrupted, presumably
               assert (zmq_msg_more (&msg));

               uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
               uint16_t event = *(uint16_t *) (data);
               if (value)
                   *value = *(uint32_t *) (data + 2);

               //  Second frame in message contains event address
               zmq_msg_init (&msg);
               if (zmq_msg_recv (&msg, monitor, 0) == -1)
                   return -1;              //  Interrupted, presumably
               assert (!zmq_msg_more (&msg));

               if (address) {
                   uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
                   size_t size = zmq_msg_size (&msg);
                   *address = (char *) malloc (size + 1);
                   memcpy (*address, data, size);
                   (*address)[size] = 0;
               }
               return event;
           }

protected:
    void run(){
        bool bStopSig = false;
        int ret = 0;
        zmq_msg_t msg;

        // disabled because: 
        //    1. it can't determine which peer disconnected, but only the underline socket FD. 
        //    2. used the draft feature of ZMQ_ROUTER_NOTIFY instead to capture peer module disconnections such as evpuser, evmlmotion.
        // thread thMon = thread([&,this](){
        //     int ret = 0;
        //     string addr = string("inproc://monitor-") + this->devSn;
        //     ret = zmq_socket_monitor(this->pRouter, addr.c_str(), ZMQ_EVENT_ALL );//ZMQ_EVENT_DISCONNECTED
        //     if(ret != 0) {
        //         spdlog::error("evmgr {} failed mon1: {},  {}", this->devSn, addr, zmq_strerror(zmq_errno()));
        //     }
        //     void *mon = zmq_socket (this->pRouterCtx, ZMQ_PAIR);
        //     ret = zmq_connect(mon, addr.c_str());
        //     if(ret != 0) {
        //         spdlog::error("evmgr {} failed mon2: {}", this->devSn, zmq_strerror(zmq_errno()));
        //     }
        //     spdlog::info("evmgr {} monitoring setup", this->devSn);
        //     while(true){
        //         int fd = 0;
        //         char *pConn = NULL;
        //         int event = get_monitor_event(mon, &fd, &pConn);
        //         cout <<"event: " << event << ", fd: "<< fd << ", conn: "<<pConn <<endl;
        //     }
        // });

        // thMon.detach();
        //

        while (true) {
            if(checkStop() == true) {
                bStopSig = true;
                break;
            }
            auto body = z_recv_multiple(pRouter,false);
            if(body.size() == 0) {
                spdlog::error("evmgr {} failed to receive multiple msg: {}", devSn, zmq_strerror(zmq_errno()));
                continue;   
            }
            // full proto msg received.
            handleMsg(body);
        }
    }
public:
    EvMgr(EvMgr &&) = delete;
    EvMgr(EvMgr &) = delete;
    EvMgr(const EvMgr &) = delete;
    EvMgr& operator=(const EvMgr &) = delete;
    EvMgr& operator=(EvMgr &&) = delete;
    EvMgr()
    {
        init();
    }
    ~EvMgr()
    {
        if(pRouter != NULL) {
            zmq_close(pRouter);
            pRouter = NULL;
        }
        if(pRouterCtx != NULL){
            zmq_ctx_destroy(pRouterCtx);
            pRouterCtx = NULL;
        }
    }
};

int main(int argc, const char *argv[])
{
    av_log_set_level(AV_LOG_ERROR);
    spdlog::set_level(spdlog::level::debug);
    EvMgr mgr;
    mgr.join();
    return 0;
}