rusty_bin/monitor/storage/

compressor.rs

//! File compression functionality for reducing storage space
//!
//! This module handles compression of old data files using zstd compression.

use crate::monitor::{
    config::storage::StorageConfig,
    storage::{Result, StorageError, naming::FileNaming},
};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime};
use tokio::fs;
use walkdir::WalkDir;

/// File compressor for reducing storage space
#[derive(Debug)]
pub struct FileCompressor {
    config: StorageConfig,
}

impl FileCompressor {
    /// Create a new file compressor
    #[must_use]
    pub const fn new(config: StorageConfig) -> Self {
        Self { config }
    }

    /// Compress a single file
    pub async fn compress_file(
        &self,
        input_path: &Path,
        output_path: &Path,
    ) -> Result<CompressionResult> {
        let start_time = std::time::Instant::now();

        // Read input file
        let input_data = fs::read(input_path).await?;
        let original_size = input_data.len() as u64;

        // Compress data
        let compressed_data = tokio::task::spawn_blocking({
            let data = input_data.clone();
            let level = self.config.compression_level as i32;
            move || -> Result<Vec<u8>> {
                zstd::bulk::compress(&data, level)
                    .map_err(|e| StorageError::Compression(e.to_string()))
            }
        })
        .await
        .map_err(|e| StorageError::Compression(e.to_string()))??;

        let compressed_size = compressed_data.len() as u64;

        // Ensure output directory exists
        if let Some(parent) = output_path.parent() {
            fs::create_dir_all(parent).await?;
        }

        // Write compressed file
        fs::write(output_path, compressed_data).await?;

        let compression_time = start_time.elapsed();
        let compression_ratio = compressed_size as f64 / original_size as f64;

        Ok(CompressionResult {
            original_size,
            compressed_size,
            compression_ratio,
            compression_time,
        })
    }

    /// Decompress a file
    pub async fn decompress_file(
        &self,
        input_path: &Path,
        output_path: &Path,
    ) -> Result<DecompressionResult> {
        let start_time = std::time::Instant::now();

        // Read compressed file
        let compressed_data = fs::read(input_path).await?;
        let compressed_size = compressed_data.len() as u64;

        // Decompress data
        let decompressed_data = tokio::task::spawn_blocking({
            let data = compressed_data.clone();
            move || -> Result<Vec<u8>> {
                zstd::bulk::decompress(&data, 10 * 1024 * 1024) // 10MB max
                    .map_err(|e| StorageError::Compression(e.to_string()))
            }
        })
        .await
        .map_err(|e| StorageError::Compression(e.to_string()))??;

        let decompressed_size = decompressed_data.len() as u64;

        // Ensure output directory exists
        if let Some(parent) = output_path.parent() {
            fs::create_dir_all(parent).await?;
        }

        // Write decompressed file
        fs::write(output_path, decompressed_data).await?;

        let decompression_time = start_time.elapsed();

        Ok(DecompressionResult {
            compressed_size,
            decompressed_size,
            decompression_time,
        })
    }

    /// Compress old files based on age threshold
    pub async fn compress_old_files(&self) -> Result<CompressionSummary> {
        let cutoff_time = SystemTime::now() - Duration::from_secs(24 * 60 * 60); // Default to 24 hours if not configured

        let mut summary = CompressionSummary::default();

        // Process raw data directory
        let files_to_compress = self
            .find_files_to_compress(&self.config.market_data_path, cutoff_time)
            .await?;

        for file_path in files_to_compress {
            match self.compress_single_file(&file_path).await {
                Ok(result) => {
                    summary.files_processed += 1;
                    summary.original_size += result.original_size;
                    summary.compressed_size += result.compressed_size;
                    summary.total_time += result.compression_time;

                    log::info!(
                        "Compressed file: {} ({} -> {}, {:.1}% reduction)",
                        file_path.display(),
                        format_bytes(result.original_size),
                        format_bytes(result.compressed_size),
                        (1.0 - result.compression_ratio) * 100.0
                    );
                }
                Err(e) => {
                    summary.errors += 1;
                    log::error!("Failed to compress file {}: {}", file_path.display(), e);
                }
            }
        }

        if summary.files_processed > 0 {
            summary.average_compression_ratio =
                summary.compressed_size as f64 / summary.original_size as f64;
            log::info!(
                "Compression summary: {} files, {:.1}% average compression",
                summary.files_processed,
                (1.0 - summary.average_compression_ratio) * 100.0
            );
        }

        Ok(summary)
    }

