gunnar-server 3.0.0

//! **What emptying an account actually means, in one place.**
//!
//! [`crate::purge_queue`] decides *when* a purge runs and guarantees the
//! lockout comes off; this module is the walk itself, and it is the only
//! implementation of it (one implementation) — the gRPC verb, the CLI verb and monetize all
//! arrive here.
//!
//! # The four classes of bytes an account holds
//!
//! The per-repository reclaim was never the hard part. **Finding everything
//! was**, and it is why a purge that looped over `repo list` would report
//! success with the account's disk still full:
//!
//! | class | which listing shows it | which one does not |
//! |---|---|---|
//! | live repository | `Repos.List` | — |
//! | soft-delete tombstone | `Repos.ListDeleted` | `List` (the row is gone) |
//! | orphan store | `Repos.ListOrphans` | both of the above |
//! | nested namespace | `Repos.ListNamespaces` | — |
//!
//! An **orphan** is the one that bites. It has no catalog row and no
//! tombstone, so before 2026-08-30 nothing on the box could see it: on the
//! appliance 2026-08-30 that was ~25-39 GB standing at `alice/linux` that
//! `list` could not show, `list-deleted` could not show, `delete` answered
//! `NoSuchRepo` for, and `create` refused the name of. A purge that skipped
//! that class would leave the single largest thing in the account behind and
//! call the account empty.
//!
//! Together those four are exhaustive over `repos/<account>/`, and that is not
//! a claim about this module's cleverness — it is `Repos.ListOrphans`'s own
//! guarantee: "every `<name>.git` directory under the repository root is in
//! exactly one of the three listings". The purge inherits it by walking all
//! three, and `tests/account_purge.rs` puts one of each on a disk and measures
//! what is left.
//!
//! # Nested namespaces, and the asymmetry that is worth knowing about
//!
//! `alice/team` is a **different** namespace from `alice`: `Scope::Namespace`
//! matches exactly, and `AuthTable::owns_account("alice", "alice/team")` is a
//! string equality that answers false. So alice does not own `alice/team` in
//! the ACL's sense.
//!
//! On the disk it is `repos/alice/team/…`, inside her account's directory, and
//! it goes with the account. That asymmetry is deliberate and it points the
//! safe way: a purge that left it behind would leave bytes on the disk, on the
//! bill, and under a name a later `Repos.Create` would refuse — after an
//! operator had been told the account was emptied. `gunnar_core::gate` closes
//! the matching hole from the other end, by walking ancestors so a gate on
//! `alice` also shuts `alice/team` while this runs.
//!
//! # Why the walk keeps going after a failure
//!
//! One unreadable directory in an account of two hundred must not leave the
//! other hundred and ninety-nine on the disk — the operator would have to run
//! the purge again anyway, and the second run would trip on the same directory.
//! So each failure is recorded by name and the walk continues; the purge as a
//! whole then FAILS, so nothing upstream records the account as finished, and
//! [`PurgeReport::failures`] is the list of what is still there.

use anyhow::{Context, Result};
use gunnar_core::RepoRef;

use crate::stores::ServerStores;

/// What `Accounts.Purge` was asked for.
#[derive(Debug, Clone)]
pub struct PurgeRequest {
    /// The account: a principal id, which is also its namespace.
    pub account: String,
    /// The principal that asked, for the audit trail.
    pub principal_id: String,
    /// Why. Carried into the log line for every store removed, because a purge
    /// is the one operation whose consequence cannot be read back off the box
    /// afterwards — the rows it removes are the rows that would have explained
    /// it.
    pub reason: String,
    /// Also remove the principal: its keys, grants, tokens and memberships.
    pub remove_principal: bool,
    /// Refuse, destroying nothing, unless the account holds no live
    /// repositories.
    pub expect_empty: bool,
}

