gunnar-core 3.0.0

//! The repository catalog: namespaces and repositories, keyed by
//! `<namespace>/<repo>`.
//!
//! Namespaces are first-class from day one (multi-tenancy): every path a client
//! sends has one, every grant can scope to one, and a repository cannot exist
//! outside one. Adding that later would have meant migrating every stored path
//! and every ACL row.
//!
//! The catalog stores **metadata only**. Creating a repository here does not
//! create its object or ref store, and deleting it does not delete them. That
//! belongs to whoever owns the backend (the storage seam), which on an appliance is one
//! `.znippy` per repository and on a normal OS is a directory. Keeping the two
//! apart is what lets `gunnar gc` and a future `repack()` run without the
//! catalog knowing anything about storage layout.

use std::sync::Arc;

use gunnar_store::git_storage_trait::GitServe;

use crate::control::ControlStore;
use crate::doc::{impl_doc, DocRef, ShardedMap};
use crate::error::{Error, Result};
use crate::model::{check_branch, HashKind, Namespace, Repo, Visibility, DEFAULT_BRANCH};
use crate::sandbox::{check_component, RepoRef};

/// **The name the catalog document is stored under** in the control store.
/// See [`crate::AUTH_REF`] for why it still looks like a reference name.
pub const CATALOG_REF: &str = "refs/gunnar/catalog";

/// The catalog as it is stored.
///
/// Both maps are sharded (`doc.rs`). Inline, `repos` held every repository on
/// the server in one blob, so `git repo create` re-serialised and re-hashed the
/// whole catalog and every SSH connection parsed it to answer one key probe.
/// Sharded, a create rewrites one leaf and the two nodes above it, and
/// a lookup reads one leaf (none at all, once it is cached).
#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct CatalogDoc {
    /// Keyed by name so the encoding is deterministic.
    #[serde(default)]
    namespaces: ShardedMap<Namespace>,
    /// Keyed by `<namespace>/<repo>`.
    #[serde(default)]
    repos: ShardedMap<Repo>,
}

impl_doc!(CatalogDoc { namespaces, repos });

/// What a new repository is created with.
///
/// Server defaults are applied here rather than by the caller, so the gRPC
/// service, the CLI and a test all get the same repository (one implementation).
#[derive(Clone, Debug)]
pub struct NewRepo {
    /// Where it lives.
    pub path: RepoRef,
    /// sha1 or sha256. Defaults to sha256.
    pub hash_kind: HashKind,
    /// Full ref name. `None` means [`DEFAULT_BRANCH`].
    pub default_branch: Option<String>,
    /// Defaults to [`Visibility::Private`].
    pub visibility: Visibility,
    /// Free text, for UIs.
    pub description: String,
}

impl NewRepo {
    /// A private sha256 repository at `path`, with the default branch.
    pub fn new(path: RepoRef) -> Self {
        NewRepo {
            path,
            hash_kind: HashKind::Sha256,
            default_branch: None,
            visibility: Visibility::Private,
            description: String::new(),
        }
    }

    /// Set the hash kind. sha1 repositories are fully supported.
    pub fn hash_kind(mut self, kind: HashKind) -> Self {
        self.hash_kind = kind;
        self
    }

    /// Set the visibility.
    pub fn visibility(mut self, visibility: Visibility) -> Self {
        self.visibility = visibility;
        self
    }

    /// Set the default branch, as a full `refs/heads/...` name.
    pub fn default_branch(mut self, name: impl Into<String>) -> Self {
        self.default_branch = Some(name.into());
        self
    }

    /// The [`Repo`] this spec would create, **without creating it**.
    ///
    /// The one place a `NewRepo` becomes a `Repo`: [`RepoCatalog::create`]
    /// writes exactly this value, and the first-repository door carries it
    /// through a push's advertisement *before* the catalog has been touched, so
    /// a client that takes one look at the advertised hash and hangs up (a sha1
    /// working tree against the sha256 default) leaves nothing behind. A
    /// door that wrote the catalog first was the 2026-08-30 ghost-repository
    /// bug: `repo get` succeeded for a repository whose push was refused.
    pub fn prospect(&self) -> Result<Repo> {
        let branch = match &self.default_branch {
            Some(name) => check_branch(name)?,
            None => DEFAULT_BRANCH.to_owned(),
        };
        Ok(Repo {
            namespace: self.path.namespace().to_owned(),
            name: self.path.name().to_owned(),
            hash_kind: self.hash_kind,
            default_branch: branch,
            visibility: self.visibility,
            created: crate::model::now_ms(),
            description: self.description.clone(),
            encrypted: false,
        })
    }
}

