//! What the operations that remove things from a repository have in
//! common: refusing to run under a backup's lock, and collecting the
//! chunks nothing names any more.
//!
//! Removing a snapshot ([`crate::prune`]) and removing paths from the
//! snapshots of one tree ([`crate::excise`]) are different decisions
//! about what should survive. Both end here, with the same three rules
//! keeping them honest.
//!
//! **Locks.** A running backup holds chunks that no snapshot names yet —
//! it has uploaded them, and will only record them when it finishes or
//! suspends. Sweeping those away would leave the run's eventual snapshot
//! full of dangling references, so this refuses while any lock is held.
//! A run that was killed leaves its lock behind on purpose: it still has
//! entries queued that no snapshot names.
//!
//! **Order.** The snapshot records the caller is replacing or removing
//! go first, then the prune record, and only then are chunks swept.
//! Interrupted at any point, what is left is a repository with
//! unreferenced chunks in it — which is untidy and entirely safe, and
//! which the next sweep collects.
//!
//! **The prune record.** Written before a single chunk is deleted, it is
//! what tells every local chunk cache, on this machine and any other,
//! that what it remembers may no longer be true.

use std::{collections::HashSet, path::PathBuf};

use repository::{ChunkId, EntryKind, Lock, PruneRecord, Repository, SnapshotId};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("{}", describe_locks(.held))]
    Locked { held: Vec<Lock> },
    #[error("{0} is not a snapshot in this repository")]
    UnknownSnapshot(SnapshotId),
    #[error("{}", describe_unknown_root(.requested, .known))]
    UnknownRoot {
        requested: PathBuf,
        known: Vec<PathBuf>,
    },
    #[error(transparent)]
    Repository(#[from] repository::Error),
}

/// The refusal a held lock earns, listing who holds it and what to do.
fn describe_locks(held: &[Lock]) -> String {
    let mut message = String::from(
        "the repository is in use by a backup that has stored chunks no \
         snapshot names yet; removing anything now could sweep them away:",
    );

    for lock in held {
        let taken = std::time::UNIX_EPOCH + std::time::Duration::from_secs(lock.taken);

        message.push_str(&format!(
            "\n  {} on {} (process {}) since {}",
            lock.root.display(),
            lock.host,
            lock.process,
            humantime::format_rfc3339_seconds(taken)
        ));
    }

    message.push_str(
        "\n\nFinish or suspend that run and it will release its lock. If it is \
         gone for good, --break-lock goes ahead anyway; the run it belonged to \
         will notice and redo whatever it had recorded.",
    );

    message
}

/// A named root with no snapshots under it is a typo far more often
/// than it is a surprise, so the answer says what the repository does
/// hold.
fn describe_unknown_root(requested: &std::path::Path, known: &[PathBuf]) -> String {
    let mut message = format!(
        "this repository holds no snapshots of {}",
        requested.display()
    );

    if known.is_empty() {
        message.push_str("; it holds no snapshots at all");

        return message;
    }

    message.push_str(". It holds snapshots of:");

    for root in known {
        message.push_str(&format!("\n  {}", root.display()));
    }

    message
}

/// Stops an operation that would sweep chunks out from under a running
/// backup.
pub(crate) fn refuse_if_locked(repository: &Repository, break_lock: bool) -> Result<(), Error> {
    let held = repository.locks()?;

    if !held.is_empty() && !break_lock {
        return Err(Error::Locked { held });
    }

    Ok(())
}

/// What a sweep collected.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Swept {
    pub swept: usize,
    pub retained: usize,
}

/// Deletes every chunk that nothing surviving names, recording that it
/// happened.
///
/// `keeping` must name every snapshot that is to survive, including the
/// ones the operation never looked at, whose chunks are just as live as
/// anyone's. Nothing else may be live: a run holding chunks no snapshot
/// names holds the repository's lock, which this refuses to run under,
/// and an excision refuses to touch a working state in that condition
/// rather than sweeping what it is still relying on. `removed` is what
/// the operation took out of the repository, for the record it leaves
/// behind.
///
/// The caller's own deletions must already have happened: this writes
/// the prune record and then sweeps, and the record is what stops a
/// local chunk cache from going on vouching for what is about to go.
pub(crate) fn sweep(
    repository: &Repository,
    keeping: &[SnapshotId],
    removed: Vec<SnapshotId>,
    progress: &mut dyn FnMut(&str),
) -> Result<Swept, Error> {
    let swept = mark(repository, keeping, progress)?;

    // The prune record goes before any sweeping, so that a sweep
    // interrupted halfway still invalidates every cache that might
    // otherwise vouch for a chunk it has already deleted
    let record = PruneRecord::new(removed, hostname());
    repository.record_prune(&record)?;

    progress(&format!(
        "sweeping {} unreferenced chunk(s)",
        swept.unreferenced.len()
    ));

    for id in &swept.unreferenced {
        repository.delete_chunk(*id)?;
    }

    // Finally the local cache, which can keep everything the sweep did
    // not touch rather than being thrown away wholesale
    if let Some(cache) = repository.chunk_cache() {
        let gone: HashSet<ChunkId> = swept.unreferenced.iter().copied().collect();
        cache.forget(&gone, repository.epoch()?)?;
    }

    Ok(swept.counts())
}

/// What a sweep would collect, without collecting it.
pub(crate) struct Marked {
    unreferenced: Vec<ChunkId>,
    stored: usize,
}

impl Marked {
    pub(crate) fn counts(&self) -> Swept {
        Swept {
            swept: self.unreferenced.len(),
            retained: self.stored - self.unreferenced.len(),
        }
    }
}

/// Reads what the surviving snapshots name, and works out what that
/// leaves unreferenced.
pub(crate) fn mark(
    repository: &Repository,
    keeping: &[SnapshotId],
    progress: &mut dyn FnMut(&str),
) -> Result<Marked, Error> {
    progress(&format!(
        "reading {} remaining snapshot(s) to see what is still needed",
        keeping.len()
    ));

    let mut referenced: HashSet<ChunkId> = HashSet::new();

    for id in keeping {
        let snapshot = repository.load_snapshot(*id)?;
        referenced.extend(snapshot.manifest.iter().copied());

        for entry in repository.manifest_entries(snapshot.manifest.clone()) {
            if let EntryKind::File { chunks, .. } = entry?.kind {
                referenced.extend(chunks);
            }
        }
    }

    progress(&format!(
        "{} chunk(s) are still referenced; listing the repository",
        referenced.len()
    ));

    let stored = repository.chunks()?;
    let unreferenced = stored
        .iter()
        .copied()
        .filter(|id| !referenced.contains(id))
        .collect();

    Ok(Marked {
        unreferenced,
        stored: stored.len(),
    })
}

pub(crate) fn hostname() -> String {
    std::fs::read_to_string("/etc/hostname")
        .map(|name| name.trim().to_string())
        .ok()
        .filter(|name| !name.is_empty())
        .unwrap_or_else(|| "unknown".to_string())
}