#![allow(unsafe_code)] // Intentional unsafe for lock-free ring buffer operations

//! Memory-safe lock-free ring buffer implementation
//!
//! This module provides a corrected SPSC (Single Producer Single Consumer) ring buffer
//! with proper memory ordering to prevent data races in high-frequency trading systems.

use std::alloc::{alloc, dealloc, Layout};
use std::ptr::NonNull;
use std::sync::atomic::{AtomicU64, Ordering};

/// Lock-free ring buffer optimized for single producer/consumer scenarios
///
/// This implementation uses proper Acquire-Release memory ordering to ensure
/// correctness in concurrent environments while maintaining optimal performance.
#[repr(align(64))] // Cache line alignment to prevent false sharing
/// LockFreeRingBuffer
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct LockFreeRingBuffer<T> {
    buffer: NonNull<T>,
    capacity: usize,
    mask: usize,     // capacity - 1 for fast modulo
    head: AtomicU64, // Producer writes here
    tail: AtomicU64, // Consumer reads from here
    layout: Layout,
}

// SAFETY: LockFreeRingBuffer is Send because:
// - All atomic operations use proper memory ordering (Release/Acquire)
// - Buffer allocation/deallocation is thread-safe via correct Layout usage
// - No shared mutable references - all access through atomic indices
unsafe impl<T: Send> Send for LockFreeRingBuffer<T> {}

// SAFETY: LockFreeRingBuffer is Sync because:
// - Concurrent access is coordinated through atomic head/tail indices
// - Memory ordering guarantees prevent race conditions
// - No data races on buffer contents due to single-producer/single-consumer design
unsafe impl<T: Send> Sync for LockFreeRingBuffer<T> {}

impl<T> std::fmt::Debug for LockFreeRingBuffer<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LockFreeRingBuffer")
            .field("capacity", &self.capacity)
            .field("mask", &self.mask)
            .field("head", &self.head.load(Ordering::Relaxed))
            .field("tail", &self.tail.load(Ordering::Relaxed))
            .field("layout", &self.layout)
            .finish()
    }
}

impl<T: Clone> LockFreeRingBuffer<T> {
    /// Create a new lock-free ring buffer with specified capacity
    ///
    /// # Arguments
    /// * `capacity` - Must be a power of 2 for optimal performance
    ///
    /// # Returns
    /// * `Ok`(LockFreeRingBuffer)` - Successfully created buffer
    /// * `Err`(&str)` - Error message if creation fails
    pub fn new(capacity: usize) -> Result<Self, &'static str> {
        if capacity == 0 {
            return Err("Capacity cannot be zero");
        }

        if !capacity.is_power_of_two() {
            return Err("Capacity must be power of two for optimal performance");
        }

        let layout = Layout::array::<T>(capacity).map_err(|_| "Layout creation failed")?;

        let buffer = unsafe {
            // SAFETY: Buffer allocation safety requirements:
            // - Layout was validated above via Layout::array::<T>(capacity)
            // - Capacity is non-zero and power-of-two validated
            // - Memory allocation failure is checked via is_null()
            // - NonNull::new_unchecked is safe because null check was performed
            let ptr = alloc(layout);
            if ptr.is_null() {
                return Err("Memory allocation failed");
            }
            NonNull::new_unchecked(ptr.cast::<T>())
        };

