rivven-core 0.0.21

//! High-performance compression layer for Rivven
//!
//! This module provides a zero-copy, adaptive compression layer supporting:
//! - **LZ4**: Ultra-fast compression for latency-sensitive paths (~3GB/s)
//! - **Snappy**: Fast compression with Kafka compatibility (~1.5GB/s)
//! - **Zstd**: High-ratio compression for storage and network (~500MB/s)
//! - **None**: Passthrough for already-compressed or tiny payloads
//!
//! # Design Principles
//!
//! 1. **Adaptive Selection**: Automatically choose algorithm based on payload characteristics
//! 2. **Zero-Copy**: Use `Bytes` throughout to avoid unnecessary copies
//! 3. **Streaming Support**: Compress/decompress incrementally for large payloads
//! 4. **Header Format**: Minimal overhead (1-byte header for algorithm + optional size)
//! 5. **Kafka Compatibility**: Full support for Kafka's compression formats
//! 6. **Checksum Verification**: Optional CRC32 checksums for data integrity
//!
//! # Wire Format
//!
//! ```text
//! +-------+----------------+----------+------------------+
//! | Flags | Original Size  | Checksum | Compressed Data  |
//! | 1 byte| 4 bytes (opt)  | 4B (opt) | N bytes          |
//! +-------+----------------+----------+------------------+
//!
//! Flags byte:
//!   bits 0-2: Algorithm (000=None, 001=LZ4, 010=Zstd, 011=Snappy)
//!   bit 3:    Reserved
//!   bit 4:    Has original size (for decompression buffer allocation)
//!   bit 5:    Has checksum (CRC32)
//!   bits 6-7: Reserved
//! ```
//!
//! # Algorithm Comparison
//!
//! | Algorithm | Compress | Decompress | Ratio | Best For |
//! |-----------|----------|------------|-------|----------|
//! | LZ4       | ~800MB/s | ~4GB/s     | 2-3x  | Real-time streaming, lowest latency |
//! | Snappy    | ~500MB/s | ~1.5GB/s   | 2-3x  | Kafka compatibility, balanced |
//! | Zstd      | ~400MB/s | ~1GB/s     | 3-5x  | Storage, network, cold data |
//!
//! # Example
//!
//! ```rust,ignore
//! use rivven_core::compression::{Compressor, CompressionConfig, CompressionAlgorithm};
//!
//! // Create compressor with default settings (LZ4, adaptive)
//! let compressor = Compressor::new();
//!
//! // Compress data
//! let data = b"Hello, World! ".repeat(100);
//! let compressed = compressor.compress(&data)?;
//!
//! // Decompress (auto-detects algorithm)
//! let decompressed = compressor.decompress(&compressed)?;
//! assert_eq!(&decompressed[..], &data[..]);
//!
//! // Use specific algorithm
//! let snappy_compressed = compressor.compress_with(&data, CompressionAlgorithm::Snappy)?;
//! ```

use bytes::{BufMut, Bytes, BytesMut};
use std::io::{Read, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use thiserror::Error;

#[cfg(feature = "compression")]
use snap;

// ============================================================================
// Error Types
// ============================================================================

/// Compression-related errors
#[derive(Debug, Error)]
pub enum CompressionError {
    #[error("LZ4 compression failed: {0}")]
    Lz4Error(String),

    #[error("Zstd compression failed: {0}")]
    ZstdError(String),

    #[error("Snappy compression failed: {0}")]
    SnappyError(String),

    #[error("Invalid compression header")]
    InvalidHeader,

    #[error("Decompression buffer too small: need {needed}, have {available}")]
    BufferTooSmall { needed: usize, available: usize },

    #[error("Unknown compression algorithm: {0}")]
    UnknownAlgorithm(u8),

    #[error("Checksum mismatch: expected {expected:#010x}, got {actual:#010x}")]
    ChecksumMismatch { expected: u32, actual: u32 },

    #[error("Data corruption detected")]
    DataCorruption,

    #[error("Decompression bomb detected: original_size {size} exceeds maximum {max}")]
    DecompressionBomb { size: usize, max: usize },

    #[error("Payload too large: {size} bytes exceeds maximum {max}")]
    PayloadTooLarge { size: usize, max: usize },

    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),
}

pub type Result<T> = std::result::Result<T, CompressionError>;

// ============================================================================
// Compression Algorithm
// ============================================================================

/// Compression algorithm selection
///
/// Supports all major compression algorithms used in distributed systems,
/// with full Kafka protocol compatibility.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Hash)]
#[repr(u8)]
pub enum CompressionAlgorithm {
    /// No compression (passthrough)
    /// Use for: Pre-compressed data, tiny payloads, or when CPU is critical
    #[default]
    None = 0,

    /// LZ4 - Ultra-fast, moderate compression ratio
    /// Best for: Real-time streaming, low-latency paths
    /// Speed: ~800 MB/s compress, ~4 GB/s decompress
    /// Ratio: 2-3x typical
    Lz4 = 1,

    /// Zstd - Excellent balance of speed and compression ratio
    /// Best for: Storage, network transfers, cold data
    /// Speed: ~400 MB/s compress, ~1 GB/s decompress
    /// Ratio: 3-5x typical (up to 10x at high levels)
    Zstd = 2,

    /// Snappy - Fast compression, Kafka-compatible
    /// Best for: Kafka compatibility, balanced workloads
    /// Speed: ~500 MB/s compress, ~1.5 GB/s decompress
    /// Ratio: 2-3x typical
    Snappy = 3,
}

impl CompressionAlgorithm {
    /// All supported algorithms
    pub const ALL: [CompressionAlgorithm; 4] = [
        CompressionAlgorithm::None,
        CompressionAlgorithm::Lz4,
        CompressionAlgorithm::Zstd,
        CompressionAlgorithm::Snappy,
    ];

