//! Integration tests for removing paths from the snapshots of one tree:
//! that what is taken out goes, that what is not stays restorable, and
//! that an excision refuses every state in which sweeping would break a
//! backup that is not finished with its work.

mod common;

use std::{
    path::{Path, PathBuf},
    sync::{Arc, atomic::AtomicBool},
};

use beeping::{
    excise::{self, Refusal},
    pipeline::{Outcome, backup},
};
use repository::{Coverage, EntryKind, Repository, SnapshotId};

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

/// A tree with something in it that should not have been backed up.
fn sample_tree(root: &Path) -> Vec<PathBuf> {
    let mut written = Vec::new();

    for (directory, files) in [("keep", 4), ("junk", 3), ("keep/nested", 2)] {
        std::fs::create_dir_all(root.join(directory)).unwrap();

        for file in 0..files {
            let relative = PathBuf::from(directory).join(format!("file-{file}.dat"));
            let bytes = deterministic_bytes(2000 + file * 97, (files * 31 + file) as u64);

            std::fs::write(root.join(&relative), &bytes).unwrap();
            written.push(relative);
        }
    }

    written
}

fn run_to_completion(repository: &Arc<Repository>, root: &Path, state: &Path) -> SnapshotId {
    let outcome = backup::run(
        repository.clone(),
        backup::BackupOptions::new(root, state),
        crossbeam_channel::unbounded().0,
        Arc::new(AtomicBool::new(false)),
    )
    .unwrap();

    match outcome {
        Outcome::Completed(snapshot) => snapshot,
        other => panic!("expected completion: {other:?}"),
    }
}

fn selecting(patterns: &[&str]) -> globset::GlobSet {
    beeping::patterns::compile_set(patterns).unwrap()
}

/// The paths go from every snapshot of the tree, their content is swept,
/// and everything else stays exactly as restorable as it was.
#[test]
fn excising_takes_a_directory_out_of_every_snapshot_of_a_tree() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    let written = sample_tree(&root);
    let repository = open_repository(dir.path());

    let first = run_to_completion(&repository, &root, &state);

    // A second backup of the same tree, so the excision has to reach
    // more than one snapshot
    std::fs::write(root.join("keep/later.dat"), b"added between backups").unwrap();
    let second = run_to_completion(&repository, &root, &state);

    // ...and a backup of a different tree, which must come through
    // untouched however alike its paths look
    let other_root = dir.path().join("other");
    std::fs::create_dir_all(other_root.join("junk")).unwrap();
    std::fs::write(other_root.join("junk/file-0.dat"), b"not mine to remove").unwrap();

    let other = run_to_completion(&repository, &other_root, &dir.path().join("other-state"));
    let before = manifest_map(&repository, other);

    let outcome = excise::excise(
        &repository,
        &root,
        &selecting(&["/junk"]),
        &state,
        false,
        false,
        |_| {},
    )
    .unwrap();

    assert_eq!(outcome.rewritten.len(), 2, "both snapshots held junk");
    assert_eq!(outcome.untouched, 0);
    assert_eq!(outcome.unqueued, 0, "no backup was suspended");
    assert!(outcome.swept > 0, "the junk's content should have gone");

    for rewritten in &outcome.rewritten {
        // Three files, and the directory that held them
        assert_eq!(rewritten.entries, 4);
        assert!(rewritten.content_bytes > 0);
    }

    // The old snapshots are gone and their replacements stand in the
    // same order, still saying when they were made
    let now: Vec<SnapshotId> = repository.snapshots().unwrap();
    assert_eq!(now.len(), 3);
    assert!(!now.contains(&first) && !now.contains(&second));

    let replacements: Vec<SnapshotId> = outcome
        .rewritten
        .iter()
        .map(|rewritten| rewritten.to.unwrap())
        .collect();

    for (original, id) in [(first, replacements[0]), (second, replacements[1])] {
        let rewritten = repository.load_snapshot(id).unwrap();

        assert_eq!(rewritten.root, root);
        assert_eq!(rewritten.coverage, Coverage::Complete);

        // Everything under junk has gone, and the directory with it
        let entries = manifest_map(&repository, id);

        assert!(!entries.contains_key(Path::new("junk")));
        assert!(entries.contains_key(Path::new("keep")));
        assert!(entries.contains_key(Path::new("keep/nested")));

        for path in &written {
            let expected = !path.starts_with("junk");
            assert_eq!(
                entries.contains_key(path),
                expected,
                "{path:?} in {original} -> {id}"
            );
        }

        // And what is left is not merely named but there
        for entry in entries.values() {
            if matches!(entry.kind, EntryKind::File { .. }) {
                assert_eq!(
                    read_back(&repository, entry),
                    std::fs::read(root.join(&entry.path)).unwrap(),
                    "{:?}",
                    entry.path
                );
            }
        }
    }

    // The other tree's snapshot is untouched, chunks and all
    assert!(now.contains(&other));
    assert_eq!(manifest_map(&repository, other), before);

    for entry in before.values() {
        if matches!(entry.kind, EntryKind::File { .. }) {
            read_back(&repository, entry);
        }
    }
}

