phytrace_sdk/reliability/

buffer.rs

//! File-based event buffer for offline reliability.
//!
//! Events are stored as JSON files in a buffer directory when the transport
//! is unavailable, then replayed when connectivity is restored.

use std::fs::{self, File};
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use flate2::read::GzDecoder;
use flate2::write::GzEncoder;
use flate2::Compression;

use crate::core::config::BufferConfig;
use crate::error::{BufferError, PhyTraceResult};
use crate::models::event::UdmEvent;

/// File-based event buffer.
#[derive(Debug)]
pub struct FileBuffer {
    config: BufferConfig,
    buffer_dir: PathBuf,
}

impl FileBuffer {
    /// Create a new file buffer with configuration.
    pub fn new(config: &BufferConfig) -> PhyTraceResult<Self> {
        let buffer_dir = PathBuf::from(&config.path);

        // Create buffer directory if it doesn't exist
        if !buffer_dir.exists() {
            fs::create_dir_all(&buffer_dir).map_err(|e| {
                BufferError::Io(format!("Failed to create buffer directory: {}", e))
            })?;
        }

        Ok(Self {
            config: config.clone(),
            buffer_dir,
        })
    }

    /// Create a buffer with default configuration at the specified path.
    pub fn at_path(path: impl Into<PathBuf>) -> PhyTraceResult<Self> {
        let path = path.into();
        let config = BufferConfig {
            path: path.to_string_lossy().to_string(),
            ..Default::default()
        };
        Self::new(&config)
    }

    /// Buffer a single event.
    pub fn buffer_event(&self, event: &UdmEvent) -> PhyTraceResult<()> {
        self.check_capacity()?;

        let filename = self.generate_filename();
        let filepath = self.buffer_dir.join(&filename);

        let json =
            serde_json::to_string(event).map_err(|e| BufferError::Serialization(e.to_string()))?;

        if self.config.compress {
            self.write_compressed(&filepath, &json)?;
        } else {
            fs::write(&filepath, &json)
                .map_err(|e| BufferError::Io(format!("Failed to write buffer file: {}", e)))?;
        }

        Ok(())
    }

    /// Buffer multiple events.
    pub fn buffer_events(&self, events: &[UdmEvent]) -> PhyTraceResult<()> {
        self.check_capacity()?;

        let filename = format!("batch_{}.jsonl", self.timestamp_nanos());
        let filepath = self.buffer_dir.join(&filename);

        if self.config.compress {
            let filepath = filepath.with_extension("jsonl.gz");
            let file = File::create(&filepath)
                .map_err(|e| BufferError::Io(format!("Failed to create buffer file: {}", e)))?;
            let encoder = GzEncoder::new(file, Compression::fast());
            let mut writer = BufWriter::new(encoder);

            for event in events {
                let json = serde_json::to_string(event)
                    .map_err(|e| BufferError::Serialization(e.to_string()))?;
                writeln!(writer, "{}", json)
                    .map_err(|e| BufferError::Io(format!("Failed to write to buffer: {}", e)))?;
            }
            writer
                .flush()
                .map_err(|e| BufferError::Io(format!("Failed to flush buffer: {}", e)))?;
        } else {
            let file = File::create(&filepath)
                .map_err(|e| BufferError::Io(format!("Failed to create buffer file: {}", e)))?;
            let mut writer = BufWriter::new(file);

            for event in events {
                let json = serde_json::to_string(event)
                    .map_err(|e| BufferError::Serialization(e.to_string()))?;
                writeln!(writer, "{}", json)
                    .map_err(|e| BufferError::Io(format!("Failed to write to buffer: {}", e)))?;
            }
            writer
                .flush()
                .map_err(|e| BufferError::Io(format!("Failed to flush buffer: {}", e)))?;
        }

        Ok(())
    }

