atomwrite 0.1.38

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

//! Prune timestamped `.bak.YYYYMMDD_HHMMSS` files by age or by count.
//! Workload: I/O-bound (readdir + filter + delete).
//! Parallelism: phase-1 readdir/retention fans out per target (`par_iter`);
//! phase-2 unlinks all selected backups with one `par_iter`. NDJSON emits
//! in path-sorted order after the join. Bound: process-wide rayon pool.
//!
//! ADR-0040 — `prune-backups` is the bulk-complement of `backup`'s
//! per-file retention. Operators invoke it once to enforce a global
//! policy across many target files.

use std::fs;
use std::io::Write;
use std::path::PathBuf;
use std::time::{Duration, Instant, SystemTime};

use anyhow::{Context, Result};
use rayon::prelude::*;

use crate::cli::{GlobalArgs, PruneBackupsArgs};
use crate::concurrency::should_parallelize;
use crate::ndjson_types::{PruneBackupEntry, PruneBackupSummary};
use crate::output::NdjsonWriter;
use crate::path_safety::validate_path;

/// Prune timestamped backups of one or more target files.
///
/// Emits one `prune-backups` event per backup handled (pruned, skipped,
/// or error) and a single `summary` event at the end.
///
/// # Errors
///
/// Returns `AtomwriteError::WorkspaceJail` if any path escapes the workspace.
/// Returns an I/O error if the parent directory cannot be listed.
#[tracing::instrument(skip_all, fields(command = "prune-backups"))]
pub fn cmd_prune_backups(
    args: &PruneBackupsArgs,
    global: &GlobalArgs,
    writer: &mut NdjsonWriter<impl Write>,
) -> Result<()> {
    let start = Instant::now();
    let workspace = global.resolve_workspace()?;
    let dry_run = args.dry_run;
    let mut total_pruned: usize = 0;

    // SAFETY (VAI-PSIQUE-CHECK per ADR-0040): refuse without a retention filter.
    if args.max_age_secs.is_none() && args.max_count.is_none() {
        return Err(crate::error::AtomwriteError::InvalidInput {
            reason: "refusing to prune without --max-age-secs or --max-count; \
                     pass at least one to define the retention policy"
                .into(),
        }
        .into());
    }

    let reason = if args.max_age_secs.is_some() {
        "age"
    } else {
        "count"
    };
    let reason_owned = reason.to_string();

    // Phase 1 — per-target retention (policy is per-file) but readdir fans out
    // across independent targets, then flatten for the unlink stage.
    let scope = PruneScope {
        max_age_secs: args.max_age_secs,
        max_count: args.max_count,
        recursive: args.recursive,
        global,
    };
    type Phase1Item = Result<(Vec<PathBuf>, Option<PruneBackupEntry>), anyhow::Error>;
    let phase1: Vec<Phase1Item> = if should_parallelize(args.paths.len()) {
        args.paths
            .par_iter()
            .map(|raw_path| collect_backups_for_target(raw_path, &workspace, &scope))
            .collect()
    } else {
        args.paths
            .iter()
            .map(|raw_path| collect_backups_for_target(raw_path, &workspace, &scope))
            .collect()
    };

    let mut to_prune: Vec<PathBuf> = Vec::new();
    // A target that does not exist is emitted here, before any backup is
    // selected, so the count is taken in this loop and not in the unlink one.
    let mut targets_missing = 0u64;
    for item in phase1 {
        let (backups, skip) = item?;
        if let Some(entry) = skip {
            if entry.reason == "not_found" {
                targets_missing += 1;
            }
            writer.write_event(&entry)?;
        }
        to_prune.extend(backups);
    }

    crate::concurrency::sort_paths_parallel(&mut to_prune);
    to_prune.dedup();
    // `--limit` bounds the work: `prune-backups --limit 1` deletes one backup.
    crate::output::cap_collected(&mut to_prune);

    // Phase 2 — one fan-out across all selected backups (multi-target).
    let outcomes: Vec<(PruneBackupEntry, Option<anyhow::Error>)> =
        if should_parallelize(to_prune.len()) {
            to_prune
                .par_iter()
                .map(|backup| prune_one(backup, dry_run, &reason_owned))
                .collect()
        } else {
            to_prune
                .iter()
                .map(|backup| prune_one(backup, dry_run, &reason_owned))
                .collect()
        };

    // Emit in path order for stable NDJSON (par_iter order is not sorted).
    let mut outcomes = outcomes;
    outcomes.sort_by(|(a, _), (b, _)| a.path.cmp(&b.path));

    // A backup that could not be removed leaves the disk as it was, so the run
    // has to say so through the exit code: the summary is a verdict about the
    // whole set, and a hole in it makes the verdict `partial`.
    let mut failures = crate::ndjson_types::FailureTally::default();
    for (entry, err) in outcomes {
        if entry.r#type == "pruned" {
            total_pruned += 1;
        }
        if let Some(err) = err {
            failures.record(std::path::Path::new(&entry.path), err);
        }
        writer.write_event(&entry)?;
    }

    let (failed_count, failed_items, first) = failures.into_parts();
    let files_failed = (failed_count > 0).then_some(failed_count);
    let summary = PruneBackupSummary {
        r#type: "summary",
        action: if dry_run { "dry_run" } else { "pruned" }.to_string(),
        total: total_pruned,
        files_budget_dropped: Some(crate::output::budget_dropped()).filter(|n| *n > 0),
        elapsed_ms: start.elapsed().as_millis() as u64,
        walker_stopped_early: crate::output::limit_reached().then_some(true),
        files_failed,
        partial: crate::ndjson_types::partial_of(files_failed),
        errors: (!failed_items.is_empty()).then_some(failed_items),
        targets_missing: (targets_missing > 0).then_some(targets_missing),
    };
    if let Some(err) = first {
        return Err(writer.publish_failure(&summary, err));
    }
    writer.write_terminal(&summary)?;

    Ok(())
}