    /// Compress a directory of files
    pub async fn compress_directory(&self, dir_path: &Path, delete_original: bool) -> Result<u64> {
        if !dir_path.exists() {
            return Ok(0);
        }

        let mut compressed_count = 0;

        for entry in WalkDir::new(dir_path).into_iter().filter_map(|e| e.ok()) {
            if entry.file_type().is_file() {
                let file_path = entry.path();

                // Skip already compressed files
                if file_path.extension().and_then(|s| s.to_str()) == Some("zst") {
                    continue;
                }

                // Skip non-data files
                if !self.is_data_file(file_path) {
                    continue;
                }

                match self.compress_single_file(file_path).await {
                    Ok(_) => {
                        compressed_count += 1;

                        if delete_original && let Err(e) = fs::remove_file(file_path).await {
                            log::warn!(
                                "Failed to delete original file {}: {}",
                                file_path.display(),
                                e
                            );
                        }
                    }
                    Err(e) => {
                        log::error!("Failed to compress file {}: {}", file_path.display(), e);
                    }
                }
            }
        }

        Ok(compressed_count)
    }

    /// Compress a single file and handle metadata
    async fn compress_single_file(&self, file_path: &Path) -> Result<CompressionResult> {
        let compressed_path = self.get_compressed_path(file_path)?;

        // Skip if already compressed
        if compressed_path.exists() {
            return Err(StorageError::Compression(
                "File already compressed".to_string(),
            ));
        }

        let result = self.compress_file(file_path, &compressed_path).await?;

        // Update metadata if it exists
        if let Some(filename) = file_path.file_name().and_then(|n| n.to_str())
            && let Some(_naming) = FileNaming::parse_filename(filename)
        {
            // Metadata update is no longer needed since we don't maintain separate metadata files
        }

        Ok(result)
    }

    /// Find files that need compression based on age
    async fn find_files_to_compress(
        &self,
        dir_path: &Path,
        cutoff_time: SystemTime,
    ) -> Result<Vec<PathBuf>> {
        if !dir_path.exists() {
            return Ok(Vec::new());
        }

        let mut files_to_compress = Vec::new();

        for entry in WalkDir::new(dir_path).into_iter().filter_map(|e| e.ok()) {
            if entry.file_type().is_file() {
                let file_path = entry.path();

                // Skip already compressed files
                if file_path.extension().and_then(|s| s.to_str()) == Some("zst") {
                    continue;
                }

                // Skip non-data files
                if !self.is_data_file(file_path) {
                    continue;
                }

                // Check file age
                if let Ok(metadata) = entry.metadata() {
                    let modified_time = metadata
                        .modified()
                        .or_else(|_| metadata.created())
                        .unwrap_or(SystemTime::UNIX_EPOCH);

                    if modified_time < cutoff_time {
                        files_to_compress.push(file_path.to_path_buf());
                    }
                }
            }
        }

        Ok(files_to_compress)
    }

    /// Check if a file is a data file that should be compressed
    fn is_data_file(&self, file_path: &Path) -> bool {
        if let Some(extension) = file_path.extension().and_then(|s| s.to_str()) {
            matches!(extension, "fb" | "json")
        } else {
            false
        }
    }

    /// Get the compressed file path for a given input path
    fn get_compressed_path(&self, input_path: &Path) -> Result<PathBuf> {
        let mut compressed_path = input_path.to_path_buf();
        let extension = compressed_path
            .extension()
            .and_then(|e| e.to_str())
            .unwrap_or("");

        // Add .zst extension if not already compressed
        if extension != "zst" {
            let new_extension = format!("{extension}.zst");
            compressed_path.set_extension(new_extension);
        }

        Ok(compressed_path)
    }

    /// Update compression statistics (no longer updates metadata files)
    #[allow(dead_code)]
    async fn update_compression_stats(
        &self,
        naming: &FileNaming,
        result: &CompressionResult,
    ) -> Result<()> {
        // Log compression statistics for monitoring
        log::info!(
            "Compressed {}: {} -> {} bytes (ratio: {:.2}%, time: {:?})",
            naming.filename(),
            result.original_size,
            result.compressed_size,
            (1.0 - result.compression_ratio) * 100.0,
            result.compression_time
        );
        Ok(())
    }
}

/// Result of a compression operation containing metrics about the compression process.
///
/// This structure provides detailed information about the compression operation,
/// including size metrics, compression ratio, and timing information.
#[derive(Debug, Clone)]
pub struct CompressionResult {
    /// The original size of the file before compression in bytes
    pub original_size: u64,
    /// The size of the compressed file in bytes
    pub compressed_size: u64,
    /// The compression ratio (compressed_size / original_size)
    /// A value of 0.5 means the file was compressed to 50% of its original size
    pub compression_ratio: f64,
    /// The time taken to complete the compression operation
    pub compression_time: std::time::Duration,
}

/// Result of a decompression operation containing metrics about the decompression process.
///
/// This structure provides detailed information about the decompression operation,
/// including size metrics and timing information.
#[derive(Debug, Clone)]
pub struct DecompressionResult {
    /// The size of the compressed file in bytes
    pub compressed_size: u64,
    /// The size of the decompressed file in bytes
    pub decompressed_size: u64,
    /// The time taken to complete the decompression operation
    pub decompression_time: std::time::Duration,
}

