atomwrite 0.1.38

// SPDX-License-Identifier: MIT OR Apache-2.0

//! WAL statistics and journal walk (CPU+I/O; rayon over independent journals).

use std::path::{Path, PathBuf};
use std::time::{Instant, SystemTime, UNIX_EPOCH};

use anyhow::Result;
use ignore::WalkBuilder;
use rayon::prelude::*;
use serde::Serialize;

use super::journal::JOURNAL_EXT;

/// Snapshot of journal state for `wal-stats` (G119 L5 local diagnostics).
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct WalStats {
    /// Envelope type discriminator for NDJSON consumers.
    #[serde(rename = "type")]
    pub r#type: &'static str,
    /// Total number of sidecar journals found in the workspace.
    pub total_journals: u64,
    /// Breakdown by terminal state.
    pub by_state: WalStateBreakdown,
    /// Age of the oldest journal in seconds (0 if none).
    pub oldest_journal_age_secs: u64,
    /// Total bytes occupied by all sidecar files.
    pub total_size_bytes: u64,
    /// Top directories by journal count (up to 10).
    pub by_directory: Vec<WalDirEntry>,
    /// True if the workspace has accumulated enough journals to warrant
    /// an auto-heal pass.
    pub auto_heal_recommended: bool,
    /// Estimated bytes that would be reclaimed by an auto-heal pass.
    pub estimated_reclaim_bytes: u64,
}

/// Breakdown of journals by terminal state.
#[derive(Debug, Clone, Default, Serialize, schemars::JsonSchema)]
pub struct WalStateBreakdown {
    /// Journals whose last entry is `Started` (potential orphans).
    pub started: u64,
    /// Journals whose last entry is `Committed` (safe to clean).
    pub committed: u64,
    /// Journals whose last entry is `Aborted` (safe to clean).
    pub aborted: u64,
    /// Journals that could not be parsed (do not auto-clean).
    pub malformed: u64,
}

/// Count of journals in a single directory.
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct WalDirEntry {
    /// Directory path relative to the workspace root.
    pub path: String,
    /// Number of journals in this directory.
    pub count: u64,
}

/// Compute a snapshot of the current journal state. Read-only and safe
/// to call from any context. Used by the `wal-stats` subcommand (G119 L5).
///
/// Discovery is `WalkParallel`; per-journal parse fans out with `par_iter`
/// when more than one sidecar is present (independent I/O).
///
/// # Errors
///
/// Returns the `anyhow::Error` propagated from [`walk_journal_paths`] when
/// discovery fails. A journal that cannot be read or parsed is counted under
/// `by_state.malformed` instead of failing the snapshot, and a missing
/// metadata size is recorded as `0`.
pub fn compute_wal_stats(workspace: &Path) -> Result<WalStats> {
    use std::collections::BTreeMap;

    let paths = walk_journal_paths(workspace)?;
    let workspace = workspace.to_path_buf();

    // Independent read+parse per journal → parallel when multi-item.
    if crate::concurrency::should_parallelize(paths.len()) {
        crate::concurrency::ensure_global_pool();
    }
    let rows: Vec<(PathBuf, u64, &'static str, u64)> =
        if crate::concurrency::should_parallelize(paths.len()) {
            paths
                .par_iter()
                .map(|path| {
                    let size = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0);
                    let (state, last_unix) = parse_journal_state(path).unwrap_or(("malformed", 0));
                    (path.clone(), size, state, last_unix)
                })
                .collect()
        } else {
            paths
                .iter()
                .map(|path| {
                    let size = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0);
                    let (state, last_unix) = parse_journal_state(path).unwrap_or(("malformed", 0));
                    (path.clone(), size, state, last_unix)
                })
                .collect()
        };

    let mut total: u64 = 0;
    let mut by_state = WalStateBreakdown::default();
    let mut oldest_unix: u64 = u64::MAX;
    let mut total_size: u64 = 0;
    let mut by_dir: BTreeMap<String, u64> = BTreeMap::new();

    for (path, size, state, last_unix) in rows {
        total += 1;
        total_size += size;

        let rel_dir = path
            .parent()
            .and_then(|p| p.strip_prefix(&workspace).ok())
            .map(|p| {
                let shown = p.display().to_string();
                // A journal sitting in the workspace root strips to an empty
                // relative path; reporting "" reads as a missing directory.
                if shown.is_empty() {
                    ".".to_string()
                } else {
                    shown
                }
            })
            .unwrap_or_else(|| ".".to_string());
        *by_dir.entry(rel_dir).or_insert(0) += 1;

        match state {
            "Committed" => by_state.committed += 1,
            "Aborted" => by_state.aborted += 1,
            "Started" => by_state.started += 1,
            _ => by_state.malformed += 1,
        }
        if state != "malformed" && last_unix < oldest_unix {
            oldest_unix = last_unix;
        }
    }

    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let oldest_age = if oldest_unix == u64::MAX {
        0
    } else {
        now.saturating_sub(oldest_unix)
    };

    let mut by_directory: Vec<WalDirEntry> = by_dir
        .into_iter()
        .map(|(path, count)| WalDirEntry { path, count })
        .collect();
    by_directory.sort_by(|a, b| b.count.cmp(&a.count));
    by_directory.truncate(crate::constants::WAL_STATS_TOP_DIRS);

    // A-013: named thresholds (constants / XDG-aligned defaults)
    let auto_heal_recommended = total > crate::constants::WAL_AUTO_HEAL_COUNT_THRESHOLD
        || oldest_age > crate::constants::WAL_AUTO_HEAL_AGE_SECS;
    let estimated_reclaim_bytes = if auto_heal_recommended { total_size } else { 0 };

    Ok(WalStats {
        r#type: "wal_stats",
        total_journals: total,
        by_state,
        oldest_journal_age_secs: oldest_age,
        total_size_bytes: total_size,
        by_directory,
        auto_heal_recommended,
        estimated_reclaim_bytes,
    })
}

