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

#include "raw_tcp.h"
#include "record_upload.h"
#include <httplib.h>
#include <ghc/filesystem.hpp>
#include <chrono>
#include <evpacket.h>

/// force Iframe
extern "C" {
#include <maque_type.h>
#include <maque_video_enc.h>
}

///
#define MAGIC_HEAD0 ((uint8_t)0xBE)
#define MAGIC_HEAD1 ((uint8_t)0xEF)
#define MATIC_TAIL0 ((uint8_t)0xFF)
#define MAGIC_TAIL1 ((uint8_t)0xFF)
#define MAGIC_TAIL2 ((uint8_t)0xDE)
#define MAGIC_TAIL3 ((uint8_t)0xAD)

#define EV_UPLOAD_MAX_BUF_SIZE 1024
static const int kMaxHeapSize = 1 * 1024 * 1024;

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

extern ev_module_config_t gConfigSystem;
extern OrderedList<int64_t> *gRecFilesList;
extern json gRecordStat;

const string kUploadMetaDir = "/mnt/sd/meta/upload/";
const int kMaxJobLoad = 20;
const int kMaxWaitSecsForJob = 5;

extern shared_ptr<spdlog::logger> rlogger;
///
string upload_video(string hostUrl, char* dev_sn, int64_t tss, int64_t tse, set<int64_t> slices, int8_t type)
{
    string rc;
    uint8_t * buf = (uint8_t *)malloc(EV_UPLOAD_MAX_BUF_SIZE);
    rlogger->info("start uploading video of camera {} from {} to {}, total {} slices, type {}", dev_sn, tss, tse, slices.size(), type);

    int s = 0;
    try {
        auto uri = httplib::Uri::Parse(hostUrl);
        if(uri.Host.empty() || uri.Port.empty()) {
            rc = fmt::format("invalid upload url configure: {}", hostUrl);
        }
        else {
            auto i = raw_connect(uri.Host, uri.Port, &s);
            if(i < 0 || s == 0) {
                rc = fmt::format("failed to connect to {}", hostUrl);
            }
            else {
                /** 
                 * META := SN_LEN, SN, START_TS, END_TS, TYPE, NUM, {FILE_META*NUM}  # 元数据
                    SN_LEN := UINT16                                        # 摄像头SN长度 一般8字节
                    SN := {UINT8*SN_LEN}                                    # 摄像头SN, 长度为SN_LEN
                    START_TS := TS                                          # 事件开始时间
                    END_TS := TS                                            # 事件结束时间
                    TYPE := UINT8                                           # 0 - 未定义, 1 - AI人形, 2 - AI人脸, 3 - 移动侦测, 
                                                                            # 4 - 覆盖, 5 - 覆盖解除, 6 - 视频获取
                                                                            # 其他 - 由发起请求的微服务自定义
                    FILE_META := FILE_TS,SIZE
                    FILE_TS := TS    # 文件开始时间
                    SIZE := UINT32   # 文件字节数
                    NUM := UINT32    # 4-byte unsigned int. 视频文件数目
                    TS := INT64      # 8-byte long. 时间戳

                    注意: 目前, 代码中写死了, SN 为 12 个字节. 
                 **/
                auto metaLen = 2 + 12 + 8 + 8 + 1 + 4 + (8 + 4)*slices.size();

                // head :  MAGIC_HEADD + META + CRC
                auto headLen = 2 + metaLen + 2;
                if(headLen > EV_UPLOAD_MAX_BUF_SIZE) {
                    rc = fmt::format("too many files ({}) to upload such that header ({}) is too large, ignore uploading.", slices.size(), headLen);
                }
                else {
                    vector<uint32_t> fsize;
                    for(auto &k:slices) {
                        struct stat st;
                        if(stat((string(consts::recFilePath) + to_string(k) + ".h264").c_str(), &st) < 0) {
                            rc = fmt::format("failed to get file {}.h264: {}", k, strerror(errno));
                            break;
                        }
                        else {
                            fsize.push_back(st.st_size);
                        }
                    }
                    if(!rc.empty()) {
                        if(buf) {
                            free(buf);
                            if(s) {
                                ::close(s);
                            }
                        }
                        throw StrException(rc);
                    }

                    int totalLen = 0;
                    buf[0] = MAGIC_HEAD0;               // 1, 0
                    buf[1] = MAGIC_HEAD1;               // 1, 1
                    totalLen += 2;
                    *((uint16_t *)(buf+totalLen)) = 12;       // 2, 2
                    totalLen += 2;
                    memcpy(buf+totalLen, dev_sn, 12);         // 12, 4
                    totalLen += 12;
                    *((uint64_t *)(buf+totalLen)) = tss;      // 8, 16
                    totalLen += 8;
                    *((uint64_t *)(buf+totalLen)) = tse;      // 8, 24
                    totalLen += 8;
                    buf[totalLen]= type;                      // 1, 32
                    totalLen += 1;
                    *((uint32_t *)(buf+totalLen)) = slices.size(); // 4, 33
                    totalLen += 4;
                    int idx = 0;
                    for(auto &k:slices) {
                        *((uint64_t *)(buf + totalLen)) = k;    // 8, 37
                        totalLen += 8;
                        *((uint32_t *)(buf + totalLen)) = fsize[idx]; // 4, 45
                        totalLen += 4;
                        idx++;
                    }
                    auto crc_ = crc16(buf, totalLen);
                    memcpy(buf+totalLen, &crc_, 2);
                    totalLen +=2;
                    assert(totalLen = headLen);
                    rlogger->info("header len: {}, crc: {}", totalLen, crc_);

                    bool file_transfer_failed = false;
                        
                    // send header
                    rc = raw_send(s, buf, totalLen);
                    if(!rc.empty()) {
                        file_transfer_failed = true;
                        rlogger->error("failed to send file header. err: " + rc);
                        rc = fmt::format("{}, {}:{}", rc, uri.Host, uri.Port);
                    }
                    else {
                        // construct body
                        idx = 0;
                        for(auto &k:slices) {
                            int readsize = 1;
                            uint32_t total_sent = 0;
                            auto fpath = string(consts::recFilePath) + to_string(k) + ".h264";
                            memcpy(buf, &fsize[idx], 4);
                            file_transfer_failed = false;
                            FILE *fp = fopen(fpath.c_str(), "rb");
                            if(!fp) {
                                rc = fmt::format("failed open file : {}", fpath);
                                break;
                            }
                            else {
                                rlogger->info("sending file {}: size {}", fpath, fsize[idx]);
                                for(int i = 0; readsize > 0; i++) {
                                    if(i == 0) {
                                        readsize = ::fread(buf+4, 1, EV_UPLOAD_MAX_BUF_SIZE - 4, fp) + 4;
                                    }
                                    else {
                                        readsize = ::fread(buf, 1, EV_UPLOAD_MAX_BUF_SIZE, fp);
                                    }

                                    if((i == 0 && readsize > 4) ||(i !=0 && readsize > 0)) {
                                        rc = raw_send(s, buf, readsize);
                                        if(!rc.empty()) {
                                            file_transfer_failed = true;
                                            rlogger->error("failed to send file data via socket. err: {}", rc);
                                            break;
                                        }
                                    }
                                    else {
                                        break;
                                    }
                                    //rlogger->info("sent bytes: {}", readsize);
                                    total_sent += readsize;
                                }

                                ::fclose(fp);

                                if(readsize < 0) {
                                    rc = fmt::format("error read file: {}", strerror(errno));
                                    rlogger->error(rc);
                                    break;
                                } else if (file_transfer_failed){
                                    rlogger->error("file transfer failed, skip rest file upload");
                                    break;
                                }
                                else {
                                    rlogger->info("sent file {}: total {}, sent {}", fpath, fsize[idx], total_sent - 4);
                                }
                            }
                            idx++;
                        }
                    }

                    if (!file_transfer_failed){
                        rlogger->info("sending upload tail");
                        // construct tail
                        buf[0] = MATIC_TAIL0;
                        buf[1] = MAGIC_TAIL1;
                        buf[2] = MAGIC_TAIL2;
                        buf[3] = MAGIC_TAIL3;
                        rc = raw_send(s, buf, 4);

                        if(!rc.empty()){
                            rlogger->error("failed to send upload tail. err: {}", rc);

                            rc = fmt::format("upload tail error: {}", rc);
                        }
                    }
                }
            }
        }
    }
    catch(exception &e) {
        rc = fmt::format("exception sending file to {}: {}, {}:{}", hostUrl, e.what(), __FILE__, __LINE__);
    }

    free(buf);
    if(s != 0) {
        ::close(s);
    }

    return rc;
}