    /// Read all buffered events, oldest first.
    pub fn read_events(&self) -> PhyTraceResult<Vec<BufferedEvent>> {
        let mut events = Vec::new();

        for entry in fs::read_dir(&self.buffer_dir)
            .map_err(|e| BufferError::Io(format!("Failed to read buffer directory: {}", e)))?
        {
            let entry = entry
                .map_err(|e| BufferError::Io(format!("Failed to read directory entry: {}", e)))?;
            let path = entry.path();

            if !path.is_file() {
                continue;
            }

            let filename = path.file_name().and_then(|n| n.to_str()).unwrap_or("");

            // Skip non-buffer files
            if !filename.ends_with(".json")
                && !filename.ends_with(".json.gz")
                && !filename.ends_with(".jsonl")
                && !filename.ends_with(".jsonl.gz")
            {
                continue;
            }

            // Check age
            if self.is_expired(&path) {
                // Delete expired files
                let _ = fs::remove_file(&path);
                continue;
            }

            // Read events from file
            let file_events = self.read_file(&path)?;
            for event in file_events {
                events.push(BufferedEvent {
                    event,
                    filepath: path.clone(),
                });
            }
        }

        // Sort by event timestamp (oldest first)
        events.sort_by(|a, b| a.event.captured_at.cmp(&b.event.captured_at));

        Ok(events)
    }

    /// Remove a specific buffered file.
    pub fn remove_file(&self, filepath: &Path) -> PhyTraceResult<()> {
        fs::remove_file(filepath)
            .map_err(|e| BufferError::Io(format!("Failed to remove buffer file: {}", e)))?;
        Ok(())
    }

    /// Get the count of buffered files.
    pub fn file_count(&self) -> usize {
        fs::read_dir(&self.buffer_dir)
            .map(|entries| {
                entries
                    .filter_map(|e| e.ok())
                    .filter(|e| e.path().is_file())
                    .filter(|e| {
                        let name = e.file_name().to_string_lossy().to_string();
                        name.ends_with(".json")
                            || name.ends_with(".json.gz")
                            || name.ends_with(".jsonl")
                            || name.ends_with(".jsonl.gz")
                    })
                    .count()
            })
            .unwrap_or(0)
    }

    /// Get total size of buffered files in bytes.
    pub fn total_size(&self) -> u64 {
        fs::read_dir(&self.buffer_dir)
            .map(|entries| {
                entries
                    .filter_map(|e| e.ok())
                    .filter_map(|e| e.metadata().ok())
                    .filter(|m| m.is_file())
                    .map(|m| m.len())
                    .sum()
            })
            .unwrap_or(0)
    }

    /// Clear all buffered events.
    pub fn clear(&self) -> PhyTraceResult<()> {
        for entry in fs::read_dir(&self.buffer_dir)
            .map_err(|e| BufferError::Io(format!("Failed to read buffer directory: {}", e)))?
        {
            let entry = entry
                .map_err(|e| BufferError::Io(format!("Failed to read directory entry: {}", e)))?;
            let path = entry.path();

            if path.is_file() {
                let _ = fs::remove_file(&path);
            }
        }
        Ok(())
    }

    /// Cleanup expired files.
    pub fn cleanup_expired(&self) -> PhyTraceResult<usize> {
        let mut removed = 0;

        for entry in fs::read_dir(&self.buffer_dir)
            .map_err(|e| BufferError::Io(format!("Failed to read buffer directory: {}", e)))?
        {
            let entry = entry
                .map_err(|e| BufferError::Io(format!("Failed to read directory entry: {}", e)))?;
            let path = entry.path();

            if path.is_file() && self.is_expired(&path) {
                let _ = fs::remove_file(&path);
                removed += 1;
            }
        }

        Ok(removed)
    }

    // =========================================================================
    // Private helpers
    // =========================================================================

    fn generate_filename(&self) -> String {
        let ts = self.timestamp_nanos();
        if self.config.compress {
            format!("event_{}.json.gz", ts)
        } else {
            format!("event_{}.json", ts)
        }
    }

