// ============================================================
// COMPLETE SHARED MEMORY LAYOUT (2000 Quote Entries)
// ============================================================

// Section 1: Change Notification Header (Cache-Line Aligned)
#[repr(C, align(64))]
struct ChangeNotification {
    write_index: AtomicU32,       // Writer increments (wraps at 512)
    read_index: AtomicU32,        // Reader updates after processing
    _padding: [u8; 56],           // Pad to 64 bytes (1 cache line)
}

// Section 2: Ring Buffer for Changed Pairs
#[repr(C, align(64))]
struct ChangedPairNotification {
    pair_index: u32,              // Index in quotes array (0-1999)
    version: u64,                 // Quote version (for double-check)
    timestamp_unix_ms: u64,       // When quote changed
    _padding: [u8; 44],           // Pad to 64 bytes (cache-line aligned)
}

// Section 3: Quote Metadata Array
#[repr(C, align(128))]
struct QuoteMetadata {
    version: AtomicU64,           // Atomic versioning (odd=writing, even=readable)
    pair_id: [u8; 32],            // BLAKE3(token_in, token_out, amount)
    input_mint: [u8; 32],         // Solana public key
    output_mint: [u8; 32],        // Solana public key
    input_amount: u64,            // Lamports
    output_amount: u64,           // Lamports
    price_impact_bps: u32,        // Basis points (0.01%)
    timestamp_unix_ms: u64,       // Unix timestamp (milliseconds)
    route_id: [u8; 32],           // BLAKE3(route_steps) -> Redis/PostgreSQL
    oracle_price_usd: f64,        // Oracle price for validation (NEW)
    staleness_flag: u8,           // 0=fresh, 1=stale, 2=very_stale (NEW)
    _padding: [u8; 15],           // Padding to 128 bytes
}

// ============================================================
// Complete Memory Map
// ============================================================
// Offset 0-63:         ChangeNotification header (64 bytes)
// Offset 64-32,831:    ChangedPairNotification[512] ring buffer (64 bytes × 512 = 32,768 bytes)
// Offset 32,832+:      QuoteMetadata[2000] array (128 bytes × 2000 = 256,000 bytes)
//
// Total size: 64 + 32,768 + 256,000 = 288,832 bytes (~282 KB)
// ✅ Fits comfortably in L2 cache (256-512 KB typical)

// Naive approach: Scan ALL 2000 quotes every time
for i in 0..2000 {
    let quote = &quotes[i];
    let version = quote.version.load(Ordering::Acquire);
    // Process quote...
}

// Performance:
// - Memory access: 2000 × 128 bytes = 256 KB sequential read
// - L2 cache hit: ~2μs (entire array in L2)
// - L3 cache hit: ~10μs (if evicted from L2)

// With ring buffer, scanner can process changes in priority order
pub fn process_changes_by_priority(&mut self) -> Vec<ArbitrageOpportunity> {
    let changed_quotes = self.read_changed_quotes();

    // Sort by priority: High-value pairs first
    let mut sorted = changed_quotes.clone();
    sorted.sort_by_key(|(idx, quote)| {
        // Priority: SOL/USDC > SOL/USDT > smaller pairs
        get_pair_priority(quote.input_mint, quote.output_mint)
    });

    // Process high-priority pairs first (get arbitrage opportunities faster)
    for (idx, quote) in sorted {
        if let Some(opp) = self.detect_arbitrage(quote) {
            return vec![opp];  // Early exit on first profitable opportunity
        }
    }
}

pub struct QuoteReader {
    local_quotes: &'static [QuoteMetadata; 2000],
    external_quotes: &'static [QuoteMetadata; 2000],
    change_notification: &'static ChangeNotification,
    changed_pairs: &'static [ChangedPairNotification; 512],

    // ✅ Adaptive strategy state
    consecutive_large_batches: u32,
    last_strategy: Strategy,
}

#[derive(Debug, Clone, Copy)]
enum Strategy {
    EarlyReturn,   // No changes
    RingBuffer,    // Low/medium changes
    FullScan,      // High changes or overflow
}

