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

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <spdlog/spdlog.h>
#include <evpacket.h>
#include <sys/time.h>
#include <mutex>
#include <map>
#include <queue>
#include <zmq.h>
#include <jsoncons/json.hpp>
#include <thread>
#include <atomic>

extern "C"
{
#include <libavutil/timestamp.h>
#include <libavformat/avformat.h>
#undef av_err2str
#define av_err2str(errnum) av_make_error_string((char*)__builtin_alloca(AV_ERROR_MAX_STRING_SIZE), AV_ERROR_MAX_STRING_SIZE, errnum)
}

using namespace std;
using namespace jsoncons;

string darwinUrl = "rtsp://evcloudsvc.ilabservice.cloud:554/";
const char *inproc = "inproc://vproxy";

void *pRouterCtx = nullptr, *pRouter = nullptr/*, *pDealerCtx = nullptr, *pDealer = nullptr*/;

typedef struct packet_processor_t {
    evpacket_t hdr;
    int state; // 0 - unkown; 1 - receiving header; 2 - header ready, receiving body; 3 - body ready, processing
    AVFormatContext *pAvCtx;
    uint64_t packetId;
    int failedCnt = 0;
    queue<uint8_t *> pktQue;
    int64_t pts;
    string sn;
    packet_processor_t():hdr{0}, state{0}, pAvCtx(0), packetId(0), pts(0) {}
} packet_processor_t;

//AVIOInterruptCB int_cb;
std::map<string, packet_processor_t *> deviceMap;
std::map<string, mutex*> deviceLock;
recursive_mutex gMutDeviceLock;

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

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


unsigned short crc16(const unsigned char* data_p, unsigned char length)
{
    unsigned char x;
    unsigned short crc = 0xFFFF;

    while (length--) {
        x = crc >> 8 ^ *data_p++;
        x ^= x>>4;
        crc = (crc << 8) ^ ((unsigned short)(x << 12)) ^ ((unsigned short)(x <<5)) ^ ((unsigned short)x);
    }
    return crc;
}

void debugHex(char *buf, int len)
{
    int i = 0;
    for (char *p = buf; i < len; i++) {
        if (i % 16 == 0) {
            fprintf(stderr, "\n\t");
        }
        fprintf(stderr, "%02hhX ", p[i]);
    }
    fprintf(stderr, "\n");
}

// av callback
int av_callback(void *ctx)
{
    return 0;
}

void clean_av_resources(packet_processor_t *pro)
{
    if(pro->pAvCtx && pro->pAvCtx->pb) {
        avio_closep(&pro->pAvCtx->pb);
    }
    if(pro->pAvCtx) {
        avformat_free_context(pro->pAvCtx);
    }
    pro->pAvCtx = nullptr;
}

AVFormatContext *rtsp_init(string rtsp_url, int codec, int height, int width, int *rc = nullptr)
{
    int ret;
    int codec_idx = 0;
    struct timeval tv, tv2;
    //int bCheckBufWrong;
    int bps = 0;
    ::gettimeofday(&tv, nullptr);
    AVFormatContext *pAVFormatRemux = nullptr;
    AVDictionary *pOptsRemux = nullptr;

    ret = av_dict_set(&pOptsRemux, "rtsp_transport", "tcp", 0);
    if ( ret < 0) {
        spdlog::error("failed set output pOptsRemux");
        if(rc) *rc = ret;
        return nullptr;
    }
    // timeout in microseconds
    av_dict_set_int(&pOptsRemux, "stimeout", (int64_t)(1000 * 1000 * 1), 0);
    ret = avformat_alloc_output_context2(&pAVFormatRemux, nullptr, "rtsp", rtsp_url.c_str());
    AVStream *out_stream = avformat_new_stream(pAVFormatRemux, nullptr);
    if (ret < 0|| !out_stream) {
        spdlog::error("Failed allocating output stream\n");
        if(rc) *rc = ret;
        return nullptr;
    }

    AVCodecParameters *out_codecpar = out_stream->codecpar;
    memset(out_codecpar, 0, sizeof(AVCodecParameters));
    out_codecpar->codec_type = AVMEDIA_TYPE_VIDEO;
    out_codecpar->codec_id = AV_CODEC_ID_H264;
    out_codecpar->bit_rate = 1024 * 1024; //1Mbps
    out_codecpar->width = width;
    out_codecpar->height = height;
    out_codecpar->codec_tag = 0;
    out_codecpar->format = AV_PIX_FMT_YUV420P;

    //int_cb = {av_callback, nullptr};
    //pAVFormatRemux->interrupt_callback = int_cb;
    ret = avformat_write_header(pAVFormatRemux, &pOptsRemux);
    if(ret < 0) {
        spdlog::error("failed to writeheader: {}", rtsp_url);
        if(rc) *rc = ret;
        return nullptr;
    }
    spdlog::info("success connect to darwin {}", rtsp_url);

    av_dict_free(&pOptsRemux);

    return pAVFormatRemux;
}

