gunnar-grpc 2.0.0

//! `Repos`: namespaces and repositories.
//!
//! The two mutations that change what exists, `Create` and `Delete`, publish
//! `RepoCreated` / `RepoDeleted` on the event bus. That is what lets a UI show
//! a new repo without polling (the event stream).

use std::pin::Pin;
use std::sync::Arc;

use gunnar_proto::repos_server::Repos;
use gunnar_proto::{
    event, ArchiveOutChunk, ArchiveOutRequest, CreateNamespaceRequest, CreateNamespaceResponse,
    CreateRepoRequest, CreateRepoResponse, DeleteNamespaceRequest, DeleteNamespaceResponse,
    DeleteRepoRequest, DeleteRepoResponse, DeletedRepo, GcRepoRequest, GcRepoResponse,
    GcStatusRequest, GcStatusResponse, GetRepoRequest, GetRepoResponse, HashKind,
    ListDeletedRequest, ListDeletedResponse,
    ListNamespacesRequest, ListNamespacesResponse, ListOrphansRequest, ListOrphansResponse,
    ListReposRequest, ListReposResponse, OrphanStore,
    PurgeRepoRequest, PurgeRepoResponse, ReclaimOrphanRequest, ReclaimOrphanResponse, RepoCreated,
    RepoDeleted, SetDefaultBranchRequest, SetDefaultBranchResponse, SetNamespaceSettingsRequest,
    RecipientKeysRequest, RecipientKeysResponse, SetEncryptedRequest, SetEncryptedResponse,
    SetNamespaceSettingsResponse, SetVisibilityRequest,
    SetVisibilityResponse,
    UndeleteRepoRequest, UndeleteRepoResponse, Visibility,
};
use tokio_stream::Stream;
use tonic::{Request, Response, Status};

use crate::auth::Access;
use crate::seam::NewRepo;
use crate::service::ReposService;
use crate::status::{opt_str, repo_path};
use gunnar_proto::GrantLevel;

/// Payload bytes one `ArchiveOutChunk` carries.
///
/// 256 KiB: comfortably inside tonic's 4 MiB default message ceiling with room
/// for framing, and big enough that a 230 MiB corpus is ~900 messages rather
/// than a syscall storm. The archive writer's own copy buffer is 1 MiB, so one
/// of its writes becomes four messages and nothing has to be re-buffered twice.
pub const ARCHIVE_CHUNK_BYTES: usize = 256 * 1024;

/// How many chunks may sit between the blocking cut and the socket.
///
/// Bounded, and small. This is the backpressure: a client that stops reading
/// blocks the `spawn_blocking` thread on `blocking_send` instead of letting the
/// server buffer a repository into memory on its behalf.
const ARCHIVE_QUEUE: usize = 4;

/// The `Write` an archive cut is handed: it chunks and sends, and it never
/// grows past one chunk.
struct ChunkSink {
    tx: tokio::sync::mpsc::Sender<Result<ArchiveOutChunk, Status>>,
    buf: Vec<u8>,
}

impl ChunkSink {
    fn new(tx: tokio::sync::mpsc::Sender<Result<ArchiveOutChunk, Status>>) -> Self {
        ChunkSink {
            tx,
            buf: Vec::with_capacity(ARCHIVE_CHUNK_BYTES),
        }
    }

    /// **A client that hung up is an error, not a quiet success.** An archive
    /// whose stream was dropped half way is a truncated archive, and the one
    /// thing a backup must never do is report a cut it did not finish.
    fn send(&mut self) -> std::io::Result<()> {
        let data = std::mem::replace(&mut self.buf, Vec::with_capacity(ARCHIVE_CHUNK_BYTES));
        self.tx
            .blocking_send(Ok(ArchiveOutChunk {
                data,
                ..ArchiveOutChunk::default()
            }))
            .map_err(|_| {
                std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "the client stopped reading the archive stream before it ended",
                )
            })
    }
}

impl std::io::Write for ChunkSink {
    fn write(&mut self, mut buf: &[u8]) -> std::io::Result<usize> {
        let total = buf.len();
        while !buf.is_empty() {
            let room = ARCHIVE_CHUNK_BYTES - self.buf.len();
            let n = room.min(buf.len());
            self.buf.extend_from_slice(&buf[..n]);
            buf = &buf[n..];
            if self.buf.len() == ARCHIVE_CHUNK_BYTES {
                self.send()?;
            }
        }
        Ok(total)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        if !self.buf.is_empty() {
            self.send()?;
        }
        Ok(())
    }
}

#[tonic::async_trait]
impl Repos for ReposService {
    type ArchiveOutStream =
        Pin<Box<dyn Stream<Item = Result<ArchiveOutChunk, Status>> + Send + 'static>>;

    /// **`Repos.ArchiveOut`: a repository's bytes, off the box.**
    ///
    /// Authorisation is `Read` on the repository — what a clone of a private
    /// repository takes over SSH, because that is what this hands over. It is
    /// spelled with [`ServiceContext::require`](crate::service::ServiceContext)
    /// like every other RPC, so a principal without the grant gets a named
    /// `PERMISSION_DENIED` before a byte is read, and the audit line is the
    /// same shape as every other denial.
    ///
    /// The cut runs on the blocking pool. The channel is bounded at
    /// [`ARCHIVE_QUEUE`], so a slow reader slows the cut instead of buffering a
    /// repository into the server's memory.
    async fn archive_out(
        &self,
        request: Request<ArchiveOutRequest>,
    ) -> Result<Response<Self::ArchiveOutStream>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        self.ctx.require(
            &caller,
            Access::repo(&path),
            GrantLevel::Read,
            "Repos.ArchiveOut",
        )?;
        // Resolve BEFORE the stream is opened. A `NOT_FOUND` that arrives as
        // the first item of an accepted stream has already made the client
        // create its output file; as the call's own status it has not.
        self.ctx.control.get_repo(&path)?;

