oo_ide/

log_store.rs

//! Log storage for task output.
//!
//! Each task gets its own log file under `.oo/cache/tasks/`:
//!
//! ```text
//! .oo/cache/tasks/<task_id>.log      # written incrementally while task runs
//! .oo/cache/tasks/<task_id>.log.gz   # compressed after task finishes
//! ```
//!
//! Logs are written **incrementally** as the task produces output so that a
//! tail-viewer can follow them in real-time.  When a task finishes the file is
//! gzip-compressed.
//!
//! A configurable disk quota is enforced after every compression: oldest logs
//! (by modification time) are deleted first until total usage is within the
//! limit.

use std::io::{BufRead as _, BufReader, BufWriter};
use std::path::{Path, PathBuf};
use std::time::SystemTime;

use anyhow::{Context, Result};
use flate2::read::GzDecoder;
use flate2::write::GzEncoder;
use flate2::Compression;
use tokio::io::AsyncWriteExt as _;

use crate::task_registry::TaskId;

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/// Default maximum total log storage: 500 MiB.
pub const DEFAULT_QUOTA_BYTES: u64 = 500 * 1024 * 1024;

// ---------------------------------------------------------------------------
// LogStore
// ---------------------------------------------------------------------------

/// Manages the on-disk log storage directory and disk-quota policy.
///
/// Cheap to clone — only holds a `PathBuf` and a `u64`.
#[derive(Clone, Debug)]
pub struct LogStore {
    /// Directory where task logs are stored.
    pub dir: PathBuf,
    /// Maximum total size of all logs before oldest-first eviction.
    pub quota_bytes: u64,
}

/// Metadata about a single stored log file.
#[derive(Debug, Clone)]
pub struct LogEntry {
    /// Task ID parsed from the file name.  `None` for unrecognised files.
    pub task_id: Option<TaskId>,
    pub path: PathBuf,
    pub size_bytes: u64,
    pub modified_at: SystemTime,
    pub compressed: bool,
}

impl LogStore {
    /// Create a new `LogStore` pointing at `dir` with the given quota.
    ///
    /// The directory is created lazily on first write.
    pub fn new(dir: PathBuf, quota_bytes: u64) -> Self {
        Self { dir, quota_bytes }
    }

    // ------------------------------------------------------------------
    // Path helpers
    // ------------------------------------------------------------------

    /// Uncompressed log path for `task_id`.
    pub fn log_path(&self, task_id: TaskId) -> PathBuf {
        self.dir.join(format!("{}.log", task_id.0))
    }

    /// Compressed log path for `task_id`.
    pub fn gz_path(&self, task_id: TaskId) -> PathBuf {
        self.dir.join(format!("{}.log.gz", task_id.0))
    }

    // ------------------------------------------------------------------
    // Writer
    // ------------------------------------------------------------------

    /// Open a new [`LogWriter`] for `task_id`, creating the directory if
    /// needed.  Any pre-existing log for this task ID is overwritten.
    pub async fn open_writer(&self, task_id: TaskId) -> Result<LogWriter> {
        tokio::fs::create_dir_all(&self.dir)
            .await
            .with_context(|| format!("create log dir {:?}", self.dir))?;

        let path = self.log_path(task_id);
        let file = tokio::fs::File::create(&path)
            .await
            .with_context(|| format!("create log file {:?}", path))?;

        Ok(LogWriter {
            path,
            writer: tokio::io::BufWriter::new(file),
        })
    }

    // ------------------------------------------------------------------
    // Directory listing
    // ------------------------------------------------------------------