int write_packet(packet_processor_t *processor, char *data, int len)
{
    AVFormatContext *ctx = nullptr;
    if(processor == nullptr||(ctx = processor->pAvCtx) == nullptr || ctx->streams == nullptr || *(ctx->streams) == nullptr) {
        spdlog::error("no avformatctx or stream available for write");
        return -1;
    }
    int ret = 0;
    AVPacket pkt;
    AVStream *out_stream = ctx->streams[0];
    av_init_packet(&pkt);

    memset(&pkt, 0, sizeof(pkt));
    pkt.stream_index = 0;
    pkt.data = (uint8_t *)data;
    pkt.size = len;
    AVRational time_base;
    char *fps = getenv("FPS");
    if(processor->hdr.vpara.fps > 0 && processor->hdr.vpara.fps < 100) {
        time_base.den = processor->hdr.vpara.fps;
    }
    else {
        time_base.den = 15;
    }
    if(fps) {
        time_base.den = atoi(fps);
    }
    time_base.num = 1;
    pkt.dts = av_rescale_q_rnd(++processor->pts, time_base, out_stream->time_base, (AV_ROUND_NEAR_INF | AV_ROUND_PASS_MINMAX));
    pkt.pts = av_rescale_q(processor->pts, time_base, out_stream->time_base);
    pkt.duration = 0; //av_rescale_q(pkt.duration, out_stream->time_base, out_stream->time_base);
    pkt.pos = -1;

    ret = av_write_frame(ctx, &pkt);
    if (ret < 0) {
        spdlog::error("Error muxing packet: {}", av_err2str(ret));
    }

    //av_packet_unref(&pkt);
    return ret;
}

typedef void *(*fn_worker_t)(void *ctx);

void dispatch(string sn, zmq_msg_t *msg)
{
    /// TODO: allocate new instance of processor
    zmq_msg_t resp;
    evpacket_t *pkt = (evpacket_t *)zmq_msg_data(msg);

    int ret = 0;

    //debugHex((char*)zmq_msg_data(&msg), 64);
    auto crc = pkt->meta.crc;
    pkt->meta.crc = 0;
    auto crc_ = crc16((unsigned char*)pkt, sizeof(evpacket_t));
    if(crc != crc_) {
        spdlog::error("invalid crc: {}", pkt->meta.packet_id);
        return;
    }
    /// we don't need lock it
    // lock_guard<mutex> lg(*deviceLock[sn]);
    auto processor = deviceMap[sn];
    if(processor->pAvCtx == nullptr) {
        int rc = 0;
        processor->pAvCtx = rtsp_init(darwinUrl + sn, 0, pkt->vpara.res.height, pkt->vpara.res.width, &rc);
        if(processor->pAvCtx == nullptr) {
            spdlog::error("failed to connect to darwin {}: {}", darwinUrl, av_err2str(rc));
            return;
        }
        else {
            memcpy((void*)&processor->hdr, zmq_msg_data(msg), sizeof(processor->hdr));
        }
    }

    ret = write_packet(processor, (char*)zmq_msg_data(msg) + sizeof(evpacket_t), zmq_msg_size(msg) - sizeof(evpacket_t) );
    if(ret < 0) {
        spdlog::error("failed to send packet!");
        clean_av_resources(processor);
    }
}

void * fn_worker(void *pArg)
{
    int ret = 0;
    packet_processor_t *processor = (packet_processor_t*)pArg;
    spdlog::info("worker {} started", processor->sn);
    zmq_msg_t *msg;
    while(1) {
        msg = nullptr;
        {
            lock_guard<mutex> lg(*deviceLock[processor->sn]);
            if(processor->pktQue.size() > 0) {
                msg = (zmq_msg_t *)processor->pktQue.front();
                processor->pktQue.pop();
            }
        }
        if(nullptr != msg) {
            dispatch(processor->sn, msg);
            zmq_msg_close(msg);
            free((char*)msg);
        }
        else {
            this_thread::sleep_for(chrono::milliseconds(200));
        }
    }

    return NULL;
}