/// Create, open, list and delete repositories.
///
/// Cheap to clone; every clone reads and writes the same document.
#[derive(Clone, Debug)]
pub struct RepoCatalog {
    doc: DocRef<CatalogDoc>,
}

impl RepoCatalog {
    /// Open the catalog held in `store` (the certificate-identity plan).
    ///
    /// This is the *control* store: the server's own, not a repository's.
    pub fn new(store: Arc<dyn ControlStore>) -> Result<Self> {
        Ok(RepoCatalog {
            doc: DocRef::new(store, CATALOG_REF)?,
        })
    }

    /// The name the catalog is stored under.
    pub fn ref_name(&self) -> &str {
        self.doc.ref_name()
    }

    /// Every version this catalog has ever had, oldest first (the certificate-identity plan). See
    /// [`crate::AuthTable::history`].
    pub fn history(&self) -> Result<Vec<crate::control::DocVersion>> {
        self.doc.history()
    }

    // ── namespaces ─────────────────────────────────────────────────────────

    /// Create a namespace.
    ///
    /// Nested names (`team/sub`) are allowed and every level is validated by
    /// the sandbox's component rules, so a namespace can never be a path
    /// escape.
    pub fn create_namespace(&self, name: &str) -> Result<Namespace> {
        for level in name.split('/') {
            check_component(name, level)?;
        }
        if name.split('/').count() >= crate::sandbox::MAX_COMPONENTS {
            // One component has to be left for the repository name.
            return Err(Error::InvalidPath {
                input: name.escape_debug().to_string(),
                reason: crate::sandbox::PathReject::TooManyComponents,
            });
        }
        let namespace = Namespace {
            name: name.to_owned(),
            // A new account takes the server's default until an admin sets a
            // plan. `None` and not a copy of the default: copying it would
            // freeze today's default into every namespace ever created, and a
            // later change to the appliance's disk would not reach them.
            explode_budget_bytes: None,
            // Private, and its events too, until its account admin says
            // otherwise (`RepoCatalog::set_namespace_public`,
            // `RepoCatalog::set_namespace_events_public`).
            public: false,
            events_public: false,
        };
        self.doc.update(|doc| {
            if doc.namespaces.contains_key(name)? {
                return Err(Error::NamespaceExists(name.to_owned()));
            }
            doc.namespaces.insert(name.to_owned(), namespace.clone())?;
            Ok(namespace.clone())
        })
    }

    /// **Set (or clear) an account's exploded-cache quota.**
    ///
    /// `None` puts the namespace back on the server's default; `Some(0)` is
    /// unlimited and is a decision rather than an absence. The quota reaches a
    /// repository the next time its store is opened, and every store already
    /// open under this namespace immediately — see
    /// `ServerStores::apply_explode_quota`.
    ///
    /// It cannot fail a read and it cannot lose an object: the cache it bounds
    /// is re-derivable from the packs, which it does not bound. Lowering a
    /// quota below what a repository already holds evicts; it never refuses.
    pub fn set_namespace_explode_budget(
        &self,
        name: &str,
        bytes: Option<u64>,
    ) -> Result<Namespace> {
        self.doc.update(|doc| {
            let Some(mut ns) = doc.namespaces.get(name)? else {
                return Err(Error::NoSuchNamespace(name.to_owned()));
            };
            ns.explode_budget_bytes = bytes;
            doc.namespaces.insert(name.to_owned(), ns.clone())?;
            Ok(ns)
        })
    }

    /// This namespace's quota, or `None` for "the server's default". A missing
    /// namespace is `None` too: a repository whose namespace row is gone is a
    /// broken catalog, and answering "the default" is the safe reading of it.
    pub fn namespace_explode_budget(&self, name: &str) -> Result<Option<u64>> {
        Ok(self
            .doc
            .load()?
            .namespaces
            .get(name)?
            .and_then(|ns| ns.explode_budget_bytes))
    }

