//! Taking paths out of the snapshots of one tree.
//!
//! Where a prune removes whole snapshots, this removes files and
//! directories from every snapshot of a given root, rewriting each one's
//! manifest without them and sweeping whatever chunks that leaves
//! unreferenced. It is how a repository gives back the space taken by
//! something that should never have been backed up — and, being
//! irreversible, it is a separate command from prune on purpose.
//!
//! What makes it safe to sweep afterwards is the rule the sweep has
//! always relied on, held to more strictly: every chunk that lives
//! without a snapshot naming it belongs to a run that has not published
//! its work, and this refuses to run against a working state in that
//! condition (see [`Refusal`]). What is left is a suspended run whose
//! every recorded byte is named by the partial snapshot it published,
//! which this rewrites along with the rest — and whose queues it filters
//! to match, so that resuming does not put back what has just been taken
//! away.

use std::path::{Path, PathBuf};

use globset::GlobSet;
use repository::{ChunkId, Coverage, Entry, EntryKind, Repository, Snapshot, SnapshotId};

use crate::{
    patterns::matches_with_ancestors,
    pipeline::backup::{SuspendedRun, Unsettled, WorkingState},
    sweep::{refuse_if_locked, sweep},
};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("{}", .0.explain())]
    Refused(Refusal),
    #[error(transparent)]
    Sweep(#[from] crate::sweep::Error),
    #[error(transparent)]
    State(#[from] crate::pipeline::Error),
    #[error(transparent)]
    Repository(#[from] repository::Error),
}

/// Why an excision would not go ahead.
///
/// Every one of these is a state in which sweeping could take away
/// something a backup still needs, or in which the snapshots do not add
/// up to what a working state believes. None is worth guessing past: an
/// excision cannot be undone, and each of these has a plain way out.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Refusal {
    /// A backup of this tree was killed rather than suspended, so it
    /// still holds work that no snapshot names.
    RunWasKilled { state_dir: PathBuf },
    /// A backup of this tree finished but has not cleared its working
    /// state away.
    RunNotTidiedUp { state_dir: PathBuf },
    /// The named state directory holds some other operation's state.
    ForeignState { state_dir: PathBuf },
    /// A partial snapshot of this tree belongs to a backup whose working
    /// state is not here to be corrected along with it.
    StraySnapshot { snapshot: SnapshotId },
}

impl Refusal {
    pub fn explain(&self) -> String {
        match self {
            Refusal::RunWasKilled { state_dir } => format!(
                "a backup of this tree was interrupted without standing down \
                 ({state_dir:?}), so it is holding content that no snapshot \
                 names and that this would sweep away. Run the backup again \
                 and let it finish or suspend, or remove that directory to \
                 abandon the run."
            ),
            Refusal::RunNotTidiedUp { state_dir } => format!(
                "a finished backup of this tree has not cleared its working \
                 state away ({state_dir:?}). Run the backup command once \
                 more, which is all it needs to tidy up."
            ),
            Refusal::ForeignState { state_dir } => {
                format!("{state_dir:?} holds the state of some other operation")
            }
            Refusal::StraySnapshot { snapshot } => format!(
                "{snapshot} is a partial snapshot, so an unfinished backup of \
                 this tree is relying on it, and its working state is not \
                 here to be corrected along with it. Point --state-dir at \
                 that backup's working state, finish the backup, or prune \
                 the snapshot to abandon it."
            ),
        }
    }
}

/// What an excision did, or would do.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Excised {
    /// The snapshots that held something the patterns selected.
    pub rewritten: Vec<Rewritten>,
    /// Snapshots of the tree that held nothing selected, and so were
    /// left exactly as they were.
    pub untouched: usize,
    /// Queued work dropped from a suspended backup of the tree.
    pub unqueued: u64,
    pub swept: usize,
    pub retained: usize,
}

/// One snapshot's share of an excision.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Rewritten {
    pub from: SnapshotId,
    /// What replaced it; `None` for a dry run, which writes nothing.
    pub to: Option<SnapshotId>,
    /// Entries taken out of it, and the content they held.
    pub entries: u64,
    pub content_bytes: u64,
}