    fn timestamp_nanos(&self) -> u128 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or(Duration::ZERO)
            .as_nanos()
    }

    fn check_capacity(&self) -> PhyTraceResult<()> {
        let current_size = self.total_size();
        if current_size >= self.config.max_size_bytes {
            return Err(BufferError::Full(format!(
                "Buffer full: {} bytes (max {})",
                current_size, self.config.max_size_bytes
            ))
            .into());
        }
        Ok(())
    }

    fn is_expired(&self, path: &Path) -> bool {
        let max_age = Duration::from_secs(self.config.max_age_secs);

        path.metadata()
            .and_then(|m| m.modified())
            .map(|modified| {
                SystemTime::now()
                    .duration_since(modified)
                    .map(|age| age > max_age)
                    .unwrap_or(false)
            })
            .unwrap_or(false)
    }

    fn write_compressed(&self, filepath: &Path, content: &str) -> PhyTraceResult<()> {
        let file = File::create(filepath)
            .map_err(|e| BufferError::Io(format!("Failed to create buffer file: {}", e)))?;
        let mut encoder = GzEncoder::new(file, Compression::fast());
        encoder
            .write_all(content.as_bytes())
            .map_err(|e| BufferError::Io(format!("Failed to write compressed data: {}", e)))?;
        encoder
            .finish()
            .map_err(|e| BufferError::Io(format!("Failed to finish compression: {}", e)))?;
        Ok(())
    }

    fn read_file(&self, path: &Path) -> PhyTraceResult<Vec<UdmEvent>> {
        let filename = path.file_name().and_then(|n| n.to_str()).unwrap_or("");

        let is_compressed = filename.ends_with(".gz");
        let is_batch = filename.contains("batch")
            || filename.ends_with(".jsonl")
            || filename.ends_with(".jsonl.gz");

        if is_batch {
            self.read_batch_file(path, is_compressed)
        } else {
            self.read_single_file(path, is_compressed).map(|e| vec![e])
        }
    }

    fn read_single_file(&self, path: &Path, compressed: bool) -> PhyTraceResult<UdmEvent> {
        let content = if compressed {
            let file = File::open(path)
                .map_err(|e| BufferError::Io(format!("Failed to open buffer file: {}", e)))?;
            let mut decoder = GzDecoder::new(file);
            let mut content = String::new();
            std::io::Read::read_to_string(&mut decoder, &mut content)
                .map_err(|e| BufferError::Io(format!("Failed to read compressed file: {}", e)))?;
            content
        } else {
            fs::read_to_string(path)
                .map_err(|e| BufferError::Io(format!("Failed to read buffer file: {}", e)))?
        };

        serde_json::from_str(&content).map_err(|e| BufferError::Serialization(e.to_string()).into())
    }

    fn read_batch_file(&self, path: &Path, compressed: bool) -> PhyTraceResult<Vec<UdmEvent>> {
        let file = File::open(path)
            .map_err(|e| BufferError::Io(format!("Failed to open buffer file: {}", e)))?;

        let reader: Box<dyn BufRead> = if compressed {
            Box::new(BufReader::new(GzDecoder::new(file)))
        } else {
            Box::new(BufReader::new(file))
        };

        let mut events = Vec::new();
        for line in reader.lines() {
            let line = line.map_err(|e| BufferError::Io(format!("Failed to read line: {}", e)))?;
            if line.trim().is_empty() {
                continue;
            }
            let event: UdmEvent = serde_json::from_str(&line)
                .map_err(|e| BufferError::Serialization(e.to_string()))?;
            events.push(event);
        }

        Ok(events)
    }
}

/// A buffered event with its file path.
#[derive(Debug)]
pub struct BufferedEvent {
    /// The event.
    pub event: UdmEvent,
    /// Path to the buffer file.
    pub filepath: PathBuf,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::domains::IdentityDomain;
    use crate::models::enums::SourceType;
    use tempfile::tempdir;

    fn test_event(id: &str) -> UdmEvent {
        UdmEvent::new(SourceType::Amr).with_identity(IdentityDomain {
            source_id: Some(id.to_string()),
            ..Default::default()
        })
    }