/// Two snapshots that differ only in what was excised still describe two
/// backups, and must not collapse into one record.
#[test]
fn snapshots_left_identical_by_an_excision_stay_separate() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    std::fs::create_dir_all(root.join("junk")).unwrap();
    std::fs::write(root.join("keep.dat"), b"unchanging").unwrap();
    std::fs::write(root.join("junk/churn.dat"), b"before").unwrap();

    let repository = open_repository(dir.path());
    let first = run_to_completion(&repository, &root, &state);

    // Only the junk changes between the two
    std::fs::write(root.join("junk/churn.dat"), b"after!").unwrap();
    let second = run_to_completion(&repository, &root, &state);

    assert_ne!(first, second);

    let made: Vec<u64> = [first, second]
        .iter()
        .map(|id| repository.load_snapshot(*id).unwrap().created)
        .collect();

    let outcome = excise::excise(
        &repository,
        &root,
        &selecting(&["/junk"]),
        &state,
        false,
        false,
        |_| {},
    )
    .unwrap();

    let ids: Vec<SnapshotId> = outcome
        .rewritten
        .iter()
        .map(|rewritten| rewritten.to.unwrap())
        .collect();

    assert_eq!(ids.len(), 2);
    assert_ne!(ids[0], ids[1], "two backups became one record");
    assert_eq!(repository.snapshots().unwrap().len(), 2);

    // Each still says when its own backup ran, which is the whole of
    // what tells them apart now
    let created: Vec<u64> = ids
        .iter()
        .map(|id| repository.load_snapshot(*id).unwrap().created)
        .collect();

    assert_eq!(created, made);
}

/// A dry run answers the only question worth asking first — did the
/// patterns select what I meant? — and changes nothing at all.
#[test]
fn a_dry_run_writes_nothing() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    sample_tree(&root);

    let repository = open_repository(dir.path());
    let snapshot = run_to_completion(&repository, &root, &state);
    let before = manifest_map(&repository, snapshot);

    let outcome = excise::excise(
        &repository,
        &root,
        &selecting(&["/junk"]),
        &state,
        true,
        false,
        |_| {},
    )
    .unwrap();

    assert_eq!(outcome.rewritten.len(), 1);
    assert_eq!(outcome.rewritten[0].from, snapshot);
    assert_eq!(outcome.rewritten[0].to, None);
    assert_eq!(outcome.rewritten[0].entries, 4);
    assert_eq!(outcome.swept, 0);

    assert_eq!(repository.snapshots().unwrap(), vec![snapshot]);
    assert_eq!(manifest_map(&repository, snapshot), before);
}

/// Nothing selected is not an error, and not a reason to rewrite
/// anything either: a snapshot that held none of it keeps its identity.
#[test]
fn a_pattern_that_selects_nothing_leaves_every_snapshot_alone() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    sample_tree(&root);

    let repository = open_repository(dir.path());
    let snapshot = run_to_completion(&repository, &root, &state);

    let outcome = excise::excise(
        &repository,
        &root,
        &selecting(&["/nothing-like-this"]),
        &state,
        false,
        false,
        |_| {},
    )
    .unwrap();

    assert!(outcome.rewritten.is_empty());
    assert_eq!(outcome.untouched, 1);
    assert_eq!(
        outcome.swept, 0,
        "an excision that removed nothing sweeps nothing"
    );
    assert_eq!(repository.snapshots().unwrap(), vec![snapshot]);
}

/// A root with no snapshots is a typo far more often than a surprise.
#[test]
fn a_root_with_no_snapshots_is_refused() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    sample_tree(&root);

    let repository = open_repository(dir.path());
    run_to_completion(&repository, &root, &state);

    let err = excise::excise(
        &repository,
        &dir.path().join("somewhere-else"),
        &selecting(&["/junk"]),
        &state,
        false,
        false,
        |_| {},
    )
    .unwrap_err();

    let message = err.to_string();

    assert!(message.contains("no snapshots of"), "{message}");
    assert!(
        message.contains(&root.display().to_string()),
        "it should say what the repository does hold: {message}"
    );
}