/// Removes every path the patterns select from every snapshot of
/// `root`, and sweeps what that leaves behind.
///
/// Patterns are matched against paths relative to the snapshot root,
/// under the rules of [`crate::patterns`]: selecting a directory selects
/// everything beneath it. `state_dir` is where a suspended backup of
/// this tree keeps its working state, which is edited to match.
///
/// With `dry_run`, nothing is written and the counts describe what would
/// have happened — except the sweep, which cannot be known in advance:
/// what a rewritten manifest occupies is not decided until it is stored.
pub fn excise(
    repository: &Repository,
    root: &Path,
    select: &GlobSet,
    state_dir: &Path,
    dry_run: bool,
    break_lock: bool,
    mut progress: impl FnMut(&str),
) -> Result<Excised, Error> {
    refuse_if_locked(repository, break_lock)?;

    // The working state comes first, before even looking at what there
    // is to rewrite: opening it is what proves no run is alive in it,
    // and a tree whose backup was killed before it published anything
    // deserves that answer rather than "no snapshots of that tree".
    let mut suspended = match SuspendedRun::open(state_dir, root)? {
        WorkingState::Suspended(run) => Some(run),
        WorkingState::Missing => None,
        WorkingState::Unsettled(reason) => {
            let state_dir = state_dir.to_owned();

            return Err(Error::Refused(match reason {
                Unsettled::Killed | Unsettled::Unpublished => Refusal::RunWasKilled { state_dir },
                Unsettled::Finished => Refusal::RunNotTidiedUp { state_dir },
                Unsettled::Foreign => Refusal::ForeignState { state_dir },
            }));
        }
    };

    let (targets, mut keeping) = snapshots_of(repository, root)?;

    let published: Vec<SnapshotId> = suspended
        .as_ref()
        .map(|run| run.published().to_vec())
        .unwrap_or_default();

    for snapshot in &targets {
        // A partial snapshot is an unfinished backup's own record of its
        // work. Rewriting one whose run cannot be told about it leaves
        // that run unable to resume, quietly, and that is not something
        // to do without being asked.
        if snapshot.coverage == Coverage::Partial && !published.contains(&snapshot.id) {
            return Err(Error::Refused(Refusal::StraySnapshot {
                snapshot: snapshot.id,
            }));
        }
    }

    let mut rewritten = Vec::new();
    let mut untouched = 0;
    let mut replaced = Vec::new();

    for snapshot in &targets {
        progress(&format!("reading the manifest of {}", snapshot.id));

        let kept = rewrite(repository, snapshot, select, dry_run)?;

        if kept.removed == 0 {
            untouched += 1;
            keeping.push(snapshot.id);
            continue;
        }

        let to = match dry_run {
            true => None,
            false => Some(publish(repository, snapshot, &kept, suspended.as_mut())?),
        };

        if let Some(id) = to {
            keeping.push(id);
            replaced.push(snapshot.id);
        }

        rewritten.push(Rewritten {
            from: snapshot.id,
            to,
            entries: kept.removed,
            content_bytes: kept.removed_bytes,
        });
    }

    let unqueued = match (&suspended, dry_run) {
        (Some(run), true) => run.count_paths(select)?,
        (Some(run), false) => run.drop_paths(select)?,
        (None, _) => 0,
    };

    if dry_run || replaced.is_empty() {
        return Ok(Excised {
            rewritten,
            untouched,
            unqueued,
            swept: 0,
            retained: 0,
        });
    }

    let counts = sweep(repository, &keeping, replaced, &mut progress)?;

    // Now that the sweep has spared everything the run still needs, its
    // record of what the repository had lost can be brought up to date.
    // Without this it would put back every file it had recorded, having
    // no way to know that what changed underneath it was its own doing.
    if let Some(run) = &mut suspended {
        run.observed(repository.epoch()?)?;
    }

    Ok(Excised {
        rewritten,
        untouched,
        unqueued,
        swept: counts.swept,
        retained: counts.retained,
    })
}