    fn test_buffer() -> (FileBuffer, tempfile::TempDir) {
        let dir = tempdir().unwrap();
        let config = BufferConfig {
            path: dir.path().to_string_lossy().to_string(),
            compress: false,
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();
        (buffer, dir)
    }

    fn test_buffer_compressed() -> (FileBuffer, tempfile::TempDir) {
        let dir = tempdir().unwrap();
        let config = BufferConfig {
            path: dir.path().to_string_lossy().to_string(),
            compress: true,
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();
        (buffer, dir)
    }

    #[test]
    fn test_buffer_single_event() {
        let (buffer, _dir) = test_buffer();
        let event = test_event("robot-001");

        buffer.buffer_event(&event).unwrap();

        assert_eq!(buffer.file_count(), 1);
    }

    #[test]
    fn test_buffer_and_read() {
        let (buffer, _dir) = test_buffer();
        let event = test_event("robot-001");

        buffer.buffer_event(&event).unwrap();
        let events = buffer.read_events().unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(
            events[0].event.identity.as_ref().unwrap().source_id,
            Some("robot-001".to_string())
        );
    }

    #[test]
    fn test_buffer_batch() {
        let (buffer, _dir) = test_buffer();
        let events = vec![
            test_event("robot-001"),
            test_event("robot-002"),
            test_event("robot-003"),
        ];

        buffer.buffer_events(&events).unwrap();

        let read = buffer.read_events().unwrap();
        assert_eq!(read.len(), 3);
    }

    #[test]
    fn test_buffer_clear() {
        let (buffer, _dir) = test_buffer();
        buffer.buffer_event(&test_event("robot-001")).unwrap();
        buffer.buffer_event(&test_event("robot-002")).unwrap();

        assert_eq!(buffer.file_count(), 2);

        buffer.clear().unwrap();
        assert_eq!(buffer.file_count(), 0);
    }

    #[test]
    fn test_compressed_buffer() {
        let dir = tempdir().unwrap();
        let config = BufferConfig {
            path: dir.path().to_string_lossy().to_string(),
            compress: true,
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();
        let event = test_event("robot-001");

        buffer.buffer_event(&event).unwrap();
        let events = buffer.read_events().unwrap();

        assert_eq!(events.len(), 1);
    }

    // Additional tests for improved coverage

    #[test]
    fn test_at_path() {
        let dir = tempdir().unwrap();
        let buffer = FileBuffer::at_path(dir.path()).unwrap();

        buffer.buffer_event(&test_event("robot-001")).unwrap();
        assert_eq!(buffer.file_count(), 1);
    }

    #[test]
    fn test_remove_file() {
        let (buffer, _dir) = test_buffer();
        buffer.buffer_event(&test_event("robot-001")).unwrap();

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 1);

        let filepath = events[0].filepath.clone();
        buffer.remove_file(&filepath).unwrap();

        assert_eq!(buffer.file_count(), 0);
    }

    #[test]
    fn test_total_size() {
        let (buffer, _dir) = test_buffer();

        assert_eq!(buffer.total_size(), 0);

        buffer.buffer_event(&test_event("robot-001")).unwrap();
        let size1 = buffer.total_size();
        assert!(size1 > 0);

        buffer.buffer_event(&test_event("robot-002")).unwrap();
        let size2 = buffer.total_size();
        assert!(size2 > size1);
    }

    #[test]
    fn test_file_count_multiple() {
        let (buffer, _dir) = test_buffer();

        assert_eq!(buffer.file_count(), 0);

        for i in 0..5 {
            buffer
                .buffer_event(&test_event(&format!("robot-{:03}", i)))
                .unwrap();
        }

        assert_eq!(buffer.file_count(), 5);
    }

    #[test]
    fn test_compressed_batch() {
        let (buffer, _dir) = test_buffer_compressed();
        let events = vec![
            test_event("robot-001"),
            test_event("robot-002"),
            test_event("robot-003"),
        ];

        buffer.buffer_events(&events).unwrap();

        let read = buffer.read_events().unwrap();
        assert_eq!(read.len(), 3);
    }

    #[test]
    fn test_uncompressed_batch() {
        let (buffer, _dir) = test_buffer();
        let events = vec![
            test_event("robot-001"),
            test_event("robot-002"),
            test_event("robot-003"),
        ];

        buffer.buffer_events(&events).unwrap();

        let read = buffer.read_events().unwrap();
        assert_eq!(read.len(), 3);

        // Verify the identity of each event
        let ids: Vec<_> = read
            .iter()
            .map(|e| {
                e.event
                    .identity
                    .as_ref()
                    .unwrap()
                    .source_id
                    .as_ref()
                    .unwrap()
                    .clone()
            })
            .collect();
        assert!(ids.contains(&"robot-001".to_string()));
        assert!(ids.contains(&"robot-002".to_string()));
        assert!(ids.contains(&"robot-003".to_string()));
    }

    #[test]
    fn test_cleanup_expired() {
        let dir = tempdir().unwrap();
        let config = BufferConfig {
            path: dir.path().to_string_lossy().to_string(),
            compress: false,
            max_age_secs: 0, // Immediately expire
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();

        buffer.buffer_event(&test_event("robot-001")).unwrap();
        buffer.buffer_event(&test_event("robot-002")).unwrap();

        assert_eq!(buffer.file_count(), 2);

        // Small delay to ensure files are considered expired
        std::thread::sleep(std::time::Duration::from_millis(10));

        let removed = buffer.cleanup_expired().unwrap();
        assert_eq!(removed, 2);
        assert_eq!(buffer.file_count(), 0);
    }

    #[test]
    fn test_read_events_removes_expired() {
        let dir = tempdir().unwrap();
        let config = BufferConfig {
            path: dir.path().to_string_lossy().to_string(),
            compress: false,
            max_age_secs: 0, // Immediately expire
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();

        buffer.buffer_event(&test_event("robot-001")).unwrap();

        // Small delay to ensure file is considered expired
        std::thread::sleep(std::time::Duration::from_millis(10));

        // read_events should skip and remove expired files
        let events = buffer.read_events().unwrap();
        assert!(events.is_empty());
    }

    #[test]
    fn test_buffer_creates_directory() {
        let dir = tempdir().unwrap();
        let nested_path = dir.path().join("nested").join("buffer").join("dir");

        let config = BufferConfig {
            path: nested_path.to_string_lossy().to_string(),
            compress: false,
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();

        assert!(nested_path.exists());
        buffer.buffer_event(&test_event("robot-001")).unwrap();
        assert_eq!(buffer.file_count(), 1);
    }

    #[test]
    fn test_buffer_full_error() {
        let dir = tempdir().unwrap();
        let config = BufferConfig {
            path: dir.path().to_string_lossy().to_string(),
            compress: false,
            max_size_bytes: 1, // Very small to trigger full
            ..Default::default()
        };
        let buffer = FileBuffer::new(&config).unwrap();

        // First write succeeds but fills the buffer
        buffer.buffer_event(&test_event("robot-001")).unwrap();

        // Second write should fail due to buffer full
        let result = buffer.buffer_event(&test_event("robot-002"));
        assert!(result.is_err());
    }

    #[test]
    fn test_read_ignores_non_buffer_files() {
        let (buffer, dir) = test_buffer();

        // Create a non-buffer file
        std::fs::write(dir.path().join("readme.txt"), "not a buffer file").unwrap();
        std::fs::write(dir.path().join("data.csv"), "a,b,c").unwrap();

        buffer.buffer_event(&test_event("robot-001")).unwrap();

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 1);
    }

    #[test]
    fn test_read_ignores_directories() {
        let (buffer, dir) = test_buffer();

        // Create a subdirectory
        std::fs::create_dir(dir.path().join("subdir")).unwrap();

        buffer.buffer_event(&test_event("robot-001")).unwrap();

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 1);
    }

    #[test]
    fn test_events_sorted_by_timestamp() {
        let (buffer, _dir) = test_buffer();

        // Buffer events with slight delays to ensure different timestamps
        for i in 0..3 {
            buffer
                .buffer_event(&test_event(&format!("robot-{:03}", i)))
                .unwrap();
            std::thread::sleep(std::time::Duration::from_millis(5));
        }

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 3);

        // Verify sorted by timestamp (oldest first)
        for i in 0..events.len() - 1 {
            assert!(events[i].event.captured_at <= events[i + 1].event.captured_at);
        }
    }

    #[test]
    fn test_clear_with_subdirectories() {
        let (buffer, dir) = test_buffer();

        // Create a subdirectory (should be ignored by clear)
        std::fs::create_dir(dir.path().join("subdir")).unwrap();

        buffer.buffer_event(&test_event("robot-001")).unwrap();
        buffer.buffer_event(&test_event("robot-002")).unwrap();

        buffer.clear().unwrap();

        // Subdirectory should still exist
        assert!(dir.path().join("subdir").exists());
        assert_eq!(buffer.file_count(), 0);
    }

    #[test]
    fn test_compressed_single_event_read_write() {
        let (buffer, _dir) = test_buffer_compressed();

        let event = test_event("compressed-robot");
        buffer.buffer_event(&event).unwrap();

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(
            events[0].event.identity.as_ref().unwrap().source_id,
            Some("compressed-robot".to_string())
        );
    }

    #[test]
    fn test_multiple_batches() {
        let (buffer, _dir) = test_buffer();

        // Buffer two separate batches
        buffer
            .buffer_events(&[test_event("batch1-1"), test_event("batch1-2")])
            .unwrap();
        std::thread::sleep(std::time::Duration::from_millis(5));
        buffer
            .buffer_events(&[test_event("batch2-1"), test_event("batch2-2")])
            .unwrap();

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 4);
    }

    #[test]
    fn test_buffered_event_contains_filepath() {
        let (buffer, _dir) = test_buffer();

        buffer.buffer_event(&test_event("robot-001")).unwrap();

        let events = buffer.read_events().unwrap();
        assert_eq!(events.len(), 1);
        assert!(events[0].filepath.exists());
        assert!(events[0].filepath.is_file());
    }

    #[test]
    fn test_empty_buffer_read() {
        let (buffer, _dir) = test_buffer();

        let events = buffer.read_events().unwrap();
        assert!(events.is_empty());
    }

    #[test]
    fn test_file_count_empty() {
        let (buffer, _dir) = test_buffer();
        assert_eq!(buffer.file_count(), 0);
    }

    #[test]
    fn test_total_size_empty() {
        let (buffer, _dir) = test_buffer();
        assert_eq!(buffer.total_size(), 0);
    }

    #[test]
    fn test_cleanup_expired_none_expired() {
        let (buffer, _dir) = test_buffer();

        buffer.buffer_event(&test_event("robot-001")).unwrap();

        // Default max_age is large, so nothing should be expired
        let removed = buffer.cleanup_expired().unwrap();
        assert_eq!(removed, 0);
        assert_eq!(buffer.file_count(), 1);
    }

    #[test]
    fn test_clear_empty_buffer() {
        let (buffer, _dir) = test_buffer();

        // Should not error on empty buffer
        buffer.clear().unwrap();
        assert_eq!(buffer.file_count(), 0);
    }

    #[test]
    fn test_buffer_events_empty_slice() {
        let (buffer, _dir) = test_buffer();

        // Should handle empty slice gracefully
        buffer.buffer_events(&[]).unwrap();
        assert_eq!(buffer.file_count(), 1); // File created but empty
    }

    #[test]
    fn test_compressed_batch_read_write() {
        let (buffer, _dir) = test_buffer_compressed();

        let events: Vec<UdmEvent> = (0..5)
            .map(|i| test_event(&format!("robot-{:03}", i)))
            .collect();

        buffer.buffer_events(&events).unwrap();

        let read = buffer.read_events().unwrap();
        assert_eq!(read.len(), 5);
    }
}