#include <catch2/catch_test_macros.hpp>
#include <kpn/scheduler.hpp>
#include <kpn/pool_node.hpp>
#include <kpn/interrupt_node.hpp>
#include <atomic>
#include <chrono>
#include <mutex>
#include <thread>

using namespace kpn;

static int double_it(int x) { return x * 2; }
static std::tuple<int, float> split_it(int x) { return {x, float(x) * 0.5f}; }
static void consume_it(int x) { (void)x; }

// ── ThreadPool ────────────────────────────────────────────────────────────────

TEST_CASE("thread pool starts and stops cleanly", "[scheduler]") {
    ThreadPool pool(2);
    pool.start();
    pool.stop();
}

TEST_CASE("thread pool executes submitted tasks", "[scheduler]") {
    ThreadPool pool(2);
    pool.start();

    std::atomic<int> counter{0};
    for (int i = 0; i < 10; ++i)
        pool.submit([&] { counter.fetch_add(1); });

    pool.drain();
    REQUIRE(counter.load() == 10);
    pool.stop();
}

TEST_CASE("thread pool drain waits for all tasks", "[scheduler]") {
    ThreadPool pool(1);
    pool.start();

    std::atomic<bool> done{false};
    pool.submit([&] {
        std::this_thread::sleep_for(std::chrono::milliseconds(20));
        done.store(true);
    });
    pool.drain();
    REQUIRE(done.load());
    pool.stop();
}

TEST_CASE("thread pool priority: higher priority tasks run first", "[scheduler]") {
    ThreadPool pool(1);  // single thread so order is deterministic
    pool.start();

    // Submit a task that blocks the worker, then queue two tasks with
    // different priorities. When the blocker finishes, the high-priority
    // task should run before the low-priority one.
    std::vector<int> order;
    std::mutex order_mutex;

    std::atomic<bool> blocker_done{false};
    pool.submit([&] {
        std::this_thread::sleep_for(std::chrono::milliseconds(30));
        blocker_done.store(true);
    }, 0.5f);

    // Wait until blocker is running, then enqueue the two ordered tasks.
    while (!blocker_done.load()) std::this_thread::sleep_for(std::chrono::milliseconds(1));

    pool.submit([&] { std::lock_guard g(order_mutex); order.push_back(1); }, 0.1f);
    pool.submit([&] { std::lock_guard g(order_mutex); order.push_back(2); }, 0.9f);

    pool.drain();
    REQUIRE(order == std::vector<int>{2, 1});
    pool.stop();
}

// ── PoolNode ──────────────────────────────────────────────────────────────────

TEST_CASE("pool node input/output counts", "[pool_node]") {
    STATIC_REQUIRE(PoolNode<double_it>::input_count  == 1);
    STATIC_REQUIRE(PoolNode<double_it>::output_count == 1);
    STATIC_REQUIRE(PoolNode<split_it>::output_count  == 2);
    STATIC_REQUIRE(PoolNode<consume_it>::output_count == 0);
}

TEST_CASE("pool node processes items end-to-end", "[pool_node]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<double_it>(pool);
    Channel<int> out_ch(10);
    node.set_output_channel<0>(&out_ch);
    node.start();

    node.input_channel<0>().push(21);
    int result = out_ch.pop();

    node.stop();
    pool->stop();

    REQUIRE(result == 42);
}

TEST_CASE("pool node processes multiple items in order", "[pool_node]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<double_it>(pool, 20);  // capacity 20
    Channel<int> out_ch(20);
    node.set_output_channel<0>(&out_ch);
    node.start();

    constexpr int N = 10;
    for (int i = 0; i < N; ++i)
        node.input_channel<0>().push(i);

    std::vector<int> results;
    for (int i = 0; i < N; ++i)
        results.push_back(out_ch.pop());

    node.stop();
    pool->stop();

    REQUIRE(results.size() == N);
    for (int i = 0; i < N; ++i)
        REQUIRE(results[i] == i * 2);
}

TEST_CASE("pool node stop is clean with no deadlock", "[pool_node]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<double_it>(pool);
    node.start();
    // Node is idle (no input pushed) — stop must return without deadlock.
    node.stop();
    REQUIRE_FALSE(node.running());

    pool->stop();
}