/// Retention policy plus the walk settings a directory argument needs.
#[derive(Clone, Copy)]
struct PruneScope<'a> {
    max_age_secs: Option<u32>,
    max_count: Option<u8>,
    recursive: bool,
    global: &'a GlobalArgs,
}

/// Per-target readdir + retention filter (safe to run in parallel across targets).
fn collect_backups_for_target(
    raw_path: &std::path::Path,
    workspace: &std::path::Path,
    scope: &PruneScope<'_>,
) -> Result<(Vec<PathBuf>, Option<PruneBackupEntry>)> {
    let PruneScope {
        max_age_secs,
        max_count,
        recursive,
        ..
    } = *scope;
    let target = validate_path(raw_path, workspace)?;

    if !target.exists() {
        return Ok((
            Vec::new(),
            Some(PruneBackupEntry {
                r#type: "skipped",
                path: target.display().to_string(),
                reason: "not_found".to_string(),
                error: None,
            }),
        ));
    }

    // A directory is not a backup subject unless the caller says `-r`. Before
    // this guard the verb derived a prefix from the directory name, found
    // nothing and exited 0, which reads as "nothing to prune" for a request
    // that named the wrong kind of path. A verb that deletes files fails closed
    // instead (ETD: the subject is the file argv named, never a tree guessed
    // from it), and the flag is the caller turning that tree into a subject.
    if target.is_dir() {
        if !recursive {
            return Err(crate::error::AtomwriteError::InvalidInput {
                reason: format!(
                    "{} is a directory [{}]; prune-backups takes the files whose backups are pruned, or a tree with --recursive",
                    target.display(),
                    crate::constants::PRUNE_DIRECTORY_REASON_MARKER
                ),
            }
            .into());
        }
        return collect_backups_in_tree(&target, scope);
    }

    let parent = target.parent().unwrap_or_else(|| std::path::Path::new("."));
    let file_name = target
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("")
        .to_string();
    let prefix = format!("{file_name}{}", crate::constants::BACKUP_INFIX);

    let mut backups: Vec<PathBuf> = fs::read_dir(parent)
        .with_context(|| format!("cannot list directory {}", parent.display()))?
        .filter_map(|e| e.ok())
        .map(|e| e.path())
        .filter(|p| {
            let named = p.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
                n.starts_with(&prefix) && crate::commands::is_backup_timestamp_name(n)
            });
            // `symlink_metadata` describes the entry itself: a directory that
            // carries a backup name matches the name and is not a backup, and
            // following a link here would offer to prune whatever it points at.
            named && fs::symlink_metadata(p).is_ok_and(|meta| meta.file_type().is_file())
        })
        .collect();

    retain_older_than(&mut backups, max_age_secs);

    if let Some(max_count) = max_count {
        if max_count > 0 {
            backups.sort();
            let to_delete = backups.len().saturating_sub(usize::from(max_count));
            backups.truncate(to_delete);
        }
    }

    Ok((backups, None))
}