        // ── the admission permit, on the control plane's one transfer ────────
        //
        // `ArchiveOut` streams a whole repository's stored bytes down this
        // channel — a transfer in everything but name — so it takes a slot from
        // the SAME gate a `git-upload-pack` does (one implementation), through the
        // `TransferGate` seam `gunnar-server` wired the SSH listener's own
        // `Arc<Admission>` into. After authorisation and the existence check,
        // for the same reason the SSH and HTTP paths gate after theirs: "the
        // server is busy" must not leak that a repository a caller may not read
        // exists. Over budget it is **refused** with `RESOURCE_EXHAUSTED` (the
        // gRPC idiom for 429) — the client and any proxy read it as transient —
        // rather than queued behind a socket that goes quiet.
        //
        // `None` is the ungated path (a plaintext test router, or a
        // listener built without a gate); the permit is then simply absent and
        // the stream runs as before. That absence is the red the test drives.
        let permit = match &self.ctx.transfer_gate {
            Some(gate) => match gate.admit_transfer(&caller.principal_id, &path.namespace) {
                Ok(permit) => Some(permit),
                Err(refused) => {
                    // The SAME audit shape the SSH and HTTP refusals file
                    // (`event="*.transfer.refused"`), on the git target so an
                    // operator greps one place across all three front doors.
                    tracing::warn!(
                        target: "gunnar::git",
                        event = "grpc.transfer.refused",
                        reason = "admission-gate-full",
                        rpc = "Repos.ArchiveOut",
                        repo = %path,
                        principal = %caller.principal_id,
                        in_flight = refused.in_flight,
                        peak_in_flight = refused.peak_in_flight,
                        limit = refused.limit,
                        reserved_bytes = refused.reserved_bytes,
                        refused_total = refused.refused_total,
                        "control-plane ArchiveOut REFUSED: the admission gate is full"
                    );
                    return Err(Status::resource_exhausted(refused.message));
                }
            },
            None => None,
        };

        let (tx, rx) = tokio::sync::mpsc::channel(ARCHIVE_QUEUE);
        let control = Arc::clone(&self.ctx.control);
        let from_generation = request.from_generation;
        let principal = caller.principal_id.clone();
        let named = format!("{path}");

        // A32-14, stays a task: one cut's producer under its permit; it ends with its stream.
        tokio::task::spawn_blocking(move || {
            use std::io::Write;
            // The permit is held for the WHOLE cut and released when this task
            // ends — the same lifetime the SSH/HTTP guard has. `entered_service`
            // files the sojourn sample the governor reads: everything before
            // here (the blocking-pool handoff) was the transfer waiting.
            if let Some(permit) = &permit {
                permit.0.entered_service();
            }
            let mut sink = ChunkSink::new(tx.clone());
            let outcome = control
                .archive_out(&path, from_generation, &mut sink)
                .map_err(Status::from)
                .and_then(|cut| {
                    sink.flush()
                        .map_err(|e| Status::aborted(format!("streaming the archive: {e}")))?;
                    Ok(cut)
                });
            let last = match outcome {
                Ok(cut) => {
                    tracing::info!(
                        target: "gunnar::archive",
                        event = "repo.archive_out",
                        repo = %named,
                        principal = %principal,
                        from_generation,
                        to_generation = cut.to_generation,
                        bytes_out = cut.bytes_out,
                        "streamed a repository archive off the box"
                    );
                    Ok(ArchiveOutChunk {
                        data: Vec::new(),
                        last: true,
                        to_generation: cut.to_generation,
                        bytes_out: cut.bytes_out,
                        file_sha256: cut.file_sha256.to_vec(),
                    })
                }
                Err(status) => {
                    tracing::error!(
                        target: "gunnar::archive",
                        event = "repo.archive_out_failed",
                        repo = %named,
                        principal = %principal,
                        from_generation,
                        error = %status.message(),
                        "a repository archive stream failed; the client has a TRUNCATED file"
                    );
                    Err(status)
                }
            };
            // The receiver is gone when the client hung up; there is nobody
            // left to tell, and that is not itself a failure to log twice.
            let _ = tx.blocking_send(last);
        });