TEST_CASE("pool node two-stage pipeline produces correct count", "[pool_node]") {
    auto pool = std::make_shared<ThreadPool>(4);
    pool->start();

    auto src       = make_pool_node<double_it>(pool);
    auto transform = make_pool_node<double_it>(pool);
    Channel<int> out_ch(20);

    // Wire src output → transform input channel, transform output → out_ch.
    src.set_output_channel<0>(&transform.input_channel<0>());
    transform.set_output_channel<0>(&out_ch);

    src.start();
    transform.start();

    constexpr int N = 5;
    for (int i = 1; i <= N; ++i)
        src.input_channel<0>().push(i);

    std::vector<int> results;
    for (int i = 0; i < N; ++i)
        results.push_back(out_ch.pop());

    // Stop nodes before they (and their channels) go out of scope.
    src.stop();
    transform.stop();
    pool->stop();

    REQUIRE(results.size() == static_cast<std::size_t>(N));
    for (int i = 0; i < N; ++i)
        REQUIRE(results[i] == (i + 1) * 4);  // double_it twice
}

// ── InterruptNode ─────────────────────────────────────────────────────────────

namespace {
static std::atomic<int> g_interrupt_counter{0};
static int interrupt_produce() { return g_interrupt_counter.fetch_add(1); }
} // namespace

TEST_CASE("interrupt node fires on each trigger", "[interrupt_node]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    g_interrupt_counter.store(0);
    auto node = make_interrupt_node<interrupt_produce>(pool, out<>{});
    Channel<int> out_ch(20);
    node.set_output_channel<0>(&out_ch);
    node.start();

    auto trigger = node.get_trigger();
    constexpr int N = 5;
    for (int i = 0; i < N; ++i) trigger();

    std::vector<int> results;
    for (int i = 0; i < N; ++i)
        results.push_back(out_ch.pop());

    node.stop();
    pool->stop();

    REQUIRE(results.size() == static_cast<std::size_t>(N));
}

TEST_CASE("interrupt node does not fire without trigger", "[interrupt_node]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    g_interrupt_counter.store(0);
    auto node = make_interrupt_node<interrupt_produce>(pool, out<>{});
    Channel<int> out_ch(5);
    node.set_output_channel<0>(&out_ch);
    node.start();

    std::this_thread::sleep_for(std::chrono::milliseconds(30));
    // No trigger fired — output channel should be empty.
    REQUIRE(out_ch.approx_size() == 0);

    node.stop();
    pool->stop();
}

TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    g_interrupt_counter.store(0);
    auto node = make_interrupt_node<interrupt_produce>(pool, out<>{});
    Channel<int> out_ch(5);
    node.set_output_channel<0>(&out_ch);
    node.start();

    auto trigger = node.get_trigger();
    node.stop();
    trigger();  // should be a no-op

    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    REQUIRE(out_ch.approx_size() == 0);
    pool->stop();
}

// ── Overflow callback ─────────────────────────────────────────────────────────

// NOTE: output overflow is no longer reachable on the data path. A node whose
// output channel is full now PARKS — it keeps the value in a hidden one-slot
// buffer, releases its scheduler worker, and is re-submitted when the consumer
// frees a slot. The overflow callback survives for other producers (a direct
// Channel::push by non-node code still throws), but a pool node cannot trigger
// it, so these cases assert the stronger property instead: nothing is dropped.
TEST_CASE("pool node parks instead of overflowing a full output", "[pool_node][overflow]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<double_it>(pool);
    // Pre-fill a tiny channel so every node push overflows.
    Channel<int> full_ch(1);
    full_ch.push(99);
    node.set_output_channel<0>(&full_ch);

    std::atomic<int> overflow_count{0};
    node.set_overflow_callback([&](auto) { overflow_count.fetch_add(1); });

    node.start();
    node.input_channel<0>().push(1);
    node.input_channel<0>().push(2);

    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    node.stop();
    pool->stop();

    // Parked, not overflowed: the value is still owned by the node.
    REQUIRE(overflow_count.load() == 0);
    // And it was never handed downstream, so nothing was lost or duplicated.
    REQUIRE(full_ch.size() == 1);
}

TEST_CASE("parking is per node, not shared", "[pool_node][overflow]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto nodeA = make_pool_node<double_it>(pool);
    auto nodeB = make_pool_node<double_it>(pool);

    std::atomic<int> a_overflows{0}, b_overflows{0};
    nodeA.set_overflow_callback([&](auto) { a_overflows.fetch_add(1); });

    Channel<int> full_ch(1);
    full_ch.push(0);
    nodeA.set_output_channel<0>(&full_ch);

    Channel<int> ok_ch(20);
    nodeB.set_output_channel<0>(&ok_ch);

    nodeA.start();
    nodeB.start();

    nodeA.input_channel<0>().push(1);
    nodeA.input_channel<0>().push(2);
    nodeB.input_channel<0>().push(10);

    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    nodeA.stop();
    nodeB.stop();
    pool->stop();

    // Neither overflows now: A parks on its full output, B runs normally. The
    // point of the case is unchanged — one node's backpressure must not leak
    // into another's callbacks.
    REQUIRE(a_overflows.load() == 0);
    REQUIRE(b_overflows.load() == 0);
}

