//! High-performance optimized complex type parsing for M3
//!
//! This module provides SIMD-optimized, vectorized parsing for complex types
//! with performance targets:
//! - Complex type parsing: <2x slower than primitive types
//! - Memory usage: <1.5x increase for complex type storage
//! - Throughput: >100 MB/s for complex type SSTable parsing
//! - Latency: <10ms additional latency for complex type queries

use super::types::{parse_cql_value, CqlTypeId};
use super::vint::{parse_vint, parse_vint_length};
use crate::types::Value;
use nom::{bytes::complete::take, combinator::map_res, number::complete::be_u8, IResult};
use std::collections::HashMap;
use std::sync::Arc;

#[cfg(target_arch = "x86_64")]
use std::arch::x86_64::*;

/// Performance-optimized complex type parser
pub struct OptimizedComplexTypeParser {
    /// Enable SIMD optimizations
    pub enable_simd: bool,
    /// Batch size for vectorized operations
    pub batch_size: usize,
    /// Performance metrics
    metrics: Arc<PerformanceMetrics>,
}

/// Performance tracking metrics
#[derive(Debug, Default)]
pub struct PerformanceMetrics {
    pub list_parse_count: std::sync::atomic::AtomicU64,
    pub list_parse_time_ns: std::sync::atomic::AtomicU64,
    pub map_parse_count: std::sync::atomic::AtomicU64,
    pub map_parse_time_ns: std::sync::atomic::AtomicU64,
    pub udt_parse_count: std::sync::atomic::AtomicU64,
    pub udt_parse_time_ns: std::sync::atomic::AtomicU64,
    pub simd_operations: std::sync::atomic::AtomicU64,
    pub cache_hits: std::sync::atomic::AtomicU64,
    pub cache_misses: std::sync::atomic::AtomicU64,
}

impl OptimizedComplexTypeParser {
    /// Create a new optimized parser
    pub fn new() -> Self {
        Self {
            enable_simd: Self::detect_simd_support(),
            batch_size: 16, // Process 16 elements at a time
            metrics: Arc::new(PerformanceMetrics::default()),
        }
    }

    /// Detect SIMD support on current platform
    fn detect_simd_support() -> bool {
        #[cfg(target_arch = "x86_64")]
        {
            is_x86_feature_detected!("sse2") && is_x86_feature_detected!("avx2")
        }
        #[cfg(not(target_arch = "x86_64"))]
        {
            false
        }
    }

    /// Parse a list with vectorized optimizations
    pub fn parse_optimized_list<'a>(&'a self, input: &'a [u8]) -> IResult<&'a [u8], Value> {
        let start_time = std::time::Instant::now();

        let (input, count) = parse_vint_length(input)?;
        let (input, element_type) = map_res(be_u8, CqlTypeId::try_from)(input)?;

        if count == 0 {
            return Ok((input, Value::List(Vec::new())));
        }

        let result = if self.enable_simd && count >= self.batch_size {
            self.parse_list_simd(input, count, element_type)
        } else {
            self.parse_list_sequential(input, count, element_type)
        };

        // Update metrics
        self.metrics
            .list_parse_count
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.metrics.list_parse_time_ns.fetch_add(
            start_time.elapsed().as_nanos() as u64,
            std::sync::atomic::Ordering::Relaxed,
        );

        result
    }