    /// Parse from flags byte
    pub fn from_flags(flags: u8) -> Result<Self> {
        match flags & 0x07 {
            0 => Ok(Self::None),
            1 => Ok(Self::Lz4),
            2 => Ok(Self::Zstd),
            3 => Ok(Self::Snappy),
            n => Err(CompressionError::UnknownAlgorithm(n)),
        }
    }

    /// Convert to flags byte
    pub fn to_flags(self, has_size: bool, has_checksum: bool) -> u8 {
        let mut flags = self as u8;
        if has_size {
            flags |= 0x10; // Set bit 4
        }
        if has_checksum {
            flags |= 0x20; // Set bit 5
        }
        flags
    }

    /// Get human-readable name
    pub fn name(&self) -> &'static str {
        match self {
            Self::None => "none",
            Self::Lz4 => "lz4",
            Self::Zstd => "zstd",
            Self::Snappy => "snappy",
        }
    }

    /// Parse from string (case-insensitive)
    pub fn parse(s: &str) -> Option<Self> {
        match s.to_lowercase().as_str() {
            "none" | "uncompressed" => Some(Self::None),
            "lz4" => Some(Self::Lz4),
            "zstd" | "zstandard" => Some(Self::Zstd),
            "snappy" => Some(Self::Snappy),
            _ => None,
        }
    }

    /// Get Kafka protocol compression type ID
    pub fn kafka_type_id(&self) -> i8 {
        match self {
            Self::None => 0,
            Self::Lz4 => 3,    // Kafka LZ4 = 3
            Self::Zstd => 4,   // Kafka Zstd = 4
            Self::Snappy => 2, // Kafka Snappy = 2
        }
    }

    /// Create from Kafka protocol compression type ID
    pub fn from_kafka_type_id(id: i8) -> Option<Self> {
        match id {
            0 => Some(Self::None),
            2 => Some(Self::Snappy),
            3 => Some(Self::Lz4),
            4 => Some(Self::Zstd),
            _ => None,
        }
    }

    /// Check if this algorithm is actually compressing (not passthrough)
    pub fn is_compressed(&self) -> bool {
        !matches!(self, Self::None)
    }
}

impl std::fmt::Display for CompressionAlgorithm {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.name())
    }
}

impl std::str::FromStr for CompressionAlgorithm {
    type Err = CompressionError;

    fn from_str(s: &str) -> Result<Self> {
        Self::parse(s).ok_or(CompressionError::UnknownAlgorithm(0))
    }
}

// ============================================================================
// Compression Level
// ============================================================================

/// Compression level presets
///
/// Different algorithms interpret levels differently:
/// - **LZ4**: Higher acceleration = faster but less compression
/// - **Zstd**: Higher level = better compression but slower (1-22)
/// - **Snappy**: Single level (no configuration)
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CompressionLevel {
    /// Fastest compression, lowest ratio
    Fast,
    /// Balanced speed and ratio (default)
    #[default]
    Default,
    /// Best compression ratio, slower
    Best,
    /// Custom level (algorithm-specific)
    Custom(i32),
}

impl CompressionLevel {
    /// Get LZ4 acceleration parameter (higher = faster, lower ratio)
    pub fn lz4_acceleration(&self) -> i32 {
        match self {
            Self::Fast => 65537, // Max acceleration
            Self::Default => 1,  // Default
            Self::Best => 1,     // LZ4 doesn't have "best", use default
            Self::Custom(n) => *n,
        }
    }

    /// Get Zstd compression level (1-22, higher = better ratio)
    pub fn zstd_level(&self) -> i32 {
        match self {
            Self::Fast => 1,
            Self::Default => 3, // Zstd default
            Self::Best => 19,   // High compression
            Self::Custom(n) => n.clamp(&1, &22).to_owned(),
        }
    }

    /// Get human-readable name
    pub fn name(&self) -> &'static str {
        match self {
            Self::Fast => "fast",
            Self::Default => "default",
            Self::Best => "best",
            Self::Custom(_) => "custom",
        }
    }
}

// ============================================================================
// Compressor Configuration
// ============================================================================

/// Configuration for the compression layer
#[derive(Debug, Clone)]
pub struct CompressionConfig {
    /// Default algorithm for new data
    pub algorithm: CompressionAlgorithm,
    /// Compression level
    pub level: CompressionLevel,
    /// Minimum payload size to compress (bytes)
    /// Payloads smaller than this are stored uncompressed
    pub min_size: usize,
    /// Compression ratio threshold (0.0-1.0)
    /// If compressed size > original * threshold, store uncompressed
    pub ratio_threshold: f32,
    /// Enable adaptive algorithm selection
    pub adaptive: bool,
    /// Enable checksum verification
    pub checksum: bool,
}

impl Default for CompressionConfig {
    fn default() -> Self {
        Self {
            algorithm: CompressionAlgorithm::Lz4,
            level: CompressionLevel::Default,
            min_size: 64,          // Don't compress < 64 bytes
            ratio_threshold: 0.95, // Must achieve at least 5% reduction
            adaptive: true,
            checksum: false, // Disabled by default for performance
        }
    }
}

impl CompressionConfig {
    /// Create a new configuration builder
    pub fn builder() -> CompressionConfigBuilder {
        CompressionConfigBuilder::default()
    }

    /// Create config optimized for low latency
    pub fn low_latency() -> Self {
        Self {
            algorithm: CompressionAlgorithm::Lz4,
            level: CompressionLevel::Fast,
            min_size: 128,
            ratio_threshold: 0.90,
            adaptive: false,
            checksum: false,
        }
    }

    /// Create config optimized for storage efficiency
    pub fn storage() -> Self {
        Self {
            algorithm: CompressionAlgorithm::Zstd,
            level: CompressionLevel::Default,
            min_size: 32,
            ratio_threshold: 0.98,
            adaptive: true,
            checksum: true, // Enable checksums for storage
        }
    }

    /// Create config optimized for network transfer
    pub fn network() -> Self {
        Self {
            algorithm: CompressionAlgorithm::Zstd,
            level: CompressionLevel::Fast,
            min_size: 64,
            ratio_threshold: 0.95,
            adaptive: true,
            checksum: false,
        }
    }