/// Every timestamped backup under `root`, aged and ranked per original file.
///
/// Only `--recursive` reaches here, and only names that carry a valid stamp
/// become candidates: the stamp is the provenance that the file is a backup
/// this tool wrote, and a directory that merely looks like one is never
/// removed by a verb whose contract is "prune backups".
fn collect_backups_in_tree(
    root: &std::path::Path,
    scope: &PruneScope<'_>,
) -> Result<(Vec<PathBuf>, Option<PruneBackupEntry>)> {
    let walker = crate::commands::build_scoped_walker(
        std::slice::from_ref(&root.to_path_buf()),
        scope.global,
        &[],
        &[],
    )?;
    let mut backups: Vec<PathBuf> = crate::concurrency::collect_mapped_parallel(&walker, |entry| {
        if !entry.file_type().is_some_and(|kind| kind.is_file()) {
            return None;
        }
        let name = entry.file_name().to_str()?;
        crate::commands::backup_names::is_backup_path(std::path::Path::new(name))
            .then(|| entry.path().to_path_buf())
    });

    retain_older_than(&mut backups, scope.max_age_secs);

    // `--max-count` is per original file, exactly like the single-target path:
    // a tree with ten targets and `--max-count 1` keeps one backup of each.
    if let Some(max_count) = scope.max_count.filter(|count| *count > 0) {
        let mut by_target: std::collections::BTreeMap<PathBuf, Vec<PathBuf>> =
            std::collections::BTreeMap::new();
        for backup in backups {
            by_target
                .entry(backup_target_of(&backup))
                .or_default()
                .push(backup);
        }
        backups = Vec::new();
        for (_, mut group) in by_target {
            group.sort();
            let excess = group.len().saturating_sub(usize::from(max_count));
            group.truncate(excess);
            backups.extend(group);
        }
    }

    Ok((backups, None))
}

/// The original file a backup name belongs to, used to group a tree by target.
fn backup_target_of(backup: &std::path::Path) -> PathBuf {
    let parent = backup.parent().unwrap_or_else(|| std::path::Path::new("."));
    backup
        .file_name()
        .and_then(|name| name.to_str())
        .and_then(|name| name.rsplit_once(crate::constants::BACKUP_INFIX))
        .map_or_else(|| backup.to_path_buf(), |(base, _)| parent.join(base))
}

/// Keep the backups strictly older than the age limit.
///
/// The age comes from the stamp in the name, because that is the moment this
/// tool wrote the copy. The mtime is only the fallback: a reflink or a clone
/// carries the source file's time, so a backup made a second ago from a year
/// old file looked a year old and was pruned in the same breath.
fn retain_older_than(backups: &mut Vec<PathBuf>, max_age_secs: Option<u32>) {
    let Some(max_age_secs) = max_age_secs else {
        return;
    };
    let cutoff = SystemTime::now() - Duration::from_secs(u64::from(max_age_secs));
    backups.retain(|backup| {
        let stamped = backup
            .file_name()
            .and_then(|name| name.to_str())
            .and_then(crate::commands::backup_names::parse_backup_timestamp);
        let created = stamped.or_else(|| backup.metadata().and_then(|m| m.modified()).ok());
        created.is_some_and(|created| created < cutoff)
    });
}

fn prune_one(
    backup: &PathBuf,
    dry_run: bool,
    reason: &str,
) -> (PruneBackupEntry, Option<anyhow::Error>) {
    if !dry_run && let Err(e) = fs::remove_file(backup) {
        let entry = PruneBackupEntry {
            r#type: "error",
            path: backup.display().to_string(),
            reason: "remove_failed".to_string(),
            error: Some(e.to_string()),
        };
        let err = crate::error::AtomwriteError::from_io_at(backup, e).into();
        return (entry, Some(err));
    }
    let entry = PruneBackupEntry {
        r#type: "pruned",
        path: backup.display().to_string(),
        reason: reason.to_string(),
        error: None,
    };
    (entry, None)
}