int setupRouter(void **ctx, void **s, string addr, int rcvQS=0)
{
    int ret = 0;
    *ctx = zmq_ctx_new();
    *s = zmq_socket(*ctx, ZMQ_ROUTER);
    ret = 1;
    zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
    ret = 30;
    zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
    zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
    ret = 2;
    zmq_setsockopt(*s, ZMQ_TCP_KEEPALIVE_CNT, &ret, sizeof (ret));
    // ret = 10 * 1000;
    // zmq_setsockopt(*s, ZMQ_SNDTIMEO, &ret, sizeof (ret));
    if(rcvQS !=0) {
        zmq_setsockopt(*s, ZMQ_RCVHWM, &rcvQS, sizeof(rcvQS));
    }
    ret = zmq_bind(*s, addr.c_str());
    if(ret < 0) {
        spdlog::error("failed to bind zmq at {} for reason: {}, retrying load configuration...", addr, zmq_strerror(zmq_errno()));
        if(*ctx) {
            if(*s) {
                zmq_close(*s);
            }
            zmq_ctx_destroy(*ctx);
        }
        *s = nullptr;
        *ctx = nullptr;
        return ret;
    }

    return ret;
}

int main()
{
    /// darwin
    auto darwinUrl_ = getenv("DARWIN_URL");
    if(darwinUrl_) {
        darwinUrl = string(darwinUrl_);
    }

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

    string addr = "tcp://0.0.0.0:7123";
    if(setupRouter(&pRouterCtx, &pRouter, addr)<0) {
        exit(1);
    }

    while(1) {
        int64_t more = 1;
        size_t more_size = sizeof(more);
        int ret = 0;
        zmq_msg_t msg;

        /// receive the first frame
        ret = zmq_msg_init(&msg);
        if(ret < 0) {
            spdlog::error("failed to receive multiple msg on zmq_msg_init: {}", zmq_strerror(zmq_errno()));
            break;
        }
        ret = zmq_msg_recv(&msg, pRouter, 0);
        if(ret < 0) {
            spdlog::error("zmq_msg_recv: {}", zmq_strerror(zmq_errno()));
            break;
        }

        /// first frame is sn
        string sn((char*)zmq_msg_data(&msg), zmq_msg_size(&msg));
        zmq_msg_close(&msg);

        /// check the second frame
        ret = zmq_getsockopt(pRouter, ZMQ_RCVMORE, &more, &more_size);
        if(ret < 0) {
            spdlog::error("zmq_getsockopt: {}", zmq_strerror(zmq_errno()));
            continue;
        }

        if(more > 0) {
            zmq_msg_t *msg = (zmq_msg_t*)malloc(sizeof(zmq_msg_t));
            ret = zmq_msg_init(msg);
            ret |= zmq_msg_recv(msg, pRouter, 0);
            if(ret < 0) {
                spdlog::error("error zmq_msg_recv seconds frame: {}", zmq_strerror(zmq_errno()));
                continue;
            }

            /// double check
            if(deviceMap.count(sn) == 0) {
                lock_guard<recursive_mutex> lg(gMutDeviceLock);
                if(deviceMap.count(sn) == 0) {
                    deviceLock.emplace(sn, new mutex());
                    deviceMap.emplace(sn, new packet_processor_t());
                    deviceMap[sn]->sn = sn;
                }
            }
            if(deviceMap[sn]->state == 0) {
                deviceMap[sn]->state = 1;
                thread thWorker = thread(fn_worker, deviceMap[sn]);
                if(thWorker.joinable()) {
                    thWorker.detach();
                }
            }

            {
                lock_guard<mutex> lg(*deviceLock[sn]);
                if(deviceMap[sn]->pktQue.size() > 10) {
                    zmq_msg_t* m = (zmq_msg_t*) (deviceMap[sn]->pktQue.front());
                    zmq_msg_close(m);
                    free((char*)m);
                    deviceMap[sn]->pktQue.pop();
                }
                deviceMap[sn]->pktQue.push((uint8_t*)msg);
            }
            //zmq_msg_close(msg);
        }
        else {
            spdlog::error("invalid packet fmt: no seconds frame");
        }

        if(ret < 0) {
            spdlog::error("failed to recv msg: {}",  ret < 0? zmq_strerror(ret): "invalid frames");
            return -1;
        }
    }

    zmq_close(pRouter);
    zmq_ctx_destroy(pRouterCtx);

    return 0;
}