        Ok(Response::new(Box::pin(
            tokio_stream::wrappers::ReceiverStream::new(rx),
        )))
    }

    async fn create_namespace(
        &self,
        request: Request<CreateNamespaceRequest>,
    ) -> Result<Response<CreateNamespaceResponse>, Status> {
        // Creating a tenant is a server-wide act: there is no namespace to
        // scope it to, which is exactly why `Scope::Server` exists (the ACL model).
        let caller = self.ctx.authorize(
            &request,
            Access::Server,
            GrantLevel::Admin,
            "Repos.CreateNamespace",
        )?;
        let request = request.into_inner();
        // The name filter (2026-10-05): a namespace is named here. Reserved
        // names are the operator's to use (a PERSON with server Admin names
        // the estate's own namespaces); a machine relaying a customer's
        // choice is held to them.
        // Only a server admin gets here, naming the estate's own namespaces
        // (the kit seeds `nordisk`): reserved names are theirs to use; an
        // offensive one is refused, and a held one waits for the account
        // admin when the namespace is inside an account.
        let account = self.ctx.account_of_namespace(&request.name);
        let last = request.name.rsplit('/').next().unwrap_or(&request.name).to_owned();
        self.ctx.filter_name(&caller, "namespace", &last, account.clone(), true)?;
        let namespace = self
            .ctx
            .control
            .create_namespace(&request.name, &request.description)?;
        self.ctx.account_event(
            account,
            crate::enterprise::AccountAuditEvent::by(&caller, "namespace.created", request.name.clone()),
        );
        Ok(Response::new(CreateNamespaceResponse {
            namespace: Some(namespace),
        }))
    }

    async fn list_namespaces(
        &self,
        request: Request<ListNamespacesRequest>,
    ) -> Result<Response<ListNamespacesResponse>, Status> {
        // **Every namespace to a server reader; to anybody else, the ones they
        // are inside and the PUBLIC ones** (owner, 2026-10-03: namespaces are
        // not public by default, "such things has to be set"; a gated account
        // or a gated caller sees no public one, see
        // `ControlPlane::namespace_for_outsider`). Inside is `Read`
        // over the namespace: its owner, a member, a grant. A private namespace
        // an outsider is not inside is not listed, even when it holds a public
        // repository — that repository is reachable by its path, and its
        // namespace shows it (and only it) to whoever asks for it by name.
        let caller = self.ctx.caller(&request)?;
        let listing = self
            .ctx
            .require_or_narrow(&caller, Access::Server, "Repos.ListNamespaces")?;
        let mut namespaces = self.ctx.control.list_namespaces()?;
        // **A server-wide reader is still not inside a customer's account**
        // (2026-10-05): a namespace closed to the caller's server grant is
        // shown as it is to any outsider — only when its admin made it public.
        let mut closed = Vec::new();
        if listing == crate::service::Listing::Everything {
            let mut open = Vec::with_capacity(namespaces.len());
            for ns in namespaces {
                if self.ctx.shielded(&caller, &ns.name)? {
                    closed.push(ns);
                } else {
                    open.push(ns);
                }
            }
            namespaces = open;
            for ns in closed.drain(..) {
                if ns.public
                    && self
                        .ctx
                        .control
                        .namespace_for_outsider(&caller.principal_id, &ns.name)?
                        .is_some_and(|outside| outside.public)
                {
                    namespaces.push(ns);
                }
            }
            namespaces.sort_by(|a, b| a.name.cmp(&b.name));
        }
        if listing != crate::service::Listing::Everything {
            let mut shown = Vec::new();
            for ns in namespaces {
                let inside = self
                    .ctx
                    .control
                    .may(&caller.principal_id, &Access::namespace(&ns.name), GrantLevel::Read)
                    .map_err(|err| {
                        Status::unavailable(format!("the identity table could not be read: {err}"))
                    })?;
                // Outside: only what the account admin opened, and nothing of
                // a gated account or to a gated caller (the gate outranks the
                // setting, as it outranks a public repository).
                let opened = inside
                    || (ns.public
                        && self
                            .ctx
                            .control
                            .namespace_for_outsider(&caller.principal_id, &ns.name)?
                            .is_some_and(|outside| outside.public));
                if opened {
                    shown.push(ns);
                }
            }
            namespaces = shown;
        }
        Ok(Response::new(ListNamespacesResponse {
            namespaces,
            // Paging is a store concern; until gunnar-store lands, one page.
            next_page_token: String::new(),
        }))
    }

    async fn delete_namespace(
        &self,
        request: Request<DeleteNamespaceRequest>,
    ) -> Result<Response<DeleteNamespaceResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        self.ctx.require(
            &caller,
            Access::namespace(&request.name),
            GrantLevel::Admin,
            "Repos.DeleteNamespace",
        )?;
        let account = self.ctx.account_of_namespace(&request.name);
        let repos_deleted = self
            .ctx
            .control
            .delete_namespace(&request.name, request.delete_repos)?;
        self.ctx.account_event(
            account,
            crate::enterprise::AccountAuditEvent::by(&caller, "namespace.deleted", request.name.clone())
                .detail(format!("{repos_deleted} repositories with it")),
        );
        Ok(Response::new(DeleteNamespaceResponse { repos_deleted }))
    }

    async fn create(
        &self,
        request: Request<CreateRepoRequest>,
    ) -> Result<Response<CreateRepoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        // Namespace-scoped, not repo-scoped: the repository does not exist
        // yet, so there is nothing narrower to hold a grant on.
        self.ctx.require(
            &caller,
            Access::namespace(&path.namespace),
            create_level(&self.ctx, &caller, &path.namespace, request.visibility),
            "Repos.Create",
        )?;
        let hash_kind = HashKind::try_from(request.hash_kind)
            .map_err(|_| Status::invalid_argument("unknown hash_kind"))?;
        let visibility = Visibility::try_from(request.visibility)
            .map_err(|_| Status::invalid_argument("unknown visibility"))?;
        // A repository born PUBLIC is a publish, and publishing is the account
        // admin's (as on `SetVisibility`): the namespace `Admin` above already
        // asked exactly that, so a member cannot create its way round it.

        // The name filter (2026-10-05): a repository is named here.
        let account = self.ctx.account_of_namespace(&path.namespace);
        self.ctx.filter_name(&caller, "repo", &path.repo, account.clone(), false)?;
        // The account's room, as a push-to-create meets it (2026-10-06): a
        // namespace its plan refuses writes to, or at its cap, gets no new
        // repository — refused with the plan's own line.
        self.ctx.control.create_room(&path.namespace, caller.machine)?;
        let repo = self.ctx.control.create_repo(NewRepo {
            path,
            hash_kind,
            default_branch: request.default_branch,
            visibility,
            description: request.description,
            adopt: request.adopt,
        })?;
        // The first commit, written by the server, when asked (a sold estate's welcome).
        if !request.readme.is_empty() {
            let path = repo_path(repo.path.clone(), "repo")?;
            let commit = self.ctx.control.write_first_commit(&path, &request.readme)?;
            tracing::info!(repo = %format!("{}/{}", path.namespace, path.repo), commit = %commit, "repos: the first commit (README.md) written by the server");
        }

        self.ctx
            .bus
            .publish(event::Payload::RepoCreated(RepoCreated {
                repo: Some(repo.clone()),
                // Named, now that there is a name to give: the event carries
                // the principal that created the repository.
                principal_id: caller.principal_id.clone(),
            }));
        self.ctx.account_event(
            account,
            crate::enterprise::AccountAuditEvent::by(
                &caller,
                "repo.created",
                format!("{}/{}", repo.path.as_ref().map(|p| p.namespace.as_str()).unwrap_or(""), repo.path.as_ref().map(|p| p.repo.as_str()).unwrap_or("")),
            )
            .detail(if repo.visibility == Visibility::Public as i32 { "born public" } else { "private" }),
        );

        Ok(Response::new(CreateRepoResponse { repo: Some(repo) }))
    }

    /// **A dated tombstone by default; a reclaim only when asked.**
    ///
    /// The event is published either way and says the same thing — the
    /// repository is gone from the catalog, and a subscriber (a UI, a mirror)
    /// must stop showing it. What differs is whether the bytes went, and that
    /// is in the response rather than the event because it is the caller's
    /// business and not the world's.
    async fn delete(
        &self,
        request: Request<DeleteRepoRequest>,
    ) -> Result<Response<DeleteRepoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        // **Repository scope, then namespace scope** — the narrowest that CAN
        // hold a grant. A plain `Access::repo` here refused everyone on the
        // appliance's orphaned `alice/linux` (`175daccc0000000b`), because
        // `may` answers `false` for a repository that is not in the catalog
        // and the store had lost its row. Delete is the verb that gets you out
        // of that, so it must not be gated on the record that has gone. Same
        // rule as `Repos.Undelete` and `Repos.Purge`, written once.
        self.ctx.require_repo_or_namespace(
            &caller,
            &path,
            GrantLevel::Admin,
            "Repos.Delete",
        )?;
        // A plain 0 is what a client that never heard of the field sends, so
        // it means "the server's default". Meaning it literally takes saying
        // so — see `zero_retention` in the proto.
        let retention_days = if request.zero_retention {
            Some(0)
        } else if request.retention_days > 0 {
            Some(request.retention_days)
        } else {
            None
        };
        let outcome = self.ctx.control.delete_repo(
            &path,
            crate::seam::DeleteOptions {
                purge: request.purge,
                retention_days,
                principal_id: caller.principal_id.clone(),
            },
        )?;
        self.ctx.account_event(
            self.ctx.account_of_namespace(&path.namespace),
            crate::enterprise::AccountAuditEvent::by(&caller, "repo.deleted", format!("{}/{}", path.namespace, path.repo))
                .detail(match &outcome {
                    crate::seam::DeleteOutcome::Purged { .. } => "purged".to_owned(),
                    crate::seam::DeleteOutcome::Tombstoned { purge_after_unix_ms, .. } => {
                        format!("kept until {purge_after_unix_ms} (undeletable)")
                    }
                }),
        );
        self.ctx
            .bus
            .publish(event::Payload::RepoDeleted(RepoDeleted {
                repo: Some(path),
                principal_id: caller.principal_id.clone(),
            }));
        Ok(Response::new(match outcome {
            crate::seam::DeleteOutcome::Purged { bytes_reclaimed } => DeleteRepoResponse {
                purged: true,
                bytes_reclaimed,
                bytes_kept: 0,
                purge_after_unix_ms: 0,
            },
            crate::seam::DeleteOutcome::Tombstoned {
                purge_after_unix_ms,
                bytes_kept,
            } => DeleteRepoResponse {
                purged: false,
                bytes_reclaimed: 0,
                bytes_kept,
                purge_after_unix_ms,
            },
        }))
    }

    /// **`Repos.Undelete`: put a soft-deleted repository back.**
    ///
    /// **The narrowest scope that CAN hold a grant**
    /// ([`ServiceContext::require_repo_or_namespace`]) — which for a deleted
    /// repository is the namespace, and that is not a widening: a deleted
    /// repository is not in the catalog, and `may` answers `false` for a grant
    /// on a repository that does not exist — deliberately, so a caller cannot
    /// probe the ACL of repositories it cannot see. A repo-scoped check here
    /// would therefore refuse *everyone*, including the account's owner.
    ///
    /// Seen red: the first cut used `Access::repo` and the drill's
    /// gunnar-admin got `principal admin has no admin grant on
    /// repo:alice/knut` on a repository it had just deleted. 2026-08-30 found
    /// the same hole still open on `Repos.Delete` and moved the rule into the
    /// one helper all three now call, so there is one answer and not three.
    async fn undelete(
        &self,
        request: Request<UndeleteRepoRequest>,
    ) -> Result<Response<UndeleteRepoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().path, "path")?;
        self.ctx.require_repo_or_namespace(
            &caller,
            &path,
            GrantLevel::Admin,
            "Repos.Undelete",
        )?;
        let repo = self.ctx.control.undelete_repo(&path)?;
        self.ctx.account_event(
            self.ctx.account_of_namespace(&path.namespace),
            crate::enterprise::AccountAuditEvent::by(&caller, "repo.undeleted", format!("{}/{}", path.namespace, path.repo)),
        );
        // The same event a create publishes, because to everything watching
        // this IS a repository appearing.
        self.ctx
            .bus
            .publish(event::Payload::RepoCreated(RepoCreated {
                repo: Some(repo.clone()),
                principal_id: caller.principal_id.clone(),
            }));
        Ok(Response::new(UndeleteRepoResponse { repo: Some(repo) }))
    }

    /// **`Repos.Purge`: end the window early.** Admin, and there is no undo
    /// after it.
    ///
    /// Scoped by [`ServiceContext::require_repo_or_namespace`] for the same
    /// reason [`Repos::undelete`] is: the repository is already out of the
    /// catalog, so there is no repo-scoped grant left for anyone to hold and
    /// the namespace is the narrowest scope that can answer.
    async fn purge(
        &self,
        request: Request<PurgeRepoRequest>,
    ) -> Result<Response<PurgeRepoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().path, "path")?;
        self.ctx.require_repo_or_namespace(
            &caller,
            &path,
            GrantLevel::Admin,
            "Repos.Purge",
        )?;
        let bytes_reclaimed = self.ctx.control.purge_repo(&path)?;
        self.ctx.account_event(
            self.ctx.account_of_namespace(&path.namespace),
            crate::enterprise::AccountAuditEvent::by(&caller, "repo.purged", format!("{}/{}", path.namespace, path.repo)),
        );
        Ok(Response::new(PurgeRepoResponse { bytes_reclaimed }))
    }

    /// **`Repos.Gc`: maintain a live repository. `Write`, deliberately, and not
    /// `Admin`.**
    ///
    /// Every other reclaim verb on this service is Admin because every other
    /// one destroys something a caller cannot get back — `Purge` ends a
    /// retention window, `ReclaimOrphan` unlinks a store. **This one destroys
    /// nothing.** `gunnar-server`'s `storage.rs` says what it takes: "every
    /// lazily-exploded object unlinked, the commit-and-tree pool kept because
    /// that is the graph". Everything unlinked is re-derivable from the packs
    /// the same run just compacted, so the entire downside is a slower next
    /// read while the cache re-explodes. There is no state a gc can reach that
    /// an operator would have to restore from a backup.
    ///
    /// So the right gate is the one that says *this is your repository*: a
    /// principal who may push to it may reclaim its cache. `Write` is exactly
    /// that, and `require_repo_or_namespace` keeps it scoped to the repository
    /// or the namespace that holds it — nobody reclaims a stranger's bytes.
    ///
    /// **The Admin gate was not a safe default, it was the bug.** With gc
    /// reachable only from a shell on the box, 26 repositories on the live
    /// estate stood at `Last repacked = —`: never once maintained, holding an
    /// exploded read cache nothing was going to reclaim, because the one
    /// principal allowed to ask was the one principal not sitting in front of
    /// the console. A verb gated tighter than its blast radius does not get
    /// used more carefully; it gets used never.
    /// **It ENQUEUES.** It does not run the gc and it returns before one has —
    /// see [`crate::seam::ControlPlane::gc_repo`] and the queue behind it. A
    /// handler that blocked for the length of a gc would hold a tonic worker
    /// for minutes and make the button that called it look broken.
    async fn gc(&self, request: Request<GcRepoRequest>) -> Result<Response<GcRepoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().path, "path")?;
        self.ctx
            .require_repo_or_namespace(&caller, &path, GrantLevel::Write, "Repos.Gc")?;
        let job = self.ctx.control.gc_repo(&path)?;
        Ok(Response::new(GcRepoResponse {
            job: Some(gc_job_out(job)),
        }))
    }

    /// **`Repos.GcStatus`: the read half, gated at `Read`.**
    ///
    /// Asking *whether* maintenance ran is not asking *for* it, so this sits a
    /// level below [`Repos::gc`]: anyone who may see the repository may see
    /// that its cache was reclaimed and what it freed. A console that could
    /// press the button but not watch it would be the worse half of the pair.
    ///
    /// An empty `job` means the server is holding nothing for this repository.
    /// That is distinct from a gc that reclaimed zero bytes, which comes back
    /// as a `DONE` job saying so.
    async fn gc_status(
        &self,
        request: Request<GcStatusRequest>,
    ) -> Result<Response<GcStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().path, "path")?;
        self.ctx
            .require(&caller, Access::repo(&path), GrantLevel::Read, "Repos.GcStatus")?;
        let job = self.ctx.control.gc_status(&path)?;
        Ok(Response::new(GcStatusResponse {
            job: job.map(gc_job_out),
        }))
    }

    /// **`Repos.ListDeleted`: what is being held, and what it costs.**
    ///
    /// Scoped exactly like `Repos.List`: one namespace needs a grant on it,
    /// every namespace needs a server-wide one, because the answer spans
    /// tenants.
    async fn list_deleted(
        &self,
        request: Request<ListDeletedRequest>,
    ) -> Result<Response<ListDeletedResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let access = match opt_str(&request.namespace) {
            Some(namespace) => Access::namespace(namespace),
            None => Access::Server,
        };
        self.ctx
            .require(&caller, access, GrantLevel::Read, "Repos.ListDeleted")?;
        let rows = self.ctx.control.list_deleted(opt_str(&request.namespace))?;
        let rows = self.ctx.retain_unshielded(&caller, rows, |row| Some(row.path.namespace.clone()))?;
        let total_bytes = rows.iter().map(|row| row.bytes).sum();
        Ok(Response::new(ListDeletedResponse {
            repos: rows
                .into_iter()
                .map(|row| DeletedRepo {
                    repo: Some(row.path),
                    deleted_at_unix_ms: row.deleted_at_unix_ms,
                    purge_after_unix_ms: row.purge_after_unix_ms,
                    principal_id: row.principal_id,
                    bytes: row.bytes,
                })
                .collect(),
            total_bytes,
        }))
    }

    /// **`Repos.ListOrphans`: the store directories nothing else can see.**
    ///
    /// Scoped exactly like [`Repos::list_deleted`] and [`Repos::list`] — one
    /// namespace needs a grant on it, every namespace needs a server-wide one,
    /// because the answer spans tenants. It is a `Read`: it says what is on
    /// the disk, it does not touch it.
    ///
    /// The three listings partition the disk. Anything under the repository
    /// root is catalogued (`List`), inside a retention window (`ListDeleted`),
    /// or stranded (here). Before this the third case had no listing, which is
    /// how ~25-39 GB stood at `alice/linux` on the appliance with every
    /// reclaim verb refusing it.
    async fn list_orphans(
        &self,
        request: Request<ListOrphansRequest>,
    ) -> Result<Response<ListOrphansResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let access = match opt_str(&request.namespace) {
            Some(namespace) => Access::namespace(namespace),
            None => Access::Server,
        };
        self.ctx
            .require(&caller, access, GrantLevel::Read, "Repos.ListOrphans")?;
        let rows = self
            .ctx
            .control
            .list_orphans(opt_str(&request.namespace))?;
        let rows = self.ctx.retain_unshielded(&caller, rows, |row| Some(row.path.namespace.clone()))?;
        let total_bytes = rows.iter().map(|row| row.bytes).sum();
        Ok(Response::new(ListOrphansResponse {
            orphans: rows
                .into_iter()
                .map(|row| OrphanStore {
                    repo: Some(row.path),
                    dir: row.dir,
                    bytes: row.bytes,
                    hash_marker: row.hash_marker,
                    modified_unix_ms: row.modified_unix_ms,
                    failed_reclaim: row.failed_reclaim,
                })
                .collect(),
            total_bytes,
        }))
    }

    /// **`Repos.ReclaimOrphan`: the exit that did not exist.** Admin, no undo.
    ///
    /// Authorised by [`ServiceContext::require_repo_or_namespace`](crate::service::ServiceContext::require_repo_or_namespace),
    /// which is the whole point: an orphan has no catalog row, so a repo-scoped
    /// check refuses everyone — including the account's owner and
    /// gunnar-admin — and that refusal is one of the three walls that made
    /// `alice/linux` inescapable.
    ///
    /// The bytes go through `purge_repo`, the same verb `--purge` and the
    /// expiry sweep use (one implementation). To KEEP an orphan's objects, use
    /// `Repos.Create` with `adopt` and the hash `ListOrphans` reports.
    async fn reclaim_orphan(
        &self,
        request: Request<ReclaimOrphanRequest>,
    ) -> Result<Response<ReclaimOrphanResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().path, "path")?;
        self.ctx.require_repo_or_namespace(
            &caller,
            &path,
            GrantLevel::Admin,
            "Repos.ReclaimOrphan",
        )?;
        let bytes_reclaimed = self.ctx.control.reclaim_orphan(&path)?;
        // The same event a delete publishes: to everything watching, a name
        // has stopped existing on this box.
        self.ctx
            .bus
            .publish(event::Payload::RepoDeleted(RepoDeleted {
                repo: Some(path),
                principal_id: caller.principal_id.clone(),
            }));
        Ok(Response::new(ReclaimOrphanResponse { bytes_reclaimed }))
    }

    async fn list(
        &self,
        request: Request<ListReposRequest>,
    ) -> Result<Response<ListReposResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        // Listing a single namespace needs a grant on that namespace; listing
        // every namespace needs a server-wide one, because the answer spans
        // tenants (multi-tenancy).
        let access = match opt_str(&request.namespace) {
            Some(namespace) => Access::namespace(namespace),
            None => Access::Server,
        };
        // **`require_or_narrow`, not `require`** — the one RPC on this service
        // that answers with a COLLECTION rather than about a thing, and
        // therefore the one where "you hold no grant here" has a better answer
        // than a refusal (see [`ServiceContext::require_or_narrow`]).
        //
        // Every other read on this service names a repository and so already
        // went through `AuthTable::may`, which has known since anonymous read that a
        // public repository is readable with no grant. This one had no
        // repository to name, so it asked about the *scope* and refused — and
        // that refusal was the whole reason a published repository could be
        // cloned by a stranger who was told its path and discovered by nobody.
        let listing = self.ctx.require_or_narrow(&caller, access, "Repos.List")?;
        let repos = self.ctx.control.list_repos(opt_str(&request.namespace))?;
        let mut repos = self.ctx.retain_readable(&caller, repos, listing)?;
        // **The caller's filters, on the caller's own listing** (2026-10-04):
        // applied AFTER the narrowing, so they can only ever remove rows — a
        // query is a question about what this caller may already see, never a
        // way to see more. Over the whole answer, so a console's search is not
        // a search of whatever first page it fetched.
        let query = request.query.trim().to_lowercase();
        if !query.is_empty() {
            repos.retain(|repo| {
                repo.path.as_ref().is_some_and(|path| {
                    format!("{}/{}", path.namespace, path.repo)
                        .to_lowercase()
                        .contains(&query)
                })
            });
        }
        if request.mirrored_only {
            let mirrored: std::collections::BTreeSet<String> = match self
                .ctx
                .control
                .mirror_status(opt_str(&request.namespace), None)
            {
                Ok(mirrors) => mirrors
                    .into_iter()
                    .filter_map(|m| m.repo.map(|p| format!("{}/{}", p.namespace, p.repo)))
                    .collect(),
                Err(crate::seam::ControlError::Unimplemented(_)) => Default::default(),
                Err(err) => return Err(err.into()),
            };
            repos.retain(|repo| {
                repo.path
                    .as_ref()
                    .is_some_and(|p| mirrored.contains(&format!("{}/{}", p.namespace, p.repo)))
            });
        }
        Ok(Response::new(ListReposResponse {
            repos,
            next_page_token: String::new(),
        }))
    }

    async fn get(
        &self,
        request: Request<GetRepoRequest>,
    ) -> Result<Response<GetRepoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().path, "path")?;
        self.ctx
            .require(&caller, Access::repo(&path), GrantLevel::Read, "Repos.Get")?;
        let repo = self.ctx.control.get_repo(&path)?;
        Ok(Response::new(GetRepoResponse { repo: Some(repo) }))
    }

    async fn set_default_branch(
        &self,
        request: Request<SetDefaultBranchRequest>,
    ) -> Result<Response<SetDefaultBranchResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        self.ctx.require(
            &caller,
            Access::repo(&path),
            GrantLevel::Admin,
            "Repos.SetDefaultBranch",
        )?;
        let repo = self
            .ctx
            .control
            .set_default_branch(&path, &request.default_branch)?;
        Ok(Response::new(SetDefaultBranchResponse { repo: Some(repo) }))
    }

    async fn set_visibility(
        &self,
        request: Request<SetVisibilityRequest>,
    ) -> Result<Response<SetVisibilityResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        // **The ACCOUNT's decision, not the repository's** (owner, 2026-10-03:
        // members "cant make repos public", only the account admin = the first
        // person; gunnar's default for every account, not a tier and not a setting). Visibility decides who reads EVERY
        // reference in the repository, on behalf of the account whose name is
        // on it, so it takes `Admin` over the NAMESPACE: the principal the
        // namespace is named after (`AuthTable::owns_account`), a
        // namespace-scoped admin grant (directly or through a group), or a
        // server-scoped one. Until 2026-10-03 it took `Admin` on the
        // repository, and a repository-level admin could publish it.
        let account = self.ctx.account_of_namespace(&path.namespace);
        if let Err(status) = self.ctx.require(
            &caller,
            Access::namespace(&path.namespace),
            GrantLevel::Admin,
            "Repos.SetVisibility",
        ) {
            if status.code() == tonic::Code::PermissionDenied {
                self.ctx.account_event(
                    account,
                    crate::enterprise::AccountAuditEvent::by(&caller, "visibility.changed", format!("{}/{}", path.namespace, path.repo))
                        .detail("only the account admin changes public and private")
                        .refused(),
                );
            }
            return Err(account_admin_only(status, &path));
        }
        let visibility = Visibility::try_from(request.visibility)
            .map_err(|_| Status::invalid_argument("unknown visibility"))?;
        // `from` is read before the write so the event can say what moved;
        // a set to the visibility the repository already has is not a change
        // and publishes nothing — a PublicOnly replica (the browse replica)
        // acts on this event, and must not be told to forget or fetch a
        // repository whose visibility did not move.
        let before = self.ctx.control.get_repo(&path)?.visibility;
        let repo = self.ctx.control.set_visibility(&path, visibility)?;
        if repo.visibility != before {
            let word = |v: i32| if v == Visibility::Public as i32 { "public" } else { "private" };
            self.ctx.account_event(
                account,
                crate::enterprise::AccountAuditEvent::by(&caller, "visibility.changed", format!("{}/{}", path.namespace, path.repo))
                    .detail(format!("{} → {}", word(before), word(repo.visibility))),
            );
            self.ctx
                .bus
                .publish(event::Payload::VisibilityChanged(
                    gunnar_proto::VisibilityChanged {
                        repo: Some(path),
                        from: before,
                        to: repo.visibility,
                        principal_id: caller.principal_id.clone(),
                    },
                ));
        }
        Ok(Response::new(SetVisibilityResponse { repo: Some(repo) }))
    }

    /// **A member's SSH public keys, for wrapping a key ring to them.**
    async fn recipient_keys(
        &self,
        request: Request<RecipientKeysRequest>,
    ) -> Result<Response<RecipientKeysResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        let repo = Access::repo(&path);
        self.ctx
            .require(&caller, repo.clone(), GrantLevel::Read, "Repos.RecipientKeys")?;
        // Only for someone who could read the repository anyway: never a
        // directory of everybody's keys.
        if !self
            .ctx
            .control
            .may(&request.principal_id, &repo, GrantLevel::Read)?
        {
            return Err(Status::permission_denied(format!(
                "{} cannot read {}/{}: grant them access before adding them as a member of its \
                 key ring",
                request.principal_id, path.namespace, path.repo
            )));
        }
        let keys = self.ctx.control.recipient_keys(&request.principal_id)?;
        Ok(Response::new(RecipientKeysResponse { keys }))
    }

    /// **Encrypted repository on or off** — the account admin's decision.
    async fn set_encrypted(
        &self,
        request: Request<SetEncryptedRequest>,
    ) -> Result<Response<SetEncryptedResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.path, "path")?;
        // The ACCOUNT's decision, as visibility is: whether the server may hold
        // plaintext for this repository is decided for every member of the
        // account, so it takes `Admin` over the namespace, not the repository.
        let account = self.ctx.account_of_namespace(&path.namespace);
        if let Err(status) = self.ctx.require(
            &caller,
            Access::namespace(&path.namespace),
            GrantLevel::Admin,
            "Repos.SetEncrypted",
        ) {
            // **In the account's audit log** (wiring tracer B 2026-10-08), as a
            // visibility change is: refused and made alike.
            if status.code() == tonic::Code::PermissionDenied {
                self.ctx.account_event(
                    account,
                    crate::enterprise::AccountAuditEvent::by(&caller, "encrypted.changed", format!("{}/{}", path.namespace, path.repo))
                        .detail("only the account admin turns encryption on or off")
                        .refused(),
                );
            }
            return Err(account_admin_only_for(status, &path.namespace, "encrypted repositories in account"));
        }
        let before = self.ctx.control.get_repo(&path)?.encrypted;
        // **A mirror is never turned encrypted** (wiring tracer B 2026-10-08):
        // its content is an upstream's, fetched with no encrypted gate, never
        // sealed by the account's members. Off is never refused.
        if request.encrypted && !before {
            let a_mirror = match self.ctx.control.mirror_status(None, Some(&path)) {
                Ok(mirrors) => !mirrors.is_empty(),
                Err(crate::seam::ControlError::NotFound(_))
                | Err(crate::seam::ControlError::Unimplemented(_)) => false,
                Err(err) => return Err(err.into()),
            };
            if a_mirror {
                return Err(Status::failed_precondition(format!(
                    "{}/{} is a mirror: its content is copied from another server and was never \
                     sealed by your members, so it cannot be encrypted (remove the mirror first, \
                     then push sealed content)",
                    path.namespace, path.repo
                )));
            }
        }
        let repo = self.ctx.control.set_encrypted(&path, request.encrypted)?;
        if repo.encrypted != before {
            let word = |on: bool| if on { "encrypted" } else { "plaintext allowed" };
            self.ctx.account_event(
                account,
                crate::enterprise::AccountAuditEvent::by(&caller, "encrypted.changed", format!("{}/{}", path.namespace, path.repo))
                    .detail(format!("{} → {}", word(before), word(repo.encrypted))),
            );
            // …and an event, as a visibility change is: a webhook or a watcher hears it.
            self.ctx.bus.publish(event::Payload::EncryptedChanged(gunnar_proto::EncryptedChanged {
                repo: Some(path.clone()),
                from: before,
                to: repo.encrypted,
                principal_id: caller.principal_id.clone(),
            }));
        }
        tracing::info!(
            target: crate::auth::AUDIT_TARGET,
            event = "control.repo_encrypted",
            principal = %caller.principal_id,
            repo = %format!("{}/{}", path.namespace, path.repo),
            from = before,
            to = repo.encrypted,
            "the repository's encrypted flag was set"
        );
        Ok(Response::new(SetEncryptedResponse { repo: Some(repo) }))
    }

    /// **The account admin's two settings**: is the namespace public, are its
    /// events public. Private, both, until the account admin says otherwise —
    /// and only the account admin may: `Admin` over the namespace, the rule
    /// `SetVisibility` keeps, refused with the same sentence.
    async fn set_namespace_settings(
        &self,
        request: Request<SetNamespaceSettingsRequest>,
    ) -> Result<Response<SetNamespaceSettingsResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let name = request.name.trim().to_owned();
        if name.is_empty() {
            return Err(Status::invalid_argument("`name`: the namespace is required"));
        }
        let account = self.ctx.account_of_namespace(&name);
        if let Err(status) = self.ctx.require(
            &caller,
            Access::namespace(&name),
            GrantLevel::Admin,
            "Repos.SetNamespaceSettings",
        ) {
            if status.code() == tonic::Code::PermissionDenied {
                self.ctx.account_event(
                    account,
                    crate::enterprise::AccountAuditEvent::by(&caller, "visibility.namespace", name.clone())
                        .detail("only the account admin changes the namespace's settings")
                        .refused(),
                );
            }
            return Err(account_admin_only_for(status, &name, "the settings of account"));
        }
        if request.public.is_none() && request.events_public.is_none() {
            return Err(Status::invalid_argument(
                "nothing to set: name `public`, `events_public` or both",
            ));
        }
        let namespace =
            self.ctx
                .control
                .set_namespace_settings(&name, request.public, request.events_public)?;
        self.ctx.account_event(
            account,
            crate::enterprise::AccountAuditEvent::by(&caller, "visibility.namespace", name.clone()).detail(format!(
                "namespace {}, events {}",
                if namespace.public { "public" } else { "private" },
                if namespace.events_public { "public" } else { "private" }
            )),
        );
        tracing::info!(
            target: crate::auth::AUDIT_TARGET,
            event = "control.namespace_settings",
            principal = %caller.principal_id,
            namespace = %name,
            public = namespace.public,
            events_public = namespace.events_public,
            "the account admin changed the namespace's settings"
        );
        Ok(Response::new(SetNamespaceSettingsResponse {
            namespace: Some(namespace),
        }))
    }
}