bool has_existing_job(string path, upload_item_t &job)
{
    bool found = false;
    if(fs::exists(path)) {
        try {
            ifstream cfg_(path);
            auto js = json::parse(cfg_);
            job.tss = js["tss"].as<int64_t>();
            job.tse = js["tse"].as<int64_t>();
            job.type = js["type"].as<uint8_t>();
            job.retried = js["retried"].as<int>();
            job.last_access = js["last_access"].as<int64_t>();
            found = true;
        }
        catch(exception &e) {
            rlogger->error("failed to load upload meta file: {}, {}", path, e.what());
            fs::remove(path);
        }
    }

    return found;
}

vector<upload_item_t> bulk_load_jobs(string path)
{
    vector<upload_item_t> ret;
    auto now = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
    const auto delta = (gConfigSystem.module.record.interval + kMaxWaitSecsForJob) * 1000;
    string fname;
    try {
        for (const auto & entry : fs::directory_iterator(path)) {
            fname = entry.path();
            if(entry.file_size() == 0 || !entry.is_regular_file()||entry.path().extension() != ".json") {
                rlogger->warn("bulk_load_jobs remove {} (empty/directory/!h264)",fname);
                fs::remove(fname);
                continue;
            }
            // auto fsTime = fs::last_write_time(fname);
            // auto cftime = decltype(fsTime)::clock::to_time_t(fsTime);
            // rlogger->info(" now {}, ft {}", now, cftime);
            // if(now - cftime < gConfigSystem.module.record.interval *2) {
            //     // maybe still in recording
            //     continue;
            // }
            if(ret.size() < kMaxJobLoad) {
                json k;
                try {
                    ifstream cfg_(fname);
                    k = json::parse(cfg_);
                    try {
                        auto tse = k["tse"].as<int64_t>();
                        auto last = k["last_access"].as<int64_t>();
                        auto retried = k["retried"].as<int>();
                        if(now - tse < delta || now - last < delta) {
                            rlogger->debug("skipped. upload meta contains videos still in recording or rescheduled interval: {}, {}", fname, k.to_string());
                        }
                        else {
                            upload_item_t item{k["tss"].as<int64_t>(), tse, k["type"].as<uint8_t>(), retried, last};
                            ret.push_back(item);
                            rlogger->info("loaded upload meta json: {}, {}", fname, k.to_string());
                        }
                    }
                    catch(exception &e) {
                        rlogger->error("failed process job: {}, {}", k.to_string(), e.what());
                        fs::remove(fname);
                    }
                }
                catch(exception &e) {
                    rlogger->error("failed to load upload meta json: {}, {}", fname, e.what());
                    fs::remove(fname);
                }
            }
        }
    }
    catch(exception &e) {
        rlogger->error("failed to load upload meta: {}", e.what());
    }

    return ret;
}