    /// List all log entries (`.log` and `.log.gz`) in the store directory.
    ///
    /// Returns an empty `Vec` if the directory does not yet exist.
    pub async fn list_entries(&self) -> Result<Vec<LogEntry>> {
        if !self.dir.exists() {
            return Ok(vec![]);
        }

        let mut read_dir = tokio::fs::read_dir(&self.dir)
            .await
            .with_context(|| format!("read log dir {:?}", self.dir))?;

        let mut entries = Vec::new();
        while let Some(entry) = read_dir.next_entry().await? {
            let path = entry.path();
            let name = match path.file_name().and_then(|n| n.to_str()) {
                Some(n) => n.to_string(),
                None => continue,
            };

            let (task_id, compressed) = if let Some(id_str) = name.strip_suffix(".log.gz") {
                (id_str.parse::<u64>().ok().map(TaskId), true)
            } else if let Some(id_str) = name.strip_suffix(".log") {
                (id_str.parse::<u64>().ok().map(TaskId), false)
            } else {
                continue;
            };

            let meta = match entry.metadata().await {
                Ok(m) => m,
                Err(_) => continue,
            };

            entries.push(LogEntry {
                task_id,
                path,
                size_bytes: meta.len(),
                modified_at: meta.modified().unwrap_or(SystemTime::UNIX_EPOCH),
                compressed,
            });
        }

        Ok(entries)
    }

    /// Total disk usage across all log files.
    pub async fn total_size(&self) -> Result<u64> {
        Ok(self.list_entries().await?.iter().map(|e| e.size_bytes).sum())
    }

    // ------------------------------------------------------------------
    // Quota enforcement
    // ------------------------------------------------------------------

    /// Delete oldest log files until total usage is within [`Self::quota_bytes`].
    ///
    /// Returns the paths of all deleted files.
    pub async fn enforce_quota(&self) -> Result<Vec<PathBuf>> {
        let mut entries = self.list_entries().await?;
        let mut total: u64 = entries.iter().map(|e| e.size_bytes).sum();
        if total <= self.quota_bytes {
            return Ok(vec![]);
        }

        entries.sort_by_key(|e| e.modified_at);

        let mut deleted = Vec::new();
        for entry in &entries {
            if total <= self.quota_bytes {
                break;
            }
            if tokio::fs::remove_file(&entry.path).await.is_ok() {
                total = total.saturating_sub(entry.size_bytes);
                deleted.push(entry.path.clone());
            }
        }
        Ok(deleted)
    }

    // ------------------------------------------------------------------
    // Reading
    // ------------------------------------------------------------------

    /// Read all lines from a stored log.
    ///
    /// Prefers `.log.gz` if present; falls back to `.log`.  Returns an empty
    /// `Vec` if neither exists.
    pub async fn read_log(&self, task_id: TaskId) -> Result<Vec<String>> {
        let gz = self.gz_path(task_id);
        if gz.exists() {
            return tokio::task::spawn_blocking(move || read_gz_lines(&gz))
                .await
                .context("spawn_blocking for gz read")?;
        }

        let plain = self.log_path(task_id);
        if plain.exists() {
            let content = tokio::fs::read_to_string(&plain)
                .await
                .with_context(|| format!("read log {:?}", plain))?;
            return Ok(content.lines().map(|l| l.to_string()).collect());
        }

        Ok(vec![])
    }
}

// ---------------------------------------------------------------------------
// LogWriter
// ---------------------------------------------------------------------------

/// An open, buffered writer for a single task's log file.
///
/// Obtained via [`LogStore::open_writer`].
pub struct LogWriter {
    /// Absolute path to the open `.log` file.
    pub path: PathBuf,
    writer: tokio::io::BufWriter<tokio::fs::File>,
}

impl LogWriter {
    /// Append one line of plain text (the `text` field of a `StyledLine`).
    ///
    /// A newline is appended automatically.
    pub async fn append_line(&mut self, text: &str) -> Result<()> {
        self.writer.write_all(text.as_bytes()).await.context("log write")?;
        self.writer.write_all(b"\n").await.context("log write newline")?;
        Ok(())
    }

    /// Flush and close the log file without compression.
    ///
    /// Returns the path of the log file.
    pub async fn close(mut self) -> Result<PathBuf> {
        self.writer.flush().await.context("log flush")?;
        Ok(self.path)
    }