TEST_CASE("interrupt node overflow callback fires on full output", "[interrupt_node][overflow]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    g_interrupt_counter.store(0);
    auto node = make_interrupt_node<interrupt_produce>(pool, out<>{});

    Channel<int> full_ch(1);
    full_ch.push(99);
    node.set_output_channel<0>(&full_ch);

    std::atomic<int> overflow_count{0};
    node.set_overflow_callback([&](auto) { overflow_count.fetch_add(1); });

    node.start();
    auto trigger = node.get_trigger();
    trigger(); trigger(); trigger();

    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    node.stop();
    pool->stop();

    REQUIRE(overflow_count.load() > 0);
}

// ── self_stop: disable inputs + outputs on crash ──────────────────────────────

static int always_throw(int) { throw std::runtime_error("node crashed"); return 0; }

TEST_CASE("pool node self_stop disables output on crash so downstream sees closed", "[pool_node][self_stop]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<always_throw>(pool, 5);
    Channel<int> out_ch(10);
    node.set_output_channel<0>(&out_ch);

    node.start();
    node.input_channel<0>().push(1);

    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    REQUIRE_FALSE(out_ch.is_accepting());

    node.stop();
    pool->stop();
}

TEST_CASE("pool node self_stop disables input on crash", "[pool_node][self_stop]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<always_throw>(pool, 5);
    Channel<int> out_ch(5);
    node.set_output_channel<0>(&out_ch);

    node.start();
    node.input_channel<0>().push(1);

    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    REQUIRE_FALSE(node.input_channel<0>().is_accepting());

    node.stop();
    pool->stop();
}

TEST_CASE("pool node closed callback fires on self_stop from crash", "[pool_node][self_stop]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<always_throw>(pool, 5);
    Channel<int> out_ch(5);
    node.set_output_channel<0>(&out_ch);

    std::atomic<bool> closed_fired{false};
    node.set_closed_callback([&](auto) { closed_fired.store(true); });

    node.start();
    node.input_channel<0>().push(1);

    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    REQUIRE(closed_fired.load());

    node.stop();
    pool->stop();
}

// ── Network-level event callbacks ─────────────────────────────────────────────

TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<double_it>(pool);

    Channel<int> full_ch(1);
    full_ch.push(0);
    node.set_output_channel<0>(&full_ch);

    std::atomic<int> net_overflows{0};
    node.set_network_overflow_callback([&](auto) { net_overflows.fetch_add(1); });

    node.start();
    node.input_channel<0>().push(1);
    node.input_channel<0>().push(2);

    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    node.stop();
    pool->stop();

    // Parking replaced overflow on the data path, so the network callback no
    // longer fires for a pool node's own output. See the note above.
    REQUIRE(net_overflows.load() == 0);
}

TEST_CASE("network_closed_callback fires on crash", "[pool_node][network]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<always_throw>(pool);
    Channel<int> out_ch(5);
    node.set_output_channel<0>(&out_ch);

    std::atomic<bool> net_closed{false};
    node.set_network_closed_callback([&](auto) { net_closed.store(true); });

    node.start();
    node.input_channel<0>().push(1);

    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    REQUIRE(net_closed.load());

    node.stop();
    pool->stop();
}

TEST_CASE("per-node and network overflow callbacks both fire independently", "[pool_node][network]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    auto node = make_pool_node<double_it>(pool);

    Channel<int> full_ch(1);
    full_ch.push(0);
    node.set_output_channel<0>(&full_ch);

    std::atomic<int> per_node{0}, network{0};
    node.set_overflow_callback([&](auto) { per_node.fetch_add(1); });
    node.set_network_overflow_callback([&](auto) { network.fetch_add(1); });

    node.start();
    node.input_channel<0>().push(1);
    node.input_channel<0>().push(2);

    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    node.stop();
    pool->stop();

    // Both zero now: the node parks rather than overflowing. The case still
    // guards that the two callbacks are wired independently.
    REQUIRE(per_node.load() == 0);
    REQUIRE(network.load() == 0);
}