bool save_job(string path, json &job)
{
    bool success = true;
    try {
        ofstream ocfg_(path, std::ios::trunc);
        ocfg_ << pretty_print(job);
        ocfg_.flush();
        ocfg_.close();
    }
    catch(exception &e) {
        auto res = fmt::format("failed save job file {}: {}, {}", path, e.what(), job.to_string());
        rlogger->error(res);
        if(gConfigSystem.pClient) {
            MqttMgr::report_response_args(gConfigSystem.pClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, res, "record", "", json());
        }
        success = false;
    }

    return success;
}

void upload_svc_entry(thread_upload_args_t *pArg)
{
    string rc;
    system(fmt::format("mkdir -p {}", kUploadMetaDir).c_str());
    auto jobs = bulk_load_jobs(kUploadMetaDir);
    for(auto &k:jobs) {
        pArg->que->push(k);
    }

    while(1) {
        bool got = false;
        auto now = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
        // copy config
        string hostUrl = (*pArg->config)[consts::kMsgConfigUpload].as<string>();
        bool bUpload = false;
        if(gConfigSystem.module.ai.enabled  && gConfigSystem.module.motion.enabled) {
            bUpload = true;
        }
        if(bUpload) {
            rlogger->debug("waiting for job");
        }
        {
            unique_lock<mutex> lk(*pArg->mut);
            got = (*pArg->cv).wait_for(lk, chrono::seconds(kMaxWaitSecsForJob), [pArg] {return pArg->que->size()> 0;});
        }// lock released

        if(got && pArg->que->size()> 0) {
            rlogger->info("process one upload job, total left {}", pArg->que->size());
            auto elem = pArg->que->front();
            pArg->que->pop();

            if(!is_sdcard_avail()) {
                continue;
            }
            try {
                fs::create_directories(kUploadMetaDir);
            }
            catch(exception &e) {
                /// TODO: report issue
                auto str = fmt::format("failed access/create directory: {}, {}", kUploadMetaDir, e.what());
                rlogger->error(str);
                continue;
            }
            // save to file
            string jfilepath = kUploadMetaDir + to_string(elem.tss) + "-" + to_string(elem.tse) + ".json";
            if(elem.retried == -1 && has_existing_job(jfilepath, elem)) {
                rlogger->warn("already has scheduled job, wait for scheduling: {}-{}.json", elem.tss, elem.tse);
                continue;
            }

            json js;

            js["tss"] = elem.tss;
            js["tse"] = elem.tse;
            js["type"] = elem.type;
            js["last_access"] = now;
            js["retried"] = elem.retried + 1;
            if(elem.retried == -1) {
                save_job(jfilepath, js);
            }

            // find video files
            if(*(pArg->recList) == nullptr) {
                rlogger->error("recfile list is nullptr");
                continue;
            }

            int64_t offs = 0;
            int64_t offe = 0;
            int status = -2;
            auto res = (*pArg->recList)->findByRange(elem.tss, elem.tse, offs, offe,gConfigSystem.module.record.interval, &status);
            if(res.empty()) {
                if(status <0 ) {
                    rlogger->warn("video file not found, they were in the past: ({},{})", elem.tss, elem.tse);
                    fs::remove(jfilepath);
                }
                else if(status > 0) {
                    rlogger->warn("video files are in the futrue: ({},{})", elem.tss, elem.tse);
                }

                continue;
            }
            try {
                // do the work
                if(bUpload) {
                    rc = upload_video(hostUrl, pArg->dev_sn, elem.tss, elem.tse, res, elem.type);
                }
            }
            catch(exception &e) {
                rc = fmt::format("exception upload_video: {}", e.what());
            }
            if(!rc.empty()) {
                rlogger->error(rc);
                /// TODO: report and retry
                // update job
                save_job(jfilepath, js);
            }
            else {
                fs::remove(jfilepath);
                rlogger->info("successfully uploaded video: {}", js.to_string());
            }
        }
        else {
            // rlogger->info("no job to process");
        }

        // preventing sleep
        this_thread::sleep_for(chrono::seconds(5));

        // do extra work: handle "scheduled/failed jobs"
        // iterate directory, and find file create older than X ms
        unique_lock<mutex> lk(*pArg->mut);
        if(pArg->que->size() == 0) {
            auto jobs = bulk_load_jobs(kUploadMetaDir);
            rlogger->info("{} scheduled jobs loaded", jobs.size());
            for(auto &k:jobs) {
                pArg->que->push(k);
            }
        }
    }
}

