KPN/include/kpn/channel.hpp

#pragma once
#include "diagnostics.hpp"
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <string>
#include <type_traits>

namespace kpn {

// ── Storage policy ────────────────────────────────────────────────────────────

template<typename T>
struct channel_storage_policy {
    static constexpr bool by_value =
        std::is_trivially_copyable_v<T> && sizeof(T) <= 8;
};

template<typename T>
using channel_storage_t = std::conditional_t<
    channel_storage_policy<T>::by_value,
    T,
    std::shared_ptr<const T>
>;

// ── Exceptions ────────────────────────────────────────────────────────────────

class ChannelOverflowError : public std::runtime_error {
public:
    explicit ChannelOverflowError(std::size_t capacity)
        : std::runtime_error("channel overflow: capacity " + std::to_string(capacity) +
                             " exceeded") {}
    ChannelOverflowError(std::size_t capacity, std::string context)
        : std::runtime_error(std::move(context) + ": capacity " + std::to_string(capacity) +
                             " exceeded") {}
};

class ChannelClosedError : public std::runtime_error {
public:
    ChannelClosedError() : std::runtime_error("channel closed") {}
};

// ── Channel ───────────────────────────────────────────────────────────────────

template<typename T>
class Channel {
public:
    using storage_type = channel_storage_t<T>;

    explicit Channel(std::size_t capacity = 5) : capacity_(capacity) {}

    Channel(const Channel&)            = delete;
    Channel& operator=(const Channel&) = delete;

    // Push a value.
    // - If channel is disabled (accepting_ == false): silently drop, return immediately.
    // - If channel is full: throw ChannelOverflowError.
    void push(T value) {
        if (!accepting_.load(std::memory_order_relaxed)) {
            stats_.record_drop();
            return;
        }
        std::unique_lock lock(mutex_);
        if (!accepting_.load(std::memory_order_relaxed)) {
            stats_.record_drop();
            return;
        }
        if (queue_.size() >= capacity_) {
            stats_.record_overflow();
            throw ChannelOverflowError(capacity_);
        }
        queue_.push(make_storage(std::move(value)));
        stats_.record_push(queue_.size());
        lock.unlock();
        cv_.notify_one();
    }

    // Blocking pop. Unblocks when an item is available or the channel is disabled.
    // Throws ChannelClosedError if disabled and queue is empty.
    T pop() {
        std::unique_lock lock(mutex_);
        cv_.wait(lock, [this] {
            return !queue_.empty() || !accepting_.load(std::memory_order_relaxed);
        });
        if (queue_.empty())
            throw ChannelClosedError{};
        T value = extract(std::move(queue_.front()));
        queue_.pop();
        stats_.record_pop();
        return value;
    }

    // Non-blocking pop with timeout. For watchdog/display use only — not used in run_loop.
    bool try_pop(T& out, std::chrono::milliseconds timeout) {
        std::unique_lock lock(mutex_);
        if (!cv_.wait_for(lock, timeout, [this] {
                return !queue_.empty() || !accepting_.load(std::memory_order_relaxed);
            }))
            return false;
        if (queue_.empty())
            return false;
        out = extract(std::move(queue_.front()));
        queue_.pop();
        stats_.record_pop();
        return true;
    }

    // Enable the channel (called by consumer node on start()).
    void enable() {
        accepting_.store(true, std::memory_order_relaxed);
    }

    // Disable the channel: drop all queued items, unblock any waiting pop().
    // Called by consumer node on stop(). Producer push() will silently drop after this.
    void disable() {
        accepting_.store(false, std::memory_order_relaxed);
        {
            std::lock_guard lock(mutex_);
            while (!queue_.empty()) queue_.pop();
        }
        cv_.notify_all();
    }

    std::size_t size() const {
        std::lock_guard lock(mutex_);
        return queue_.size();
    }

    std::size_t capacity()     const { return capacity_; }
    bool        is_accepting() const { return accepting_.load(std::memory_order_relaxed); }
    const ChannelStats& stats() const { return stats_; }

    ChannelSnapshot snapshot(const std::string& name) const {
        std::lock_guard lock(mutex_);
        return {
            name,
            capacity_,
            queue_.size(),
            stats_.peak_fill.load(std::memory_order_relaxed),
            stats_.pushes.load(std::memory_order_relaxed),
            stats_.drops.load(std::memory_order_relaxed),
            stats_.overflows.load(std::memory_order_relaxed),
            stats_.pops.load(std::memory_order_relaxed),
            sizeof(T),   // payload bytes — sizeof(T) regardless of storage policy
        };
    }

private:
    static storage_type make_storage(T&& v) {
        if constexpr (channel_storage_policy<T>::by_value)
            return std::move(v);
        else
            return std::make_shared<const T>(std::move(v));
    }

    static T extract(storage_type&& s) {
        if constexpr (channel_storage_policy<T>::by_value)
            return std::move(s);
        else
            return *s;
    }

    const std::size_t            capacity_;
    std::queue<storage_type>     queue_;
    std::atomic<bool>            accepting_{true};
    mutable std::mutex           mutex_;
    std::condition_variable      cv_;
    ChannelStats                 stats_;
};

} // namespace kpn