#include <gtest/gtest.h>

#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <thread>
#include <vector>

#include "renderive/render_graph/External_Operation.hpp"
#include "renderive/render_graph/Render_Plan.hpp"
#include "renderive/render_graph/detail/Render_Graph_Runtime.hpp"
#include "renderive/scheduling/detail/OneTBB_Runtime.hpp"

namespace {

std::shared_ptr<const Render_Plan> two_node_plan() {
    Render_Graph_Builder builder;
    const auto submit = builder.emplace(
        301, 30, "Submit GPU", Render_Node_Kind::render);
    const auto publish = builder.emplace(
        302, 30, "Publish Frame", Render_Node_Kind::composite);
    builder.precede(submit, publish);
    Render_Plan_History history;
    return history.publish(std::move(builder).finish());
}

std::vector<Node_Execution*> execution_slots(
    const Render_Plan& plan, std::vector<Node_Execution>& storage) {
    storage.resize(plan.graph.nodes.size());
    std::vector<Node_Execution*> result(storage.size());
    for (const auto& node : plan.graph.nodes) {
        storage[node.execution_index].node_id = node.node_id;
        result[node.execution_index] = &storage[node.execution_index];
    }
    return result;
}

} // namespace

TEST(external_operation_test, completion_before_subscription_is_delivered_once) {
    External_Operation_Source source;
    const auto operation = source.operation();
    EXPECT_TRUE(source.complete());
    EXPECT_FALSE(source.complete());

    int completion_count{};
    operation.on_complete([&](std::exception_ptr error) {
        EXPECT_FALSE(error);
        ++completion_count;
    });
    EXPECT_EQ(completion_count, 1);
}

TEST(external_operation_test,
     callback_exceptions_are_contained_for_both_completion_orders) {
    External_Operation_Source completed_first;
    const auto completed_operation = completed_first.operation();
    ASSERT_TRUE(completed_first.complete());
    EXPECT_NO_THROW(completed_operation.on_complete(
        [](std::exception_ptr) { throw std::runtime_error("late callback"); }));

    External_Operation_Source subscribed_first;
    const auto subscribed_operation = subscribed_first.operation();
    subscribed_operation.on_complete(
        [](std::exception_ptr) { throw std::runtime_error("early callback"); });
    EXPECT_TRUE(subscribed_first.complete());
}

TEST(render_graph_runtime_test, reusable_topology_executes_multiple_frames) {
    const auto plan = two_node_plan();
    renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
    std::array<std::size_t, 2> execution_count{};
    const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
        ++execution_count.at(index);
        return Node_Execution_Result::completed();
    };
    runtime.execute({}, execute_node);
    runtime.execute({}, execute_node);
    EXPECT_EQ(execution_count[0], 2U);
    EXPECT_EQ(execution_count[1], 2U);
}

TEST(render_graph_runtime_test,
     external_successor_stays_blocked_until_operation_completes) {
    const auto plan = two_node_plan();
    External_Operation_Source source;
    std::vector<Node_Execution> execution_storage;
    auto slots = execution_slots(*plan, execution_storage);
    std::atomic<bool> submit_started{};
    std::atomic<bool> publish_executed{};

    renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
    const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
        if (index == 0) {
            submit_started.store(true, std::memory_order_release);
            submit_started.notify_all();
            return Node_Execution_Result::external(source.operation());
        }
        publish_executed.store(true, std::memory_order_release);
        return Node_Execution_Result::completed();
    };

    std::thread execution([&] { runtime.execute(slots, execute_node); });
    submit_started.wait(false, std::memory_order_acquire);
    EXPECT_FALSE(publish_executed.load(std::memory_order_acquire));
    EXPECT_TRUE(source.complete());
    execution.join();

    EXPECT_TRUE(publish_executed.load(std::memory_order_acquire));
    EXPECT_EQ(execution_storage[0].status, Node_Execution_Status::complete);
    EXPECT_EQ(execution_storage[1].status, Node_Execution_Status::complete);
    EXPECT_GT(execution_storage[0].cpu_end_time_ns, 0U);
    EXPECT_GE(execution_storage[0].external_end_time_ns,
              execution_storage[0].external_start_time_ns);
}

TEST(render_graph_runtime_test,
     failed_external_node_prevents_its_successor_from_running) {
    const auto plan = two_node_plan();
    External_Operation_Source source;
    std::atomic<bool> submit_started{};
    std::atomic<bool> publish_executed{};
    std::exception_ptr graph_error;

    renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
    const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
        if (index == 0) {
            submit_started.store(true, std::memory_order_release);
            submit_started.notify_all();
            return Node_Execution_Result::external(source.operation());
        }
        publish_executed.store(true, std::memory_order_release);
        return Node_Execution_Result::completed();
    };

    std::thread execution([&] {
        try {
            runtime.execute({}, execute_node);
        } catch (...) {
            graph_error = std::current_exception();
        }
    });
    submit_started.wait(false, std::memory_order_acquire);
    EXPECT_TRUE(source.fail(
        std::make_exception_ptr(std::runtime_error("GPU submission failed"))));
    execution.join();

    ASSERT_TRUE(graph_error);
    EXPECT_THROW(std::rethrow_exception(graph_error), std::runtime_error);
    EXPECT_FALSE(publish_executed.load(std::memory_order_acquire));
}