impl QuoteReader {
    /// ✅ MAIN ENTRY POINT: Adaptive change detection
    pub fn read_changed_quotes(&mut self) -> Vec<(u32, &QuoteMetadata)> {
        let read_index = self.change_notification.read_index.load(Ordering::Acquire);
        let write_index = self.change_notification.write_index.load(Ordering::Acquire);

        // Calculate number of changes
        let num_changes = self.calculate_changes(read_index, write_index);

        // ✅ ADAPTIVE STRATEGY: Choose best approach
        let strategy = self.select_strategy(num_changes);

        match strategy {
            Strategy::EarlyReturn => {
                // No changes - fastest path (10ns)
                Vec::new()
            }
            Strategy::RingBuffer => {
                // Low/medium changes - process ring buffer
                self.ring_buffer_scan(read_index, write_index, num_changes)
            }
            Strategy::FullScan => {
                // High changes or overflow - full scan
                self.full_scan_optimized()
            }
        }
    }

    /// ✅ Strategy selection with adaptive thresholds
    fn select_strategy(&mut self, num_changes: u32) -> Strategy {
        const RING_BUFFER_SIZE: u32 = 512;
        const FULL_SCAN_THRESHOLD: u32 = 400;  // 20% of 2000 quotes

        if num_changes == 0 {
            self.consecutive_large_batches = 0;
            self.last_strategy = Strategy::EarlyReturn;
            Strategy::EarlyReturn
        } else if num_changes >= RING_BUFFER_SIZE {
            // Overflow - full scan fallback
            log::warn!("Ring buffer overflow ({} changes), full scan", num_changes);
            self.last_strategy = Strategy::FullScan;
            Strategy::FullScan
        } else if num_changes > FULL_SCAN_THRESHOLD {
            // High update rate - full scan is faster
            self.consecutive_large_batches += 1;

            // If consistently high update rate (>10 consecutive), stay in full scan mode
            if self.consecutive_large_batches > 10 {
                log::info!("Switching to full scan mode (consistently high updates)");
            }

            self.last_strategy = Strategy::FullScan;
            Strategy::FullScan
        } else {
            // Low/medium update rate - ring buffer is faster
            self.consecutive_large_batches = 0;
            self.last_strategy = Strategy::RingBuffer;
            Strategy::RingBuffer
        }
    }

    /// Calculate number of changes from ring buffer indices
    fn calculate_changes(&self, read_index: u32, write_index: u32) -> u32 {
        if write_index >= read_index {
            write_index - read_index
        } else {
            512 - read_index + write_index  // Wrapped
        }
    }

    /// ❗ CRITICAL: Safe quote read with torn read prevention
    ///
    /// Implements double-read verification to prevent reading partially-written structs.
    /// This is MANDATORY for correctness in lock-free shared memory.
    ///
    /// Protocol:
    /// 1. Read version (v1) before reading struct
    /// 2. Skip if v1 is odd (write in progress)
    /// 3. Copy entire struct
    /// 4. Read version (v2) after copying struct
    /// 5. Accept only if v1 == v2 (no concurrent write)
    fn read_quote_safe(&self, quote: &QuoteMetadata) -> Option<QuoteMetadata> {
        // Standard lock-free read protocol with retry
        for _ in 0..10 {  // Max 10 retries (prevents infinite loop)
            // Step 1: Read version BEFORE reading struct
            let v1 = quote.version.load(Ordering::Acquire);

            // Step 2: Reject if version is odd (write in progress)
            if v1 % 2 != 0 {
                std::hint::spin_loop();  // Writer is actively writing
                continue;
            }

            // Step 3: Read the entire struct
            // ⚠️ This may be a torn read if writer starts during copy
            let quote_copy = QuoteMetadata {
                version: AtomicU64::new(v1),
                pair_id: quote.pair_id,
                input_mint: quote.input_mint,
                output_mint: quote.output_mint,
                input_amount: quote.input_amount,
                output_amount: quote.output_amount,
                price_impact_bps: quote.price_impact_bps,
                timestamp_unix_ms: quote.timestamp_unix_ms,
                route_id: quote.route_id,
                oracle_price_usd: quote.oracle_price_usd,
                staleness_flag: quote.staleness_flag,
                _padding: [0u8; 15],
            };

            // Step 4: Read version AGAIN after copying
            let v2 = quote.version.load(Ordering::Acquire);

            // Step 5: Verify consistency
            if v1 == v2 {
                // ✅ SUCCESS: Struct was not modified during read
                return Some(quote_copy);
            }

            // ❌ RETRY: Writer modified version while we were reading
            std::hint::spin_loop();
        }

        // Failed after 10 retries (extremely rare, indicates heavy write contention)
        None
    }