    /// Create config for Kafka compatibility
    pub fn kafka_compatible() -> Self {
        Self {
            algorithm: CompressionAlgorithm::Snappy,
            level: CompressionLevel::Default,
            min_size: 64,
            ratio_threshold: 0.95,
            adaptive: false,
            checksum: false, // Kafka has its own checksums
        }
    }
}

/// Builder for CompressionConfig
#[derive(Debug, Default)]
pub struct CompressionConfigBuilder {
    config: CompressionConfig,
}

impl CompressionConfigBuilder {
    pub fn algorithm(mut self, algorithm: CompressionAlgorithm) -> Self {
        self.config.algorithm = algorithm;
        self
    }

    pub fn level(mut self, level: CompressionLevel) -> Self {
        self.config.level = level;
        self
    }

    pub fn min_size(mut self, size: usize) -> Self {
        self.config.min_size = size;
        self
    }

    pub fn ratio_threshold(mut self, threshold: f32) -> Self {
        self.config.ratio_threshold = threshold.clamp(0.0, 1.0);
        self
    }

    pub fn adaptive(mut self, enabled: bool) -> Self {
        self.config.adaptive = enabled;
        self
    }

    pub fn checksum(mut self, enabled: bool) -> Self {
        self.config.checksum = enabled;
        self
    }

    pub fn build(self) -> CompressionConfig {
        self.config
    }
}

// ============================================================================
// Core Compression Functions
// ============================================================================

/// Compress data using LZ4 (pure Rust via lz4_flex)
fn compress_lz4(data: &[u8], _level: CompressionLevel) -> Result<Vec<u8>> {
    // lz4_flex uses the standard LZ4 block format at maximum speed.
    // Compression level is not configurable — LZ4 is designed for speed.
    Ok(lz4_flex::block::compress_prepend_size(data))
}

/// Decompress LZ4 data
///
/// Validates the prepended uncompressed size against `MAX_DECOMPRESSION_SIZE`
/// before allocating to prevent decompression-bomb DoS attacks.
fn decompress_lz4(data: &[u8], _original_size: Option<usize>) -> Result<Vec<u8>> {
    // The first 4 bytes of LZ4 block format contain the uncompressed size
    // (little-endian u32). Validate before letting lz4_flex allocate.
    if data.len() >= 4 {
        let claimed_size = u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;
        if claimed_size > MAX_DECOMPRESSION_SIZE {
            return Err(CompressionError::DecompressionBomb {
                size: claimed_size,
                max: MAX_DECOMPRESSION_SIZE,
            });
        }
    }
    lz4_flex::block::decompress_size_prepended(data)
        .map_err(|e| CompressionError::Lz4Error(e.to_string()))
}

/// Compress data using Zstd
fn compress_zstd(data: &[u8], level: CompressionLevel) -> Result<Vec<u8>> {
    let level = level.zstd_level();

    zstd::bulk::compress(data, level).map_err(|e| CompressionError::ZstdError(e.to_string()))
}

/// Maximum decompression output size (256 MiB) to prevent decompression bombs.
const MAX_DECOMPRESSION_SIZE: usize = 256 * 1024 * 1024;

/// Decompress Zstd data.
///
/// Uses `original_size` from the header when available (capped at `MAX_DECOMPRESSION_SIZE`)
/// to allocate the correct buffer. Falls back to a conservative 16 MB limit when the
/// original size is unknown.
fn decompress_zstd(data: &[u8], original_size: Option<usize>) -> Result<Vec<u8>> {
    let capacity = match original_size {
        Some(size) => {
            // Validate original_size against maximum to prevent decompression bombs
            if size > MAX_DECOMPRESSION_SIZE {
                return Err(CompressionError::DecompressionBomb {
                    size,
                    max: MAX_DECOMPRESSION_SIZE,
                });
            }
            size
        }
        None => 16 * 1024 * 1024, // 16 MB fallback when size unknown
    };
    zstd::bulk::decompress(data, capacity).map_err(|e| CompressionError::ZstdError(e.to_string()))
}

/// Compress data using Snappy
fn compress_snappy(data: &[u8]) -> Result<Vec<u8>> {
    let mut encoder = snap::raw::Encoder::new();
    encoder
        .compress_vec(data)
        .map_err(|e| CompressionError::SnappyError(e.to_string()))
}

/// Decompress Snappy data
///
/// Validates decompressed length against `MAX_DECOMPRESSION_SIZE` before
/// allocating to prevent decompression-bomb DoS attacks.
fn decompress_snappy(data: &[u8]) -> Result<Vec<u8>> {
    // snap::raw::decompress_len reads the varint header without allocating.
    let decompressed_len = snap::raw::decompress_len(data)
        .map_err(|e| CompressionError::SnappyError(e.to_string()))?;
    if decompressed_len > MAX_DECOMPRESSION_SIZE {
        return Err(CompressionError::DecompressionBomb {
            size: decompressed_len,
            max: MAX_DECOMPRESSION_SIZE,
        });
    }
    let mut decoder = snap::raw::Decoder::new();
    decoder
        .decompress_vec(data)
        .map_err(|e| CompressionError::SnappyError(e.to_string()))
}

/// Calculate CRC32 checksum
#[inline]
fn crc32_checksum(data: &[u8]) -> u32 {
    crc32fast::hash(data)
}

// ============================================================================
// Compressor
// ============================================================================

/// High-performance compressor with configurable algorithms
///
/// The compressor supports adaptive algorithm selection, checksums,
/// and various optimization presets for different use cases.
#[derive(Debug, Clone)]
pub struct Compressor {
    config: CompressionConfig,
    stats: Arc<CompressionStatsCollector>,
}

impl Compressor {
    /// Create compressor with default config
    pub fn new() -> Self {
        Self {
            config: CompressionConfig::default(),
            stats: Arc::new(CompressionStatsCollector::new()),
        }
    }