void print_ts_files(OrderedList<int64_t> &_list)
{
    rlogger->info("print file ts {}:", _list.items().size());
    int cnt = 0;
    printf("\t");
    for(auto &k:_list.items()) {
        cnt++;
        printf("%lld, ", k);
        if(cnt %6 == 0) {
            printf("\n\t");
        }
    }
    rlogger->info("\nlist_videos done!");
}


void load_sd_files(OrderedList<int64_t> &_list)
{
    auto now = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
    try {
        string fname, baseName;
        for (const auto & entry : fs::directory_iterator(consts::recFilePath)) {
            fname = entry.path().c_str();
            if(entry.file_size() == 0 || !entry.is_regular_file()||entry.path().extension() != ".h264") {
                rlogger->warn("load_sd_files skipped {} (empty/directory/!h264)", entry.path().c_str());
                fs::remove(fname);
                continue;
            }

            int64_t ts = stoll(entry.path().stem().string());
            if(ts < consts::TS_2020) {
                /// TODO: files records when offline with no valid time got
                fs::remove(entry.path());
                rlogger->warn("remove records having invalid timestamp: {}", ts);
                continue;
            }
            _list.insert(ts);
        }
    }
    catch(exception &e) {
        rlogger->error("{}:{} load_sd_files exception : {}", __FILE__, __LINE__, e.what());
    }
}

