/**
 * @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>

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;

#define DEFAULT_BACKLOG 128
#define MAX_FRAME_SIZE 500000
#define MAX_RETRY_PUSH_INTV 450

string darwinUrl = "rtsp://evcloudsvc.ilabservice.cloud:554/";
string bindPort = "7123";
string bindAddr = "0.0.0.0";

typedef struct {
    evpacket_t hdr;
    int state; // 0 - unkown; 1 - receiving header; 2 - header ready, receiving body; 3 - body ready, processing
    AVFormatContext *pAvCtx;
    uint64_t packetId;
    mutex mut;
    int failedCnt = 0;
    int64_t pts;
} packet_processor_t;


AVIOInterruptCB int_cb;

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

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

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

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 = 200 * 100 * 8; //400000;
    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;
}


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 = 5;
    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));
    if(rcvQS !=0) {
        zmq_setsockopt(*s, ZMQ_RCVHWM, &rcvQS, sizeof(rcvQS));
    }
    ret = zmq_bind(*s, addr.c_str());
    if(ret < 0) {
        spdlog::debug("failed to bind zmq at {} for reason: {}, retrying load configuration...", addr, zmq_strerror(zmq_errno()));
    }
    return ret;
}

int z_recv_multiple(void *s, vector<uint8_t> &buf, vector<int>&frameIdx)
{
    int64_t more = 1;
    size_t more_size = sizeof(more);
    int ret = 0;
    int cnt = 0;
    while(more > 0) {
        cnt++;
        zmq_msg_t msg;
        ret = zmq_msg_init(&msg);
        if(ret < 0) {
            spdlog::debug("failed to receive multiple msg on zmq_msg_init: {}", zmq_strerror(zmq_errno()));
            break;
        }
        ret = zmq_recvmsg(s, &msg, 0);
        if(ret < 0) {
            spdlog::debug("z_recv_multiple: {}", zmq_strerror(zmq_errno()));
            break;
        }
        buf.insert(buf.end(), (uint8_t*)zmq_msg_data(&msg), (uint8_t*)zmq_msg_data(&msg)+ret);
        // spdlog::info("buff size: {}", buf.size());
        frameIdx.push_back(buf.size());
        zmq_msg_close(&msg);
        ret = zmq_getsockopt(s, ZMQ_RCVMORE, &more, &more_size);
        if(ret < 0) {
            spdlog::debug("z_recv_multiple: {}", zmq_strerror(zmq_errno()));
            break;
        }
    }

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

    return 0;
}

packet_processor_t globalProcess = {0};

void dispatch(string sn, zmq_msg_t &msg)
{
    evpacket_t *pkt = (evpacket_t *)zmq_msg_data(&msg);
    globalProcess.packetId++;

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

    if(globalProcess.pAvCtx == nullptr && globalProcess.packetId %15*10 == 0) {
        int rc = 0;
        globalProcess.pAvCtx = rtsp_init(darwinUrl + sn, 0, 1080, 1920, &rc);
        if(globalProcess.pAvCtx == nullptr || rc < 0) {
            spdlog::error("failed to connect to darwin: {}", darwinUrl);
        }
        memcpy(&globalProcess.hdr, zmq_msg_data(&msg), sizeof(globalProcess.hdr));
    }

    int ret = write_packet(&globalProcess, (char*)zmq_msg_data(&msg) + sizeof(evpacket_t), zmq_msg_size(&msg) - sizeof(evpacket_t) );
    if(ret < 0) {
        spdlog::error("failed to send packet!");
        if(globalProcess.pAvCtx && globalProcess.pAvCtx->pb) {
            avio_closep(&globalProcess.pAvCtx->pb);
        }
        if(globalProcess.pAvCtx) {
            avformat_free_context(globalProcess.pAvCtx);
        }
        globalProcess.pAvCtx = nullptr;
    }
}

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";
    void *pRouterCtx = nullptr, *pRouter = nullptr/*, *pDealerCtx = nullptr, *pDealer = nullptr*/;
    setupRouter(&pRouterCtx, &pRouter, addr, 10);

    while(1) {
        int64_t more = 1;
        size_t more_size = sizeof(more);
        int ret = 0;
        zmq_msg_t msg;
        ret = zmq_msg_init(&msg);
        if(ret < 0) {
            spdlog::debug("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::debug("z_recv_multiple: {}", zmq_strerror(zmq_errno()));
            break;
        }
        string sn((char*)zmq_msg_data(&msg), zmq_msg_size(&msg));
        zmq_msg_close(&msg);
        // ignore the first frame
        ret = zmq_getsockopt(pRouter, ZMQ_RCVMORE, &more, &more_size);
        if(ret < 0) {
            spdlog::debug("z_recv_multiple: {}", zmq_strerror(zmq_errno()));
            break;
        }
        if(more > 0) {
            ret = zmq_msg_init(&msg);
            ret |= zmq_msg_recv(&msg, pRouter, 0);
            if(ret < 0) {
                spdlog::debug("z_recv_multiple: {}", zmq_strerror(zmq_errno()));
                break;
            }
            dispatch(sn, msg);
        }
        else {
            spdlog::error("invalid packet fmt");
        }

        zmq_msg_close(&msg);

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

    this_thread::sleep_for(2s);

    zmq_close(pRouter);
    zmq_ctx_destroy(pRouterCtx);

    return 0;
}