    /// Create compressor with custom config
    pub fn with_config(config: CompressionConfig) -> Self {
        Self {
            config,
            stats: Arc::new(CompressionStatsCollector::new()),
        }
    }

    /// Get the current configuration
    pub fn config(&self) -> &CompressionConfig {
        &self.config
    }

    /// Get compression statistics
    pub fn stats(&self) -> CompressionStatsSnapshot {
        self.stats.snapshot()
    }

    /// Compress data, returning compressed bytes with header
    pub fn compress(&self, data: &[u8]) -> Result<Bytes> {
        // Skip compression for small payloads
        if data.len() < self.config.min_size {
            self.stats.record_skipped(data.len());
            return Ok(self.encode_uncompressed(data));
        }

        // Select algorithm (adaptive or configured)
        let algorithm = if self.config.adaptive {
            self.select_algorithm(data)
        } else {
            self.config.algorithm
        };

        // Compress based on algorithm
        let compressed = match algorithm {
            CompressionAlgorithm::None => {
                self.stats.record_skipped(data.len());
                return Ok(self.encode_uncompressed(data));
            }
            CompressionAlgorithm::Lz4 => compress_lz4(data, self.config.level)?,
            CompressionAlgorithm::Zstd => compress_zstd(data, self.config.level)?,
            CompressionAlgorithm::Snappy => compress_snappy(data)?,
        };

        // Check if compression was worthwhile
        let ratio = compressed.len() as f32 / data.len() as f32;
        if ratio > self.config.ratio_threshold {
            // Compression didn't help enough, store uncompressed
            self.stats.record_skipped(data.len());
            return Ok(self.encode_uncompressed(data));
        }

        // Record stats
        self.stats
            .record_compression(algorithm, data.len(), compressed.len());

        // Encode with header
        self.encode_compressed(algorithm, data.len(), &compressed, data)
    }

    /// Compress data with explicit algorithm choice
    pub fn compress_with(&self, data: &[u8], algorithm: CompressionAlgorithm) -> Result<Bytes> {
        if algorithm == CompressionAlgorithm::None || data.len() < self.config.min_size {
            self.stats.record_skipped(data.len());
            return Ok(self.encode_uncompressed(data));
        }

        let compressed = match algorithm {
            CompressionAlgorithm::None => unreachable!(),
            CompressionAlgorithm::Lz4 => compress_lz4(data, self.config.level)?,
            CompressionAlgorithm::Zstd => compress_zstd(data, self.config.level)?,
            CompressionAlgorithm::Snappy => compress_snappy(data)?,
        };

        self.stats
            .record_compression(algorithm, data.len(), compressed.len());
        self.encode_compressed(algorithm, data.len(), &compressed, data)
    }

    /// Decompress data (auto-detects algorithm from header)
    pub fn decompress(&self, data: &[u8]) -> Result<Bytes> {
        if data.is_empty() {
            return Err(CompressionError::InvalidHeader);
        }

        let flags = data[0];
        let algorithm = CompressionAlgorithm::from_flags(flags)?;
        let has_size = (flags & 0x10) != 0;
        let has_checksum = (flags & 0x20) != 0;

        let mut offset = 1;

        // Parse original size if present
        let original_size = if has_size {
            if data.len() < offset + 4 {
                return Err(CompressionError::InvalidHeader);
            }
            let size_bytes: [u8; 4] = data[offset..offset + 4].try_into().unwrap();
            offset += 4;
            Some(u32::from_le_bytes(size_bytes) as usize)
        } else {
            None
        };

        // Parse checksum if present
        let expected_checksum = if has_checksum {
            if data.len() < offset + 4 {
                return Err(CompressionError::InvalidHeader);
            }
            let checksum_bytes: [u8; 4] = data[offset..offset + 4].try_into().unwrap();
            offset += 4;
            Some(u32::from_le_bytes(checksum_bytes))
        } else {
            None
        };

        let payload = &data[offset..];

        let decompressed = match algorithm {
            CompressionAlgorithm::None => payload.to_vec(),
            CompressionAlgorithm::Lz4 => decompress_lz4(payload, original_size)?,
            CompressionAlgorithm::Zstd => decompress_zstd(payload, original_size)?,
            CompressionAlgorithm::Snappy => decompress_snappy(payload)?,
        };

        // Verify checksum if present
        if let Some(expected) = expected_checksum {
            let actual = crc32_checksum(&decompressed);
            if actual != expected {
                return Err(CompressionError::ChecksumMismatch { expected, actual });
            }
        }

        self.stats
            .record_decompression(algorithm, payload.len(), decompressed.len());

        Ok(Bytes::from(decompressed))
    }

    /// Try to decompress data, returning the original bytes unchanged if
    /// decompression fails (e.g. for pre-existing uncompressed messages).
    ///
    /// backward-compatible decompression for consumer side.
    /// Messages produced with compression have a self-describing flags byte.
    /// Messages produced before compression was enabled lack this header and
    /// will fail decompression — `try_decompress` returns them as-is.
    pub fn try_decompress(&self, data: &[u8]) -> Bytes {
        match self.decompress(data) {
            Ok(decompressed) => decompressed,
            Err(_) => Bytes::copy_from_slice(data),
        }
    }

    /// Get compression analysis for data (without actually storing)
    pub fn analyze(&self, data: &[u8]) -> CompressionAnalysis {
        let lz4_result = compress_lz4(data, self.config.level);
        let zstd_result = compress_zstd(data, self.config.level);
        let snappy_result = compress_snappy(data);

        CompressionAnalysis {
            original_size: data.len(),
            lz4_size: lz4_result.as_ref().map(|v| v.len()).ok(),
            zstd_size: zstd_result.as_ref().map(|v| v.len()).ok(),
            snappy_size: snappy_result.as_ref().map(|v| v.len()).ok(),
            entropy: estimate_entropy(data),
            recommended: self.select_algorithm(data),
        }
    }