    /// SIMD-optimized list parsing for supported types
    #[cfg(target_arch = "x86_64")]
    fn parse_list_simd<'a>(
        &self,
        input: &'a [u8],
        count: usize,
        element_type: CqlTypeId,
    ) -> IResult<&'a [u8], Value> {
        match element_type {
            CqlTypeId::Int => self.parse_int_list_simd(input, count),
            CqlTypeId::Float => self.parse_float_list_simd(input, count),
            CqlTypeId::BigInt => self.parse_bigint_list_simd(input, count),
            _ => self.parse_list_sequential(input, count, element_type),
        }
    }

    #[cfg(not(target_arch = "x86_64"))]
    fn parse_list_simd<'a>(
        &self,
        input: &'a [u8],
        count: usize,
        element_type: CqlTypeId,
    ) -> IResult<&'a [u8], Value> {
        self.parse_list_sequential(input, count, element_type)
    }

    /// SIMD-optimized integer list parsing
    #[cfg(target_arch = "x86_64")]
    fn parse_int_list_simd<'a>(
        &self,
        mut input: &'a [u8],
        count: usize,
    ) -> IResult<&'a [u8], Value> {
        let mut elements = Vec::with_capacity(count);
        let mut remaining = count;

        unsafe {
            // Process 8 integers at a time using AVX2
            while remaining >= 8 && input.len() >= 32 {
                if is_x86_feature_detected!("avx2") {
                    // Load 8 32-bit big-endian integers
                    let chunk = _mm256_loadu_si256(input.as_ptr() as *const __m256i);

                    // Convert from big-endian to little-endian
                    let swapped = self.simd_bswap_epi32(chunk);

                    // Extract values
                    let values: [i32; 8] = std::mem::transmute(swapped);
                    for &val in &values {
                        elements.push(Value::Integer(val));
                    }

                    input = &input[32..];
                    remaining -= 8;

                    self.metrics
                        .simd_operations
                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }
        }

        // Handle remaining elements sequentially
        for _ in 0..remaining {
            let (new_input, value) = parse_cql_value(input, CqlTypeId::Int)?;
            elements.push(value);
            input = new_input;
        }

        Ok((input, Value::List(elements)))
    }

    /// SIMD byte swap for 32-bit integers
    #[cfg(target_arch = "x86_64")]
    unsafe fn simd_bswap_epi32(&self, chunk: __m256i) -> __m256i {
        // AVX2 byte swap using shuffle
        unsafe {
            let shuffle_mask = _mm256_set_epi8(
                12, 13, 14, 15, 8, 9, 10, 11, 4, 5, 6, 7, 0, 1, 2, 3, 12, 13, 14, 15, 8, 9, 10, 11,
                4, 5, 6, 7, 0, 1, 2, 3,
            );
            _mm256_shuffle_epi8(chunk, shuffle_mask)
        }
    }

    /// SIMD-optimized float list parsing
    #[cfg(target_arch = "x86_64")]
    fn parse_float_list_simd<'a>(
        &self,
        mut input: &'a [u8],
        count: usize,
    ) -> IResult<&'a [u8], Value> {
        let mut elements = Vec::with_capacity(count);
        let mut remaining = count;

        unsafe {
            // Process 8 floats at a time using AVX2
            while remaining >= 8 && input.len() >= 32 {
                if is_x86_feature_detected!("avx2") {
                    // Load 8 32-bit big-endian floats
                    let chunk = _mm256_loadu_si256(input.as_ptr() as *const __m256i);

                    // Convert from big-endian to little-endian
                    let swapped = self.simd_bswap_epi32(chunk);
                    let floats = _mm256_castsi256_ps(swapped);

                    // Extract values
                    let values: [f32; 8] = std::mem::transmute(floats);
                    for &val in &values {
                        elements.push(Value::Float(val as f64));
                    }

                    input = &input[32..];
                    remaining -= 8;

                    self.metrics
                        .simd_operations
                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }
        }

        // Handle remaining elements sequentially
        for _ in 0..remaining {
            let (new_input, value) = parse_cql_value(input, CqlTypeId::Float)?;
            elements.push(value);
            input = new_input;
        }

        Ok((input, Value::List(elements)))
    }

    /// SIMD-optimized big integer list parsing
    #[cfg(target_arch = "x86_64")]
    fn parse_bigint_list_simd<'a>(
        &self,
        mut input: &'a [u8],
        count: usize,
    ) -> IResult<&'a [u8], Value> {
        let mut elements = Vec::with_capacity(count);
        let mut remaining = count;

        unsafe {
            // Process 4 64-bit integers at a time using AVX2
            while remaining >= 4 && input.len() >= 32 {
                if is_x86_feature_detected!("avx2") {
                    // Load 4 64-bit big-endian integers
                    let chunk = _mm256_loadu_si256(input.as_ptr() as *const __m256i);

                    // Convert from big-endian to little-endian for 64-bit values
                    let swapped = self.simd_bswap_epi64(chunk);

                    // Extract values
                    let values: [i64; 4] = std::mem::transmute(swapped);
                    for &val in &values {
                        elements.push(Value::BigInt(val));
                    }

                    input = &input[32..];
                    remaining -= 4;

                    self.metrics
                        .simd_operations
                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }
        }

        // Handle remaining elements sequentially
        for _ in 0..remaining {
            let (new_input, value) = parse_cql_value(input, CqlTypeId::BigInt)?;
            elements.push(value);
            input = new_input;
        }

        Ok((input, Value::List(elements)))
    }

    /// SIMD byte swap for 64-bit integers
    #[cfg(target_arch = "x86_64")]
    unsafe fn simd_bswap_epi64(&self, chunk: __m256i) -> __m256i {
        // AVX2 byte swap for 64-bit values
        unsafe {
            let shuffle_mask = _mm256_set_epi8(
                8, 9, 10, 11, 12, 13, 14, 15, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15,
                0, 1, 2, 3, 4, 5, 6, 7,
            );
            _mm256_shuffle_epi8(chunk, shuffle_mask)
        }
    }

    /// Sequential list parsing fallback
    fn parse_list_sequential<'a>(
        &self,
        mut input: &'a [u8],
        count: usize,
        element_type: CqlTypeId,
    ) -> IResult<&'a [u8], Value> {
        let mut elements = Vec::with_capacity(count);

        for _ in 0..count {
            let (new_input, element) = parse_cql_value(input, element_type)?;
            elements.push(element);
            input = new_input;
        }

        Ok((input, Value::List(elements)))
    }

    /// Parse a map with optimized memory layout
    pub fn parse_optimized_map<'a>(&self, input: &'a [u8]) -> IResult<&'a [u8], Value> {
        let start_time = std::time::Instant::now();

        let (input, count) = parse_vint_length(input)?;
        let (input, key_type) = map_res(be_u8, CqlTypeId::try_from)(input)?;
        let (input, value_type) = map_res(be_u8, CqlTypeId::try_from)(input)?;

        if count == 0 {
            return Ok((input, Value::Map(Vec::new())));
        }

        // Use pre-allocated buffer for efficiency
        let mut map = Vec::with_capacity(count);
        let mut remaining_input = input;

        // Batch processing for better cache locality
        let batch_size = self.batch_size.min(count);
        for chunk_start in (0..count).step_by(batch_size) {
            let chunk_end = (chunk_start + batch_size).min(count);

            for _ in chunk_start..chunk_end {
                // Parse key with length prefix (same as regular map parser)
                let (new_input, key_length) = parse_vint(remaining_input)?;
                if key_length < 0 {
                    return Err(nom::Err::Error(nom::error::Error::new(
                        remaining_input,
                        nom::error::ErrorKind::Verify,
                    )));
                }
                let (new_input, key_data) = take(key_length as usize)(new_input)?;
                let key = super::types::parse_cql_value_raw(key_data, key_type)?.1;

                // Parse value with length prefix (same as regular map parser)
                let (new_input, value_length) = parse_vint(new_input)?;
                if value_length < 0 {
                    return Err(nom::Err::Error(nom::error::Error::new(
                        new_input,
                        nom::error::ErrorKind::Verify,
                    )));
                }
                let (new_input, value_data) = take(value_length as usize)(new_input)?;
                let value = super::types::parse_cql_value_raw(value_data, value_type)?.1;

                map.push((key, value));
                remaining_input = new_input;
            }
        }

        // Update metrics
        self.metrics
            .map_parse_count
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.metrics.map_parse_time_ns.fetch_add(
            start_time.elapsed().as_nanos() as u64,
            std::sync::atomic::Ordering::Relaxed,
        );

        Ok((remaining_input, Value::Map(map)))
    }

    /// Parse a UDT with field caching optimization
    pub fn parse_optimized_udt<'a>(&self, input: &'a [u8]) -> IResult<&'a [u8], Value> {
        let start_time = std::time::Instant::now();

        let (input, type_name_len) = parse_vint_length(input)?;
        let (input, type_name_bytes) = take(type_name_len)(input)?;
        let type_name = String::from_utf8(type_name_bytes.to_vec()).map_err(|_| {
            nom::Err::Error(nom::error::Error::new(input, nom::error::ErrorKind::Verify))
        })?;

        let (input, field_count) = parse_vint_length(input)?;

        // Use HashMap with pre-allocated capacity for better performance
        let mut fields = HashMap::with_capacity(field_count);
        let mut remaining_input = input;

        for _ in 0..field_count {
            let (new_input, field_name_len) = parse_vint_length(remaining_input)?;
            let (new_input, field_name_bytes) = take(field_name_len)(new_input)?;
            let field_name = String::from_utf8(field_name_bytes.to_vec()).map_err(|_| {
                nom::Err::Error(nom::error::Error::new(
                    new_input,
                    nom::error::ErrorKind::Verify,
                ))
            })?;

            let (new_input, field_type) = map_res(be_u8, CqlTypeId::try_from)(new_input)?;
            let (new_input, field_value) = parse_cql_value(new_input, field_type)?;

            fields.insert(field_name, field_value);
            remaining_input = new_input;
        }

        // Update metrics
        self.metrics
            .udt_parse_count
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.metrics.udt_parse_time_ns.fetch_add(
            start_time.elapsed().as_nanos() as u64,
            std::sync::atomic::Ordering::Relaxed,
        );

        let udt_value = crate::types::UdtValue {
            type_name,
            keyspace: "default".to_string(), // TODO: Get from context
            fields: fields
                .into_iter()
                .map(|(name, value)| crate::types::UdtField {
                    name,
                    value: Some(value),
                })
                .collect(),
        };
        Ok((remaining_input, Value::Udt(udt_value)))
    }

    /// Parse a set (optimized similar to list)
    pub fn parse_optimized_set<'a>(&'a self, input: &'a [u8]) -> IResult<&'a [u8], Value> {
        // Sets are stored similarly to lists, so we can reuse list optimization
        let (input, list_value) = self.parse_optimized_list(input)?;

        if let Value::List(elements) = list_value {
            Ok((input, Value::Set(elements)))
        } else {
            Err(nom::Err::Error(nom::error::Error::new(
                input,
                nom::error::ErrorKind::Verify,
            )))
        }
    }

    /// Parse a tuple with memory-efficient layout
    pub fn parse_optimized_tuple<'a>(&self, input: &'a [u8]) -> IResult<&'a [u8], Value> {
        let (input, element_count) = parse_vint_length(input)?;

        let mut elements = Vec::with_capacity(element_count);
        let mut remaining_input = input;

        for _ in 0..element_count {
            let (new_input, element_type) = map_res(be_u8, CqlTypeId::try_from)(remaining_input)?;
            let (new_input, element) = parse_cql_value(new_input, element_type)?;

            elements.push(element);
            remaining_input = new_input;
        }

        Ok((remaining_input, Value::Tuple(elements)))
    }

    /// Get performance metrics
    pub fn get_metrics(&self) -> &PerformanceMetrics {
        &self.metrics
    }

    /// Generate performance report
    pub fn generate_performance_report(&self) -> String {
        use std::sync::atomic::Ordering;

        let list_count = self.metrics.list_parse_count.load(Ordering::Relaxed);
        let list_time = self.metrics.list_parse_time_ns.load(Ordering::Relaxed);
        let map_count = self.metrics.map_parse_count.load(Ordering::Relaxed);
        let map_time = self.metrics.map_parse_time_ns.load(Ordering::Relaxed);
        let udt_count = self.metrics.udt_parse_count.load(Ordering::Relaxed);
        let udt_time = self.metrics.udt_parse_time_ns.load(Ordering::Relaxed);
        let simd_ops = self.metrics.simd_operations.load(Ordering::Relaxed);

        format!(
            "Complex Type Parser Performance Report\n\
            =======================================\n\
            SIMD Support: {}\n\
            Batch Size: {}\n\n\
            Lists Parsed: {} (avg: {:.2} μs each)\n\
            Maps Parsed: {} (avg: {:.2} μs each)\n\
            UDTs Parsed: {} (avg: {:.2} μs each)\n\
            SIMD Operations: {}\n\n\
            Total Parse Time: {:.2} ms\n\
            Average Throughput: {:.2} MB/s",
            self.enable_simd,
            self.batch_size,
            list_count,
            if list_count > 0 {
                list_time as f64 / list_count as f64 / 1000.0
            } else {
                0.0
            },
            map_count,
            if map_count > 0 {
                map_time as f64 / map_count as f64 / 1000.0
            } else {
                0.0
            },
            udt_count,
            if udt_count > 0 {
                udt_time as f64 / udt_count as f64 / 1000.0
            } else {
                0.0
            },
            simd_ops,
            (list_time + map_time + udt_time) as f64 / 1_000_000.0,
            self.calculate_throughput()
        )
    }

    /// Calculate average throughput in MB/s
    fn calculate_throughput(&self) -> f64 {
        use std::sync::atomic::Ordering;

        let total_time_ns = self.metrics.list_parse_time_ns.load(Ordering::Relaxed)
            + self.metrics.map_parse_time_ns.load(Ordering::Relaxed)
            + self.metrics.udt_parse_time_ns.load(Ordering::Relaxed);

        if total_time_ns > 0 {
            // Estimate data processed (rough approximation)
            let estimated_bytes = (self.metrics.list_parse_count.load(Ordering::Relaxed) * 100)
                + (self.metrics.map_parse_count.load(Ordering::Relaxed) * 200)
                + (self.metrics.udt_parse_count.load(Ordering::Relaxed) * 150);

            (estimated_bytes as f64 * 1_000_000_000.0) / (total_time_ns as f64 * 1_000_000.0)
        } else {
            0.0
        }
    }
}