    /// **Make a namespace public, or private again** ([`Namespace::public`]).
    ///
    /// Only the stored fact: who may change it (the account admin) is the gRPC
    /// layer's question (`Repos.SetNamespaceSettings`). It touches no
    /// repository — a private repository stays private in a public namespace.
    pub fn set_namespace_public(&self, name: &str, public: bool) -> Result<Namespace> {
        self.update_namespace(name, |ns| ns.public = public)
    }

    /// **Make an account's emitted events public, or private again**
    /// ([`Namespace::events_public`]). The stored fact only, as
    /// [`RepoCatalog::set_namespace_public`].
    pub fn set_namespace_events_public(&self, name: &str, public: bool) -> Result<Namespace> {
        self.update_namespace(name, |ns| ns.events_public = public)
    }

    /// One namespace row, changed in one compare-and-swap.
    fn update_namespace(&self, name: &str, change: impl Fn(&mut Namespace)) -> Result<Namespace> {
        self.doc.update(|doc| {
            let Some(mut ns) = doc.namespaces.get(name)? else {
                return Err(Error::NoSuchNamespace(name.to_owned()));
            };
            change(&mut ns);
            doc.namespaces.insert(name.to_owned(), ns.clone())?;
            Ok(ns)
        })
    }

    /// **The catalog's version**: changes exactly when any namespace or
    /// repository row does. One pointer read — what lets a reader that keeps
    /// its own per-namespace index of the catalog (the server's usage ledger)
    /// rebuild it only when something changed, instead of listing every
    /// repository on every read.
    pub fn version(&self) -> Result<Option<crate::control::BlobId>> {
        self.doc.version()
    }

    /// Every namespace, by name. **O(namespaces)**, the listing path.
    pub fn namespaces(&self) -> Result<Vec<Namespace>> {
        self.doc.load()?.namespaces.values()
    }

    /// One namespace, if it exists.
    pub fn namespace(&self, name: &str) -> Result<Option<Namespace>> {
        self.doc.load()?.namespaces.get(name)
    }

    /// Delete a namespace, and return the repositories that went with it.
    ///
    /// Refuses a namespace that still holds repositories unless `delete_repos`.
    /// Deleting a tenant's repositories has to be asked for explicitly.
    pub fn delete_namespace(&self, name: &str, delete_repos: bool) -> Result<Vec<Repo>> {
        self.doc.update(|doc| {
            if !doc.namespaces.contains_key(name)? {
                return Err(Error::NoSuchNamespace(name.to_owned()));
            }
            // O(repositories on the server): a tenant's repositories are not
            // indexed apart from every other tenant's, because the connection
            // path looks a repository up by its full key and never by namespace.
            let doomed: Vec<Repo> = doc
                .repos
                .values()?
                .into_iter()
                .filter(|repo| repo.namespace == name)
                .collect();
            if !doomed.is_empty() && !delete_repos {
                return Err(Error::NamespaceNotEmpty {
                    name: name.to_owned(),
                    repos: doomed.len(),
                });
            }
            for repo in &doomed {
                doc.repos.remove(&repo.key())?;
            }
            doc.namespaces.remove(name)?;
            Ok(doomed)
        })
    }

    // ── repositories ───────────────────────────────────────────────────────

    /// Create a repository.
    ///
    /// The namespace must already exist. A repository is never allowed to
    /// conjure one, because then a typo in a push URL would silently become a
    /// tenant.
    pub fn create(&self, spec: NewRepo) -> Result<Repo> {
        let repo = spec.prospect()?;
        let key = spec.path.key();
        self.doc.update(|doc| {
            if !doc.namespaces.contains_key(repo.namespace.as_str())? {
                return Err(Error::NoSuchNamespace(repo.namespace.clone()));
            }
            if doc.repos.contains_key(&key)? {
                return Err(Error::RepoExists(spec.path.clone()));
            }
            doc.repos.insert(key.clone(), repo.clone())?;
            Ok(repo.clone())
        })
    }

    /// Look a repository up, failing if it is not there.
    ///
    /// This is what the SSH layer calls after the sandbox has produced a
    /// [`RepoRef`]: a path that parses is not a repository that exists.
    pub fn open(&self, path: &RepoRef) -> Result<Repo> {
        self.get(path)?
            .ok_or_else(|| Error::NoSuchRepo(path.clone()))
    }

