#include "Web_Server.hpp"
#include "Gallery_Video_Stream.hpp"
#include "Gallery_WebSocket.hpp"
#include "Graph_WebSocket.hpp"
#include "Gallery_Plots.hpp"
#include <drogon/drogon.h>
#include <nlohmann/json.hpp>
#include <algorithm>
#include <cstddef>
#include <functional>
#include <iterator>
#include <memory>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>

namespace aethera::web {
namespace {
using Plot_Map = std::unordered_map<std::string, std::shared_ptr<Plot>>;
using Gallery_Stream_Map =
    std::unordered_map<std::string, std::shared_ptr<Gallery_Video_Stream>>;

drogon::HttpResponsePtr json_response(nlohmann::json value) {
    auto response = drogon::HttpResponse::newHttpResponse();
    response->setContentTypeCode(drogon::CT_APPLICATION_JSON);
    response->setBody(value.dump());
    return response;
}

drogon::HttpResponsePtr error_response(drogon::HttpStatusCode status, std::string message) {
    auto response = json_response({{"error", std::move(message)}});
    response->setStatusCode(status);
    return response;
}

std::shared_ptr<Plot> find_plot(const Plot_Map& plots, std::string_view id) {
    const auto found = plots.find(std::string(id));
    return found == plots.end() ? nullptr : found->second;
}

nlohmann::json taskflow_runtime_json() {
    const auto state = aethera::task_runtime_state();
    nlohmann::json workers = nlohmann::json::array();
    for (const auto& worker : state.workers)
        workers.push_back({
            {"id", worker.id}, {"task_count", worker.task_count},
            {"current_queue_size", worker.current_queue_size},
            {"current_queue_capacity", worker.current_queue_capacity},
            {"peak_queue_size", worker.peak_observed_queue_size},
            {"max_queue_capacity", worker.max_observed_queue_capacity},
            {"active_task", {{"native_id", std::to_string(worker.active_task_hash)},
                              {"type", worker.active_task_type},
                              {"time_ns", worker.active_task_time_ns}}},
            {"task_time_ns", worker.task_time_ns},
            {"busy_time_ns", worker.busy_time_ns},
            {"cpu_time_ns", worker.cpu_time_ns},
            {"non_cpu_time_ns", worker.non_cpu_time_ns},
            {"idle_time_ns", worker.idle_time_ns},
            {"min_task_time_ns", worker.min_task_time_ns},
            {"max_task_time_ns", worker.max_task_time_ns},
            {"utilization", worker.utilization},
            {"cpu_utilization", worker.cpu_utilization}});
    nlohmann::json task_types = nlohmann::json::array();
    for (const auto& type : state.task_types)
        task_types.push_back({
            {"name", type.name}, {"count", type.count},
            {"total_time_ns", type.total_time_ns},
            {"min_time_ns", type.min_time_ns},
            {"max_time_ns", type.max_time_ns}});
    return {
        {"protocol", "aethera.taskflow.runtime"}, {"version", 1},
        {"worker_count", state.worker_count},
        {"active_topologies", state.active_topology_count},
        {"active_taskflows", state.active_taskflow_count},
        {"peak_active_taskflows", state.peak_active_taskflow_count},
        {"completed_taskflows", state.completed_taskflow_count},
        {"failed_taskflows", state.failed_taskflow_count},
        {"active_tasks", state.active_task_count},
        {"peak_active_tasks", state.peak_active_task_count},
        {"active_workers", state.active_worker_count},
        {"peak_active_workers", state.peak_active_worker_count},
        {"observed_tasks", state.observed_task_count},
        {"named_tasks", state.named_task_count},
        {"peak_worker_queue_size", state.peak_observed_worker_queue_size},
        {"max_worker_queue_capacity", state.max_observed_worker_queue_capacity},
        {"max_predecessors", state.max_predecessors},
        {"max_successors", state.max_successors},
        {"max_strong_dependencies", state.max_strong_dependencies},
        {"max_weak_dependencies", state.max_weak_dependencies},
        {"longest_task", {{"native_id", std::to_string(state.longest_task_hash)},
                           {"name", state.longest_task_name},
                           {"type", state.longest_task_type},
                           {"time_ns", state.longest_task_time_ns}}},
        {"total_task_time_ns", state.total_task_time_ns},
        {"worker_busy_time_ns", state.worker_busy_time_ns},
        {"worker_cpu_time_ns", state.worker_cpu_time_ns},
        {"observed_wall_time_ns", state.observed_wall_time_ns},
        {"worker_utilization", state.worker_utilization},
        {"worker_cpu_utilization", state.worker_cpu_utilization},
        {"task_types", std::move(task_types)},
        {"workers", std::move(workers)}};
}

bool taskflow_graph_contains_gallery_media(const nlohmann::json& graph) {
    if (!graph.is_object() || !graph.contains("nodes") ||
        !graph["nodes"].is_array())
        return false;
    for (const auto& node : graph["nodes"]) {
        if (!node.is_object()) continue;
        const auto name = node.value("name", std::string{});
        if (name == "gallery.sample.capture" ||
            name == "FFmpeg.H264.encode" ||
            name == "gallery.ffmpeg.webrtc.publish" ||
            name == "gallery.websocket_pixels.pack" ||
            name == "gallery.websocket_pixels.publish" ||
            name == "gallery.sample.complete")
            return true;
    }
    return false;
}

/*
 * Gallery 的媒体 DAG 实际只在组内唯一 encode-source Plot 上执行。诊断接口
 * 在服务端把该真实执行帧中由 Task_Graph 包装器捕获到的媒体 graph 附加到
 * 当前 Plot 的 trace 响应；不复制执行、不伪造 FFmpeg 节点，也不让前端知道
 * encode-source 是哪个 Plot。
 */
nlohmann::json merge_gallery_media_trace(nlohmann::json plot_trace,
                                         const nlohmann::json& media_trace) {
    if (!plot_trace.is_object() || !media_trace.is_object() ||
        !plot_trace.contains("frames") || !plot_trace["frames"].is_array() ||
        !media_trace.contains("frames") || !media_trace["frames"].is_array())
        return plot_trace;

    auto& plot_frames = plot_trace["frames"];
    const auto& media_frames = media_trace["frames"];
    const auto count = std::min(plot_frames.size(), media_frames.size());

    /*
     * 只把真实媒体 graph 附加到已有 Plot trace；绝不裁剪 Plot 自己的帧，
     * 也不改写 Plot trace 的 requested/captured/remaining/complete 语义。
     */
    for (std::size_t index = 0; index < count; ++index) {
        auto& output = plot_frames[index];
        const auto& media = media_frames[index];
        if (!output.contains("graphs") || !output["graphs"].is_array() ||
            !output.contains("executions") || !output["executions"].is_array() ||
            !media.contains("graphs") || !media["graphs"].is_array() ||
            !media.contains("executions") || !media["executions"].is_array())
            continue;

        std::vector<std::string> media_native_ids;
        for (const auto& graph : media["graphs"]) {
            if (!taskflow_graph_contains_gallery_media(graph)) continue;
            auto appended = graph;
            appended["stage"] = "gallery.media";
            for (const auto& node : graph["nodes"])
                if (node.is_object() && node.contains("native_id") &&
                    node["native_id"].is_string())
                    media_native_ids.push_back(node["native_id"].get<std::string>());
            output["graphs"].push_back(std::move(appended));
        }
        if (media_native_ids.empty()) continue;
        for (const auto& execution : media["executions"]) {
            if (!execution.is_object() || !execution.contains("native_id") ||
                !execution["native_id"].is_string())
                continue;
            const auto native_id = execution["native_id"].get<std::string>();
            if (std::ranges::find(media_native_ids, native_id) !=
                media_native_ids.end())
                output["executions"].push_back(execution);
        }
        output["gallery_media_sequence"] = media.value("sequence", 0ULL);
        output["gallery_media_correlation_id"] =
            media.value("correlation_id", 0ULL);
    }

    return plot_trace;
}
}

int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) {
    auto plots = std::make_shared<Plot_Map>();
    plots->emplace("axes", make_axes_plot());
    plots->emplace("spectrum", make_spectrum_plot());
    plots->emplace("frequency_trace", make_frequency_trace_plot());
    plots->emplace("sweep_spectrum", make_sweep_spectrum_plot());
    plots->emplace("afterglow", make_afterglow_plot());
    plots->emplace("waterfall", make_waterfall_plot());
    plots->emplace("constellation", make_constellation_plot());
    plots->emplace("selection_overlay", make_selection_overlay_plot());
    plots->emplace("datoviz_point", make_datoviz_point_plot());
    plots->emplace("datoviz_splat", make_datoviz_splat_plot());
    plots->emplace("datoviz_pixel", make_datoviz_pixel_plot());
    plots->emplace("datoviz_marker", make_datoviz_marker_plot());
    plots->emplace("datoviz_sphere", make_datoviz_sphere_plot());
    plots->emplace("datoviz_segment", make_datoviz_segment_plot());
    plots->emplace("datoviz_vector", make_datoviz_vector_plot());
    plots->emplace("datoviz_primitive", make_datoviz_primitive_plot());
    plots->emplace("datoviz_mesh", make_datoviz_mesh_plot());
    plots->emplace("datoviz_spectrogram", make_datoviz_spectrogram_plot());
    plots->emplace("datoviz_path", make_datoviz_path_plot());
    plots->emplace("datoviz_image", make_datoviz_image_plot());
    plots->emplace("datoviz_labels", make_datoviz_labels_plot());
    plots->emplace("datoviz_glyph", make_datoviz_glyph_plot());
    plots->emplace("datoviz_text", make_datoviz_text_plot());
    plots->emplace("datoviz_volume", make_datoviz_volume_plot());

    std::vector<Gallery_Video_Stream::Plot_Entry> gallery_2d;
    std::vector<Gallery_Video_Stream::Plot_Entry> gallery_3d;
    gallery_2d.reserve(8);
    gallery_3d.reserve(16);
    for (const auto& [id, plot] : *plots)
        (id.starts_with("datoviz_") ? gallery_3d : gallery_2d)
            .push_back({id, plot});
    const auto entry_order = [](const auto& left, const auto& right) {
        return left.id < right.id;
    };
    std::ranges::sort(gallery_2d, entry_order);
    std::ranges::sort(gallery_3d, entry_order);

    auto gallery_streams = std::make_shared<Gallery_Stream_Map>();
    auto plot_media = std::make_shared<
        std::unordered_map<std::string, std::string>>();
    auto plot_media_source = std::make_shared<
        std::unordered_map<std::string, std::string>>();
    const auto add_media_group = [&gallery_streams, &plot_media, &plot_media_source](
        std::string id,
        std::vector<Gallery_Video_Stream::Plot_Entry> entries) {
        if (entries.empty()) return;
        const auto path = "/ws/gallery?group=" + id;
        const auto media_source = entries.back().id;
        for (const auto& entry : entries) {
            plot_media->emplace(entry.id, path);
            plot_media_source->emplace(entry.id, media_source);
        }
        gallery_streams->emplace(
            std::move(id), Gallery_Video_Stream::create(std::move(entries)));
    };
    constexpr std::size_t plots_per_media_group{4};
    const auto add_media_groups = [&add_media_group](
        std::string_view prefix,
        std::vector<Gallery_Video_Stream::Plot_Entry> entries) {
        std::size_t group_index{};
        for (std::size_t offset = 0; offset < entries.size();
             offset += plots_per_media_group) {
            const auto end = std::min(
                entries.size(), offset + plots_per_media_group);
            std::vector<Gallery_Video_Stream::Plot_Entry> group(
                std::make_move_iterator(entries.begin() +
                    static_cast<std::ptrdiff_t>(offset)),
                std::make_move_iterator(entries.begin() +
                    static_cast<std::ptrdiff_t>(end)));
            add_media_group(std::string{prefix} + "-" +
                std::to_string(++group_index), std::move(group));
        }
    };
    add_media_groups("2d", std::move(gallery_2d));
    add_media_groups("3d", std::move(gallery_3d));
    auto resolve_plot = [plots](std::string_view id) { return find_plot(*plots, id); };
    auto websocket = std::make_shared<Graph_WebSocket_Controller>(resolve_plot);
    auto gallery_websocket = std::make_shared<Gallery_WebSocket_Controller>(
        [gallery_streams](std::string_view id) {
            const auto found = gallery_streams->find(std::string{id});
            return found == gallery_streams->end() ? nullptr : found->second;
        });
    auto& app = drogon::app();

    app.registerHandler("/plot", [plots, plot_media](const drogon::HttpRequestPtr&,
                                         std::function<void(const drogon::HttpResponsePtr&)>&& callback) {
        nlohmann::json result = nlohmann::json::array();
        for (const auto& [id, plot] : *plots) {
            static_cast<void>(plot);
            const auto& media = plot_media->at(id);
            result.push_back({
                {"id", id}, {"title", id}, {"category", "Plots"}, {"description", ""},
                {"dimension", id.starts_with("datoviz_") ? "3D" : "2D"}, {"websocket", "/ws/plot/" + id},
                {"media", media},
                {"schema", "/plot/" + id + "/schema"},
                {"diagnostics", "/plot/" + id + "/diagnostics"},
                {"taskflow", "/plot/" + id + "/taskflow"}});
        }
        callback(json_response(std::move(result)));
    }, {drogon::Get});

    /* Taskflow is a browser route, not a server-side resource.  A direct load or
     * refresh must enter the same SPA document as `/`; Vite then resolves the
     * requested Plot id without creating a second HTTP/session protocol. */
    app.registerHandler("/taskflow/{1}", [index = asset_root / "index.html"](
        const drogon::HttpRequestPtr& request,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string) {
        auto response = drogon::HttpResponse::newFileResponse(
            index.string(), {}, drogon::CT_TEXT_HTML, {}, request);
        response->addHeader("Cache-Control", "no-store");
        callback(std::move(response));
    }, {drogon::Get});

    app.registerHandler("/plot/{1}/diagnostics", [plots](
        const drogon::HttpRequestPtr& request,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string plot_id) {
        auto plot = find_plot(*plots, plot_id);
        if (!plot) {
            callback(error_response(drogon::k404NotFound, "unknown plot"));
            return;
        }
        if (request->method() == drogon::Delete) {
            plot->reset_diagnostics();
            callback(json_response({{"success", true}}));
            return;
        }
        callback(json_response(plot->diagnostics()));
    }, {drogon::Get, drogon::Delete});

    app.registerHandler("/taskflow/diagnostics", [](
        const drogon::HttpRequestPtr&,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback) {
        callback(json_response(taskflow_runtime_json()));
    }, {drogon::Get});

    app.registerHandler("/plot/{1}/taskflow", [plots, plot_media_source](
        const drogon::HttpRequestPtr& request,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string plot_id) {
        auto plot = find_plot(*plots, plot_id);
        if (!plot) {
            callback(error_response(drogon::k404NotFound, "unknown plot"));
            return;
        }
        const auto source_found = plot_media_source->find(plot_id);
        auto media_plot = source_found == plot_media_source->end()
            ? plot : find_plot(*plots, source_found->second);
        if (!media_plot) media_plot = plot;
        try {
            if (request->method() == drogon::Post) {
                const auto input = nlohmann::json::parse(request->body());
                if (!input.is_object() || !input.contains("frame_count") ||
                    !input["frame_count"].is_number_unsigned()) {
                    callback(error_response(drogon::k400BadRequest,
                        "Taskflow trace requires unsigned frame_count"));
                    return;
                }
                const auto frame_count = input["frame_count"].get<std::size_t>();
                const auto trace_active = [](const nlohmann::json& state) {
                    return state.value("requested", std::size_t{}) >
                           state.value("captured", std::size_t{});
                };
                if (trace_active(plot->taskflow_trace()) ||
                    trace_active(media_plot->post_publish_taskflow_trace()))
                    throw std::logic_error(
                        "A Taskflow frame trace request is already active");
                /*
                 * 非 encode-source Plot 只让真实媒体源捕获 post-publish DAG。
                 * 不再额外追踪另一张 Plot 的 Render/Prepare/Paint，避免诊断本身
                 * 放大 Executor 压力。FFmpeg 仍来自真实 Task_Graph 包装节点。
                 */
                media_plot->request_post_publish_taskflow_trace(frame_count);
                plot->request_taskflow_trace(frame_count);
            }
            auto output = plot->taskflow_trace();
            const auto media = media_plot->post_publish_taskflow_trace();
            output = merge_gallery_media_trace(std::move(output), media);
            const bool plot_complete = output.value("complete", false);
            const bool media_complete = media.value("complete", false);
            output["media_requested"] = media.value("requested", 0U);
            output["media_captured"] = media.value("captured", 0U);
            output["media_remaining"] = media.value("remaining", 0U);
            /* Scene 与异步媒体图分别完成捕获后才结束轮询。 */
            output["complete"] = plot_complete && media_complete;
            output["remaining"] = std::max(
                output.value("remaining", 0U),
                media.value("remaining", 0U));
            callback(json_response(std::move(output)));
        }
        catch (const nlohmann::json::exception&) {
            callback(error_response(drogon::k400BadRequest,
                                    "invalid Taskflow trace request"));
        }
        catch (const std::invalid_argument& failure) {
            callback(error_response(drogon::k400BadRequest, failure.what()));
        }
        catch (const std::logic_error& failure) {
            callback(error_response(drogon::k409Conflict, failure.what()));
        }
    }, {drogon::Get, drogon::Post});

    app.registerHandler("/gallery/{1}/diagnostics", [gallery_streams](
        const drogon::HttpRequestPtr&,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string group_id) {
        const auto found = gallery_streams->find(group_id);
        if (found == gallery_streams->end()) {
            callback(error_response(drogon::k404NotFound,
                                    "unknown gallery media group"));
            return;
        }
        callback(json_response(found->second->diagnostics()));
    }, {drogon::Get});

    app.registerHandler("/plot/{1}/schema", [plots](
        const drogon::HttpRequestPtr&,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string plot_id) {
        auto plot = find_plot(*plots, plot_id);
        if (!plot) {
            callback(error_response(drogon::k404NotFound, "unknown plot"));
            return;
        }
        try { callback(json_response(plot->schema())); }
        catch (const std::exception& failure) {
            callback(error_response(drogon::k500InternalServerError,
                                    failure.what()));
        }
    }, {drogon::Get});

    app.registerHandler("/plot/{1}/component/{2}/prop/{3}", [plots](
        const drogon::HttpRequestPtr& request,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string plot_id,
        std::string component,
        std::string key) {
        auto plot = find_plot(*plots, plot_id);
        if (!plot) {
            callback(error_response(drogon::k404NotFound, "unknown plot"));
            return;
        }
        nlohmann::json value;
        try {
            value = nlohmann::json::parse(request->body());
        } catch (const nlohmann::json::exception&) {
            callback(error_response(drogon::k400BadRequest, "invalid JSON value"));
            return;
        }
        try {
            callback(json_response(plot->write_prop(component, key, value)));
        }
        catch (const std::exception& failure) {
            callback(error_response(drogon::k500InternalServerError,
                                    failure.what()));
        }
    }, {drogon::Put});

    app.registerHandler("/plot/{1}/component/{2}/state", [plots](
        const drogon::HttpRequestPtr&,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string plot_id,
        std::string component) {
        auto plot = find_plot(*plots, plot_id);
        if (!plot) {
            callback(error_response(drogon::k404NotFound, "unknown plot"));
            return;
        }
        try {
            auto result = plot->component_state(component);
            if (result.value("success", true)) callback(json_response(std::move(result)));
            else callback(error_response(drogon::k404NotFound,
                                         result.value("error", "unknown component")));
        }
        catch (const std::exception& failure) {
            callback(error_response(drogon::k500InternalServerError,
                                    failure.what()));
        }
    }, {drogon::Get});

    app.registerHandler("/plot/{1}/data/generate", [plots](
        const drogon::HttpRequestPtr& request,
        std::function<void(const drogon::HttpResponsePtr&)>&& callback,
        std::string plot_id) {
        auto plot = find_plot(*plots, plot_id);
        if (!plot) {
            callback(error_response(drogon::k404NotFound, "unknown plot"));
            return;
        }
        nlohmann::json input;
        try {
            input = nlohmann::json::parse(request->body());
        } catch (const nlohmann::json::exception&) {
            callback(error_response(drogon::k400BadRequest, "invalid data generation request"));
            return;
        }
        if (!input.is_object()) {
            callback(error_response(drogon::k400BadRequest, "data generation input must be an object"));
            return;
        }
        try { callback(json_response(plot->generate_data(input))); }
        catch (const std::exception& failure) {
            callback(error_response(drogon::k500InternalServerError,
                                    failure.what()));
        }
    }, {drogon::Post});

    app.registerController(websocket)
        .registerController(gallery_websocket)
        .setDocumentRoot(asset_root.string())
        .setHomePage("index.html")
        .setStaticFileHeaders({{"Cache-Control", "no-store"}})
        .addListener("127.0.0.1", port)
        .setThreadNum(std::min(8U, std::max(2U, std::thread::hardware_concurrency())))
        .setIdleConnectionTimeout(90)
        .run();
    /* 先解除 Plot 完成帧订阅并停止页面时钟；Taskflow 任务只捕获弱所有权，
     * 因而销毁期不会访问已经释放的图集，也不需要额外业务线程池排空。 */
    for (const auto& [id, stream] : *gallery_streams) {
        static_cast<void>(id);
        stream->shutdown();
    }
    return 0;
}
}