/// The snapshots of `root`, oldest first, and the ids of all the others
/// — which survive untouched, and whose chunks are no less live for it.
fn snapshots_of(
    repository: &Repository,
    root: &Path,
) -> Result<(Vec<Snapshot>, Vec<SnapshotId>), Error> {
    let mut targets = Vec::new();
    let mut others = Vec::new();
    let mut roots: Vec<PathBuf> = Vec::new();

    for id in repository.snapshots()? {
        let snapshot = repository.load_snapshot(id)?;

        if snapshot.root == root {
            targets.push(snapshot);
            continue;
        }

        if !roots.contains(&snapshot.root) {
            roots.push(snapshot.root.clone());
        }

        others.push(id);
    }

    if targets.is_empty() {
        roots.sort();

        return Err(crate::sweep::Error::UnknownRoot {
            requested: root.to_owned(),
            known: roots,
        }
        .into());
    }

    targets.sort_by_key(|snapshot| snapshot.created);

    Ok((targets, others))
}

/// What one snapshot's manifest looks like with the selected paths gone.
struct Kept {
    chunks: Vec<ChunkId>,
    entries: u64,
    content_bytes: u64,
    removed: u64,
    removed_bytes: u64,
}

/// Reads a manifest, storing a new one without the selected paths.
///
/// The entries stream straight from the old manifest into the new, so a
/// manifest of any size costs its own read and no more memory than one
/// entry holds. A dry run reads the same stream and stores nothing.
fn rewrite(
    repository: &Repository,
    snapshot: &Snapshot,
    select: &GlobSet,
    dry_run: bool,
) -> Result<Kept, Error> {
    let mut kept = Kept {
        chunks: Vec::new(),
        entries: 0,
        content_bytes: 0,
        removed: 0,
        removed_bytes: 0,
    };
    let mut failure = None;

    {
        let entries = repository
            .manifest_entries(snapshot.manifest.clone())
            .map_while(|entry| match entry {
                Ok(entry) => Some(entry),
                Err(err) => {
                    failure = Some(err);
                    None
                }
            })
            .filter(|entry| {
                let bytes = content_bytes(entry);

                if matches_with_ancestors(select, &entry.path) {
                    kept.removed += 1;
                    kept.removed_bytes += bytes;

                    return false;
                }

                kept.entries += 1;
                kept.content_bytes += bytes;

                true
            });

        match dry_run {
            true => entries.for_each(drop),
            false => kept.chunks = repository.store_manifest(entries)?,
        }
    }

    // A manifest that could not be read through is no basis for
    // replacing anything, and the chunks stored above are simply
    // unreferenced — the next sweep collects them.
    if let Some(err) = failure {
        return Err(err.into());
    }

    Ok(kept)
}

/// Stores a snapshot's replacement and takes the original away.
///
/// In that order, always: the two are separate objects, so there is no
/// moment when this backup has no snapshot at all. A crash between them
/// leaves the original in place with its replacement beside it, which
/// the next excision resolves — it derives the same replacement again,
/// finds it already stored, and goes on to the deletion.
fn publish(
    repository: &Repository,
    snapshot: &Snapshot,
    kept: &Kept,
    suspended: Option<&mut SuspendedRun>,
) -> Result<SnapshotId, Error> {
    let id = repository.rewritten_snapshot_id(snapshot, &kept.chunks);

    repository.store_snapshot(&snapshot.rewritten(
        id,
        kept.chunks.clone(),
        kept.entries,
        kept.content_bytes,
    ))?;

    // A suspended run's own record of what it published goes next, so
    // that the deletion below never takes away a snapshot it is still
    // pointing at
    if let Some(run) = suspended
        && run.published().contains(&snapshot.id)
    {
        run.republished(id, kept.chunks.clone(), kept.entries, kept.content_bytes)?;
    }

    repository.delete_snapshot(snapshot.id)?;

    Ok(id)
}

fn content_bytes(entry: &Entry) -> u64 {
    match entry.kind {
        EntryKind::File { len, .. } => len,
        EntryKind::Directory | EntryKind::Symlink { .. } => 0,
    }
}