//! Removing snapshots and sweeping what nothing needs any more, and the
//! machinery that keeps that from taking chunks out from under a backup
//! that is still going to need them.

mod common;

use std::sync::{Arc, atomic::AtomicBool};

use beeping::{
    pipeline::{Outcome, backup},
    prune::prune,
};
use repository::{Coverage, Lock, Repository};

use common::{deterministic_bytes, manifest_map, open_repository, read_back};

/// Backs up `root` into `repository`, suspending at once if asked.
fn run(
    repository: &Arc<Repository>,
    root: &std::path::Path,
    state: &std::path::Path,
    suspend: bool,
) -> Outcome {
    let (events, receiver) = crossbeam_channel::unbounded();
    drop(receiver);

    backup::run(
        repository.clone(),
        backup::BackupOptions::new(root, state),
        events,
        Arc::new(AtomicBool::new(suspend)),
    )
    .unwrap()
}

fn tree(root: &std::path::Path, files: std::ops::Range<u64>) {
    std::fs::create_dir_all(root).unwrap();

    for index in files {
        std::fs::write(
            root.join(format!("file-{index}.dat")),
            deterministic_bytes(20_000, index),
        )
        .unwrap();
    }
}

/// Pruning takes away the chunks nothing needs any more, and nothing
/// else: what a surviving snapshot shares with the removed one stays.
#[test]
fn pruning_sweeps_only_what_nothing_needs() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let repository = open_repository(dir.path());

    // One snapshot, then a second with some of the same content and
    // some new
    tree(&root, 0..30);
    let Outcome::Completed(first) = run(&repository, &root, &dir.path().join("state-1"), false)
    else {
        panic!("expected completion");
    };

    tree(&root, 30..50);
    for index in 0..10 {
        std::fs::remove_file(root.join(format!("file-{index}.dat"))).unwrap();
    }

    let Outcome::Completed(second) = run(&repository, &root, &dir.path().join("state-2"), false)
    else {
        panic!("expected completion");
    };

    let before = repository.chunks().unwrap().len();
    let survives = manifest_map(&repository, second);

    let outcome = prune(&repository, &[first], false, false, |_| {}).unwrap();

    assert_eq!(outcome.removed, vec![first]);
    assert_eq!(outcome.kept, 1);
    assert!(
        outcome.swept > 0,
        "the removed snapshot had content of its own"
    );
    assert!(
        outcome.retained < before,
        "the repository should have shrunk"
    );
    assert_eq!(repository.chunks().unwrap().len(), outcome.retained);

    // The removed snapshot is gone; the surviving one is untouched, and
    // every byte of it still reads back
    assert!(matches!(
        repository.load_snapshot(first),
        Err(repository::Error::MissingSnapshot(_))
    ));

    let after = manifest_map(&repository, second);
    assert_eq!(after.len(), survives.len());

    for (path, entry) in &after {
        if let repository::EntryKind::File { .. } = entry.kind {
            assert_eq!(
                read_back(&repository, entry),
                std::fs::read(root.join(path)).unwrap(),
                "{path:?} did not survive the prune intact"
            );
        }
    }
}

/// A backup that has stored chunks no snapshot names yet must not have
/// them swept away underneath it.
#[test]
fn pruning_refuses_while_a_backup_holds_the_repository() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let repository = open_repository(dir.path());

    tree(&root, 0..10);
    let Outcome::Completed(snapshot) = run(&repository, &root, &dir.path().join("state"), false)
    else {
        panic!("expected completion");
    };

    // A run that was killed leaves its lock behind, which is exactly the
    // state that should stop a prune
    let lock = Lock::new(root.clone(), "some-other-host");
    repository.take_lock(&lock).unwrap();

    let refused = prune(&repository, &[snapshot], false, false, |_| {}).unwrap_err();
    let message = format!("{refused}");

    assert!(message.contains("some-other-host"), "{message}");
    assert!(message.contains("--break-lock"), "{message}");
    assert!(
        repository.load_snapshot(snapshot).is_ok(),
        "a refused prune must not have removed anything"
    );

    // Breaking the lock is the documented way through
    prune(&repository, &[snapshot], false, true, |_| {}).unwrap();
    assert!(matches!(
        repository.load_snapshot(snapshot),
        Err(repository::Error::MissingSnapshot(_))
    ));

    // A dry run says what it would do and leaves the lock's owner alone
    repository.take_lock(&lock).unwrap();
    assert!(prune(&repository, &[], true, false, |_| {}).is_err());
}