    /// Select best algorithm based on payload characteristics
    fn select_algorithm(&self, data: &[u8]) -> CompressionAlgorithm {
        // Heuristics for algorithm selection:
        // 1. Very small data: No compression
        // 2. Detect high entropy (random/encrypted): No compression
        // 3. Text-like data / large payloads: Zstd (better ratio)
        // 4. Medium entropy / medium size: Snappy or LZ4

        if data.len() < self.config.min_size {
            return CompressionAlgorithm::None;
        }

        // Quick entropy estimation using byte frequency
        let entropy = estimate_entropy(data);

        if entropy > 7.5 {
            // High entropy - likely already compressed or encrypted
            return CompressionAlgorithm::None;
        }

        if entropy < 4.5 || data.len() > 64 * 1024 {
            // Low entropy or large payload - Zstd shines
            return CompressionAlgorithm::Zstd;
        }

        if entropy < 6.0 {
            // Medium-low entropy - Snappy is a good balance
            return CompressionAlgorithm::Snappy;
        }

        // Default to LZ4 for speed
        CompressionAlgorithm::Lz4
    }

    /// Encode uncompressed data with header
    fn encode_uncompressed(&self, data: &[u8]) -> Bytes {
        let has_checksum = self.config.checksum;
        let header_size = 1 + if has_checksum { 4 } else { 0 };

        let mut buf = BytesMut::with_capacity(header_size + data.len());
        buf.put_u8(CompressionAlgorithm::None.to_flags(false, has_checksum));

        if has_checksum {
            buf.put_u32_le(crc32_checksum(data));
        }

        buf.put_slice(data);
        buf.freeze()
    }

    /// Encode compressed data with header
    ///
    /// `original_data` is used to compute the checksum over the *uncompressed*
    /// payload, matching what `decompress()` verifies after decompression.
    fn encode_compressed(
        &self,
        algorithm: CompressionAlgorithm,
        original_size: usize,
        compressed: &[u8],
        original_data: &[u8],
    ) -> Result<Bytes> {
        let has_checksum = self.config.checksum;

        // reject payloads > u32::MAX before truncating to 4-byte size header
        if original_size > u32::MAX as usize {
            return Err(CompressionError::PayloadTooLarge {
                size: original_size,
                max: u32::MAX as usize,
            });
        }

        // Header: flags (1) + size (4) + optional checksum (4)
        let header_size = 5 + if has_checksum { 4 } else { 0 };

        let mut buf = BytesMut::with_capacity(header_size + compressed.len());
        buf.put_u8(algorithm.to_flags(true, has_checksum));
        buf.put_u32_le(original_size as u32);

        if has_checksum {
            // Checksum of the original (uncompressed) data — decompress() verifies
            // against decompressed output, so the checksum must match that.
            buf.put_u32_le(crc32_checksum(original_data));
        }

        buf.put_slice(compressed);
        Ok(buf.freeze())
    }
}

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

// ============================================================================
// Compression Statistics & Analysis
// ============================================================================

/// Thread-safe statistics collector for compression operations
#[derive(Debug, Default)]
pub struct CompressionStatsCollector {
    // Compression stats by algorithm
    lz4_compressed_bytes: AtomicU64,
    lz4_original_bytes: AtomicU64,
    lz4_operations: AtomicU64,
    zstd_compressed_bytes: AtomicU64,
    zstd_original_bytes: AtomicU64,
    zstd_operations: AtomicU64,
    snappy_compressed_bytes: AtomicU64,
    snappy_original_bytes: AtomicU64,
    snappy_operations: AtomicU64,
    // Decompression stats
    decompressed_bytes: AtomicU64,
    decompress_operations: AtomicU64,
    // Skipped (too small or poor ratio)
    skipped_bytes: AtomicU64,
    skipped_operations: AtomicU64,
}

impl CompressionStatsCollector {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn record_compression(
        &self,
        algorithm: CompressionAlgorithm,
        original: usize,
        compressed: usize,
    ) {
        match algorithm {
            CompressionAlgorithm::Lz4 => {
                self.lz4_original_bytes
                    .fetch_add(original as u64, Ordering::Relaxed);
                self.lz4_compressed_bytes
                    .fetch_add(compressed as u64, Ordering::Relaxed);
                self.lz4_operations.fetch_add(1, Ordering::Relaxed);
            }
            CompressionAlgorithm::Zstd => {
                self.zstd_original_bytes
                    .fetch_add(original as u64, Ordering::Relaxed);
                self.zstd_compressed_bytes
                    .fetch_add(compressed as u64, Ordering::Relaxed);
                self.zstd_operations.fetch_add(1, Ordering::Relaxed);
            }
            CompressionAlgorithm::Snappy => {
                self.snappy_original_bytes
                    .fetch_add(original as u64, Ordering::Relaxed);
                self.snappy_compressed_bytes
                    .fetch_add(compressed as u64, Ordering::Relaxed);
                self.snappy_operations.fetch_add(1, Ordering::Relaxed);
            }
            CompressionAlgorithm::None => {}
        }
    }