/// What a purge did.
#[derive(Debug, Clone, Default)]
pub struct PurgeReport {
    /// Stores the walk found across all four classes.
    pub stores_found: u32,
    /// How many are gone.
    pub stores_purged: u32,
    /// Bytes that left the filesystem, measured as `du` would.
    pub bytes_reclaimed: u64,
    /// Whether the principal itself was removed.
    pub principal_removed: bool,
    /// One line per store still on the disk, naming it and why. **The recovery
    /// list.**
    pub failures: Vec<String>,
    /// **Frames erased from the transaction log** — the purged repositories'
    /// packs, refs and LFS objects, rewritten in place as `Redacted` (owner
    /// ruling g-nonpay). The twin erases the same frames when it stores the
    /// `Erase` frame this purge appended.
    pub log_frames_erased: u64,
    /// Their blob bytes, gone from the segment files.
    pub log_bytes_erased: u64,
}

/// Progress a caller wants to see while the walk runs.
///
/// A purge of a large account runs for minutes with a console polling the whole
/// time, so it reports as it goes rather than returning a summary that can only
/// be read once nobody needs it. The queue's handler re-stamps the gate's lease
/// on [`Step::Reclaimed`], which is what keeps a long purge from unlocking
/// underneath itself.
#[derive(Debug, Clone)]
pub enum Step {
    /// The walk finished counting: this many stores, before any were removed.
    Found(u32),
    /// One store is gone, and this is what it was holding.
    Reclaimed { store: String, bytes: u64 },
    /// One store could not be removed, and this is why.
    Failed(String),
    /// The principal was removed.
    PrincipalRemoved,
}