void remove_ts_file(int64_t ts)
{
    string fname = consts::recFilePath + to_string(ts) + ".h264";
    fs::remove(fname);
}


/**
 * write to sd
 * if immediate is false, copy to heap buffer, dump heap to file when the heap if full
 * if immediate is true, directly write to file.
 * return: 0 - OK, -1: exception;
 */
static int cached_sd_write(fstream *fp, char *buf, int sz, bool immediate = false)
{
    int ret = 0;
    static char * heap_buff = (char*)malloc(kMaxHeapSize *3);
    static int current_heap_size = 0;

    try {
        if(!immediate) {
            if(current_heap_size + sz > kMaxHeapSize) {
                rlogger->info("write full cached buffer to recording file. size {}", current_heap_size);
                fp->write(heap_buff, current_heap_size);
                current_heap_size = 0;
            }

            memcpy(heap_buff + current_heap_size, buf, sz);
            current_heap_size += sz;
        }
        else {
            if(current_heap_size > 0) {
                rlogger->info("write cached buffer to recording file. size {}", current_heap_size);
                fp->write(heap_buff, current_heap_size);
                current_heap_size = 0;
            }
            rlogger->info("write to recording file. size {}", sz);
            fp->write(buf, sz);
        }
    }
    catch(exception &e) {
        auto res = fmt::format("failed to write sd: {}", e.what());
        rlogger->error(res);
        if(gConfigSystem.pClient) {
            MqttMgr::report_response_args(gConfigSystem.pClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, res, "record", "", json());
        }
    }

    return ret;
}

extern XM_S32 LibXmMaQue_VideoEnc_forceIFrame(XM_S32 channel, MaQueStreamChannel_e eStreamChn);