    /// Look a repository up.
    ///
    /// **This is the connection path**. `Control::resolve` calls it for every
    /// SSH session. One shard probe, and no object read at all once the catalog
    /// has been read since it last changed.
    pub fn get(&self, path: &RepoRef) -> Result<Option<Repo>> {
        self.doc.load()?.repos.get(&path.key())
    }

    /// **Many rows, one read of the catalog** (round 5.3.1): `get` for each
    /// path, the document's pointer read once rather than once per row — a
    /// listing's access check asks about every row it lists.
    pub fn get_many(&self, paths: &[RepoRef]) -> Result<Vec<Option<Repo>>> {
        let doc = self.doc.load()?;
        paths.iter().map(|p| doc.repos.get(&p.key())).collect()
    }

    /// Is this repository in the catalog?
    pub fn contains(&self, path: &RepoRef) -> Result<bool> {
        self.doc.load()?.repos.contains_key(&path.key())
    }

    /// Every repository, or every repository in one namespace, ordered by key.
    ///
    /// **O(repositories)** either way; see [`RepoCatalog::delete_namespace`].
    pub fn list(&self, namespace: Option<&str>) -> Result<Vec<Repo>> {
        let doc = self.doc.load()?;
        Ok(doc
            .repos
            .values()?
            .into_iter()
            .filter(|repo| namespace.is_none_or(|ns| repo.namespace == ns))
            .collect())
    }

    /// Remove a repository's metadata and hand it back.
    ///
    /// The caller owns the storage: whatever backend held its objects and refs
    /// is still there afterwards and must be dealt with separately.
    pub fn delete(&self, path: &RepoRef) -> Result<Repo> {
        let key = path.key();
        self.doc.update(|doc| {
            doc.repos
                .remove(&key)?
                .ok_or_else(|| Error::NoSuchRepo(path.clone()))
        })
    }

    /// Record another default branch for a repository.
    ///
    /// This writes the **catalog** and nothing else. The repository's own
    /// `HEAD` is a projection of this value and is moved by
    /// [`point_head_at`], which whoever owns the repository's stores calls
    /// with the `Repo` this hands back. Doing it here is not possible: the
    /// catalog holds metadata only and has never had a repository's ref store
    /// (see this module's opening comment).
    pub fn set_default_branch(&self, path: &RepoRef, branch: &str) -> Result<Repo> {
        let branch = check_branch(branch)?;
        let key = path.key();
        self.doc.update(|doc| {
            let mut repo = doc
                .repos
                .get(&key)?
                .ok_or_else(|| Error::NoSuchRepo(path.clone()))?;
            repo.default_branch = branch.clone();
            doc.repos.insert(key.clone(), repo.clone())?;
            Ok(repo)
        })
    }

    /// Change a repository's visibility.
    pub fn set_visibility(&self, path: &RepoRef, visibility: Visibility) -> Result<Repo> {
        let key = path.key();
        self.doc.update(|doc| {
            let mut repo = doc
                .repos
                .get(&key)?
                .ok_or_else(|| Error::NoSuchRepo(path.clone()))?;
            repo.visibility = visibility;
            doc.repos.insert(key.clone(), repo.clone())?;
            Ok(repo)
        })
    }

    /// Turn a repository's "encrypted" flag on or off (`Repos.SetEncrypted`;
    /// the account admin's decision, checked by the caller).
    pub fn set_encrypted(&self, path: &RepoRef, encrypted: bool) -> Result<Repo> {
        let key = path.key();
        self.doc.update(|doc| {
            let mut repo = doc
                .repos
                .get(&key)?
                .ok_or_else(|| Error::NoSuchRepo(path.clone()))?;
            repo.encrypted = encrypted;
            doc.repos.insert(key.clone(), repo.clone())?;
            Ok(repo)
        })
    }

    /// How many namespaces and repositories there are, for `Admin.Stats`.
    pub fn counts(&self) -> Result<(usize, usize)> {
        let doc = self.doc.load()?;
        Ok((doc.namespaces.len(), doc.repos.len()))
    }
}