    /// Flush, close, gzip-compress, delete the original `.log`, then enforce
    /// the store's disk quota.
    ///
    /// Returns the path of the compressed `.log.gz` file.
    pub async fn close_and_compress(mut self, store: &LogStore) -> Result<PathBuf> {
        self.writer.flush().await.context("log flush before compress")?;

        let log_path = self.path.clone();

        // Derive the .log.gz path from the .log stem.
        let stem = log_path
            .file_stem()
            .and_then(|s| s.to_str())
            .unwrap_or("unknown");
        let gz_path = store.dir.join(format!("{stem}.log.gz"));

        // Drop the writer to ensure the OS file handle is closed before we
        // open it again for reading inside spawn_blocking.
        drop(self.writer);

        let lp = log_path.clone();
        let gp = gz_path.clone();
        tokio::task::spawn_blocking(move || compress_sync(&lp, &gp))
            .await
            .context("spawn_blocking for compression")??;

        store.enforce_quota().await?;

        Ok(gz_path)
    }
}

// ---------------------------------------------------------------------------
// Sync helpers (run inside spawn_blocking)
// ---------------------------------------------------------------------------

fn compress_sync(src: &Path, dst: &Path) -> Result<()> {
    let input =
        std::fs::File::open(src).with_context(|| format!("open for compress {:?}", src))?;
    let output =
        std::fs::File::create(dst).with_context(|| format!("create gz {:?}", dst))?;
    let mut reader = BufReader::new(input);
    let mut encoder = GzEncoder::new(BufWriter::new(output), Compression::default());
    std::io::copy(&mut reader, &mut encoder).context("compress: io::copy")?;
    encoder.finish().context("compress: gz finish")?;
    // Remove the original only after successful compression so we don't lose
    // data if compression fails mid-stream.
    std::fs::remove_file(src)
        .with_context(|| format!("remove after compress {:?}", src))?;
    Ok(())
}