/// **Empty an account.** The one implementation.
///
/// Returns `Err` only when the purge could not be *attempted* — an
/// `expect_empty` refusal, or a listing that could not be read. A walk that ran
/// and hit unremovable stores returns `Ok` with [`PurgeReport::failures`]
/// populated, and it is the caller that turns a non-empty `failures` into a
/// FAILED job. The split is deliberate: a refusal means nothing was destroyed,
/// and a partial means some of it was, and a caller that could not tell those
/// apart would either retry a refusal for ever or report a half-emptied account
/// as untouched.
pub fn purge_account(
    stores: &ServerStores,
    request: PurgeRequest,
    report: &mut dyn FnMut(Step),
) -> Result<PurgeReport> {
    let account = request.account.trim().to_owned();
    anyhow::ensure!(
        !account.is_empty(),
        "an account purge needs an account name"
    );

    // ── the census, all four classes, before anything is removed ───────────
    //
    // Counted first and in full so `stores_found` is a number the walk can be
    // measured against. A count that grew as the walk discovered things would
    // make "purged 5 of 5" true of a purge that missed half the account.
    let namespaces = account_namespaces(stores, &account)?;
    let mut targets: Vec<Target> = Vec::new();

    for namespace in &namespaces {
        for repo in stores
            .catalog()
            .list(Some(namespace))
            .with_context(|| format!("listing the repositories in {namespace}"))?
        {
            targets.push(Target {
                path: repo.path(),
                kind: Kind::Live,
            });
        }
    }
    for deleted in stores
        .tombstones()
        .context("listing the soft-deleted repositories")?
    {
        if covers(&namespaces, deleted.path.namespace()) {
            targets.push(Target {
                path: deleted.path,
                kind: Kind::Tombstoned,
            });
        }
    }
    for orphan in stores.orphans().context("listing the orphan stores")? {
        if covers(&namespaces, orphan.path.namespace()) {
            targets.push(Target {
                path: orphan.path,
                kind: Kind::Orphan,
            });
        }
    }
    // A directory can be BOTH catalogued and tombstoned only in the broken
    // states the reclaim rules describe, and the three listings are otherwise disjoint —
    // but a purge that removed a directory twice would count its bytes twice
    // and report a reclaim that never happened. Deduplicated by name, cheaply,
    // rather than trusted.
    targets.sort_by_key(|a| a.path.key());
    targets.dedup_by(|a, b| a.path.key() == b.path.key());

    if request.expect_empty {
        let live: Vec<String> = targets
            .iter()
            .filter(|t| t.kind == Kind::Live)
            .map(|t| t.path.to_string())
            .collect();
        anyhow::ensure!(
            live.is_empty(),
            "{account} still holds {} live repositor{}: {}. `expect_empty` refuses rather than \
             destroys, so nothing was touched; run the purge without it to reclaim them.",
            live.len(),
            if live.len() == 1 { "y" } else { "ies" },
            live.join(", ")
        );
    }

    let mut out = PurgeReport {
        stores_found: targets.len() as u32,
        ..PurgeReport::default()
    };
    report(Step::Found(out.stores_found));

    // ── the walk ──────────────────────────────────────────────────────────
    let mut purged_keys: Vec<String> = Vec::new();
    for target in targets {
        let name = target.path.to_string();
        // The catalog row first, so a store that is reclaimed but whose row
        // survives cannot be served afterwards. `delete` on a name with no row
        // — a tombstone or an orphan — is expected and is not a failure.
        if target.kind == Kind::Live {
            if let Err(err) = stores.catalog().delete(&target.path) {
                out.failures.push(format!(
                    "{name}: the catalog row could not be removed: {err}"
                ));
                report(Step::Failed(
                    out.failures.last().cloned().unwrap_or_default(),
                ));
                continue;
            }
        }
        match stores.purge_repo(&target.path) {
            Ok((before, after)) => {
                let bytes = before.saturating_sub(after);
                purged_keys.push(target.path.key());
                out.stores_purged = out.stores_purged.saturating_add(1);
                out.bytes_reclaimed = out.bytes_reclaimed.saturating_add(bytes);
                tracing::info!(
                    target: "gunnar::store",
                    event = "account.store_purged",
                    account = %account,
                    store = %name,
                    kind = target.kind.as_str(),
                    bytes,
                    principal = %request.principal_id,
                    reason = %request.reason,
                    "a store was reclaimed by an account purge"
                );
                report(Step::Reclaimed { store: name, bytes });
            }
            Err(err) => {
                // `purge_repo` leaves a ReclaimFailed tombstone of its own, so
                // what is left here is visible to `Repos.ListOrphans` and is
                // not lost when this report is pruned.
                let note = format!("{name}: {err:#}");
                out.failures.push(note.clone());
                report(Step::Failed(note));
            }
        }
    }

    // ── the log ───────────────────────────────────────────────────────────
    //
    // The stores are gone from the disk, and every pack and LFS object they ever
    // took is still in the transaction log: on this box until retention trims
    // it, and on the twin for ever, because the twin's log IS the copy (owner
    // ruling g-nonpay 2026-09-29: "deleted on main AND twin"). So an `Erase`
    // frame goes into the log — the twin erases its own frames when it stores
    // it, a restore when it replays it — and this box erases its own now. Only
    // the stores that WERE purged: a store the walk could not remove still has
    // its data on the disk, and a restore must be able to rebuild it.
    if !purged_keys.is_empty() && !stores.txlog().is_follower() {
        match erase_from_log(stores, &account, &purged_keys, &request.reason) {
            Ok(erased) => {
                out.log_frames_erased = erased.frames;
                out.log_bytes_erased = erased.bytes;
                tracing::info!(
                    target: "gunnar::store",
                    event = "account.log_erased",
                    account = %account,
                    frames = erased.frames,
                    bytes = erased.bytes,
                    principal = %request.principal_id,
                    reason = %request.reason,
                    "an account purge erased the account's frames from the transaction log"
                );
            }
            Err(err) => out.failures.push(format!(
                "the account's frames could not be erased from the transaction log: {err:#}"
            )),
        }
    }

    // ── the namespace rows, deepest first ─────────────────────────────────
    //
    // Deepest first because `delete_namespace` refuses a namespace that still
    // holds repositories, and `alice` holding `alice/team`'s rows would make
    // the outer delete fail on an account that is otherwise empty. Sorting by
    // length descending is enough: a nested name is strictly longer than the
    // one it nests under.
    let mut ordered = namespaces.clone();
    ordered.sort_by_key(|n| std::cmp::Reverse(n.len()));
    for namespace in ordered {
        match stores.catalog().delete_namespace(&namespace, false) {
            Ok(_) => {}
            // Nothing there is the normal case for an account whose namespace
            // row never existed, and it is not a failure of the purge.
            Err(gunnar_core::Error::NoSuchNamespace(_)) => {}
            Err(err) => out.failures.push(format!(
                "the namespace {namespace} could not be removed: {err}"
            )),
        }
    }

    // ── the principal, last ───────────────────────────────────────────────
    //
    // Last, and only if asked. Last because it takes the account's gate with
    // it (`AuthTable::remove_principal`), and a gate removed while stores were
    // still being reclaimed would unlock an account mid-walk — the exact race
    // the lockout exists to prevent. By the time this runs there is nothing
    // left to protect.
    if request.remove_principal {
        match stores.auth().remove_principal(&account) {
            Ok(_) => {
                out.principal_removed = true;
                tracing::info!(
                    target: "gunnar::audit",
                    event = "account.principal_removed",
                    account = %account,
                    principal = %request.principal_id,
                    reason = %request.reason,
                    "an account purge removed the principal, its keys, grants and memberships"
                );
                report(Step::PrincipalRemoved);
            }
            // A principal that is not there is the outcome that was asked for.
            Err(gunnar_core::Error::NoSuchPrincipal(_)) => out.principal_removed = true,
            Err(err) => out.failures.push(format!(
                "the principal {account} could not be removed: {err}"
            )),
        }
    }

    Ok(out)
}