/// A suspended backup is corrected along with the snapshots: what it has
/// published is rewritten, what it still has queued loses the selected
/// paths, and it resumes to a finished snapshot without them — with no
/// need to re-store what it had already done.
#[test]
fn a_suspended_backup_is_corrected_along_with_its_snapshots() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    // Enough of it that a suspension lands partway through
    for outer in 0..10 {
        let sub = root.join(format!("dir-{outer}"));
        std::fs::create_dir_all(&sub).unwrap();

        for file in 0..20 {
            let bytes = deterministic_bytes(3000, (outer * 40 + file) as u64);
            std::fs::write(sub.join(format!("file-{file}.dat")), &bytes).unwrap();
        }
    }

    std::fs::create_dir_all(root.join("junk")).unwrap();
    for file in 0..6 {
        let bytes = deterministic_bytes(5000, 900 + file);
        std::fs::write(root.join(format!("junk/file-{file}.dat")), &bytes).unwrap();
    }

    let repository = open_repository(dir.path());

    let run = |suspend: bool| {
        backup::run(
            repository.clone(),
            backup::BackupOptions::new(&root, &state),
            crossbeam_channel::unbounded().0,
            Arc::new(AtomicBool::new(suspend)),
        )
        .unwrap()
    };

    let Outcome::Suspended { restorable } = run(true) else {
        panic!("expected a suspension");
    };

    let partial = restorable.expect("a suspension publishes what it has");
    assert!(state.exists());

    let outcome = excise::excise(
        &repository,
        &root,
        &selecting(&["/junk"]),
        &state,
        false,
        false,
        |_| {},
    )
    .unwrap();

    // Whatever the suspension had reached of the junk is gone from its
    // partial snapshot, and whatever it had not is gone from its queues
    assert!(
        outcome.unqueued > 0 || outcome.rewritten.len() == 1,
        "the junk was neither queued nor stored: {outcome:?}"
    );

    let published = repository.snapshots().unwrap();
    assert_eq!(published.len(), 1, "one snapshot, rewritten in place");
    assert_eq!(
        repository.load_snapshot(published[0]).unwrap().coverage,
        Coverage::Partial,
        "an unfinished backup's snapshot is still unfinished"
    );

    if !outcome.rewritten.is_empty() {
        assert_ne!(published[0], partial);
    }

    // The run resumes — it is not told to start again — and finishes
    // without any of the junk
    let Outcome::Completed(complete) = run(false) else {
        panic!("expected the resumed backup to finish");
    };

    let entries = manifest_map(&repository, complete);

    for (path, entry) in &entries {
        assert!(
            !path.starts_with("junk"),
            "the resumed run put {path:?} back"
        );

        if matches!(entry.kind, EntryKind::File { .. }) {
            assert_eq!(
                read_back(&repository, entry),
                std::fs::read(root.join(path)).unwrap(),
                "{path:?}"
            );
        }
    }

    // ...and it did finish: everything outside the junk is there
    assert_eq!(entries.len(), 10 + 200, "10 directories and their files");
    assert_eq!(repository.snapshots().unwrap(), vec![complete]);
}

/// The queue filtering itself: what a suspended run still has to do
/// loses exactly the selected paths, and nothing else.
#[test]
fn a_suspended_runs_queues_lose_the_selected_paths() {
    use beeping::pipeline::backup::{SuspendedRun, WorkingState};

    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    for outer in 0..10 {
        let sub = root.join(format!("dir-{outer}"));
        std::fs::create_dir_all(&sub).unwrap();

        for file in 0..20 {
            let bytes = deterministic_bytes(3000, (outer * 40 + file) as u64);
            std::fs::write(sub.join(format!("file-{file}.dat")), &bytes).unwrap();
        }
    }

    let repository = open_repository(dir.path());

    let outcome = backup::run(
        repository.clone(),
        backup::BackupOptions::new(&root, &state),
        crossbeam_channel::unbounded().0,
        Arc::new(AtomicBool::new(true)),
    )
    .unwrap();

    assert!(matches!(outcome, Outcome::Suspended { .. }));

    let opened = |state: &Path| match SuspendedRun::open(state, &root).unwrap() {
        WorkingState::Suspended(run) => run,
        other => panic!("expected a suspended run, not {}", suspended_kind(&other)),
    };

    let everything = selecting(&["**"]);
    let queued = {
        let run = opened(&state);
        let queued = run.count_paths(&everything).unwrap();

        assert!(queued > 0, "the suspension left nothing to do");

        // Selecting one directory takes only that directory's share
        let some = run.count_paths(&selecting(&["/dir-3"])).unwrap();
        assert!(some < queued, "selecting one directory selected all of it");

        // Counting is not doing
        assert_eq!(run.count_paths(&everything).unwrap(), queued);

        assert_eq!(run.drop_paths(&selecting(&["/dir-3"])).unwrap(), some);
        assert_eq!(run.count_paths(&everything).unwrap(), queued - some);

        queued - some
    };

    // ...and what it left behind is what the next session finds
    let run = opened(&state);
    assert_eq!(run.count_paths(&everything).unwrap(), queued);
    assert_eq!(run.count_paths(&selecting(&["/dir-3"])).unwrap(), 0);
}

