#include <chrono>
#include <string>
#include <mxl/flow.h>
#include <mxl/time.h>
#include "NodeBase.hpp"

class FakeSinkNode : public dmf::NodeBase {
    void run() override {
        const auto flow_info = config().at("video_flow_id");
        const auto flow_id   = flow_info.at("id").get<std::string>();
        const int  fps_num   = flow_info.value("fps_num", 25);
        const int  fps_den   = flow_info.value("fps_den", 1);

        log("flow=%s", flow_id.c_str());

        log("waiting for flow to become active...");
        bool active = false;
        while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
            mxlIsFlowActive(instance(), flow_id.c_str(), &active);
            if (!active) mxlSleepForNs(100'000'000);
        }
        if (!dmf::g_running) return;
        log("flow active — starting read");

        mxlFlowReader reader{};
        mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &reader);
        if (st != MXL_STATUS_OK) {
            log("mxlCreateFlowReader failed (%s)", dmf::mxl_status_str(st));
            return;
        }

        const mxlRational rate = {fps_num, fps_den};

        uint64_t index          = mxlGetCurrentIndex(&rate);
        uint64_t frame_count    = 0;
        uint64_t invalid_count  = 0;
        uint64_t late_count     = 0;
        auto     wall_start     = std::chrono::steady_clock::now();
        auto     last_log_time  = wall_start;

        while (dmf::g_running.load(std::memory_order_relaxed)) {
            mxlGrainInfo grain{};
            uint8_t*     buf = nullptr;

            st = mxlFlowReaderGetGrainNonBlocking(reader, index, &grain, &buf);

            if (st == MXL_STATUS_OK) {
                frame_count++;
                if (grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
                index++;

            } else if (st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
                mxlSleepForNs(1'000'000);  // 1 ms poll

            } else if (st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
                late_count++;
                // Jump to the most recent frame in the ring buffer
                mxlFlowRuntimeInfo ri{};
                mxlFlowReaderGetRuntimeInfo(reader, &ri);
                index = ri.headIndex;

            } else if (st == MXL_ERR_FLOW_INVALID) {
                log("flow invalidated — reconnecting...");
                if (reader) mxlReleaseFlowReader(instance(), reader);
                reader = nullptr;
                mxlSleepForNs(100'000'000);
                if (mxlCreateFlowReader(instance(), flow_id.c_str(), "", &reader) == MXL_STATUS_OK) {
                    log("flow reconnected");
                    index = mxlGetCurrentIndex(&rate);
                }

            } else {
                log("unexpected status (%s) on index=%llu", dmf::mxl_status_str(st), index);
                break;
            }

            // Log stats every second (wall clock) — avoid per-frame fprintf
            auto now = std::chrono::steady_clock::now();
            if (std::chrono::duration<double>(now - last_log_time).count() >= 1.0) {
                const double elapsed = std::chrono::duration<double>(now - wall_start).count();
                log("frames=%llu  invalid=%llu  late=%llu  avg=%.2f fps",
                    frame_count, invalid_count, late_count,
                    static_cast<double>(frame_count) / elapsed);
                last_log_time = now;
            }
        }

        log("stopped — total frames=%llu  invalid=%llu  late=%llu",
            frame_count, invalid_count, late_count);
        if (reader) mxlReleaseFlowReader(instance(), reader);
    }
};

int main() {
    FakeSinkNode node;
    return node.execute();
}