fn read_gz_lines(path: &Path) -> Result<Vec<String>> {
    let file =
        std::fs::File::open(path).with_context(|| format!("open gz {:?}", path))?;
    let decoder = GzDecoder::new(BufReader::new(file));
    let reader = BufReader::new(decoder);
    let mut lines = Vec::new();
    for line in reader.lines() {
        lines.push(line.context("read gz line")?);
    }
    Ok(lines)
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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

    fn tmp_store(name: &str) -> LogStore {
        let dir = std::env::temp_dir().join(format!("oo_log_test_{name}"));
        // Clean up any run from a previous test invocation.
        let _ = std::fs::remove_dir_all(&dir);
        LogStore::new(dir, DEFAULT_QUOTA_BYTES)
    }

    fn task(n: u64) -> TaskId {
        TaskId(n)
    }

    #[tokio::test]
    async fn creates_dir_and_file() {
        let store = tmp_store("creates_dir");
        assert!(!store.dir.exists(), "dir should not exist yet");
        let writer = store.open_writer(task(1)).await.unwrap();
        assert!(store.dir.exists(), "dir should be created on open");
        assert!(writer.path.exists(), "log file should be created");
        writer.close().await.unwrap();
    }

    #[tokio::test]
    async fn append_and_close_writes_lines() {
        let store = tmp_store("append_close");
        let mut writer = store.open_writer(task(2)).await.unwrap();
        writer.append_line("hello").await.unwrap();
        writer.append_line("world").await.unwrap();
        let path = writer.close().await.unwrap();
        let content = std::fs::read_to_string(&path).unwrap();
        assert_eq!(content, "hello\nworld\n");
    }

    #[tokio::test]
    async fn list_entries_nonexistent_dir() {
        let store = tmp_store("list_nonexistent");
        let entries = store.list_entries().await.unwrap();
        assert!(entries.is_empty());
    }

    #[tokio::test]
    async fn list_entries_sees_log_file() {
        let store = tmp_store("list_log");
        let mut w = store.open_writer(task(10)).await.unwrap();
        w.append_line("test").await.unwrap();
        w.close().await.unwrap();

        let entries = store.list_entries().await.unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].task_id, Some(task(10)));
        assert!(!entries[0].compressed);
    }

    #[tokio::test]
    async fn compress_and_read_back() {
        let store = tmp_store("compress_read");
        let mut w = store.open_writer(task(20)).await.unwrap();
        for i in 0..10 {
            w.append_line(&format!("line {i}")).await.unwrap();
        }
        let gz = w.close_and_compress(&store).await.unwrap();

        assert!(gz.exists(), ".log.gz should exist");
        assert!(
            !store.log_path(task(20)).exists(),
            "original .log should be removed after compression"
        );

        let lines = store.read_log(task(20)).await.unwrap();
        assert_eq!(lines.len(), 10);
        assert_eq!(lines[0], "line 0");
        assert_eq!(lines[9], "line 9");
    }

    #[tokio::test]
    async fn read_log_plain() {
        let store = tmp_store("read_plain");
        let mut w = store.open_writer(task(30)).await.unwrap();
        w.append_line("alpha").await.unwrap();
        w.close().await.unwrap();

        let lines = store.read_log(task(30)).await.unwrap();
        assert_eq!(lines, vec!["alpha"]);
    }

    #[tokio::test]
    async fn read_log_missing_returns_empty() {
        let store = tmp_store("read_missing");
        let lines = store.read_log(task(99)).await.unwrap();
        assert!(lines.is_empty());
    }

    #[tokio::test]
    async fn list_entries_sees_compressed_file() {
        let store = tmp_store("list_compressed");
        let mut w = store.open_writer(task(50)).await.unwrap();
        w.append_line("compressed data").await.unwrap();
        w.close_and_compress(&store).await.unwrap();

        let entries = store.list_entries().await.unwrap();
        assert_eq!(entries.len(), 1);
        assert!(entries[0].compressed);
        assert_eq!(entries[0].task_id, Some(task(50)));
    }

    #[tokio::test]
    async fn total_size_is_sum_of_entries() {
        let store = tmp_store("total_size");
        let mut w = store.open_writer(task(60)).await.unwrap();
        w.append_line("some content").await.unwrap();
        w.close().await.unwrap();

        let total = store.total_size().await.unwrap();
        assert!(total > 0);
        let sum: u64 = store.list_entries().await.unwrap().iter().map(|e| e.size_bytes).sum();
        assert_eq!(total, sum);
    }

    #[tokio::test]
    async fn enforce_quota_deletes_oldest() {
        let store = LogStore::new(
            std::env::temp_dir().join("oo_log_test_quota"),
            1, // 1-byte quota: any file exceeds it
        );
        let _ = std::fs::remove_dir_all(&store.dir);

        let mut w1 = store.open_writer(task(100)).await.unwrap();
        w1.append_line("file one").await.unwrap();
        w1.close().await.unwrap();

        // Allow mtime to differ on fast filesystems.
        tokio::time::sleep(Duration::from_millis(20)).await;

        let mut w2 = store.open_writer(task(101)).await.unwrap();
        w2.append_line("file two").await.unwrap();
        w2.close().await.unwrap();

        let deleted = store.enforce_quota().await.unwrap();
        assert!(!deleted.is_empty(), "expected at least one deletion");
        // Oldest file (task 100) must be deleted first.
        assert!(
            deleted[0].to_string_lossy().contains("100"),
            "oldest file should be deleted first; got {:?}",
            deleted
        );
    }

    #[tokio::test]
    async fn multiple_writers_are_independent() {
        let store = tmp_store("multi_writer");
        let mut w1 = store.open_writer(task(200)).await.unwrap();
        let mut w2 = store.open_writer(task(201)).await.unwrap();
        w1.append_line("task 200").await.unwrap();
        w2.append_line("task 201").await.unwrap();
        w1.close().await.unwrap();
        w2.close().await.unwrap();

        let l1 = store.read_log(task(200)).await.unwrap();
        let l2 = store.read_log(task(201)).await.unwrap();
        assert_eq!(l1, vec!["task 200"]);
        assert_eq!(l2, vec!["task 201"]);
    }
}