fn suspended_kind(state: &beeping::pipeline::backup::WorkingState) -> &'static str {
    use beeping::pipeline::backup::WorkingState;

    match state {
        WorkingState::Missing => "no state at all",
        WorkingState::Suspended(_) => "a suspended run",
        WorkingState::Unsettled(_) => "an unsettled run",
    }
}

/// A backup that was killed still holds content that no snapshot names.
/// Sweeping around it would leave its eventual snapshot naming chunks
/// that are gone, so an excision refuses until the run is settled — and
/// this drives the real binary, because a killed run is the one state
/// that cannot be arranged politely.
#[test]
fn a_killed_backup_refuses_the_excision_until_it_is_settled() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");

    // Big enough that the run is still going when it is killed
    for outer in 0..12 {
        let sub = root.join(format!("dir-{outer:02}"));
        std::fs::create_dir_all(&sub).unwrap();

        for file in 0..25 {
            std::fs::write(
                sub.join(format!("file-{file:02}.dat")),
                deterministic_bytes(20_000, (outer * 100 + file) as u64),
            )
            .unwrap();
        }
    }

    std::fs::create_dir_all(root.join("junk")).unwrap();
    std::fs::write(root.join("junk/waste.dat"), b"never wanted").unwrap();

    beeping_command(dir.path())
        .args(["init", "--repository", "repo"])
        .status()
        .unwrap();

    let backup = [
        "backup",
        "tree",
        "--repository",
        "repo",
        "--state-dir",
        "state",
        "--plain",
    ];

    let mut child = beeping_command(dir.path()).args(backup).spawn().unwrap();

    // Kill it once it has claimed the repository, which is the state
    // that matters here: a lock object written and never given back
    let locks = dir.path().join("repo/locks");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);

    while std::fs::read_dir(&locks).map(|d| d.count()).unwrap_or(0) == 0 {
        assert!(
            std::time::Instant::now() < deadline,
            "the backup never claimed the repository"
        );
        std::thread::sleep(std::time::Duration::from_millis(50));
    }

    child.kill().unwrap();
    child.wait().unwrap();

    let excising = |extra: &[&str]| {
        let mut args = vec![
            "excise",
            "tree",
            "/junk",
            "--repository",
            "repo",
            "--state-dir",
            "state",
        ];
        args.extend_from_slice(extra);

        let output = beeping_command(dir.path()).args(args).output().unwrap();

        (
            output.status.success(),
            String::from_utf8_lossy(&output.stderr).into_owned(),
        )
    };

    // The lock stops it first...
    let (ok, message) = excising(&[]);
    assert!(!ok, "the excision should have been refused");
    assert!(message.contains("in use by a backup"), "{message}");

    // ...and breaking the lock does not get past the working state,
    // which is the thing that would actually be broken
    let (ok, message) = excising(&["--break-lock"]);
    assert!(!ok, "breaking the lock should not be enough");
    assert!(message.contains("without standing down"), "{message}");

    // Settling the run is what makes it safe, and then it goes through
    let finished = beeping_command(dir.path()).args(backup).status().unwrap();
    assert!(finished.success());

    let (ok, message) = excising(&[]);
    assert!(ok, "{message}");
}

fn beeping_command(dir: &Path) -> std::process::Command {
    let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_beeping"));

    command
        .current_dir(dir)
        .env("BEEPING_PASSWORD", "test-password")
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());

    command
}

/// A partial snapshot whose backup's working state is not here belongs
/// to a run that cannot be told what happened. Rewriting it would leave
/// that run unable to resume, so the excision says so instead.
#[test]
fn a_partial_snapshot_without_its_working_state_refuses_the_excision() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path().join("tree");
    let state = dir.path().join("state");

    sample_tree(&root);

    let repository = open_repository(dir.path());

    let Outcome::Suspended { restorable } = backup::run(
        repository.clone(),
        backup::BackupOptions::new(&root, &state),
        crossbeam_channel::unbounded().0,
        Arc::new(AtomicBool::new(true)),
    )
    .unwrap() else {
        panic!("expected a suspension");
    };

    let partial = restorable.expect("a suspension publishes what it has");

    // The state is there, but this excision is not looking at it
    let err = excise::excise(
        &repository,
        &root,
        &selecting(&["/junk"]),
        &dir.path().join("not-the-state-dir"),
        false,
        false,
        |_| {},
    )
    .unwrap_err();

    assert!(
        matches!(
            err,
            excise::Error::Refused(Refusal::StraySnapshot { snapshot }) if snapshot == partial
        ),
        "{err:?}"
    );

    assert_eq!(repository.snapshots().unwrap(), vec![partial]);
}