void record_video_entry(ev_module_config_t *pArgs)
{
    int ret = 0;
    // this thread will never exit
    while(1) {
        if(*(pArgs->pPubCtx) == nullptr) {
            rlogger->error("zmq is not started up");
            this_thread::sleep_for(10s);
            continue;
        }

        if(!is_sdcard_avail()) {
            rlogger->error("no sd card valial. can't record local videos");
            this_thread::sleep_for(10s);
            continue;
        }
        uint64_t total = 0, avail = 0;
        if(!get_sdcard_megabytes(total, avail)) {
            rlogger->error("failed to get sd card size");
            continue;
        }

        rlogger->info("sd card: {}MB total, {}MB free", total, avail);
        // calc num of slices
        // reserve 0.5GB space
        // multipled by 1.5 because videos take less space in midnight (variable bitrate)
        ssize_t maxSlices = (ssize_t)((total - 512)/10*1.5);
        gRecFilesList = new OrderedList<int64_t>(maxSlices, remove_ts_file);
        // load existing videos
        load_sd_files(*gRecFilesList);

        // setup sub
        void *pSub = zmq_socket(*(pArgs->pPubCtx), ZMQ_SUB);
        // int onlyOne = 1;
        // zmq_setsockopt(pSub, ZMQ_CONFLATE, &onlyOne, sizeof(onlyOne));
        // if(ret < 0) {
        //     rlogger->error("failed to set ZMQ_CONFLATE on record {}", zmq_strerror(ret));
        // }
        // ret = 10;
        // ret = zmq_setsockopt(pSub, ZMQ_RCVHWM, &ret, sizeof(ret));
        // if(ret < 0) {
        //     rlogger->error("failed to set ZMQ_RCVHWM on record sub {}", zmq_strerror(zmq_errno()));
        // }
        ret = zmq_connect(pSub, consts::strPubUrl.c_str());
        if(ret != 0) {
            rlogger->error("failed connect frame pub: {}", consts::strPubUrl);
            exit(1);
        }
        ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
        rlogger->info("zmq sub successed (record)");

        zmq_msg_t msg;
        int64_t preTs = 0;
        int64_t nowTs = 0;
        uint32_t cnt = 0;
        fstream * fp = nullptr;
        pArgs->isRecordRunning = true;
        LibXmMaQue_VideoEnc_forceIFrame(0, (MaQueStreamChannel_e) 0);
        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(ret));
                continue;
            }
            ssize_t sz = zmq_msg_size(&msg);
            char *ptr = (char*)zmq_msg_data(&msg);
            /// we calc timestamps about every 2s
            if(cnt % 30 == 0) {
                nowTs = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
            }
            if(nowTs - preTs > pArgs->module.record.interval *1000) {
                if(fp != nullptr) {
                    if(fp->is_open()) {
                        cached_sd_write(fp, ptr, sz, true);
                        fp->close();
                        delete fp;
                        fp = nullptr;
                    }
                }
                fp = new fstream(consts::recFilePath+to_string(nowTs) + ".h264", std::ios::out|std::ios::binary);
                if(!fp->is_open()) {
                    auto res = "failed to create record video file";
                    rlogger->error(res);
                    if(gConfigSystem.pClient) {
                        MqttMgr::report_response_args(gConfigSystem.pClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, res, "record", "", json());
                    }
                    delete fp;
                    fp = nullptr;
                }
                else {
                    gRecFilesList->insert(nowTs);
                }
                preTs = nowTs;

                ///
                LibXmMaQue_VideoEnc_forceIFrame(0, (MaQueStreamChannel_e) 0);
                /// next for new file
                continue;
            }

            if(fp != nullptr) {
                /// we had packed it with leading evpacket_t, ignore before saving files
#ifdef RAW_FORMAT
                cached_sd_write(fp, ptr, sz, false);
#else
                fp->write(ptr + sizeof(evpacket_t), sz - sizeof(evpacket_t));
#endif
            }
            cnt++;
            zmq_msg_close(&msg);
        }
        if(nullptr != fp) {
            fp->close();
            delete fp;
        }
        zmq_close (pSub);
        /// DO NOT! do this
        //zmq_ctx_destroy(pPubCtx);
    }
}