// Regression: NodeSnapshot's fields must line up with what nodes initialise.
//
// The snapshot is an aggregate that every node type fills positionally, and
// a8cfe73 appended queued/wake_pending/total_exec_ms to it in an order no call
// site used: each node supplies total_exec_ms as the element straight after
// queue_wait_ms, but the struct declared the two bools there. So the exec total
// landed in `queued`, `queued` landed in `wake_pending`, and `wake_pending`
// landed in total_exec_ms. The compiler said so (-Wnarrowing, bool to double,
// once per node instantiation) and the build carried on.
//
// It matters more than a cosmetic mix-up: these three fields exist to diagnose a
// wedge, and a wedged pipeline reported total_exec_ms as 0 or 1 and `queued` as
// "did this node ever run". Reading them would have pointed at the wrong node.
//
// Asserted against ema_exec_ms because that field is independently computed and
// was already correct: a true sum over several frames cannot be below the
// exponentially-weighted average of the same samples.
TEST_CASE("node snapshot fields line up with the values nodes supply",
          "[pool_node][diagnostics]") {
    auto pool = std::make_shared<ThreadPool>(1);
    pool->start();

    auto node = make_pool_node<double_it>(pool, 64);
    Channel<int> out(64);
    node.set_output_channel<0>(&out);
    node.start();

    for (int i = 0; i < 8; ++i) node.input_channel<0>().push(i);
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    auto snap = node.node_snapshot("n", 1.0);
    node.stop();
    pool->stop();

    INFO("frames=" << snap.frames_processed
         << " ema=" << snap.ema_exec_ms
         << " total=" << snap.total_exec_ms);
    REQUIRE(snap.frames_processed == 8);
    // The mis-ordered aggregate put wake_pending here, so this was 0.0 or 1.0.
    CHECK(snap.total_exec_ms >= snap.ema_exec_ms);
    // ...and the exec total here, which is non-zero, so `queued` read true for
    // any node that had ever run — including one asleep with nothing to do.
    CHECK_FALSE(snap.queued);
    CHECK_FALSE(snap.wake_pending);
}

// Regression: a value parked twice must keep its payload.
//
// push_outputs ends with
//
//     else pending_ = std::move(result);
//
// and the retry path calls it as push_outputs(std::move(*pending_), …), so on
// that path `result` is the parked tuple itself. The assignment was therefore a
// self-move-assignment. std::tuple's is elementwise, and libstdc++'s
// std::vector does not guard against self-move: it swaps its data into a
// temporary and leaves the vector empty. So the first park was clean (the
// argument is a local temporary) and the second erased the payload.
//
// The value was still delivered, still in order, still counted — just empty.
// Downstream cannot distinguish that from a frame on which the node genuinely
// found nothing, which is why it never surfaced as an error: in
// scene-actor-extraction it reads as "no faces in this frame" and the run
// completes with a quietly wrong answer.
//
// Reaching it needs *two* outputs. With one, the only thing that resubmits a
// parked node is that output's own space callback, which by definition fires
// when there is room — so the retry always succeeds and never reassigns. With
// two, output A draining resubmits the node while output B is still full: the
// retry skips A (already delivered, tracked in pending_done_) and fails on B,
// and that is the reassignment that eats B's payload.
//
// Driven through raw channels rather than consumer nodes so each step is
// forced rather than raced: B is pre-filled and stays full for exactly as long
// as the test wants it to.
namespace {

struct TwoPayloads {
    static constexpr std::string_view label() { return "two_payloads"; }
    std::tuple<std::vector<int>, std::vector<int>> operator()() {
        return {std::vector<int>(4, 1), std::vector<int>(4, 2)};
    }
};

}  // namespace

TEST_CASE("a twice-parked value keeps its payload", "[pool_node][backpressure]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();

    TwoPayloads fn;
    auto node = make_pool_node(fn, pool);

    // Both capacity 1. A must be *full* for its pop to signal space at all —
    // Channel fires the space callback only on the full->not-full edge, so a
    // roomy A would never resubmit the node and the retry would never happen.
    Channel<std::vector<int>> out_a(1), out_b(1);
    node.set_output_channel<0>(&out_a);
    node.set_output_channel<1>(&out_b);

    // B is full before the node ever runs, so the very first firing parks.
    out_b.push(std::vector<int>(4, 99));

    node.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(50));

    // Draining A resubmits the node while B is still full: this is the retry
    // that reassigned the tuple to itself.
    (void)out_a.pop();
    std::this_thread::sleep_for(std::chrono::milliseconds(50));

    // Now let B through and collect what the node had been holding for it.
    (void)out_b.pop();                       // the pre-fill
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    std::vector<int> parked = out_b.pop();   // the value parked across two tries

    node.stop();
    pool->stop();

    INFO("parked payload size " << parked.size());
    CHECK(parked.size() == 4);
    if (parked.size() == 4) CHECK(parked[0] == 2);
}