    /// ✅ Ring buffer scan (low/medium changes) with torn read prevention
    fn ring_buffer_scan(
        &mut self,
        read_index: u32,
        write_index: u32,
        num_changes: u32,
    ) -> Vec<(u32, QuoteMetadata)> {
        let mut changed_quotes = Vec::with_capacity(num_changes as usize);

        let mut current_index = (read_index + 1) % 512;
        while current_index != ((write_index + 1) % 512) {
            let notification = &self.changed_pairs[current_index as usize];
            let pair_index = notification.pair_index;

            if pair_index >= 2000 {
                // Invalid index, skip
                current_index = (current_index + 1) % 512;
                continue;
            }

            let quote = &self.local_quotes[pair_index as usize];

            // ❗ CRITICAL: Use safe read with torn read prevention
            if let Some(quote_copy) = self.read_quote_safe(quote) {
                // ✅ Additional validation after safe read
                // 1. Version matches notification (consistency)
                // 2. Quote is not very stale
                let version = quote_copy.version.load(Ordering::Relaxed);
                if version == notification.version && quote_copy.staleness_flag < 2 {
                    changed_quotes.push((pair_index, quote_copy));
                }
            }

            current_index = (current_index + 1) % 512;
        }

        // ✅ Update read index (reader caught up)
        self.change_notification.read_index.store(write_index, Ordering::Release);

        changed_quotes
    }

    /// ✅ Full scan (high changes or overflow fallback)
    fn full_scan_optimized(&self) -> Vec<(u32, &QuoteMetadata)> {
        let mut changed_quotes = Vec::with_capacity(2000);

        // ✅ Optimized full scan with SIMD hints and early filtering
        for (idx, quote) in self.local_quotes.iter().enumerate() {
            let version = quote.version.load(Ordering::Acquire);

            // Fast path: Skip empty or being-written quotes
            if version == 0 || version % 2 != 0 {
                continue;
            }

            // Skip very stale quotes
            if quote.staleness_flag >= 2 {
                continue;
            }

            changed_quotes.push((idx as u32, quote));
        }

        changed_quotes
    }
}

pub struct ArbitrageScanner {
    quote_reader: QuoteReader,
    min_profit_bps: u32,  // Minimum profit (basis points)
}

impl ArbitrageScanner {
    /// Main arbitrage detection loop
    pub fn scan_for_arbitrage(&mut self) -> Vec<ArbitrageOpportunity> {
        let mut opportunities = Vec::new();

        // ✅ ADAPTIVE: Get changed quotes (uses ring buffer or full scan)
        let changed_quotes = self.quote_reader.read_changed_quotes();

        if changed_quotes.is_empty() {
            return opportunities;  // No updates, return early (10ns path)
        }

        // ✅ Process only changed quotes (1-10 typically, 100 max)
        for (pair_index, quote) in changed_quotes {
            // Fast oracle validation (no external calls - data in shared memory)
            let quote_price = (quote.output_amount as f64) / (quote.input_amount as f64);
            let deviation = (quote_price - quote.oracle_price_usd).abs() / quote.oracle_price_usd;

            if deviation > 0.01 {
                continue;  // >1% deviation from oracle, skip
            }

            // Check for arbitrage opportunities
            if let Some(opp) = self.detect_arbitrage(quote) {
                opportunities.push(opp);
            }
        }

        opportunities
    }
}

// Adjust based on production metrics
const RING_BUFFER_SIZE: u32 = 512;           // Ring buffer capacity
const FULL_SCAN_THRESHOLD: u32 = 400;        // 20% of 2000 quotes
const CONSECUTIVE_FULL_SCAN_THRESHOLD: u32 = 10;  // Switch to full scan mode

// If metrics show:
// - Ring buffer overflow >1% → Increase RING_BUFFER_SIZE to 1024
// - Full scan triggered >50% → Decrease FULL_SCAN_THRESHOLD to 300
// - Early return <80% → Normal (good sign of sparse updates)