/// Recursively walk a directory and yield all `*.atomwrite.journal.json`
/// sidecar paths. Returns an empty Vec if the workspace does not exist.
///
/// Uses `WalkParallel` (hidden sidecars included) so discovery scales on
/// monorepos; paths are sorted for stable stats/heal order.
///
/// # Errors
///
/// The signature admits `Err` for a future fallible walker, but the current
/// body never returns one: a missing workspace answers `Ok(Vec::new())`, and
/// directories the walk cannot open are skipped by the collector rather than
/// surfaced.
pub fn walk_journal_paths(workspace: &Path) -> Result<Vec<PathBuf>> {
    if !workspace.exists() {
        return Ok(Vec::new());
    }
    let mut builder = WalkBuilder::new(workspace);
    // Journals are hidden-name sidecars (`.atomwrite.journal.*`).
    builder.hidden(false).git_ignore(false);
    // Bound by process-wide pool (configured in run via --threads).
    crate::concurrency::apply_walk_threads(&mut builder, None);
    let mut out = crate::concurrency::collect_mapped_parallel(&builder, |entry| {
        if !entry.file_type().is_some_and(|ft| ft.is_file()) {
            return None;
        }
        let name = entry.file_name().to_str()?;
        if name.ends_with(JOURNAL_EXT) {
            Some(entry.path().to_path_buf())
        } else {
            None
        }
    });
    crate::concurrency::sort_paths_parallel(&mut out);
    Ok(out)
}

/// One directory the startup heal may scan.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HealScope {
    /// Directory to scan (never a file).
    pub root: PathBuf,
    /// Walk depth below `root`; `None` means the whole subtree.
    pub max_depth: Option<usize>,
}

/// Discover journals under `scopes` with a wall-clock cutoff checked per entry.
///
/// Unlike [`walk_journal_paths`], this walk is sequential (it never spawns
/// more threads than `--threads` allows), honours `.gitignore`/`.ignore`, never
/// enters [`crate::constants::WAL_STARTUP_HEAL_SKIP_DIRS`], and stops as soon as
/// `deadline` passes or the run is cancelled, so the cost is bounded by the
/// budget and not by the size of the tree.
pub fn walk_journal_paths_bounded(scopes: &[HealScope], deadline: Instant) -> Vec<PathBuf> {
    let mut out = Vec::new();
    for scope in scopes {
        if !scope.root.is_dir() {
            continue;
        }
        let mut builder = WalkBuilder::new(&scope.root);
        builder
            .hidden(false)
            .max_depth(scope.max_depth)
            .filter_entry(|entry| {
                let is_dir = entry.file_type().is_some_and(|ft| ft.is_dir());
                !(is_dir
                    && entry.depth() > 0
                    && entry.file_name().to_str().is_some_and(|name| {
                        crate::constants::WAL_STARTUP_HEAL_SKIP_DIRS.contains(&name)
                    }))
            });
        for result in builder.build() {
            if Instant::now() >= deadline || crate::signal::is_global_shutdown() {
                tracing::debug!("startup wal-heal walk cut at the time budget");
                return out;
            }
            let Ok(entry) = result else { continue };
            if !entry.file_type().is_some_and(|ft| ft.is_file()) {
                continue;
            }
            if entry
                .file_name()
                .to_str()
                .is_some_and(|name| name.ends_with(JOURNAL_EXT))
            {
                out.push(entry.into_path());
            }
        }
    }
    out
}

/// Parse a sidecar and return `(terminal_state, last_unix)` for the last
/// entry. Used by `wal-stats` to classify journals without paying the
/// full `OrphanJournalReport` cost.
pub(crate) fn parse_journal_state(path: &Path) -> Option<(&'static str, u64)> {
    let content = std::fs::read_to_string(path).ok()?;
    let mut state = "malformed";
    let mut last_unix: u64 = 0;
    for line in content.lines() {
        let val: serde_json::Value = match serde_json::from_str(line) {
            Ok(v) => v,
            Err(_) => {
                state = "malformed";
                continue;
            }
        };
        let phase = val.get("phase").and_then(|v| v.as_str()).unwrap_or("");
        match phase {
            "started" => {
                state = "Started";
                last_unix = val
                    .get("started_at_unix")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(0);
            }
            "committed" => {
                state = "Committed";
                last_unix = val
                    .get("committed_at_unix")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(0);
            }
            "aborted" => {
                state = "Aborted";
                last_unix = val
                    .get("aborted_at_unix")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(0);
            }
            _ => {
                state = "malformed";
            }
        }
    }
    Some((state, last_unix))
}