impl Default for OptimizedComplexTypeParser {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::parser::vint::encode_vint;

    #[test]
    fn test_optimized_list_parsing() {
        let parser = OptimizedComplexTypeParser::new();

        // Create test data for integer list
        let mut data = Vec::new();
        data.extend_from_slice(&encode_vint(3)); // count = 3
        data.push(CqlTypeId::Int as u8); // element type

        // Add 3 integers
        for i in 1i32..=3 {
            data.extend_from_slice(&i.to_be_bytes());
        }

        let (_, result) = parser.parse_optimized_list(&data).unwrap();
        if let Value::List(elements) = result {
            assert_eq!(elements.len(), 3);
            assert_eq!(elements[0], Value::Integer(1));
            assert_eq!(elements[1], Value::Integer(2));
            assert_eq!(elements[2], Value::Integer(3));
        } else {
            panic!("Expected List value");
        }
    }

    #[test]
    fn test_optimized_map_parsing() {
        let parser = OptimizedComplexTypeParser::new();

        // Create test data for string->int map
        let mut data = Vec::new();
        data.extend_from_slice(&encode_vint(2)); // count = 2
        data.push(CqlTypeId::Varchar as u8); // key type
        data.push(CqlTypeId::Int as u8); // value type

        // Add key-value pairs
        // "key1" -> 42
        data.extend_from_slice(&encode_vint(4)); // key length
        data.extend_from_slice(b"key1"); // key data
        data.extend_from_slice(&encode_vint(4)); // value length (int is 4 bytes)
        data.extend_from_slice(&42i32.to_be_bytes()); // value data

        // "key2" -> 84
        data.extend_from_slice(&encode_vint(4)); // key length
        data.extend_from_slice(b"key2"); // key data
        data.extend_from_slice(&encode_vint(4)); // value length (int is 4 bytes)
        data.extend_from_slice(&84i32.to_be_bytes()); // value data

        let (_, result) = parser.parse_optimized_map(&data).unwrap();
        if let Value::Map(pairs) = result {
            assert_eq!(pairs.len(), 2);
        } else {
            panic!("Expected Map value");
        }
    }

    #[test]
    fn test_performance_metrics() {
        let parser = OptimizedComplexTypeParser::new();

        // Parse some data to generate metrics
        let mut data = Vec::new();
        data.extend_from_slice(&encode_vint(1));
        data.push(CqlTypeId::Int as u8);
        data.extend_from_slice(&42i32.to_be_bytes());

        let _ = parser.parse_optimized_list(&data).unwrap();

        let metrics = parser.get_metrics();
        assert_eq!(
            metrics
                .list_parse_count
                .load(std::sync::atomic::Ordering::Relaxed),
            1
        );
    }

    #[test]
    fn test_simd_detection() {
        let simd_support = OptimizedComplexTypeParser::detect_simd_support();
        println!("SIMD support detected: {}", simd_support);
        // This test just verifies the detection doesn't panic
    }

    #[test]
    fn test_performance_report() {
        let parser = OptimizedComplexTypeParser::new();
        let report = parser.generate_performance_report();
        assert!(report.contains("Complex Type Parser Performance Report"));
        assert!(report.contains("SIMD Support:"));
    }
}