/// A `PERMISSION_DENIED` from `Repos.SetVisibility`'s namespace check, said as
/// what it is: the ACCOUNT admin's decision. gunnar's own sentence (which
/// names the principal) is kept in front; every other refusal — a browse
/// replica, a narrowed principal, an unreadable identity table — passes
/// through untouched.
fn account_admin_only(status: Status, path: &gunnar_proto::RepoPath) -> Status {
    if status.code() != tonic::Code::PermissionDenied {
        return status;
    }
    Status::permission_denied(format!(
        "{}: changing the visibility of {}/{} is the account admin's decision (admin of account \
         {}); an admin grant on the repository alone does not reach it",
        status.message(),
        path.namespace,
        path.repo,
        path.namespace
    ))
}

/// [`account_admin_only`] for an account-wide decision (`what` names it).
pub(crate) fn account_admin_only_for(status: Status, account: &str, what: &str) -> Status {
    if status.code() != tonic::Code::PermissionDenied {
        return status;
    }
    Status::permission_denied(format!(
        "{}: {what} {account} are the account admin's decision (admin of account {account})",
        status.message()
    ))
}

/// **The one place a [`crate::seam::GcJob`] becomes wire bytes.**
///
/// The seam's job type and `gunnar.v1.GcJob` are deliberately separate — an
/// adapter implements the trait without linking the protos — so exactly one
/// function knows both, and it is this one. A second copy of this mapping is
/// how a state ends up meaning `UNSPECIFIED` on one path and `DONE` on another.
fn gc_job_out(job: crate::seam::GcJob) -> gunnar_proto::GcJob {
    gunnar_proto::GcJob {
        job_id: job.job_id,
        path: Some(job.path),
        state: match job.state {
            crate::seam::GcState::Queued => gunnar_proto::GcState::Queued,
            crate::seam::GcState::Running => gunnar_proto::GcState::Running,
            crate::seam::GcState::Done => gunnar_proto::GcState::Done,
            crate::seam::GcState::Failed => gunnar_proto::GcState::Failed,
        } as i32,
        queue_position: job.queue_position,
        bytes_reclaimed: job.bytes_reclaimed,
        error: job.error.unwrap_or_default(),
        enqueued_at_unix_ms: job.enqueued_at_unix_ms,
        finished_at_unix_ms: job.finished_at_unix_ms,
    }
}

/// **The level `Repos.Create` asks** (clans, 2026-10-05): `Admin` over the
/// namespace — the account admin's — except that a CLAN's member creates
/// PRIVATE repositories in the clan's namespace with the `Write` its
/// membership gives it. A repository born public is a publish, and that is
/// the clan admin's, as on an account. The decision who is in the clan is the
/// identity table's; this only picks which question to ask it.
fn create_level(
    ctx: &crate::service::ServiceContext,
    caller: &crate::auth::Caller,
    namespace: &str,
    visibility: i32,
) -> GrantLevel {
    if visibility == gunnar_proto::Visibility::Public as i32 {
        return GrantLevel::Admin;
    }
    let member = ctx
        .control
        .clans()
        .and_then(|clans| clans.clan_role(&caller.principal_id, namespace).ok().flatten())
        .is_some_and(|role| role == gunnar_proto::ClanRole::Member);
    if member {
        GrantLevel::Write
    } else {
        GrantLevel::Admin
    }
}