//! Zero-copy ring buffer for relay packet forwarding (R-02).
//!
//! A lock-free, fixed-capacity ring buffer that avoids heap allocation
//! on the packet forwarding hot path. Packets are written into pre-allocated
//! slots; readers receive a reference rather than a copy.

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

/// Fixed-size packet slot — pre-allocated to avoid hot-path malloc.
const MAX_PACKET_SIZE: usize = 1500; // MTU-sized slots
const RING_CAPACITY: usize = 4096; // Must be power of 2

pub struct PacketSlot {
    pub len: usize,
    pub data: [u8; MAX_PACKET_SIZE],
}

impl Default for PacketSlot {
    fn default() -> Self {
        Self {
            len: 0,
            data: [0; MAX_PACKET_SIZE],
        }
    }
}

/// Single-producer, single-consumer ring buffer for relay forwarding.
pub struct RingBuffer {
    slots: Box<[PacketSlot; RING_CAPACITY]>,
    head: AtomicUsize, // writer advances head
    tail: AtomicUsize, // reader advances tail
}

impl RingBuffer {
    pub fn new() -> Box<Self> {
        // SAFETY: PacketSlot is all-zero safe
        let slots = unsafe {
            let layout = std::alloc::Layout::array::<PacketSlot>(RING_CAPACITY).unwrap();
            let ptr = std::alloc::alloc_zeroed(layout) as *mut [PacketSlot; RING_CAPACITY];
            Box::from_raw(ptr)
        };
        Box::new(Self {
            slots,
            head: AtomicUsize::new(0),
            tail: AtomicUsize::new(0),
        })
    }

    /// Write a packet into the ring buffer.
    /// Returns false if the buffer is full (producer must drop or wait).
    pub fn push(&mut self, data: &[u8]) -> bool {
        assert!(data.len() <= MAX_PACKET_SIZE, "packet exceeds MTU");

        let head = self.head.load(Ordering::Relaxed);
        let next_head = (head + 1) & (RING_CAPACITY - 1);

        if next_head == self.tail.load(Ordering::Acquire) {
            return false; // Buffer full
        }

        let slot = &mut self.slots[head];
        slot.len = data.len();
        slot.data[..data.len()].copy_from_slice(data);

        self.head.store(next_head, Ordering::Release);
        true
    }

    /// Read the next packet from the ring buffer.
    /// Returns None if the buffer is empty.
    pub fn pop(&mut self) -> Option<&[u8]> {
        let tail = self.tail.load(Ordering::Relaxed);

        if tail == self.head.load(Ordering::Acquire) {
            return None; // Buffer empty
        }

        let slot = &self.slots[tail];
        let data = &slot.data[..slot.len];

        let next_tail = (tail + 1) & (RING_CAPACITY - 1);
        self.tail.store(next_tail, Ordering::Release);

        Some(data)
    }

    /// Returns the number of packets currently in the buffer.
    pub fn len(&self) -> usize {
        let head = self.head.load(Ordering::Acquire);
        let tail = self.tail.load(Ordering::Acquire);
        (head + RING_CAPACITY - tail) & (RING_CAPACITY - 1)
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn capacity() -> usize {
        RING_CAPACITY
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_push_pop_single() {
        let mut buf = RingBuffer::new();
        assert!(buf.push(b"hello"));
        let data = buf.pop().unwrap();
        assert_eq!(data, b"hello");
        assert!(buf.pop().is_none());
    }

    #[test]
    fn test_fifo_ordering() {
        let mut buf = RingBuffer::new();
        for i in 0..5u8 {
            assert!(buf.push(&[i; 10]));
        }
        for i in 0..5u8 {
            let data = buf.pop().unwrap();
            assert_eq!(data, &[i; 10]);
        }
    }

    #[test]
    fn test_mtu_sized_packet() {
        let mut buf = RingBuffer::new();
        let big = vec![0xFFu8; 1500];
        assert!(buf.push(&big));
        let out = buf.pop().unwrap();
        assert_eq!(out.len(), 1500);
        assert!(out.iter().all(|&b| b == 0xFF));
    }

    #[test]
    fn test_len_tracking() {
        let mut buf = RingBuffer::new();
        assert_eq!(buf.len(), 0);
        buf.push(b"a");
        buf.push(b"b");
        assert_eq!(buf.len(), 2);
        buf.pop();
        assert_eq!(buf.len(), 1);
    }
}