    pub fn record_decompression(
        &self,
        _algorithm: CompressionAlgorithm,
        _compressed: usize,
        decompressed: usize,
    ) {
        self.decompressed_bytes
            .fetch_add(decompressed as u64, Ordering::Relaxed);
        self.decompress_operations.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_skipped(&self, size: usize) {
        self.skipped_bytes.fetch_add(size as u64, Ordering::Relaxed);
        self.skipped_operations.fetch_add(1, Ordering::Relaxed);
    }

    pub fn snapshot(&self) -> CompressionStatsSnapshot {
        CompressionStatsSnapshot {
            lz4_compressed_bytes: self.lz4_compressed_bytes.load(Ordering::Relaxed),
            lz4_original_bytes: self.lz4_original_bytes.load(Ordering::Relaxed),
            lz4_operations: self.lz4_operations.load(Ordering::Relaxed),
            zstd_compressed_bytes: self.zstd_compressed_bytes.load(Ordering::Relaxed),
            zstd_original_bytes: self.zstd_original_bytes.load(Ordering::Relaxed),
            zstd_operations: self.zstd_operations.load(Ordering::Relaxed),
            snappy_compressed_bytes: self.snappy_compressed_bytes.load(Ordering::Relaxed),
            snappy_original_bytes: self.snappy_original_bytes.load(Ordering::Relaxed),
            snappy_operations: self.snappy_operations.load(Ordering::Relaxed),
            decompressed_bytes: self.decompressed_bytes.load(Ordering::Relaxed),
            decompress_operations: self.decompress_operations.load(Ordering::Relaxed),
            skipped_bytes: self.skipped_bytes.load(Ordering::Relaxed),
            skipped_operations: self.skipped_operations.load(Ordering::Relaxed),
        }
    }
}

/// Snapshot of compression statistics
#[derive(Debug, Clone, Default)]
pub struct CompressionStatsSnapshot {
    pub lz4_compressed_bytes: u64,
    pub lz4_original_bytes: u64,
    pub lz4_operations: u64,
    pub zstd_compressed_bytes: u64,
    pub zstd_original_bytes: u64,
    pub zstd_operations: u64,
    pub snappy_compressed_bytes: u64,
    pub snappy_original_bytes: u64,
    pub snappy_operations: u64,
    pub decompressed_bytes: u64,
    pub decompress_operations: u64,
    pub skipped_bytes: u64,
    pub skipped_operations: u64,
}

impl CompressionStatsSnapshot {
    /// Get overall compression ratio for LZ4
    pub fn lz4_ratio(&self) -> Option<f64> {
        if self.lz4_original_bytes > 0 {
            Some(self.lz4_compressed_bytes as f64 / self.lz4_original_bytes as f64)
        } else {
            None
        }
    }

    /// Get overall compression ratio for Zstd
    pub fn zstd_ratio(&self) -> Option<f64> {
        if self.zstd_original_bytes > 0 {
            Some(self.zstd_compressed_bytes as f64 / self.zstd_original_bytes as f64)
        } else {
            None
        }
    }

    /// Get overall compression ratio for Snappy
    pub fn snappy_ratio(&self) -> Option<f64> {
        if self.snappy_original_bytes > 0 {
            Some(self.snappy_compressed_bytes as f64 / self.snappy_original_bytes as f64)
        } else {
            None
        }
    }

    /// Get total bytes saved by compression
    pub fn bytes_saved(&self) -> u64 {
        let original =
            self.lz4_original_bytes + self.zstd_original_bytes + self.snappy_original_bytes;
        let compressed =
            self.lz4_compressed_bytes + self.zstd_compressed_bytes + self.snappy_compressed_bytes;
        original.saturating_sub(compressed)
    }
}

/// Analysis result for a data payload
#[derive(Debug, Clone)]
pub struct CompressionAnalysis {
    pub original_size: usize,
    pub lz4_size: Option<usize>,
    pub zstd_size: Option<usize>,
    pub snappy_size: Option<usize>,
    pub entropy: f32,
    pub recommended: CompressionAlgorithm,
}

impl CompressionAnalysis {
    pub fn lz4_ratio(&self) -> Option<f32> {
        self.lz4_size.map(|s| s as f32 / self.original_size as f32)
    }

    pub fn zstd_ratio(&self) -> Option<f32> {
        self.zstd_size.map(|s| s as f32 / self.original_size as f32)
    }

    pub fn snappy_ratio(&self) -> Option<f32> {
        self.snappy_size
            .map(|s| s as f32 / self.original_size as f32)
    }

    /// Get the best compression result
    pub fn best_size(&self) -> Option<usize> {
        [self.lz4_size, self.zstd_size, self.snappy_size]
            .into_iter()
            .flatten()
            .min()
    }

    /// Get the best compression algorithm based on actual results
    pub fn best_algorithm(&self) -> CompressionAlgorithm {
        let mut best = (CompressionAlgorithm::None, self.original_size);

        if let Some(size) = self.lz4_size {
            if size < best.1 {
                best = (CompressionAlgorithm::Lz4, size);
            }
        }
        if let Some(size) = self.zstd_size {
            if size < best.1 {
                best = (CompressionAlgorithm::Zstd, size);
            }
        }
        if let Some(size) = self.snappy_size {
            if size < best.1 {
                best = (CompressionAlgorithm::Snappy, size);
            }
        }

        best.0
    }
}

/// Estimate Shannon entropy of data (bits per byte, 0-8)
fn estimate_entropy(data: &[u8]) -> f32 {
    if data.is_empty() {
        return 0.0;
    }

    // Sample for large data to keep this fast
    let sample_size = data.len().min(4096);
    let sample = &data[..sample_size];

    // Count byte frequencies
    let mut freq = [0u32; 256];
    for &byte in sample {
        freq[byte as usize] += 1;
    }

    // Calculate entropy
    let len = sample.len() as f32;
    let mut entropy = 0.0f32;

    for count in freq.iter() {
        if *count > 0 {
            let p = *count as f32 / len;
            entropy -= p * p.log2();
        }
    }

    entropy
}

// ============================================================================
// Streaming Compression (for large payloads)
// ============================================================================

/// Streaming compressor for large data
pub struct StreamingCompressor<W: Write> {
    encoder: StreamingEncoder<W>,
}

enum StreamingEncoder<W: Write> {
    Lz4(lz4_flex::frame::FrameEncoder<W>),
    Zstd(zstd::Encoder<'static, W>),
    Snappy(Box<snap::write::FrameEncoder<W>>),
    None(W),
}

impl<W: Write> StreamingCompressor<W> {
    /// Create streaming compressor
    pub fn new(
        writer: W,
        algorithm: CompressionAlgorithm,
        level: CompressionLevel,
    ) -> Result<Self> {
        let encoder = match algorithm {
            CompressionAlgorithm::None => StreamingEncoder::None(writer),
            CompressionAlgorithm::Lz4 => {
                let encoder = lz4_flex::frame::FrameEncoder::new(writer);
                StreamingEncoder::Lz4(encoder)
            }
            CompressionAlgorithm::Zstd => {
                let encoder = zstd::Encoder::new(writer, level.zstd_level())
                    .map_err(|e| CompressionError::ZstdError(e.to_string()))?;
                StreamingEncoder::Zstd(encoder)
            }
            CompressionAlgorithm::Snappy => {
                let encoder = snap::write::FrameEncoder::new(writer);
                StreamingEncoder::Snappy(Box::new(encoder))
            }
        };

        Ok(Self { encoder })
    }