/// The lock is held for exactly as long as the run has references no
/// snapshot names: dropped when it finishes, and dropped when it
/// suspends, because suspending publishes them.
#[test]
fn a_backup_holds_the_repository_only_while_it_owes_it_something() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let repository = open_repository(dir.path());

    tree(&root, 0..40);

    let state = dir.path().join("state");
    assert!(matches!(
        run(&repository, &root, &state, true),
        Outcome::Suspended { .. }
    ));

    assert!(
        repository.locks().unwrap().is_empty(),
        "a suspended run has published what it recorded, so it holds nothing"
    );

    assert!(matches!(
        run(&repository, &root, &state, false),
        Outcome::Completed(_)
    ));

    assert!(
        repository.locks().unwrap().is_empty(),
        "a finished run holds nothing either"
    );
}

/// A prune between one session of a backup and the next moves the
/// repository's epoch, and the resumed run puts back whatever it had
/// recorded rather than assuming those chunks are still there.
#[test]
fn a_prune_between_sessions_makes_the_run_re_store_what_it_recorded() {
    let dir = tempfile::tempdir().unwrap();
    let repository = open_repository(dir.path());

    // Something else in the repository, for the prune to remove
    let other = dir.path().join("other");
    tree(&other, 100..120);

    let Outcome::Completed(disposable) =
        run(&repository, &other, &dir.path().join("state-o"), false)
    else {
        panic!("expected completion");
    };

    // The backup we care about, suspended part-way
    let root = dir.path().join("tree");
    tree(&root, 0..60);

    let state = dir.path().join("state");
    assert!(matches!(
        run(&repository, &root, &state, true),
        Outcome::Suspended { .. }
    ));

    let epoch_before = repository.epoch().unwrap();
    prune(&repository, &[disposable], false, false, |_| {}).unwrap();
    assert_ne!(
        repository.epoch().unwrap(),
        epoch_before,
        "a prune must move the epoch, or no cache would know to doubt itself"
    );

    // Resuming notices and finishes correctly regardless
    let Outcome::Completed(snapshot) = run(&repository, &root, &state, false) else {
        panic!("expected completion");
    };

    let map = manifest_map(&repository, snapshot);

    for index in 0..60 {
        let path = std::path::PathBuf::from(format!("file-{index}.dat"));
        let entry = map.get(&path).unwrap_or_else(|| panic!("{path:?} missing"));

        assert_eq!(
            read_back(&repository, entry),
            std::fs::read(root.join(&path)).unwrap(),
            "{path:?} did not survive the prune intact"
        );
    }
}

/// A suspended backup's partial snapshot is what keeps its recorded work
/// findable. Pruning it leaves the run unable to finish, and it must say
/// so rather than producing a snapshot full of dangling references.
#[test]
fn a_pruned_partial_snapshot_stops_its_backup_rather_than_finishing_it_wrong() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let repository = open_repository(dir.path());

    tree(&root, 0..60);

    let state = dir.path().join("state");
    let Outcome::Suspended { restorable } = run(&repository, &root, &state, true) else {
        panic!("expected suspension");
    };

    let partial = restorable.expect("suspending publishes what it has stored");
    assert_eq!(
        repository.load_snapshot(partial).unwrap().coverage,
        Coverage::Partial
    );

    // Pruning it is allowed — it is a snapshot like any other — but it
    // takes the resumed run's recorded work with it
    let mut warnings = Vec::new();
    prune(&repository, &[partial], false, false, |note| {
        warnings.push(note.to_string())
    })
    .unwrap();

    assert!(
        warnings
            .iter()
            .any(|note| note.contains("partial snapshot")),
        "the prune should have said what it was removing: {warnings:?}"
    );

    let (events, receiver) = crossbeam_channel::unbounded();
    drop(receiver);

    let failed = backup::run(
        repository.clone(),
        backup::BackupOptions::new(&root, &state),
        events,
        Arc::new(AtomicBool::new(false)),
    )
    .unwrap_err();

    assert!(
        matches!(failed, beeping::pipeline::Error::RecordedWorkPruned { .. }),
        "expected a refusal to finish, got {failed:?}"
    );

    // And it says what to do about it
    let message = format!("{failed}");
    assert!(message.contains("run the backup again"), "{message}");
}