TEST(render_graph_runtime_test,
     pending_external_operation_does_not_occupy_scheduler_worker) {
    Render_Graph_Builder builder;
    builder.emplace(401, 40, "GPU Fence", Render_Node_Kind::render);
    Render_Plan_History history;
    const auto plan = history.publish(std::move(builder).finish());
    External_Operation_Source source;
    std::atomic<bool> external_started{};
    std::mutex probe_mutex;
    std::condition_variable probe_completed;
    bool probe_ran{};

    renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
    const auto execute_node = [&](std::size_t, Node_Execution_Metrics*) {
        external_started.store(true, std::memory_order_release);
        external_started.notify_all();
        return Node_Execution_Result::external(source.operation());
    };

    std::thread execution([&] { runtime.execute({}, execute_node); });
    external_started.wait(false, std::memory_order_acquire);
    renderive::scheduling::detail::OneTBB_Runtime::instance().enqueue([&] {
        {
            std::lock_guard lock(probe_mutex);
            probe_ran = true;
        }
        probe_completed.notify_one();
    });
    {
        std::unique_lock lock(probe_mutex);
        EXPECT_TRUE(probe_completed.wait_for(
            lock, std::chrono::seconds(1), [&] { return probe_ran; }));
    }
    EXPECT_TRUE(source.complete());
    execution.join();
}

TEST(external_operation_test, cancellation_is_terminal_and_delivered_once) {
    External_Operation_Source source;
    const auto operation = source.operation();
    EXPECT_TRUE(operation.cancel());
    EXPECT_FALSE(source.complete());
    EXPECT_EQ(operation.status(), External_Operation_Status::cancelled);

    int completion_count{};
    std::exception_ptr completion_error;
    operation.on_complete([&](std::exception_ptr error) {
        completion_error = std::move(error);
        ++completion_count;
    });
    EXPECT_EQ(completion_count, 1);
    ASSERT_TRUE(completion_error);
    EXPECT_THROW(std::rethrow_exception(completion_error),
                 External_Operation_Cancelled);
}

TEST(external_operation_test, deadline_cancels_pending_operation) {
    External_Operation_Source source;
    const auto operation = source.operation();
    std::mutex mutex;
    std::condition_variable condition;
    bool completed{};
    std::exception_ptr completion_error;
    operation.on_complete([&](std::exception_ptr error) {
        {
            std::lock_guard lock(mutex);
            completion_error = std::move(error);
            completed = true;
        }
        condition.notify_one();
    });
    source.set_deadline(External_Operation_Source::Clock::now());
    {
        std::unique_lock lock(mutex);
        ASSERT_TRUE(condition.wait_for(lock, std::chrono::seconds(1),
                                       [&] { return completed; }));
    }
    ASSERT_TRUE(completion_error);
    EXPECT_THROW(std::rethrow_exception(completion_error),
                 External_Operation_Deadline_Exceeded);
    EXPECT_EQ(operation.status(), External_Operation_Status::cancelled);
}

TEST(render_graph_runtime_test,
     cancellation_releases_external_wait_and_blocks_successor) {
    const auto plan = two_node_plan();
    External_Operation_Source source;
    std::vector<Node_Execution> execution_storage;
    auto slots = execution_slots(*plan, execution_storage);
    std::atomic<bool> submit_started{};
    std::atomic<bool> publish_executed{};
    std::exception_ptr graph_error;

    renderive::render_graph::detail::Render_Graph_Runtime runtime(*plan);
    const auto execute_node = [&](std::size_t index, Node_Execution_Metrics*) {
        if (index == 0) {
            submit_started.store(true, std::memory_order_release);
            submit_started.notify_all();
            return Node_Execution_Result::external(source.operation());
        }
        publish_executed.store(true, std::memory_order_release);
        return Node_Execution_Result::completed();
    };

    std::thread execution([&] {
        try {
            runtime.execute(slots, execute_node);
        } catch (...) {
            graph_error = std::current_exception();
        }
    });
    submit_started.wait(false, std::memory_order_acquire);
    runtime.cancel_pending();
    execution.join();

    ASSERT_TRUE(graph_error);
    EXPECT_THROW(std::rethrow_exception(graph_error),
                 External_Operation_Cancelled);
    EXPECT_FALSE(publish_executed.load(std::memory_order_acquire));
    EXPECT_EQ(execution_storage[0].status, Node_Execution_Status::cancelled);
    EXPECT_EQ(source.operation().status(), External_Operation_Status::cancelled);
}

TEST(external_operation_test, completion_commit_is_exclusive_with_cancellation) {
    External_Operation_Source source;
    const auto operation = source.operation();
    std::atomic<int> published{};
    EXPECT_TRUE(operation.cancel());
    EXPECT_FALSE(source.complete([&] {
        published.fetch_add(1, std::memory_order_relaxed);
    }));
    EXPECT_EQ(published.load(std::memory_order_relaxed), 0);
}

TEST(external_operation_test, completion_commit_precedes_completion_notification) {
    External_Operation_Source source;
    const auto operation = source.operation();
    std::atomic<int> published{};
    int observed{};
    operation.on_complete([&](std::exception_ptr error) {
        EXPECT_FALSE(error);
        observed = published.load(std::memory_order_acquire);
    });

    EXPECT_TRUE(source.complete([&] {
        published.store(1, std::memory_order_release);
        EXPECT_EQ(operation.status(), External_Operation_Status::pending);
    }));
    EXPECT_EQ(observed, 1);
    EXPECT_EQ(operation.status(), External_Operation_Status::completed);
}

TEST(external_operation_test, completion_commit_failure_fails_operation) {
    External_Operation_Source source;
    const auto operation = source.operation();
    std::exception_ptr completion_error;
    operation.on_complete([&](std::exception_ptr error) {
        completion_error = std::move(error);
    });

    EXPECT_TRUE(source.complete([] { throw std::runtime_error("publish failed"); }));
    ASSERT_TRUE(completion_error);
    EXPECT_THROW(std::rethrow_exception(completion_error), std::runtime_error);
    EXPECT_EQ(operation.status(), External_Operation_Status::failed);
}