    /// Write data to compressor
    pub fn write(&mut self, data: &[u8]) -> Result<usize> {
        match &mut self.encoder {
            StreamingEncoder::None(w) => Ok(w.write(data)?),
            StreamingEncoder::Lz4(e) => Ok(e.write(data)?),
            StreamingEncoder::Zstd(e) => Ok(e.write(data)?),
            StreamingEncoder::Snappy(e) => Ok(e.write(data)?),
        }
    }

    /// Finish compression and return the underlying writer
    pub fn finish(self) -> Result<W> {
        match self.encoder {
            StreamingEncoder::None(w) => Ok(w),
            StreamingEncoder::Lz4(e) => e
                .finish()
                .map_err(|e| CompressionError::Lz4Error(e.to_string())),
            StreamingEncoder::Zstd(e) => e
                .finish()
                .map_err(|e| CompressionError::ZstdError(e.to_string())),
            StreamingEncoder::Snappy(e) => e
                .into_inner()
                .map_err(|e| CompressionError::SnappyError(e.to_string())),
        }
    }
}

/// Streaming decompressor for large data
pub struct StreamingDecompressor<R: Read> {
    decoder: StreamingDecoder<R>,
}

enum StreamingDecoder<R: Read> {
    Lz4(lz4_flex::frame::FrameDecoder<R>),
    Zstd(zstd::Decoder<'static, std::io::BufReader<R>>),
    Snappy(snap::read::FrameDecoder<R>),
    None(R),
}

impl<R: Read> StreamingDecompressor<R> {
    /// Create streaming decompressor
    pub fn new(reader: R, algorithm: CompressionAlgorithm) -> Result<Self> {
        let decoder = match algorithm {
            CompressionAlgorithm::None => StreamingDecoder::None(reader),
            CompressionAlgorithm::Lz4 => {
                let decoder = lz4_flex::frame::FrameDecoder::new(reader);
                StreamingDecoder::Lz4(decoder)
            }
            CompressionAlgorithm::Zstd => {
                let decoder = zstd::Decoder::new(reader)
                    .map_err(|e| CompressionError::ZstdError(e.to_string()))?;
                StreamingDecoder::Zstd(decoder)
            }
            CompressionAlgorithm::Snappy => {
                let decoder = snap::read::FrameDecoder::new(reader);
                StreamingDecoder::Snappy(decoder)
            }
        };

        Ok(Self { decoder })
    }

    /// Read decompressed data
    pub fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        match &mut self.decoder {
            StreamingDecoder::None(r) => Ok(r.read(buf)?),
            StreamingDecoder::Lz4(d) => Ok(d.read(buf)?),
            StreamingDecoder::Zstd(d) => Ok(d.read(buf)?),
            StreamingDecoder::Snappy(d) => Ok(d.read(buf)?),
        }
    }
}

// ============================================================================
// Batch Compression (for message batches)
// ============================================================================

/// Compress multiple messages as a batch for better compression ratio
pub struct BatchCompressor {
    compressor: Compressor,
    buffer: BytesMut,
    message_offsets: Vec<u32>,
}

impl BatchCompressor {
    /// Create new batch compressor
    pub fn new(config: CompressionConfig) -> Self {
        Self {
            compressor: Compressor::with_config(config),
            buffer: BytesMut::with_capacity(64 * 1024),
            message_offsets: Vec::with_capacity(100),
        }
    }

    /// Add message to batch
    pub fn add(&mut self, data: &[u8]) -> Result<()> {
        // guard against accumulated batch exceeding u32::MAX
        let new_len = self
            .buffer
            .len()
            .checked_add(data.len())
            .and_then(|l| l.checked_add(4)) // length prefix
            .ok_or(CompressionError::PayloadTooLarge {
                size: self.buffer.len() + data.len() + 4,
                max: u32::MAX as usize,
            })?;
        if new_len > u32::MAX as usize {
            return Err(CompressionError::PayloadTooLarge {
                size: new_len,
                max: u32::MAX as usize,
            });
        }
        self.message_offsets.push(self.buffer.len() as u32);
        // Write length-prefixed message
        self.buffer.put_u32_le(data.len() as u32);
        self.buffer.put_slice(data);
        Ok(())
    }

    /// Compress the batch and return compressed data with metadata
    pub fn finish(self) -> Result<CompressedBatch> {
        let message_count = self.message_offsets.len();
        let uncompressed_size = self.buffer.len();

        // Compress the batch
        let compressed = self.compressor.compress(&self.buffer)?;

        Ok(CompressedBatch {
            data: compressed,
            message_count,
            uncompressed_size,
        })
    }

    /// Reset for reuse
    pub fn reset(&mut self) {
        self.buffer.clear();
        self.message_offsets.clear();
    }
}

/// A compressed batch of messages
#[derive(Debug, Clone)]
pub struct CompressedBatch {
    pub data: Bytes,
    pub message_count: usize,
    pub uncompressed_size: usize,
}

impl CompressedBatch {
    /// Decompress and iterate over messages
    pub fn decompress(&self) -> Result<BatchIterator> {
        let compressor = Compressor::new();
        let decompressed = compressor.decompress(&self.data)?;

        Ok(BatchIterator {
            data: decompressed,
            position: 0,
        })
    }