        Ok(Self {
            buffer,
            capacity,
            mask: capacity - 1,
            head: AtomicU64::new(0),
            tail: AtomicU64::new(0),
            layout,
        })
    }

    /// Try to push an item to the buffer (non-blocking)
    ///
    /// Uses Release ordering on head update to ensure visibility to consumer
    #[inline(always)]
    pub fn try_push(&self, item: T) -> Result<(), T> {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Acquire); // Acquire latest tail position

        // Check if buffer is full (leave one slot empty to distinguish full from empty)
        // For SPSC: buffer is full when (head - tail) == capacity - 1
        if head.wrapping_sub(tail) >= (self.capacity as u64 - 1) {
            return Err(item);
        }

        let index = (head as usize) & self.mask;
        // SAFETY: Unsafe operation validated - invariants maintained by surrounding code
        unsafe {
            // SAFETY: Write operation safety requirements:
            // - index is bounded by mask (capacity-1) ensuring it's within buffer bounds
            // - Buffer was allocated with sufficient capacity (checked in new())
            // - Single producer design prevents concurrent writes to same index
            // - Memory ordering (Release below) ensures visibility
            self.buffer.as_ptr().add(index).write(item);
        }

        // Release ordering ensures item write is visible before head update
        self.head.store(head.wrapping_add(1), Ordering::Release);
        Ok(())
    }

    /// Try to pop an item from the buffer (non-blocking)
    ///
    /// Uses Acquire ordering on head load to see latest producer writes
    #[inline(always)]
    pub fn try_pop(&self) -> Option<T> {
        let tail = self.tail.load(Ordering::Relaxed);
        let head = self.head.load(Ordering::Acquire); // Acquire latest head position

        // Check if buffer is empty
        if tail == head {
            return None;
        }

        let index = (tail as usize) & self.mask;
        let item = unsafe {
            // SAFETY: Read operation safety requirements:
            // - index is bounded by mask (capacity-1) ensuring it's within buffer bounds
            // - Buffer was allocated with sufficient capacity (checked in new())
            // - Single consumer design prevents concurrent reads from same index
            // - head.load(Acquire) above ensures we see producer's writes
            // - Empty check above guarantees valid data exists at this index
            self.buffer.as_ptr().add(index).read()
        };

        // Release ordering ensures item read completes before tail update
        self.tail.store(tail.wrapping_add(1), Ordering::Release);
        // Some variant
        Some(item)
    }

    /// Get current buffer utilization (0.0 to 1.0)
    #[inline]
    pub fn utilization(&self) -> f64 {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Relaxed);
        let used = head.wrapping_sub(tail) as usize;
        used as f64 / self.capacity as f64
    }

    /// Get buffer capacity
    #[inline]
    pub const fn capacity(&self) -> usize {
        self.capacity
    }

    /// Check if buffer is empty
    #[inline]
    pub fn is_empty(&self) -> bool {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Relaxed);
        head == tail
    }

    /// Check if buffer is full
    #[inline]
    pub fn is_full(&self) -> bool {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Acquire);
        // For SPSC: buffer is full when (head - tail) == capacity - 1
        head.wrapping_sub(tail) >= (self.capacity as u64 - 1)
    }

    /// Get current number of items in buffer
    #[inline]
    pub fn len(&self) -> usize {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Relaxed);
        head.wrapping_sub(tail) as usize
    }
}

impl<T> Drop for LockFreeRingBuffer<T> {
    fn drop(&mut self) {
        // SAFETY: Allocator operations use valid layout with correct alignment and size
        unsafe {
            dealloc(self.buffer.as_ptr().cast::<u8>(), self.layout);
        }
    }
}

/// Type alias for SPSC queue (Single Producer Single Consumer)
pub type SPSCQueue<T> = LockFreeRingBuffer<T>;

