#pragma once
#include <algorithm>
#include <array>
#include <cstdint>
#include <cstring>
#include <cmath>
#include <memory>
#include <memory_resource>
#include <optional>
#include <rigtorp/MPMCQueue.h>
#include <span>
#include <system_error>
#include <utility>
#include <vector>
#include "../Axis/Frequency_Axis_p.h"
#include "../Axis/Time_Axis_p.h"
#include "../architecture/Render_Time.h"
#include "../architecture/Renderable.h"
#include "../base/Memory.h"
#include "../base/global.h"
#include "../render/Canvas.h"
#include "../render/Image.h"
#include "../render/Render_Frame_Snapshot.h"
#include "Psc_Cpp_Core/Base/RingBuffer.hpp"
#include "Interpolation_p.h"
#include "Waterfall.h"
namespace renderive {
struct Waterfall_Data {
    int tick{};
    std::pmr::vector<double> frequency{memory_resource(Memory_Domain::Waterfall)};
    std::shared_ptr<Update_State> completion_state;
};
struct Waterfall_Input_Data : Input_Data {};
struct Waterfall_Render_State : Render_State, Hover_Tooltip_State, Waterfall_Prop {};
struct Waterfall_Ring_Buffer {
    Psc::StreamRingBuffer_ST buffer{memory_resource(Memory_Domain::Waterfall)};
    std::pmr::vector<std::uint8_t> record_buffer{memory_resource(Memory_Domain::Waterfall)};
    std::pmr::vector<std::uint8_t> snapshot_buffer{memory_resource(Memory_Domain::Waterfall)};
    std::pmr::vector<double> row_buffer{memory_resource(Memory_Domain::Waterfall)};
    std::size_t record_byte_size{};
    std::size_t row_byte_size{};
    int row_count{};
    int col_count{};
    int data_count{};
    bool match(int col, int row) const {
        return col_count == col && row_count == row && record_byte_size;
    }
    void resize(int col, int row) {
        col_count = col;
        row_count = row;
        data_count = 0;
        row_byte_size = sizeof(double) * col_count;
        record_byte_size = row_byte_size + sizeof(int);
        record_buffer.resize(record_byte_size);
        snapshot_buffer.resize(record_byte_size * row_count);
        row_buffer.resize(col_count);
        buffer.init(record_byte_size * row_count);
    }
    void push_data(const double* data, int tick) {
        if (data_count == row_count) {
            buffer.skip(record_byte_size);
            --data_count;
        }
        std::memcpy(record_buffer.data(), data, row_byte_size);
        std::memcpy(record_buffer.data() + row_byte_size, &tick, sizeof(tick));
        buffer.write(record_buffer.data(), record_byte_size);
        ++data_count;
    }
    int snapshot() {
        buffer.peek_best_effort(snapshot_buffer.data(), record_byte_size * data_count);
        return data_count;
    }
    const double* frequency_data(int row, int first_col, int last_col) {
        std::size_t offset = static_cast<std::size_t>(first_col) * sizeof(double);
        std::size_t size = static_cast<std::size_t>(last_col - first_col + 1) * sizeof(double);
        std::memcpy(reinterpret_cast<std::uint8_t*>(row_buffer.data()) + offset, snapshot_buffer.data() + row * record_byte_size + offset, size);
        return row_buffer.data();
    }
    const double* frequency_data(int row) {
        std::memcpy(row_buffer.data(), snapshot_buffer.data() + row * record_byte_size, row_byte_size);
        return row_buffer.data();
    }
    int tick(int row) const {
        int value;
        std::memcpy(&value, snapshot_buffer.data() + row * record_byte_size + row_byte_size, sizeof(value));
        return value;
    }
};
struct Waterfall_Image_Cache_Key {
    int start_tick{};
    int first_col{};
    int last_col{};
    Range frequency_range{};
    Range power_range{};
    Range visible_range{};
    Color background;
    std::uint64_t color_map_hash{};
    int width{};
    int height{};
    bool operator==(const Waterfall_Image_Cache_Key& other) const {
        return start_tick == other.start_tick &&
        first_col == other.first_col &&
        last_col == other.last_col &&
        frequency_range == other.frequency_range &&
        power_range == other.power_range &&
        visible_range == other.visible_range &&
        background == other.background &&
        color_map_hash == other.color_map_hash &&
        width == other.width &&
        height == other.height;
    }
};
struct Waterfall_Image_Snapshot {
    Waterfall_Image_Snapshot() : update_states(memory_resource(Memory_Domain::Update_Completion)) {}
    Waterfall_Image_Cache_Key key;
    Image image;
    Range frequency_range{};
    Range time_range{};
    RectF source_rect{};
    std::pmr::vector<std::shared_ptr<Update_State>> update_states;
};
struct Waterfall_Image_Request {
    Waterfall_Image_Request()
        : rows(memory_resource(Memory_Domain::Waterfall)),
          update_states(memory_resource(Memory_Domain::Update_Completion)) {}
    Waterfall_Image_Cache_Key key;
    Color_Map color_map;
    double value_start_coord{};
    double rate{};
    int lower_stream_tick{};
    int upper_stream_tick{};
    int visible_time_count{};
    bool newest_at_axis_start{};
    Range draw_frequency_range{};
    Range draw_time_range{};
    RectF draw_source_rect{};
    std::pmr::vector<Waterfall_Data> rows;
    std::pmr::vector<std::shared_ptr<Update_State>> update_states;
};
struct Waterfall_Image_Tile {
    int x{};
    int y{};
    Image image;
};
struct Waterfall_Image_Job {
    explicit Waterfall_Image_Job(std::shared_ptr<Waterfall_Image_Request> request, int columns, int rows)
        : request(std::move(request)), tile_columns(columns), tile_rows(rows), remaining(columns * rows) {
        tiles.resize(static_cast<std::size_t>(columns * rows));
    }
    std::shared_ptr<Waterfall_Image_Request> request;
    int tile_columns{};
    int tile_rows{};
    std::vector<Waterfall_Image_Tile> tiles;
    std::atomic<int> remaining;
};
static void complete_waterfall_request_states(Waterfall_Image_Request& request, std::error_code error, Update_Outcome outcome) {
    for (auto& state : request.update_states) {
        if (state)
            state->complete_all(error, outcome);
    }
    request.update_states.clear();
}
static void complete_waterfall_result_states(Waterfall_Image_Snapshot& snapshot, std::error_code error, Update_Outcome outcome) {
    for (const auto& state : snapshot.update_states) {
        if (state)
            state->complete_all(error, outcome);
    }
    snapshot.update_states.clear();
}
static bool waterfall_request_coord_by_stream_tick(const Waterfall_Image_Request& request, int stream_tick, int& coord) {
    if (request.visible_time_count <= 0)
        return false;
    if (stream_tick < request.lower_stream_tick || stream_tick > request.upper_stream_tick)
        return false;
    if (request.newest_at_axis_start)
        coord = request.key.start_tick + (request.upper_stream_tick - stream_tick);
    else
        coord = request.key.start_tick + (request.key.height - request.visible_time_count) + (stream_tick - request.lower_stream_tick);
    return coord >= request.key.start_tick && coord < request.key.start_tick + request.key.height;
}
static int waterfall_tile_column_count(int width) {
    static constexpr int Tile_Width = 128;
    return std::max(1, (width + Tile_Width - 1) / Tile_Width);
}
static int waterfall_tile_row_count(int height) {
    static constexpr int Tile_Height = 64;
    return std::max(1, (height + Tile_Height - 1) / Tile_Height);
}
static void write_waterfall_tile(Waterfall_Image_Job& job, int tile_index) {
    const Waterfall_Image_Request& request = *job.request;
    int tile_x = tile_index % job.tile_columns;
    int tile_y = tile_index / job.tile_columns;
    int x_begin = request.key.width * tile_x / job.tile_columns;
    int x_end = request.key.width * (tile_x + 1) / job.tile_columns;
    int y_begin = request.key.height * tile_y / job.tile_rows;
    int y_end = request.key.height * (tile_y + 1) / job.tile_rows;
    Waterfall_Image_Tile& tile = job.tiles[static_cast<std::size_t>(tile_index)];
    tile.x = x_begin;
    tile.y = y_begin;
    int tile_width = std::max(0, x_end - x_begin);
    int tile_height = std::max(0, y_end - y_begin);
    tile.image.resize(tile_width, tile_height);
    tile.image.fill(request.key.background);
    for (const Waterfall_Data& row : request.rows) {
        int coord{};
        if (!waterfall_request_coord_by_stream_tick(request, row.tick, coord))
            continue;
        int image_y = coord - request.key.start_tick;
        if (image_y < y_begin || image_y >= y_end)
            continue;
        Pixel* line = tile.image.row(image_y - y_begin);
        if (!line)
            continue;
        int first_col = std::max(request.key.first_col, x_begin);
        int last_col = std::min(request.key.last_col, x_end - 1);
        for (int col = first_col; col <= last_col; ++col) {
            std::size_t source_col = static_cast<std::size_t>(col - request.key.first_col);
            if (source_col >= row.frequency.size())
                continue;
            int color_offset = static_cast<int>((row.frequency[source_col] - request.value_start_coord) * request.rate);
            line[col - x_begin] = request.color_map.at_offset(color_offset);
        }
    }
}
static void write_waterfall_tiles(Waterfall_Image_Job& job) {
    int tile_count = static_cast<int>(job.tiles.size());
    for (int tile_index = 0; tile_index < tile_count; ++tile_index)
        write_waterfall_tile(job, tile_index);
}
static Waterfall_Image_Snapshot build_waterfall_image_snapshot(const std::shared_ptr<Waterfall_Image_Request>& request_ptr) {
    Waterfall_Image_Request& request = *request_ptr;
    Waterfall_Image_Job job(
        request_ptr,
        waterfall_tile_column_count(request.key.width),
        waterfall_tile_row_count(request.key.height));
    write_waterfall_tiles(job);
    Waterfall_Image_Snapshot snapshot;
    snapshot.key = request.key;
    snapshot.frequency_range = request.draw_frequency_range;
    snapshot.time_range = request.draw_time_range;
    snapshot.source_rect = request.draw_source_rect;
    snapshot.image.resize(request.key.width, request.key.height);
    for (const auto& tile : job.tiles) {
        for (int y = 0; y < tile.image.height(); ++y) {
            Pixel* target = snapshot.image.row(tile.y + y);
            const Pixel* source = tile.image.row(y);
            if (target && source)
                std::memcpy(target + tile.x, source, static_cast<std::size_t>(tile.image.width()) * sizeof(Pixel));
        }
    }
    snapshot.update_states = std::move(request.update_states);
    return snapshot;
}
struct Waterfall_Private : Typed_Render_Data<Waterfall, Waterfall_Render_State, Waterfall_Input_Data>, Hit_Testable, Hover_Interactive {
    static constexpr std::size_t Row_Queue_Capacity = 256;
    Waterfall_Ring_Buffer ring_buffer;
    std::weak_ptr<Color_Bar> color_bar;
    std::optional<Waterfall_Data> pending_latest_row;
    std::pmr::vector<std::shared_ptr<Update_State>> pending_image_update_states{memory_resource(Memory_Domain::Update_Completion)};
    rigtorp::MPMCQueue<std::shared_ptr<Waterfall_Data>> row_queue;
    std::atomic_uint64_t dropped_row_count{0};
    std::uint64_t last_line_commit_ns{};
    std::optional<Waterfall_Image_Snapshot> image_snapshot;
    std::optional<Waterfall_Image_Cache_Key> submitted_image_key;
    std::uint64_t pending_latest_deadline_ns{};
    Waterfall_Private()
        : row_queue(Row_Queue_Capacity) {}
    bool select_test(const PointF& pos) override {
        return true;
    }
    void set_hover_state(Point pos, bool active) override {
        Render_Edit_Lease<Waterfall_Private, Waterfall_Render_State> edit(this);
        edit->hover_position = pos;
        edit->hover_info_active = active;
    }
    ~Waterfall_Private() override {
        if (image_snapshot)
            complete_waterfall_result_states(*image_snapshot, std::make_error_code(std::errc::operation_canceled), Update_Outcome::Cancelled);
        std::shared_ptr<Waterfall_Data> row;
        while (row_queue.try_pop(row)) {
            if (row && row->completion_state)
                row->completion_state->complete_all(std::make_error_code(std::errc::operation_canceled), Update_Outcome::Cancelled);
        }
        if (pending_latest_row && pending_latest_row->completion_state)
            pending_latest_row->completion_state->complete_all(std::make_error_code(std::errc::operation_canceled), Update_Outcome::Cancelled);
        for (auto& state : pending_image_update_states) {
            if (state)
                state->complete_all(std::make_error_code(std::errc::operation_canceled), Update_Outcome::Cancelled);
        }
        pending_image_update_states.clear();
    }
    void enqueue_row(int tick, std::span<const double> data, std::shared_ptr<Update_State> completion_state = {}) {
        auto row = std::make_shared<Waterfall_Data>();
        row->tick = tick;
        row->frequency.assign(data.begin(), data.end());
        row->completion_state = std::move(completion_state);
        enqueue_row(std::move(row));
    }
    void enqueue_row(int tick, std::pmr::vector<double>&& data, std::shared_ptr<Update_State> completion_state = {}) {
        auto row = std::make_shared<Waterfall_Data>();
        row->tick = tick;
        row->frequency = std::move(data);
        row->completion_state = std::move(completion_state);
        enqueue_row(std::move(row));
    }
    void enqueue_row(std::shared_ptr<Waterfall_Data> row) {
        if (!row)
            return;
        if (!row_queue.try_push(row)) {
            if (row->completion_state)
                row->completion_state->complete_all({}, Update_Outcome::Superseded);
            dropped_row_count.fetch_add(1, std::memory_order_relaxed);
            return;
        }
        if (row->completion_state)
            row->completion_state->complete_until(Update_Stage::Input_Released);
        if (q())
            q()->mark_render_dirty();
    }
    void consume_row_queue(std::pmr::vector<Waterfall_Data>& output) {
        std::shared_ptr<Waterfall_Data> row;
        while (row_queue.try_pop(row)) {
            if (!row)
                continue;
            output.push_back(std::move(*row));
        }
    }
    void commit_row_ticket(Waterfall_Data& row) {
        if (!row.completion_state)
            return;
        row.completion_state->complete_until(Update_Stage::Committed);
        pending_image_update_states.push_back(std::move(row.completion_state));
    }
    void schedule_pending_latest_deadline(std::uint64_t deadline_ns) {
        if (!deadline_ns || pending_latest_deadline_ns == deadline_ns)
            return;
        pending_latest_deadline_ns = deadline_ns;
        std::weak_ptr<Renderable> owner = q() ? std::weak_ptr<Renderable>(q()->shared_from_this()) : std::weak_ptr<Renderable>();
        Global::instance()->render_scheduler().post_at(deadline_ns, [owner]() {
            if (auto renderable = owner.lock())
                renderable->mark_render_dirty();
        });
    }
    bool push_row(std::span<const double> frequency_data, int tick, int col_count) {
        if (frequency_data.size() != col_count) {
            return false;
        }
        ring_buffer.push_data(frequency_data.data(), tick);
        return true;
    }
    bool push_owned_row(std::pmr::vector<double>& frequency_data, int tick, int col_count, Waterfall_Row_Update_Policy update_policy) {
        if (frequency_data.size() != col_count) {
            return false;
        }
        ring_buffer.push_data(frequency_data.data(), tick);
        (void)update_policy;
        return true;
    }
    void prepare_data(const Render_Frame_Snapshot& snapshot) override {
        sync_state_pipeline();
        const Waterfall_Render_State* s = render_state();
        auto time_axis = s->time_axis.lock();
        if (!time_axis)
            return;
        auto timeline = Time_Axis_Render_Access::capture_timeline(time_axis.get(), snapshot);
        if (!timeline)
            return;
        int col_count = s->frequency_bin_count, row_count = timeline->visible_time_point_count;
        if (col_count <= 0 || row_count <= 0)
            return;
        if (!ring_buffer.match(col_count, row_count)) {
            ring_buffer.resize(col_count, row_count);
        }
        std::pmr::vector<Waterfall_Data> queued_rows(memory_resource(Memory_Domain::Waterfall));
        consume_row_queue(queued_rows);
        switch (s->row_update_policy) {
        case Waterfall_Row_Update_Policy::All_Pending:
            for (auto& row : queued_rows) {
                if (push_row(row.frequency, row.tick, col_count))
                    commit_row_ticket(row);
                else if (row.completion_state)
                    row.completion_state->complete_all(std::make_error_code(std::errc::invalid_argument), Update_Outcome::Cancelled);
            }
            break;
        case Waterfall_Row_Update_Policy::Latest_Only:
        case Waterfall_Row_Update_Policy::Rate_Limited_Latest: {
            for (auto& row : queued_rows) {
                if (row.frequency.size() != col_count) {
                    if (row.completion_state)
                        row.completion_state->complete_all(std::make_error_code(std::errc::invalid_argument), Update_Outcome::Cancelled);
                    continue;
                }
                if (pending_latest_row && pending_latest_row->completion_state)
                    pending_latest_row->completion_state->complete_all({}, Update_Outcome::Superseded);
                pending_latest_row = std::move(row);
            }
            if (pending_latest_row && pending_latest_row->frequency.size() != col_count) {
                if (pending_latest_row->completion_state)
                    pending_latest_row->completion_state->complete_all(std::make_error_code(std::errc::invalid_argument), Update_Outcome::Cancelled);
                pending_latest_row.reset();
            }
            if (pending_latest_row) {
                bool allow_commit = true;
                std::uint64_t now_ns = steady_now_ns();
                std::uint64_t deadline_ns = 0;
                if (s->row_update_policy == Waterfall_Row_Update_Policy::Rate_Limited_Latest && last_line_commit_ns != 0) {
                    std::uint64_t interval_ns = static_cast<std::uint64_t>(std::max(1, s->latest_row_min_interval_ms)) * 1000000ull;
                    deadline_ns = last_line_commit_ns + interval_ns;
                    allow_commit = now_ns >= deadline_ns;
                }
                if (allow_commit) {
                    if (push_owned_row(pending_latest_row->frequency, pending_latest_row->tick, col_count, s->row_update_policy)) {
                        commit_row_ticket(*pending_latest_row);
                        last_line_commit_ns = now_ns;
                        pending_latest_deadline_ns = 0;
                        pending_latest_row.reset();
                    }
                }
                else {
                    schedule_pending_latest_deadline(deadline_ns);
                }
            }
            break;
        }
        }
        update_image_item(snapshot, *timeline);
    }
    void update_image_item(const Render_Frame_Snapshot& snapshot, const Timeline_Stream_Snapshot& timeline) {
        Waterfall_Render_State* s = render_state();
        auto frequency_axis = s->frequency_axis.lock();
        auto time_axis = s->time_axis.lock();
        if (!frequency_axis || !time_axis)
            return;
        auto color_scale = s->color_scale.snapshot();
        Abs_Axis* h_axis = frequency_axis.get();
        Range h_range = s->frequency_range;
        Range v_range{static_cast<double>(timeline.coordinate_begin), static_cast<double>(timeline.coordinate_begin + timeline.visible_time_point_count)};
        if (s->visible_range_only && !intersect_range(s->frequency_range, Axis_Render_Access::coord_range(h_axis), h_range)) {
            image_snapshot.reset();
            return;
        }
        int col_count = s->frequency_bin_count, row_count = timeline.visible_time_point_count;
        if (col_count <= 0 || row_count <= 0 || s->frequency_range.length() == 0.0 || color_scale->range.size() == 0.0)
            return;
        double source_start = static_cast<double>(col_count) * (h_range.origin - s->frequency_range.origin) / s->frequency_range.length();
        double source_end = static_cast<double>(col_count) * (h_range.target - s->frequency_range.origin) / s->frequency_range.length();
        double source_min = std::min(source_start, source_end);
        double source_max = std::max(source_start, source_end);
        int first_col = std::clamp(static_cast<int>(std::floor(source_min)), 0, col_count - 1);
        int last_col = std::clamp(static_cast<int>(std::ceil(source_max)) - 1, first_col, col_count - 1);
        double value_start_coord = color_scale->range.origin;
        double rate = static_cast<double>(color_scale->color_map.size()) / color_scale->range.size();
        int buffer_data_count = ring_buffer.snapshot();
        int start_tick = timeline.coordinate_begin;
        std::uint64_t color_map_hash = color_scale->color_map.hash();
        Waterfall_Image_Cache_Key key{
            start_tick,
            first_col,
            last_col,
            s->frequency_range,
            color_scale->range,
            h_range,
            snapshot.background_color,
            color_map_hash,
            col_count,
            row_count
        };
        bool should_submit = !submitted_image_key || !(*submitted_image_key == key) || !pending_image_update_states.empty();
        if (should_submit) {
            submitted_image_key = key;
            auto request = std::make_shared<Waterfall_Image_Request>();
            request->key = key;
            request->color_map = color_scale->color_map;
            request->value_start_coord = value_start_coord;
            request->rate = rate;
            request->lower_stream_tick = timeline.first_stream_tick;
            request->upper_stream_tick = timeline.last_stream_tick;
            request->visible_time_count = static_cast<int>(timeline.times.size());
            request->newest_at_axis_start = timeline.newest_at_axis_start;
            request->draw_frequency_range = h_range;
            request->draw_time_range = v_range;
            request->draw_source_rect = RectF(source_start, 0.0, source_end - source_start, static_cast<double>(row_count));
            request->rows.reserve(static_cast<std::size_t>(buffer_data_count));
            for (int row = 0; row < buffer_data_count; ++row) {
                int buffer_row = buffer_data_count - row - 1;
                int stream_tick = ring_buffer.tick(buffer_row);
                int image_coord{};
                if (!waterfall_request_coord_by_stream_tick(*request, stream_tick, image_coord))
                    continue;
                const double* list = ring_buffer.frequency_data(buffer_row, first_col, last_col);
                Waterfall_Data image_row;
                image_row.tick = stream_tick;
                image_row.frequency.assign(list + first_col, list + last_col + 1);
                request->rows.push_back(std::move(image_row));
            }
            request->update_states = std::move(pending_image_update_states);
            pending_image_update_states = std::pmr::vector<std::shared_ptr<Update_State>>(memory_resource(Memory_Domain::Update_Completion));
            image_snapshot = build_waterfall_image_snapshot(request);
        }
    }
    void draw_image(Canvas& canvas, Waterfall_Render_State* s, Abs_Axis* h_axis, Time_Axis* v_axis) {
        if (!image_snapshot || image_snapshot->image.empty())
            return;
        if (image_snapshot->frequency_range.length() == 0.0 || image_snapshot->time_range.length() == 0.0)
            return;
        if (image_snapshot->source_rect.width == 0.0 || image_snapshot->source_rect.height == 0.0)
            return;
        Axis_Mapping_2D mapping = Axis_Render_Access::mapping(h_axis, v_axis);
        Axis_Basis_2D basis = Axis_Basis_2D::from_ranges(mapping, image_snapshot->frequency_range, image_snapshot->time_range);
        canvas.save();
        canvas.set_image_interpolation(s->interpolation_mode);
        canvas.transform(
            basis.domain_vector.x / image_snapshot->source_rect.width,
            basis.domain_vector.y / image_snapshot->source_rect.width,
            basis.value_vector.x / image_snapshot->source_rect.height,
            basis.value_vector.y / image_snapshot->source_rect.height,
            basis.origin.x,
            basis.origin.y);
        canvas.draw_image(RectF{0.0, 0.0, image_snapshot->source_rect.width, image_snapshot->source_rect.height}, image_snapshot->image, image_snapshot->source_rect);
        canvas.restore();
    }
    void draw(Canvas& canvas, const Render_Frame_Snapshot& snapshot) override {
        Waterfall_Render_State* s = render_state();
        auto frequency_axis = s->frequency_axis.lock();
        auto time_axis = s->time_axis.lock();
        if (!frequency_axis || !time_axis)
            return;
        if (snapshot.frame_update_states && image_snapshot) {
            for (const auto& state : image_snapshot->update_states)
                snapshot.frame_update_states->push_back(state);
            image_snapshot->update_states.clear();
        }
        Abs_Axis* h_axis = frequency_axis.get();
        Time_Axis* v_axis = time_axis.get();
        draw_image(canvas, s, h_axis, v_axis);
        if (q()->hover_ok(q())) {
            q()->draw_hover_tooltip(&canvas, h_axis, v_axis);
        }
    }
};
} // namespace renderive