/// Frame the erase, then erase this box's own frames of `repos` before it.
fn erase_from_log(
    stores: &ServerStores,
    account: &str,
    repos: &[String],
    reason: &str,
) -> Result<gunnar_core::txlog::Erased> {
    use gunnar_core::txlog::{Kind, Meta};
    let log = stores.txlog();
    let seq = {
        let _framed = log.mutation();
        log.append(
            Kind::Erase,
            &Meta::Erase {
                account: account.to_owned(),
                repos: repos.to_vec(),
                reason: reason.to_owned(),
            },
            &[],
        )
        .context("framing the erase")?
    };
    log.erase(seq, repos)
        .context("erasing the account's frames")
}

/// What a target is, for the log line and for `expect_empty`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Kind {
    Live,
    Tombstoned,
    Orphan,
}

impl Kind {
    fn as_str(self) -> &'static str {
        match self {
            Kind::Live => "live",
            Kind::Tombstoned => "tombstoned",
            Kind::Orphan => "orphan",
        }
    }
}

struct Target {
    path: RepoRef,
    kind: Kind,
}

/// **Every namespace that belongs to `account`**: the account itself, and every
/// namespace nested under it.
///
/// The account's own name is always in the list even when no namespace row
/// exists for it — an orphan store can stand under a name the catalog never
/// knew, which is what an orphan IS, and a purge that keyed off the namespace
/// rows would skip exactly the class it most needs to reach.
fn account_namespaces(stores: &ServerStores, account: &str) -> Result<Vec<String>> {
    let mut out = vec![account.to_owned()];
    let prefix = format!("{account}/");
    for namespace in stores
        .catalog()
        .namespaces()
        .context("listing the namespaces")?
    {
        if namespace.name.starts_with(&prefix) {
            out.push(namespace.name);
        }
    }
    out.sort();
    out.dedup();
    Ok(out)
}

/// Is `namespace` one of the account's, or nested under one of them?
///
/// On PATH COMPONENTS. `alicehacks` starts with `alice` and is a different
/// account; a purge that caught it by string prefix would take a stranger's
/// repositories, which is the one mistake this whole module must not make.
fn covers(namespaces: &[String], namespace: &str) -> bool {
    namespaces
        .iter()
        .any(|owned| namespace == owned || namespace.starts_with(&format!("{owned}/")))
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The prefix rule, on its own, because it is the difference between
    /// emptying one account and emptying two.
    #[test]
    fn covers_matches_path_components_and_not_characters() {
        let owned = vec!["alice".to_owned()];
        assert!(covers(&owned, "alice"));
        assert!(covers(&owned, "alice/team"));
        assert!(covers(&owned, "alice/team/deep"));
        assert!(
            !covers(&owned, "alicehacks"),
            "`alicehacks` is a different account and a purge of `alice` must not reach it"
        );
        assert!(!covers(&owned, "bob"));
        assert!(
            !covers(&owned, "notalice"),
            "a suffix match would be worse still"
        );
    }
}