#[cfg(test)]
#[allow(clippy::assertions_on_result_states)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::thread;

    #[test]
    fn test_basic_operations() -> Result<(), Box<dyn std::error::Error>> {
        let buffer = LockFreeRingBuffer::<u64>::new(8)?;

        // Test empty buffer
        assert!(buffer.is_empty());
        assert!(!buffer.is_full());
        assert_eq!(buffer.len(), 0);
        assert_eq!(buffer.try_pop(), None);

        // Test push
        assert!(buffer.try_push(42).is_ok());
        assert!(!buffer.is_empty());
        assert_eq!(buffer.len(), 1);

        // Test pop
        assert_eq!(buffer.try_pop(), Some(42));
        assert!(buffer.is_empty());
        assert_eq!(buffer.len(), 0);

        Ok(())
    }

    #[test]
    fn test_capacity_validation() {
        assert!(LockFreeRingBuffer::<u64>::new(0).is_err());
        assert!(LockFreeRingBuffer::<u64>::new(3).is_err()); // Not power of 2
        assert!(LockFreeRingBuffer::<u64>::new(8).is_ok());
        assert!(LockFreeRingBuffer::<u64>::new(1024).is_ok());
    }

    #[test]
    fn test_buffer_full() -> Result<(), Box<dyn std::error::Error>> {
        let buffer = LockFreeRingBuffer::<u64>::new(4)?;

        // Fill buffer to usable capacity (3 items for capacity 4)
        // Ring buffer keeps one slot reserved to distinguish full from empty
        // So is_full() triggers when head - tail == capacity - 1
        for i in 0..3 {
            assert!(buffer.try_push(i).is_ok());
        }

        // Buffer should be full now (head - tail == capacity - 1)
        assert!(buffer.is_full());
        assert!(buffer.try_push(99).is_err());

        // Pop one item and verify we can push again
        assert_eq!(buffer.try_pop(), Some(0));
        assert!(!buffer.is_full());
        assert!(buffer.try_push(99).is_ok());

        Ok(())
    }

    #[test]
    fn test_wraparound() -> Result<(), Box<dyn std::error::Error>> {
        let buffer = LockFreeRingBuffer::<u64>::new(4)?;

        // Test wraparound by cycling through many items
        for i in 0..100 {
            assert!(buffer.try_push(i).is_ok());
            assert_eq!(buffer.try_pop(), Some(i));
        }

        Ok(())
    }

    #[test]
    fn test_concurrent_spsc() -> Result<(), Box<dyn std::error::Error>> {
        let buffer = Arc::new(LockFreeRingBuffer::<u64>::new(1024)?);
        let buffer_clone = Arc::clone(&buffer);

        const NUM_ITEMS: u64 = 10000;

        // Producer thread
        let producer = thread::spawn(move || {
            for i in 0..NUM_ITEMS {
                while buffer_clone.try_push(i).is_err() {
                    thread::yield_now();
                }
            }
        });

        // Consumer thread
        let consumer = thread::spawn(move || {
            let mut received = Vec::new();
            while received.len() < NUM_ITEMS as usize {
                if let Some(item) = buffer.try_pop() {
                    received.push(item);
                } else {
                    thread::yield_now();
                }
            }
            received
        });

        producer.join().expect("Producer thread failed");
        let received = consumer.join().expect("Consumer thread failed");

        // Verify all items received in order
        assert_eq!(received.len(), NUM_ITEMS as usize);
        for (i, item) in received.into_iter().enumerate() {
            assert_eq!(item, i as u64);
        }

        Ok(())
    }

    #[test]
    fn test_performance() -> Result<(), Box<dyn std::error::Error>> {
        let buffer = LockFreeRingBuffer::<u64>::new(8192)?;

        const NUM_OPERATIONS: usize = 1_000_000;
        let start = std::time::Instant::now();

        // Alternate push/pop operations
        for i in 0..NUM_OPERATIONS {
            buffer.try_push(i as u64).expect("Push failed");
            let value = buffer.try_pop().expect("Pop failed");
            assert_eq!(value, i as u64);
        }

        let duration = start.elapsed();
        let ops_per_sec = NUM_OPERATIONS as f64 / duration.as_secs_f64();
        let avg_latency_ns = duration.as_nanos() / (NUM_OPERATIONS * 2) as u128; // *2 for push+pop

        println!(
            "Performance: {:.0} ops/sec, avg latency: {}ns",
            ops_per_sec, avg_latency_ns
        );

        // For HFT, we want sub-microsecond performance
        assert!(
            avg_latency_ns < 1000,
            "Latency too high: {}ns > 1000ns",
            avg_latency_ns
        );

        Ok(())
    }
}