/// Make a repository's `HEAD` name the default branch its catalog entry
/// carries, and say whether it had to move.
///
/// # The bug this exists to make impossible
///
/// The catalog is the **authority** for a repository's default branch. A
/// repository's own `HEAD` is a **projection** of that value, and this is the
/// only writer of the projection (one implementation).
///
/// Until 2026-08-06 there was no projection at all, so the two disagreed the
/// moment anyone moved the authority. `FsStore::open_or_init` wrote
/// `ref: refs/heads/main` into a new repository and nothing ever wrote it
/// again; `Repos.SetDefaultBranch` wrote the catalog and nothing else. So
/// `set-default-branch master` answered `refs/heads/master`, `repo get` read
/// `refs/heads/master` back — and the reference advertisement, which reads the
/// repository's `HEAD` and not the catalog, still resolved
/// `refs/heads/main`. That branch had never been created, so `HEAD` had no
/// object, so it was **dropped from the advertisement entirely** and every
/// `master`-default donor (`foot`, `ripgrep`, `mdBook`) pushed cleanly into a
/// repository nobody could clone back. MEASURED that day: `ls-remote` returned
/// one line, `refs/heads/master`, and no `HEAD` line at all.
///
/// # Why it takes a `&dyn GitServe` and not a `&RepoCatalog`
///
/// The catalog stores metadata only and has never held a repository's stores;
/// the caller that opens those is the one that can hand both halves over. That
/// caller is `ServerStores::repo_store`, which runs on create, on
/// `set-default-branch` and on the first transfer after a restart — so an
/// already-broken repository on disk is repaired the next time it is opened,
/// with no migration step.
///
/// A symbolic `HEAD` is written whether or not the branch exists. An unborn
/// `HEAD` is legal and is exactly what a freshly created, never-pushed
/// repository has; what is not legal is a `HEAD` naming a branch the owner
/// never asked for.
///
/// # ⚠ The contract point this depended on is SETTLED — 2026-08-10, the `GitServe` HEAD ruling
///
/// This used to be `RefStore::head()` / `set_head()`, and when that pair died it
/// was rewritten against `GitOps::refs()` / `put_refs()` with a warning attached:
/// *whether `refs()` includes a row for `HEAD` is an open question … both cannot
/// be right*. It is answered. **`HEAD` is not a row.** Every backend excludes
/// pseudo-refs from the ref stream, and the accessor pair came back on the
/// reading half of the contract as [`GitServe::head`] / [`GitServe::set_head`].
///
/// So this reads and writes through that pair, and the pessimism the old comment
/// described is gone with it: against a conformant backend the comparison below
/// could never match, so *"did it move"* was `true` on every open of every
/// healthy repository. It is now exact. That is also why the parameter widened
/// from `&dyn GitOps`: `GitServe: GitOps`, so this is the same store and the same
/// single handle a caller already holds, viewed through the half that can answer
/// the question.
pub fn point_head_at(git: &dyn GitServe, default_branch: &str) -> Result<bool> {
    let want = check_branch(default_branch)?;
    let current = git
        .head()
        .map_err(|err| backend(err.context("reading HEAD to see where it points")))?
        .and_then(|row| row.symref_target);
    if current.as_deref() == Some(want.as_str()) {
        return Ok(false);
    }
    // `set_head_projected`, not `set_head`, and the difference is a promise
    // this function is the proof of: the authority for HEAD is the catalog
    // row's `default_branch` column, and this runs on **every** open of every
    // repository (`Stores::repo_store`), so a HEAD frame a power cut took is
    // written again the next time anything touches the store. That lets the
    // znippy arm skip a barrier — 6.0 ms of a 68.5 ms push-create on the 2 vCPU
    // rig — for a fact it would rebuild anyway. A backend with nothing cheaper
    // to do gets the default, which is `set_head`.
    git.set_head_projected(&want)
        .map_err(|err| backend(err.context(format!("pointing HEAD at {default_branch}"))))?;
    Ok(true)
}

/// A backend fault from [`GitOps`](gunnar_store::git_storage_trait::GitOps), which
/// answers in `anyhow`, as this crate's
/// [`Error::Store`].
///
/// `GitOps` has no typed rejection — `update_ref`'s lost compare-and-swap and a
/// broken disk arrive as the same `anyhow::Error`. Nothing in
/// this file needs to tell them apart; anything that does must not do it by
/// matching on the message.
fn backend(err: anyhow::Error) -> Error {
    Error::Store(gunnar_store::Error::Backend(err))
}