/// Summary of multiple compression operations providing aggregate statistics.
///
/// This structure accumulates metrics from multiple compression operations,
/// providing an overview of the overall compression process including success
/// rate, space savings, and performance metrics.
#[derive(Debug, Clone, Default)]
pub struct CompressionSummary {
    /// The total number of files successfully processed
    pub files_processed: u64,
    /// The total original size of all files before compression in bytes
    pub original_size: u64,
    /// The total compressed size of all files in bytes
    pub compressed_size: u64,
    /// The average compression ratio across all files (compressed_size / original_size)
    pub average_compression_ratio: f64,
    /// The total time spent on all compression operations
    pub total_time: std::time::Duration,
    /// The number of compression operations that failed
    pub errors: u64,
}

/// Format bytes in human-readable format (B, KB, MB, GB, TB).
///
/// Converts a byte count into a human-readable string with appropriate units.
/// Uses binary prefixes (1024 bytes = 1 KB) for consistency with storage systems.
///
/// # Arguments
///
/// * `bytes` - The number of bytes to format
///
/// # Returns
///
/// A formatted string with the size and appropriate unit
///
/// # Examples
///
/// ```
/// assert_eq!(format_bytes(1024), "1.0 KB");
/// assert_eq!(format_bytes(1536), "1.5 KB");
/// assert_eq!(format_bytes(1024 * 1024), "1.0 MB");
/// ```
fn format_bytes(bytes: u64) -> String {
    const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"];
    let mut size = bytes as f64;
    let mut unit_index = 0;

    while size >= 1024.0 && unit_index < UNITS.len() - 1 {
        size /= 1024.0;
        unit_index += 1;
    }

    if unit_index == 0 {
        format!("{} {}", bytes, UNITS[unit_index])
    } else {
        format!("{:.1} {}", size, UNITS[unit_index])
    }
}

#[cfg(test)]
mod tests {
    use crate::monitor::config::storage::StorageConfig;

    use super::*;
    use tempfile::tempdir;

    #[tokio::test]
    async fn test_file_compressor_creation() {
        let config = StorageConfig::default();
        let _compressor = FileCompressor::new(config);

        // Just test that it can be created without panic
    }

    #[tokio::test]
    async fn test_compress_file() {
        let temp_dir = tempdir().unwrap();
        let config = StorageConfig {
            market_data_path: temp_dir.path().to_path_buf(),
            compression_level: 3,
            ..Default::default()
        };
        let compressor = FileCompressor::new(config);

        // Create test file
        let input_path = temp_dir.path().join("test.fb");
        let output_path = temp_dir.path().join("test.fb.zst");

        let test_data = b"This is test data that should compress well. ".repeat(100);
        fs::write(&input_path, &test_data).await.unwrap();

        // Compress file
        let result = compressor
            .compress_file(&input_path, &output_path)
            .await
            .unwrap();

        assert!(output_path.exists());
        assert!(result.compressed_size < result.original_size);
        assert!(result.compression_ratio < 1.0);
        assert_eq!(result.original_size, test_data.len() as u64);
    }

    #[tokio::test]
    async fn test_decompress_file() {
        let temp_dir = tempdir().unwrap();
        let config = StorageConfig {
            market_data_path: temp_dir.path().to_path_buf(),
            compression_level: 3,
            ..Default::default()
        };
        let compressor = FileCompressor::new(config);

        // Create and compress test file
        let original_path = temp_dir.path().join("original.fb");
        let compressed_path = temp_dir.path().join("compressed.fb.zst");
        let decompressed_path = temp_dir.path().join("decompressed.fb");

        let test_data = b"This is test data for compression/decompression test.";
        fs::write(&original_path, test_data).await.unwrap();

        // Compress
        compressor
            .compress_file(&original_path, &compressed_path)
            .await
            .unwrap();

        // Decompress
        let result = compressor
            .decompress_file(&compressed_path, &decompressed_path)
            .await
            .unwrap();

        assert!(decompressed_path.exists());
        assert_eq!(result.decompressed_size, test_data.len() as u64);

        // Verify content
        let decompressed_data = fs::read(&decompressed_path).await.unwrap();
        assert_eq!(decompressed_data, test_data);
    }

    #[test]
    fn test_format_bytes() {
        assert_eq!(format_bytes(512), "512 B");
        assert_eq!(format_bytes(1024), "1.0 KB");
        assert_eq!(format_bytes(1536), "1.5 KB");
        assert_eq!(format_bytes(1024 * 1024), "1.0 MB");
        assert_eq!(format_bytes(1024 * 1024 * 1024), "1.0 GB");
    }
}