KPN/include/kpn/interrupt_node.hpp

#pragma once
#include "channel.hpp"
#include "diagnostics.hpp"
#include "fixed_string.hpp"
#include "inode.hpp"
#include "port.hpp"
#include "scheduler.hpp"
#include "traits.hpp"

#include <array>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <iostream>
#include <memory>
#include <string>
#include <tuple>
#include <type_traits>

namespace kpn {

// ── InterruptNode ─────────────────────────────────────────────────────────────
//
// A source node (zero inputs) driven by an external event — camera frame ready,
// timer tick, socket data, etc. — rather than self-submission.
//
// Usage:
//   auto node = make_interrupt_node<produce_frame>(pool, out<"frame">{});
//   camera_sdk.on_frame_ready(node.get_trigger());  // register with external source
//   network.add("camera", node).connect(...).build().start();
//
// The trigger callable is safe to call from any thread, including signal handlers,
// provided the underlying scheduler's submit() is signal-safe. After fire_once()
// completes the node is idle until the next trigger fires — it does NOT busy-loop.

template<auto Func,
         typename OutputTag = out<>,
         fixed_string Label    = "",
         std::size_t UniqueTag = 0>
class InterruptNode;

template<auto Func, fixed_string... OutNames, fixed_string Label, std::size_t UniqueTag>
class InterruptNode<Func, out<OutNames...>, Label, UniqueTag> : public INode {
public:
    using F            = decltype(Func);
    using return_raw   = return_t<F>;
    using return_tuple = normalised_return_t<return_raw>;

    static_assert(arity_v<F> == 0,
        "InterruptNode function must take no arguments (it has no input channels)");

    static constexpr std::string_view label()     { return Label.view(); }
    static constexpr std::size_t      unique_tag  = UniqueTag;
    static constexpr std::size_t      input_count  = 0;
    static constexpr std::size_t      output_count = std::tuple_size_v<return_tuple>;

    static_assert(
        sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
        "make_interrupt_node: number of output names must match return tuple size, or provide none"
    );

    explicit InterruptNode(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5)
        : scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
    {}

    ~InterruptNode() override { stop(); }

    // ── INode ─────────────────────────────────────────────────────────────────

    void start() override {
        stop_flag_.store(false, std::memory_order_relaxed);
        pending_.store(0, std::memory_order_relaxed);
        // Does NOT self-submit — waits for first external trigger.
    }

    void stop() override {
        stop_flag_.store(true, std::memory_order_seq_cst);
        // In-flight fire_once() observes stop_flag_ on its next check.
    }

    bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); }
    void set_name(std::string name) override { name_ = std::move(name); }
    void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
    void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }

    void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
    void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
    void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
    void set_network_closed_callback(NodeEventCallback cb)   override { closed_callbacks_[1] = std::move(cb); }

    const NodeStats& stats() const override { return stats_; }

    NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
        uint64_t frames   = stats_.frames_processed.load(std::memory_order_relaxed);
        double exec_ms    = stats_.ema_exec_us.load(std::memory_order_relaxed) / 1000.0;
        double qwait_ms   = stats_.queue_wait_us.load(std::memory_order_relaxed) / 1000.0;
        double total_ms   = exec_ms; // no blocked time for interrupt nodes
        return {
            name, frames, exec_ms,
            stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0,
            0.0,  // blocked_ms — not applicable
            elapsed_s > 0 ? frames / elapsed_s : 0.0,
            stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
            total_ms > 0 ? 100.0 : 0.0,
            qwait_ms,
            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
        };
    }

    // ── Port access — by index ────────────────────────────────────────────────

    template<std::size_t I>
    OutputPort<InterruptNode, I> output() {
        static_assert(I < output_count, "output index out of range");
        return {*this};
    }

    template<fixed_string Name>
    auto output() {
        constexpr std::size_t idx = index_of<Name, OutNames...>();
        static_assert(idx != npos, "unknown output port name");
        return output<idx>();
    }

    template<std::size_t I>
    void set_output_channel(Channel<std::tuple_element_t<I, return_tuple>>* ch) {
        std::get<I>(output_channels_) = ch;
    }

    // ── Trigger ───────────────────────────────────────────────────────────────

    // Returns a callable that fires this node when called.
    // Pass it to a camera SDK, timer, or any external event source.
    // Thread-safe; may be called from any thread.
    std::function<void()> get_trigger() {
        return [this] { trigger(); };
    }