    /// Get compression ratio
    pub fn ratio(&self) -> f32 {
        self.data.len() as f32 / self.uncompressed_size as f32
    }
}

/// Iterator over messages in a decompressed batch
pub struct BatchIterator {
    data: Bytes,
    position: usize,
}

impl Iterator for BatchIterator {
    type Item = Bytes;

    fn next(&mut self) -> Option<Self::Item> {
        if self.position + 4 > self.data.len() {
            return None;
        }

        let len_bytes: [u8; 4] = self.data[self.position..self.position + 4]
            .try_into()
            .ok()?;
        let len = u32::from_le_bytes(len_bytes) as usize;
        self.position += 4;

        if self.position + len > self.data.len() {
            return None;
        }

        let message = self.data.slice(self.position..self.position + len);
        self.position += len;

        Some(message)
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    #[test]
    fn test_compress_decompress_lz4() {
        let data = b"Hello, World! This is a test of LZ4 compression. ".repeat(100);
        let compressor = Compressor::with_config(CompressionConfig {
            algorithm: CompressionAlgorithm::Lz4,
            adaptive: false,
            ..Default::default()
        });

        let compressed = compressor.compress(&data).unwrap();
        assert!(compressed.len() < data.len());

        let decompressed = compressor.decompress(&compressed).unwrap();
        assert_eq!(&decompressed[..], &data[..]);
    }

    #[test]
    fn test_compress_decompress_zstd() {
        let data = b"Hello, World! This is a test of Zstd compression. ".repeat(100);
        let compressor = Compressor::with_config(CompressionConfig {
            algorithm: CompressionAlgorithm::Zstd,
            adaptive: false,
            ..Default::default()
        });

        let compressed = compressor.compress(&data).unwrap();
        assert!(compressed.len() < data.len());

        let decompressed = compressor.decompress(&compressed).unwrap();
        assert_eq!(&decompressed[..], &data[..]);
    }

    #[test]
    fn test_small_payload_not_compressed() {
        let data = b"tiny";
        let compressor = Compressor::new();

        let compressed = compressor.compress(data).unwrap();
        // Should be flags (1) + data (4) = 5 bytes
        assert_eq!(compressed.len(), 5);

        let decompressed = compressor.decompress(&compressed).unwrap();
        assert_eq!(&decompressed[..], &data[..]);
    }

    #[test]
    fn test_adaptive_algorithm_selection() {
        let compressor = Compressor::with_config(CompressionConfig {
            adaptive: true,
            ..Default::default()
        });

        // Low entropy text should use Zstd
        let text = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        let algo = compressor.select_algorithm(text.as_bytes());
        assert_eq!(algo, CompressionAlgorithm::Zstd);
    }

    #[test]
    fn test_batch_compression() {
        let config = CompressionConfig::default();
        let mut batch = BatchCompressor::new(config);

        for i in 0..100 {
            let msg = format!("Message {} with some content to compress", i);
            batch.add(msg.as_bytes()).unwrap();
        }

        let compressed = batch.finish().unwrap();
        assert!(compressed.ratio() < 0.5); // Should achieve at least 50% compression

        let messages: Vec<_> = compressed.decompress().unwrap().collect();
        assert_eq!(messages.len(), 100);
        assert_eq!(&messages[0][..], b"Message 0 with some content to compress");
    }

    #[test]
    fn test_entropy_estimation() {
        // Low entropy
        let low = b"aaaaaaaaaaaaaaaa";
        assert!(estimate_entropy(low) < 1.0);

        // High entropy (random-ish)
        let high: Vec<u8> = (0..=255).collect();
        assert!(estimate_entropy(&high) > 7.0);
    }

    #[test]
    fn test_compression_stats() {
        let data = b"Test data for compression statistics analysis ".repeat(50);
        let compressor = Compressor::new();

        // Compress some data to generate stats
        let _ = compressor.compress(&data).unwrap();
        let stats = compressor.stats();
        assert!(
            stats.lz4_operations > 0 || stats.zstd_operations > 0 || stats.snappy_operations > 0
        );
    }

    #[test]
    fn test_compress_decompress_snappy() {
        let data = b"Hello, World! This is a test of Snappy compression. ".repeat(100);
        let compressor = Compressor::with_config(CompressionConfig {
            algorithm: CompressionAlgorithm::Snappy,
            adaptive: false,
            ..Default::default()
        });

        let compressed = compressor.compress(&data).unwrap();
        assert!(compressed.len() < data.len());

        let decompressed = compressor.decompress(&compressed).unwrap();
        assert_eq!(&decompressed[..], &data[..]);
    }

    #[test]
    fn test_compression_analysis() {
        let data = b"Test data for compression analysis with multiple algorithms ".repeat(50);
        let compressor = Compressor::new();

        let analysis = compressor.analyze(&data);
        assert!(analysis.lz4_size.is_some());
        assert!(analysis.zstd_size.is_some());
        assert!(analysis.snappy_size.is_some());
        assert!(analysis.best_size().unwrap() < data.len());
    }

    #[test]
    fn test_kafka_type_ids() {
        assert_eq!(CompressionAlgorithm::None.kafka_type_id(), 0);
        assert_eq!(CompressionAlgorithm::Snappy.kafka_type_id(), 2);
        assert_eq!(CompressionAlgorithm::Lz4.kafka_type_id(), 3);
        assert_eq!(CompressionAlgorithm::Zstd.kafka_type_id(), 4);

        assert_eq!(
            CompressionAlgorithm::from_kafka_type_id(0),
            Some(CompressionAlgorithm::None)
        );
        assert_eq!(
            CompressionAlgorithm::from_kafka_type_id(2),
            Some(CompressionAlgorithm::Snappy)
        );
        assert_eq!(
            CompressionAlgorithm::from_kafka_type_id(3),
            Some(CompressionAlgorithm::Lz4)
        );
        assert_eq!(
            CompressionAlgorithm::from_kafka_type_id(4),
            Some(CompressionAlgorithm::Zstd)
        );
    }
}