private:
    // Each trigger() increments pending_. When going 0→1 a task is submitted.
    // Each fire_once() handles one pending event and decrements; if more remain
    // (old value > 1) it resubmits itself. This guarantees every trigger produces
    // exactly one execution even if triggers arrive faster than fire_once completes.
    void trigger() {
        if (stop_flag_.load(std::memory_order_relaxed)) return;
        if (pending_.fetch_add(1, std::memory_order_acq_rel) == 0)
            scheduler_->submit([this] { fire_once(); });
    }

    void fire_once() {
        if (stop_flag_.load(std::memory_order_relaxed)) {
            pending_.store(0, std::memory_order_release);
            return;
        }

        auto t0 = clock_t::now();
        int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
            t0.time_since_epoch()).count();
        stats_.exec_start_us.store(now_us, std::memory_order_relaxed);

        bool fatal = false;
        try {
            auto t1 = clock_t::now();
            stats_.record_queue_wait(duration_t(t1 - t0));
            auto cpu0 = NodeStats::cpu_now();

            if constexpr (std::is_void_v<return_raw>) {
                Func();
            } else {
                auto result = Func();
                push_outputs(normalise(std::move(result)),
                             std::make_index_sequence<output_count>{});
            }

            auto cpu1 = NodeStats::cpu_now();
            auto t2   = clock_t::now();
            stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
        } catch (const ChannelOverflowError&) {
            fire_callbacks(event_callbacks_);
        } catch (...) {
            if (!error_handler_ || !error_handler_(name_, std::current_exception()))
                fatal = true;
        }

        stats_.exec_start_us.store(0, std::memory_order_relaxed);

        if (fatal) {
            fire_callbacks(closed_callbacks_);
            disable_outputs(std::make_index_sequence<output_count>{});
            pending_.store(0, std::memory_order_release);
            stop_flag_.store(true, std::memory_order_relaxed);
            return;
        }

        // Decrement and resubmit only if more triggers are queued.
        // fetch_sub returns old value; old > 1 means new > 0.
        if (pending_.fetch_sub(1, std::memory_order_acq_rel) > 1)
            scheduler_->submit([this] { fire_once(); });
    }

    template<typename R = return_raw>
    static return_tuple normalise(R&& r) {
        if constexpr (is_tuple_v<R>) return std::move(r);
        else                         return std::make_tuple(std::move(r));
    }

    template<std::size_t... Is>
    void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
        (push_one<Is>(std::get<Is>(std::move(result))), ...);
    }

    template<std::size_t I>
    void push_one(std::tuple_element_t<I, return_tuple>&& val) {
        auto* ch = std::get<I>(output_channels_);
        if (!ch) return;
        try { ch->push(std::move(val)); }
        catch (const ChannelOverflowError&) {
            throw ChannelOverflowError(ch->capacity(),
                "interrupt node '" + name_ + "' " + output_port_label<I>());
        }
    }

    template<std::size_t I>
    static std::string output_port_label() {
        if constexpr (sizeof...(OutNames) > 0) {
            constexpr std::array<std::string_view, sizeof...(OutNames)> names{OutNames.view()...};
            return std::string("output['") + std::string(names[I]) + "']";
        } else {
            return "output[" + std::to_string(I) + "]";
        }
    }

    template<typename Tup, std::size_t... Is>
    static auto make_output_channel_tuple(std::index_sequence<Is...>)
        -> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;

    using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
        std::make_index_sequence<output_count>{}));

    template<std::size_t... Is>
    void disable_outputs(std::index_sequence<Is...>) {
        auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
        (disable_one(std::get<Is>(output_channels_)), ...);
    }

    static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
        const auto ts = std::chrono::steady_clock::now();
        for (auto& cb : cbs) if (cb) cb(ts);
    }

    std::shared_ptr<IScheduler>              scheduler_;
    std::string                              name_;
    std::size_t                              fifo_capacity_;
    output_channels_t                        output_channels_{};
    std::atomic<bool>                        stop_flag_{true};
    std::atomic<int>                         pending_{0};
    NodeStats                                stats_;
    NodeErrorHandler                         error_handler_;
    std::chrono::milliseconds                max_exec_time_{0};
    std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
    std::array<NodeEventCallback, 2>         closed_callbacks_{};
};

// ── make_interrupt_node factory ───────────────────────────────────────────────

template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0>
auto make_interrupt_node(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5) {
    return InterruptNode<Func, out<>, Label, UniqueTag>(std::move(sched), fifo_capacity);
}

template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
         fixed_string... OutNames>
auto make_interrupt_node(std::shared_ptr<IScheduler> sched, out<OutNames...>,
                         std::size_t fifo_capacity = 5) {
    return InterruptNode<Func, out<OutNames...>, Label, UniqueTag>(
        std::move(sched), fifo_capacity);
}

} // namespace kpn