gunnar-grpc 2.0.0

//! The seven service implementations (the gRPC control plane).
//!
//! Each is a thin shell: validate the request, call the storage seam, map the
//! error, and, where the operation changes the world, publish an event
//! (the event stream). No service holds state of its own; all of it is behind
//! [`ControlPlane`](crate::seam::ControlPlane) or on the bus.
//!
//! **Every one of them authorises before it acts.** The first statement of
//! every RPC body is a [`ServiceContext::authorize`] (or
//! [`ServiceContext::caller`] followed by [`ServiceContext::require`], where
//! the scope is inside the message). There is no RPC in this crate that
//! reaches the storage seam without one, and
//! `tests/auth.rs::every_rpc_refuses_an_unauthenticated_caller` drives all 58
//! of them **with no interceptor in the way** so a handler that forgot goes
//! red rather than serving.
//!
//! `clippy::result_large_err` is allowed for the same reason it is allowed in
//! [`crate::status`]: `tonic::Status` is a large error type by design and every
//! RPC in the surface returns one. Boxing it would fight the framework for no
//! gain.
#![allow(clippy::result_large_err)]

use std::sync::Arc;
use std::time::Instant;

use gunnar_proto::GrantLevel;
use tonic::{Request, Status};

use crate::auth::{audit_allowed, audit_denied, Access, AuthMode, Caller};
use crate::bus::EventBus;
use crate::config::ServerConfig;
use crate::seam::ControlPlane;

pub mod access;
pub mod accounts;
pub mod admin;
#[cfg(feature = "metered")]
pub mod billing;
pub mod bootstrap;
/// Clans (2026-10-05): the person's own verbs, decided from the clan.
pub mod clans;
#[cfg(feature = "metered")]
pub mod entitlement;
/// The named sealed slot (2026-10-09): bytes under a name, its owner's alone.
#[cfg(feature = "metered")]
pub mod sealed;
/// The owner's enterprise list: the account audit log, the operator's aggregates and look, the name filter.
pub mod enterprise;
pub mod events;
pub mod forge;
/// Always compiled, and on purpose: the community appliance reports the same
/// seven meters and has the same three doors, so the table describing them is
/// not a metered thing. Only [`meters::caps_gunnar_cannot_honour`]'s CALLER is
/// (`Entitlement.Set` does not exist without `metered`).
pub mod meters;
pub mod mirror;
pub mod policy;
pub mod pulls;
pub mod refs;
pub mod repos;
pub mod sso;
pub mod usage;
pub mod usage_batch;
pub mod webhook;

/// **How much of a collection a caller may be shown.**
///
/// The answer [`ServiceContext::require_or_narrow`] gives where
/// [`ServiceContext::require`] would have given a `PERMISSION_DENIED`, and the
/// input [`ServiceContext::retain_readable`] filters on.
///
/// It is an enum with two variants rather than a `bool` because the two are not
/// degrees of the same thing: `Everything` means *the grant already covers the
/// scope, ask nothing further*, and `PublicSubset` means *ask about every row
/// individually*. A `bool` named `filter` would read as an optimisation that
/// could be skipped, and skipping it is a disclosure.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Listing {
    /// The caller holds `Read` over the scope that was asked about. Every row
    /// is served, private ones included. **This is the pre-existing behaviour
    /// and the only one an operator ever sees.**
    Everything,
    /// The caller holds no grant over the scope. Rather than being refused,
    /// each row is put to `AuthTable::may` on its own; what survives is what
    /// a public repository is readable without a grant, plus anything the caller owns or was
    /// granted at repository scope.
    PublicSubset,
}

/// What every service needs: the storage seam, the event bus, and the config
/// (so `Admin.Health` can report which addresses are actually bound).
#[derive(Clone)]
pub struct ServiceContext {
    pub control: Arc<dyn ControlPlane>,
    pub bus: Arc<EventBus>,
    pub config: ServerConfig,
    pub started: Instant,
    /// **The admission gate the transfer-shaped RPCs take a permit from**
    /// (one implementation), reached through the [`crate::TransferGate`] seam.
    ///
    /// `Some` when `gunnar-server` handed in the SSH listener's own
    /// `Arc<Admission>` (via [`ServiceContext::with_transfer_gate`]), so a
    /// `Repos.ArchiveOut` counts against the same server-wide limit, the same
    /// per-principal share and the same memory budget as a `git-upload-pack`
    /// over SSH or HTTP. `None` on a plaintext test router or a listener built
    /// without a gate: the transfer-shaped RPCs then run **ungated**, which is
    /// the exact ungated state the red half of the test drives — a green
    /// there proves nothing, so the gate has to be wired for the refusal to
    /// exist at all.
    pub transfer_gate: Option<Arc<dyn crate::TransferGate>>,
    /// **Is this box a browse replica**? The server's
    /// `ServerStores::browse()` flag, shared by `Arc`, flipped by the
    /// replication slot. `None` (a bare context, a test) is never a browse
    /// box. While `true`, [`ServiceContext::require`] refuses every mutator
    /// by name — see [`BROWSE_STILL_ANSWERS`].
    pub browse: Option<Arc<std::sync::atomic::AtomicBool>>,
    /// **The durable audit trail** (GUN-044), or `None` for the `tracing`
    /// record alone.
    ///
    /// `Some` when `gunnar-server` handed in the hash-chained ledger it keeps
    /// beside the rest of the control state (via
    /// [`ServiceContext::with_audit_ledger`]), so every decision the audit
    /// functions record is also written somewhere a removed line shows up as a
    /// broken chain. `None` on a test router: the `tracing` record is unchanged
    /// either way, which is what makes the ledger additive rather than a second
    /// trail that can disagree with the first.
    pub audit: Option<Arc<dyn crate::auth::AuditLedger>>,
    /// **The method-blind credentials this listener has been shown**
    /// ([`crate::auth::UnboundCredentialLog`]).
    ///
    /// Minted here rather than inside [`crate::auth::AuthGate`] because the
    /// gate writes it and `Admin.Health` reads it, and those are two different
    /// objects on one listener: `router_with_trust` hands this same `Arc` to
    /// the gate it builds. A context whose gate was built elsewhere reports an
    /// empty log, which is the honest answer — it has seen nothing — and never
    /// a wrong one.
    pub unbound: Arc<crate::auth::UnboundCredentialLog>,
    /// **The dual heartbeat's primary half** ([`crate::heartbeat::Registry`]):
    /// every `Mirror.LogHead` carrying a `TwinReport` is recorded here and
    /// `Admin.Health` reads it back. One per listener, like `unbound`.
    pub heartbeats: Arc<crate::heartbeat::Registry>,
    /// **The machine principals that call this listener**, newest call each
    /// ([`crate::topology::CallerLog`]): written by [`ServiceContext::caller`],
    /// read by `Admin.Topology` — the consoles and the front's other keys are
    /// boxes of the estate, and their calls are the evidence that they are up.
    pub callers: Arc<crate::topology::CallerLog>,
    /// **The spent `(fingerprint, nonce)` pairs of this listener**, one guard
    /// for both doors that check a credential: `router_with_trust` hands this
    /// `Arc` to the gate, and `Bootstrap.Restore` checks its own credential
    /// against it ([`crate::auth::authenticate_restore`]) — so a header spent
    /// at one door is spent at the other.
    pub replay: Arc<crate::auth::ReplayGuard>,
    /// **The web challenges this listener has issued** (`Access.WebChallenge`,
    /// Stage L) and not yet seen answered: single use, five minutes.
    pub web_challenges: Arc<crate::webauthn::Challenges>,
    /// **The web sessions this listener enrolled, and whether their login
    /// verified the user** (A34-10): the step-up the SSO write verbs demand.
    pub web_sessions: Arc<WebSessions>,
    /// **The machine enrolment requests waiting for an admin**
    /// (`Access.RequestMachineEnrolment`, 2026-10-05): fifteen minutes each,
    /// in this listener's memory, like the challenges.
    pub machine_enrolments: Arc<crate::machine_enrolment::Requests>,
    /// **Company SSO's memory on this listener** ([`crate::sso::SsoState`]):
    /// the sign-ins begun, the second-factor tickets, the SAML assertions
    /// consumed, the IdPs' keys, the rate windows — and the egress
    /// `gunnar-server` hands in.
    pub sso: Arc<crate::sso::SsoState>,
    /// **This estate SELLS room** (the feed's `sells_room`, LIVE; [`crate::selling`]): a
    /// person's own account with no fact holds no room here until its fact lands. The same
    /// `Arc` the entitlement slot's push gate reads.
    pub sells_room: Arc<std::sync::atomic::AtomicBool>,
    /// **The runtime the handlers run on** (R5.1-G), or `None` to run them on
    /// the runtime that serves the connections.
    ///
    /// The control-plane handlers are `async fn`s that make SYNCHRONOUS calls
    /// into the store: a `StorageInfo` disk walk, a txlog fsync, a snapshot.
    /// Each one holds the worker it runs on until it returns. When they ran on
    /// the runtime that also drives the HTTP/2 connections, enough of them at
    /// once held every worker, nothing polled the connection, a client's
    /// keepalive PING went unanswered, and the client killed the connection —
    /// failing every call in flight on it, the slow ones and the quick ones
    /// alike (`KeepAliveTimedOut` on the console, an `Internal` stream reset
    /// on monetize's `Entitlement.Set`). `gunnar-server` hands in a runtime of
    /// its own here ([`ServiceContext::with_handler_runtime`]), so the
    /// connections are driven by workers no handler can hold.
    pub handlers: Option<tokio::runtime::Handle>,
}

/// **What a browse replica's control plane still answers at Write/Admin
/// level.** Every other method that needs more than `Read` is refused with
/// `FAILED_PRECONDITION` while the box is a browse replica: it holds only
/// public repositories, serves strangers, and must never be promoted or
/// reshaped by anyone — a stolen admin key on the public-facing box buys
/// nothing but a read. The list is the Admin-level READS (listings the
/// operator uses to check the box), plus the two writes the design needs:
///
/// * `Mirror.SetRole` — so the box can be narrowed to `STANDBY`; the handler
///   refuses `PRIMARY` on a browse box itself.
/// * `Repos.Delete` — **from a `Machine` only**: the primary's PublicOnly
///   replication deletes a repository here when it goes Private. A person's
///   admin key cannot delete on a browse box; the primary's deploy key can.
/// * `Repos.Create` — **from a `Machine` only**, the delete's other half
///   (2026-09-22): a repository the browse box forgot comes back by a push,
///   and a push-create is sha1 by construction (`push_create_spec`), so a
///   sha256 repository could never return on its own. The primary's
///   replicator pre-creates it here, in the repository's own object format,
///   and then pushes. Same key, same standing as the delete.
pub const BROWSE_STILL_ANSWERS: &[&str] = &[
    "Access.ListGrants",
    "Access.ListGroupMembers",
    "Access.ListGroupMembershipEvents",
    "Access.ListKeys",
    "Access.ListPrincipals",
    "Mirror.ReplicationStatus",
    // The twin (2026-09-15): a twin is frozen like a browse replica, and its log is what a
    // restore reads. A read of the log mutates nothing; the level is Admin because the
    // log carries every sealed secret, not because it writes.
    "Mirror.LogFetch",
    "Mirror.SnapshotFetch",
    "Webhooks.ListWebhooks",
    "Mirror.SetRole",
    "Repos.Delete",
    "Repos.Create",
];

/// **What a TWIN answers beyond [`BROWSE_STILL_ANSWERS`]**: its OWN data-set growth (owner
/// ruling 2026-09-22, T21 — the twin, built from the same medium, grows EXACTLY like main and
/// is never restarted for it). A twin is frozen like a browse replica, so without this the grow
/// verbs were refused there and the only way left to grow it was stop → resize → start. Passed
/// only on a twin ([`ControlPlane::is_twin`]), never on a public browse box; the go-ahead is
/// still checked by the grow machine as on main.
pub const TWIN_STILL_ANSWERS: &[&str] = &["Admin.GrowStart", "Admin.GrowResume", "Admin.GrowStatus"];

/// **What a principal is to the account it acts in** — see
/// [`ServiceContext::standing`]. `role` is `admin`, `individual`, `employee`,
/// or empty with no account.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Standing {
    pub account: String,
    pub role: &'static str,
    /// The account admin's display name: whom a member asks for more room.
    pub admin: String,
}

impl ServiceContext {
    pub fn new(control: Arc<dyn ControlPlane>, bus: Arc<EventBus>, config: ServerConfig) -> Self {
        Self {
            control,
            bus,
            config,
            started: Instant::now(),
            transfer_gate: None,
            browse: None,
            audit: None,
            unbound: Arc::new(crate::auth::UnboundCredentialLog::default()),
            heartbeats: Arc::new(crate::heartbeat::Registry::default()),
            callers: Arc::new(crate::topology::CallerLog::default()),
            replay: Arc::new(crate::auth::ReplayGuard::default()),
            web_challenges: Arc::new(crate::webauthn::Challenges::default()),
            web_sessions: Arc::new(WebSessions::default()),
            machine_enrolments: Arc::new(crate::machine_enrolment::Requests::default()),
            sso: Arc::new(crate::sso::SsoState::default()),
            sells_room: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            handlers: None,
        }
    }

    /// The process's one "this estate sells room" flag ([`ServiceContext::sells_room`]).
    pub fn with_sells_room(mut self, flag: Arc<std::sync::atomic::AtomicBool>) -> Self {
        self.sells_room = flag;
        self
    }

    /// Run every control-plane handler on `handlers` rather than on the
    /// runtime that drives the HTTP/2 connections — see
    /// [`ServiceContext::handlers`].
    pub fn with_handler_runtime(mut self, handlers: tokio::runtime::Handle) -> Self {
        self.handlers = Some(handlers);
        self
    }

    /// Write every decision to `ledger` as well as to `tracing`.
    pub fn with_audit_ledger(mut self, ledger: Arc<dyn crate::auth::AuditLedger>) -> Self {
        self.audit = Some(ledger);
        self
    }

    /// Use `registry` for the heartbeat (a test shortens the bounds).
    pub fn with_heartbeats(mut self, registry: Arc<crate::heartbeat::Registry>) -> Self {
        self.heartbeats = registry;
        self
    }

    /// **Record a change BEFORE it is applied** (RA2-9), in the hash-chained ledger and on the
    /// audit target: what `method` (already allowed by [`Self::require`]) is about to change.
    pub fn audit_change(&self, caller: &Caller, access: Access, level: GrantLevel, method: &'static str, what: &str) {
        crate::auth::audit_change(method, caller, &access, level, self.config.auth, what, self.ledger());
    }

    /// [`ServiceContext::audit_change`] for a change that is not applied unless its record is
    /// kept (RA4-8): `Err` is the `audit.unavailable` refusal.
    pub fn audit_change_durable(&self, caller: &Caller, access: Access, level: GrantLevel, method: &'static str, what: &str) -> Result<(), Status> {
        crate::auth::audit_change_durable(method, caller, &access, level, self.config.auth, what, self.ledger())
    }

    /// The ledger, as the audit functions take it.
    fn ledger(&self) -> Option<&dyn crate::auth::AuditLedger> {
        self.audit.as_deref()
    }

    /// Share the browse flag in (`ServerStores::browse()`).
    pub fn with_browse_flag(mut self, browse: Arc<std::sync::atomic::AtomicBool>) -> Self {
        self.browse = Some(browse);
        self
    }

    /// Is this box a browse replica right now?
    pub fn is_browse(&self) -> bool {
        self.browse
            .as_ref()
            .is_some_and(|flag| flag.load(std::sync::atomic::Ordering::SeqCst))
    }

    /// **The browse replica's mutator refusal**,
    /// asked before the grant: a `Read` always passes; anything higher passes
    /// only if [`BROWSE_STILL_ANSWERS`] names the method — and `Repos.Delete`
    /// and `Repos.Create` only from a `Machine`. Before the grant so the refusal is the same for
    /// everyone, admin or not: the box's answer, not the caller's standing.
    fn refuse_on_browse(
        &self,
        caller: &Caller,
        level: GrantLevel,
        method: &'static str,
    ) -> Result<(), Status> {
        if level == GrantLevel::Read || !self.is_browse() {
            return Ok(());
        }
        let allowed = match method {
            "Repos.Delete" | "Repos.Create" => caller.machine,
            other => BROWSE_STILL_ANSWERS.contains(&other) || (TWIN_STILL_ANSWERS.contains(&other) && self.control.is_twin()),
        };
        if allowed {
            return Ok(());
        }
        tracing::info!(
            target: crate::auth::AUDIT_TARGET,
            event = "control.browse_refused",
            method,
            principal = %caller.principal_id,
            machine = caller.machine,
            "a browse replica refused a control-plane mutator"
        );
        Err(Status::failed_precondition(format!(
            "this box is a browse replica: it holds public repositories read-only and cannot be \
             promoted or changed here — {method} is refused; run it against the primary"
        )))
    }

    /// Route this context's transfer-shaped RPCs (`Repos.ArchiveOut`) through an
    /// **existing** admission gate — the SSH listener's own (one implementation).
    ///
    /// `gunnar-server` passes the same `Arc<Admission>` the SSH and HTTP fronts
    /// hold, wrapped in its [`crate::TransferGate`] adapter, so all three front
    /// doors meter against one ceiling. Without this call the control plane's
    /// archive stream is the one transfer on the box with no backpressure.
    pub fn with_transfer_gate(mut self, gate: Arc<dyn crate::TransferGate>) -> Self {
        self.transfer_gate = Some(gate);
        self
    }

    /// The verified caller [`crate::auth::AuthGate`] attached to this request.
    ///
    /// **This is where deny-by-default lives.** No extension means no verified
    /// credential, whether because the caller sent none or because someone
    /// mounted these services without the gate in front of them. The answer to
    /// that is `UNAUTHENTICATED`, never "assume an admin".
    pub fn caller<T>(&self, request: &Request<T>) -> Result<Caller, Status> {
        if let Some(caller) = request.extensions().get::<Caller>() {
            // A MACHINE's call is evidence that its box is up (`Admin.Topology`);
            // a person's laptop is not a box of the estate, so people are not
            // recorded.
            if caller.machine && caller.authenticated {
                let address = request.remote_addr().map(|a| a.to_string()).unwrap_or_default();
                self.callers.note(&caller.principal_id, &address, estate_clock::now_unix_ms(), false);
            }
            return Ok(caller.clone());
        }
        if self.config.auth == AuthMode::AllowAllInsecure {
            return Ok(Caller {
                principal_id: Caller::INSECURE_PRINCIPAL.to_owned(),
                fingerprint: String::new(),
                machine: false,
                authenticated: false,
            });
        }
        Err(crate::auth::AuthFailure::Missing.status())
    }

    /// **Is this call about the caller's OWN principal?** (2026-10-04, the
    /// self-scope.) True only for a caller the gate authenticated by one of
    /// `target`'s own keys — never for the insecure escape hatch's stand-in,
    /// never for an empty target.
    pub fn is_self(&self, caller: &Caller, target: &str) -> bool {
        caller.authenticated
            && !caller.principal_id.is_empty()
            && caller.principal_id == target
            && caller.principal_id != Caller::INSECURE_PRINCIPAL
    }

    /// **The self-scope check, and its audit record** (owner, 2026-10-04: the
    /// console calls gunnar AS THE PERSON). A principal may manage its OWN keys
    /// — add one, list them, revoke one — without any grant: the call is
    /// already that principal's own signed word, which is exactly the consent
    /// `key_attestation::Requirement::CallerIsTarget` names. What still
    /// applies is everything that is not a grant: a browse replica refuses the
    /// mutators it refuses for everybody, and a NARROWED principal acts only
    /// through the methods on its list.
    ///
    /// No account gate is consulted, on purpose: a gate shuts the ACCOUNT's
    /// data, and a key is not data — a locked-out person who adds a key reaches
    /// nothing through it that the gate does not still refuse.
    pub fn require_self(
        &self,
        caller: &Caller,
        level: GrantLevel,
        method: &'static str,
    ) -> Result<(), Status> {
        self.refuse_on_browse(caller, level, method)?;
        if self.config.auth != AuthMode::AllowAllInsecure {
            match self.control.narrowed_methods(&caller.principal_id) {
                Ok(Some(allowed)) if !allowed.iter().any(|m| m == method) => {
                    audit_denied(
                        method,
                        caller,
                        &Access::Server,
                        level,
                        self.config.auth,
                        "outside its narrowed grant",
                        self.ledger(),
                    );
                    return Err(Status::permission_denied(format!(
                        "principal {} is narrowed to [{}]: {method} is outside its narrowed grant",
                        caller.principal_id,
                        allowed.join(", ")
                    )));
                }
                Ok(_) => {}
                Err(err) => {
                    return Err(Status::unavailable(format!(
                        "the identity table could not be read: {err}"
                    )))
                }
            }
        }
        crate::auth::audit_self(method, caller, self.config.auth, self.ledger());
        Ok(())
    }

    /// **The decision [`ServiceContext::require`] makes, asked without acting**
    /// (`Access.Check`, 2026-10-04). Same order — a browse replica, a NARROWED
    /// principal's list, the insecure escape hatch, then
    /// [`crate::seam::Identity::may`] (grants, groups, account ownership, the
    /// account gate) — and the same sentence for a refusal. `Ok(None)` is
    /// allowed, `Ok(Some(why))` is not; a storage fault is an error, never a
    /// `no`.
    ///
    /// **Nothing is written to the audit trail.** A "may I?" asked before a
    /// button is drawn is not a call that was refused, and a `decision="deny"`
    /// WARN line per page view would drown the refusals an operator greps for.
    /// The verb, when it is called, records itself as it always did.
    ///
    /// A browse replica answers nothing above `Read` but its own listings, and
    /// a check does not know which verb is coming, so above `Read` it says no
    /// there; a narrowed principal's grants are never consulted, so it says no
    /// to every scope.
    pub fn check(
        &self,
        caller: &Caller,
        access: Access,
        level: GrantLevel,
    ) -> Result<Option<String>, Status> {
        if level != GrantLevel::Read && self.is_browse() {
            return Ok(Some(
                "this box is a browse replica: it holds public repositories read-only, and \
                 nothing above Read is done here; run it against the primary"
                    .to_owned(),
            ));
        }
        if self.config.auth == AuthMode::AllowAllInsecure {
            return Ok(None);
        }
        match self.control.narrowed_methods(&caller.principal_id) {
            Ok(Some(allowed)) => {
                return Ok(Some(format!(
                    "principal {} is narrowed to [{}]: its grants are not consulted",
                    caller.principal_id,
                    allowed.join(", ")
                )))
            }
            Ok(None) => {}
            Err(err) => {
                return Err(Status::unavailable(format!(
                    "the identity table could not be read: {err}"
                )))
            }
        }
        match self.control.may(&caller.principal_id, &access, level) {
            Ok(true) => Ok(None),
            Ok(false) => Ok(Some(format!(
                "principal {} has no {} grant on {access}",
                caller.principal_id,
                crate::auth::level_name(level)
            ))),
            Err(err) => Err(Status::unavailable(format!(
                "the identity table could not be read: {err}"
            ))),
        }
    }

    /// **The grant check, and the audit record.** One call site shape for all
    /// 37 RPCs (one implementation), so the decision and the trail cannot drift apart.
    ///
    /// The decision itself is [`crate::seam::Identity::may`], which is
    /// `AuthTable::may` in the real server. It does not compute it; it asks,
    /// records the answer, and turns a `false` into `PERMISSION_DENIED`.
    pub fn require(
        &self,
        caller: &Caller,
        access: Access,
        level: GrantLevel,
        method: &'static str,
    ) -> Result<(), Status> {
        self.refuse_on_browse(caller, level, method)?;
        // A NARROWED principal (f-principal-reach): its list, and nothing else
        // — whatever grants it might hold are not consulted.
        if self.config.auth != AuthMode::AllowAllInsecure {
            match self.control.narrowed_methods(&caller.principal_id) {
                Ok(Some(allowed)) => {
                    if allowed.iter().any(|m| m == method) {
                        audit_allowed(method, caller, &access, level, self.config.auth, self.ledger());
                        return Ok(());
                    }
                    audit_denied(method, caller, &access, level, self.config.auth, "outside its narrowed grant", self.ledger());
                    return Err(Status::permission_denied(format!(
                        "principal {} is narrowed to [{}]: {method} is outside its narrowed grant",
                        caller.principal_id,
                        allowed.join(", ")
                    )));
                }
                Ok(None) => {}
                Err(err) => {
                    audit_denied(method, caller, &access, level, self.config.auth, "grant lookup failed", self.ledger());
                    return Err(Status::unavailable(format!("the identity table could not be read: {err}")));
                }
            }
        }
        if self.config.auth == AuthMode::AllowAllInsecure {
            audit_allowed(
                method,
                caller,
                &access,
                level,
                self.config.auth,
                self.ledger(),
            );
            return Ok(());
        }
        match self.control.may(&caller.principal_id, &access, level) {
            Ok(true) => {
                audit_allowed(
                    method,
                    caller,
                    &access,
                    level,
                    self.config.auth,
                    self.ledger(),
                );
                Ok(())
            }
            Ok(false) => {
                audit_denied(
                    method,
                    caller,
                    &access,
                    level,
                    self.config.auth,
                    "no grant",
                    self.ledger(),
                );
                Err(Status::permission_denied(format!(
                    "principal {} has no {} grant on {access}",
                    caller.principal_id,
                    crate::auth::level_name(level)
                )))
            }
            // A storage fault is not a denial. Reporting it as one would turn
            // an outage into a silent, confusing permission problem (the ACL model: `may`
            // returns `Result<bool>` for exactly this reason).
            Err(err) => {
                audit_denied(
                    method,
                    caller,
                    &access,
                    level,
                    self.config.auth,
                    "grant lookup failed",
                    self.ledger(),
                );
                Err(Status::unavailable(format!(
                    "the identity table could not be read: {err}"
                )))
            }
        }
    }

    /// **The narrowest scope that CAN hold a grant**: the repository, and the
    /// namespace when the repository cannot hold one.
    ///
    /// MEASURED on the appliance 2026-08-30 (`v1.1.3-227-g6a2367a`, requests
    /// `175daccc0000000b` and `…0e`):
    ///
    /// ```text
    /// repo delete alice/linux --purge --yes
    ///   -> principal alice has no admin grant on repo:alice/linux
    /// ```
    ///
    /// …on a store that was on the disk, holding tens of gigabytes, with no
    /// catalog row. [`crate::seam::Identity::may`] answers `false` for a grant
    /// on a repository that is not in the catalog — deliberately, so nobody
    /// can probe the ACL of repositories they cannot see — so a repo-scoped
    /// check on a repository the catalog has lost refuses **everyone**,
    /// gunnar-admin included. The verb that reclaims a lost repository was
    /// gated on the record whose absence is the whole problem.
    ///
    /// The soft-delete work hit exactly this on `Repos.Undelete` and `Repos.Purge` and
    /// answered it by scoping them to the namespace. This is that same answer,
    /// written once for every verb that has to work on a repository the
    /// catalog may not hold (one implementation) rather than a second rule beside it.
    ///
    /// **It is not a widening.** `AuthTable::may` already takes the maximum
    /// over every grant whose scope `covers` the repository, and
    /// `Scope::Namespace(n).covers(repo)` is `n == repo.namespace`; account
    /// ownership is the same one rule on both sides. So for a repository that
    /// IS in the catalog, `may_namespace(ns) => may(repo)` and the second
    /// question can only ever be answered `true` where the first already was.
    /// What it adds is the case where the first question has no repository to
    /// ask about — and there it is the *narrower* answer, because a
    /// repo-scoped grant cannot be evaluated against a row that is gone.
    ///
    /// The refusal names **the repository**, never the namespace, so which of
    /// the two questions was asked is not an existence oracle for a caller
    /// with no grant at all.
    pub fn require_repo_or_namespace(
        &self,
        caller: &Caller,
        path: &gunnar_proto::RepoPath,
        level: GrantLevel,
        method: &'static str,
    ) -> Result<(), Status> {
        let asked = Access::repo(path);
        self.refuse_on_browse(caller, level, method)?;
        if self.config.auth == AuthMode::AllowAllInsecure {
            audit_allowed(
                method,
                caller,
                &asked,
                level,
                self.config.auth,
                self.ledger(),
            );
            return Ok(());
        }
        let unavailable = |err: crate::seam::ControlError| {
            audit_denied(
                method,
                caller,
                &asked,
                level,
                self.config.auth,
                "grant lookup failed",
                self.ledger(),
            );
            Status::unavailable(format!("the identity table could not be read: {err}"))
        };
        match self.control.may(&caller.principal_id, &asked, level) {
            Ok(true) => {
                audit_allowed(
                    method,
                    caller,
                    &asked,
                    level,
                    self.config.auth,
                    self.ledger(),
                );
                return Ok(());
            }
            Ok(false) => {}
            Err(err) => return Err(unavailable(err)),
        }
        let wider = Access::namespace(&path.namespace);
        match self.control.may(&caller.principal_id, &wider, level) {
            Ok(true) => {
                audit_allowed(
                    method,
                    caller,
                    &wider,
                    level,
                    self.config.auth,
                    self.ledger(),
                );
                Ok(())
            }
            Ok(false) => {
                audit_denied(
                    method,
                    caller,
                    &asked,
                    level,
                    self.config.auth,
                    "no grant",
                    self.ledger(),
                );
                Err(Status::permission_denied(format!(
                    "principal {} has no {} grant on {asked}",
                    caller.principal_id,
                    crate::auth::level_name(level)
                )))
            }
            Err(err) => Err(unavailable(err)),
        }
    }

    /// **The grant question a LISTING asks**: not "may you?" but "how much of
    /// it may you see?"
    ///
    /// [`ServiceContext::require`] has exactly two answers, and for a listing
    /// the wrong one is the refusal. A caller with no grant on a namespace is
    /// not a caller who may not *ask* what is in it — since anonymous read it is a caller
    /// who may see the [`Visibility::Public`](gunnar_core) rows and no others.
    /// Refusing outright is what made `Repo.visibility` undiscoverable: a
    /// stranger could clone `nordisk/gunnar` if somebody told them the path,
    /// and had no way to learn the path existed.
    ///
    /// So this returns a [`Listing`] rather than `()`, and the caller feeds it
    /// to [`ServiceContext::retain_readable`].
    ///
    /// # Why the scope check still runs first
    ///
    /// [`Listing::Everything`] is the *fast* answer and the *unchanged* one: a
    /// caller holding `Read` over the whole scope gets the whole collection in
    /// one grant lookup, exactly as before this existed (one implementation). The narrowing
    /// is reached only where a refusal would have been, so no caller that could
    /// see a row yesterday stops seeing it today, and no caller pays the
    /// per-row cost unless it is the only thing standing between them and a
    /// `PERMISSION_DENIED`.
    ///
    /// # Why a storage fault is still a refusal
    ///
    /// An `Err` from the identity table is `UNAVAILABLE`, never a narrowing.
    /// Degrading an unreadable ACL into "show them the public subset" would
    /// make an outage look like a permissions change, and — worse — would make
    /// the *safest-looking* branch the one that runs when the check is broken.
    pub fn require_or_narrow(
        &self,
        caller: &Caller,
        access: Access,
        method: &'static str,
    ) -> Result<Listing, Status> {
        let level = GrantLevel::Read;
        self.refuse_on_browse(caller, level, method)?;
        if self.config.auth == AuthMode::AllowAllInsecure {
            audit_allowed(
                method,
                caller,
                &access,
                level,
                self.config.auth,
                self.ledger(),
            );
            return Ok(Listing::Everything);
        }
        match self.control.may(&caller.principal_id, &access, level) {
            Ok(true) => {
                audit_allowed(
                    method,
                    caller,
                    &access,
                    level,
                    self.config.auth,
                    self.ledger(),
                );
                Ok(Listing::Everything)
            }
            // NOT `audit_denied`: nothing was denied. A record saying `deny`
            // here would put a warning in the log for every visitor reading the
            // front page, and would hide the real denials an operator greps
            // for. Its own decision word instead, so "who browsed anonymously"
            // and "who was refused" stay different questions.
            Ok(false) => {
                crate::auth::audit_narrowed(
                    method,
                    caller,
                    &access,
                    level,
                    self.config.auth,
                    self.ledger(),
                );
                Ok(Listing::PublicSubset)
            }
            Err(err) => {
                audit_denied(
                    method,
                    caller,
                    &access,
                    level,
                    self.config.auth,
                    "grant lookup failed",
                    self.ledger(),
                );
                Err(Status::unavailable(format!(
                    "the identity table could not be read: {err}"
                )))
            }
        }
    }

    /// Drop the rows `caller` may not read, one repository at a time.
    ///
    /// # Why this asks `may` again instead of reading `visibility`
    ///
    /// Because `AuthTable::may` is **the** ACL (one implementation), and "public" is only one
    /// of the four things it decides. A filter here that kept the rows whose
    /// `Repo.visibility` field said `PUBLIC` would be a second grant decision
    /// living beside the first, free to disagree with it — and the one-sandbox rule records
    /// which one wins when two disagree. Concretely, that filter would be wrong
    /// in three directions at once:
    ///
    /// * it would **hide a caller's own account** from them. A principal is
    ///   unconditionally admin of the namespace named after it, with no grant
    ///   anywhere, so `vetra` listing repositories must see `vetra/secrets`.
    /// * it would **ignore the account gate**. `may` reads
    ///   [`crate::seam::ControlPlane`]'s gate *above* every other rule, in both
    ///   directions, precisely so a stranger cannot pull bytes out of an
    ///   account a purge is walking. A visibility test would put that account's
    ///   public repositories straight back on the list.
    /// * it would **miss a repo-scoped grant**, so a caller granted `Read` on
    ///   one private repository would not find it in a listing that named it.
    ///
    /// All three are arms of `tests/public_listing.rs` in `gunnar-cli`, driven
    /// against the real `AuthTable` rather than a double, because a double that
    /// answered `may` out of a grant map would be green on every one of them.
    ///
    /// # The cost, and who pays it
    ///
    /// One `may` per row, and `AuthTable::may` reads the identity document each
    /// time. Only [`Listing::PublicSubset`] pays it — a caller with a grant
    /// returns on the first line — so the per-row cost falls exactly on the
    /// grantless caller reading a front page, and never on an operator listing
    /// a large server.
    pub fn retain_readable(
        &self,
        caller: &Caller,
        repos: Vec<gunnar_proto::Repo>,
        listing: Listing,
    ) -> Result<Vec<gunnar_proto::Repo>, Status> {
        let mut readable = Vec::new();
        let mut closed_ns: std::collections::HashMap<String, bool> = std::collections::HashMap::new();
        // ★ The rows that need the ACL, asked about TOGETHER (round 5.3.1,
        // test5 2026-10-08: one `may` per row read the catalog and the identity
        // table per row; at 160 rps of the login page's list main's
        // `Repos.List` p50 was 2.7 s on a half-idle box).
        let mut ask: Vec<(usize, gunnar_proto::RepoPath)> = Vec::new();
        let mut slots: Vec<Option<gunnar_proto::Repo>> = Vec::new();
        for repo in repos {
            let Some(path) = repo.path.as_ref() else {
                // A catalogue row with no path cannot be authorised, so it is
                // not shown. Dropping it is the only safe reading of a row this
                // code cannot ask a question about.
                continue;
            };
            // **Everything, but a customer's account** (2026-10-05): a row
            // inside an account closed to the caller's server grant is asked
            // about on its own, like any outsider's — a public one is shown.
            if listing == Listing::Everything {
                let closed = match closed_ns.get(&path.namespace) {
                    Some(known) => *known,
                    None => {
                        let known = self.shielded(caller, &path.namespace)?;
                        closed_ns.insert(path.namespace.clone(), known);
                        known
                    }
                };
                if !closed {
                    slots.push(Some(repo));
                    continue;
                }
            }
            ask.push((slots.len(), path.clone()));
            slots.push(Some(repo));
        }
        if !ask.is_empty() {
            let paths: Vec<gunnar_proto::RepoPath> = ask.iter().map(|(_, p)| p.clone()).collect();
            let answers = self
                .control
                .may_each_repo(&caller.principal_id, &paths, GrantLevel::Read)
                .map_err(|err| Status::unavailable(format!("the identity table could not be read: {err}")))?;
            for ((slot, _), may) in ask.iter().zip(answers) {
                if !may {
                    slots[*slot] = None;
                }
            }
        }
        readable.extend(slots.into_iter().flatten());
        Ok(readable)
    }

    /// [`ServiceContext::caller`] then [`ServiceContext::require`], for the RPCs
    /// whose scope is known before the message is parsed.
    pub fn authorize<T>(
        &self,
        request: &Request<T>,
        access: Access,
        level: GrantLevel,
        method: &'static str,
    ) -> Result<Caller, Status> {
        let caller = self.caller(request)?;
        self.require(&caller, access, level, method)?;
        Ok(caller)
    }

    // ── The owner's enterprise list (2026-10-05) ───────────────────────────

    /// **Write `event` into `account`'s audit log** — best effort: a log
    /// that cannot be written is one WARN line, never a refused RPC (the
    /// control plane's own trail makes the same promise, GUN-044).
    pub fn account_event(&self, account: Option<String>, event: crate::enterprise::AccountAuditEvent) {
        let Some(account) = account else {
            return;
        };
        let Some(plane) = self.control.enterprise() else {
            return;
        };
        if let Err(err) = plane.account_audit_record(&account, &event) {
            tracing::warn!(
                target: crate::auth::AUDIT_TARGET,
                event = "account_audit.not_written",
                account = %account,
                kind = %event.kind,
                reason = %err,
                "an event could not be written to its account's audit log"
            );
        }
    }

    /// **The enterprise half of the plane**, or the refusal a bare plane gives.
    pub fn enterprise(&self) -> Result<&dyn crate::enterprise::EnterprisePlane, Status> {
        self.control.enterprise().ok_or_else(|| crate::enterprise::no_plane().into())
    }

    /// The account `namespace` belongs to (a person's), if any.
    pub fn account_of_namespace(&self, namespace: &str) -> Option<String> {
        self.control.enterprise()?.account_of_namespace(namespace).ok().flatten()
    }

    /// **The account that enrolled `principal_id` as a MEMBER**, or `None`
    /// (2026-10-07, the members model). A plane that keeps no employers
    /// answers `None`: nobody is anybody's member there.
    pub fn employer_of(&self, principal_id: &str) -> Result<Option<String>, Status> {
        match self.control.employer_of(principal_id) {
            Ok(employer) => Ok(employer),
            Err(crate::seam::ControlError::Unimplemented(_)) => Ok(None),
            Err(err) => Err(err.into()),
        }
    }

    /// **What `principal_id` is to the account it acts in** (2026-10-07, the
    /// members model: a customer is an ACCOUNT that pays; its members use its
    /// space and never buy) — `Access.WhoAmI`'s `account` / `account_role` /
    /// `account_admin`, read off the identity table and nothing else:
    ///
    /// * a MEMBER (an employer row): the employer's account, `employee`;
    /// * a PERSON whose own namespace exists: that account — `admin` when it
    ///   has enrolled members or holds seats, `individual` otherwise;
    /// * anything else (a machine, a deploy key, a group, a principal with no
    ///   namespace): none — it speaks for no account and buys nothing.
    pub fn standing(&self, principal_id: &str) -> Result<Standing, Status> {
        if let Some(account) = self.employer_of(principal_id)? {
            let admin = self.display_name(&account)?;
            // **An APPOINTED account admin** (owner 2026-10-08): a member the account admin gave
            // Admin over the account's namespace decides for its repositories (publish, encrypt)
            // and reads its audit log — the rule SetVisibility keeps — and never buys
            // (`Billing.MintActorTicket` refuses every member, this one too).
            let appointed = match self.control.may(principal_id, &Access::namespace(&account), gunnar_proto::GrantLevel::Admin) {
                Ok(yes) => yes,
                Err(crate::seam::ControlError::Unimplemented(_)) => false,
                Err(err) => return Err(err.into()),
            };
            return Ok(Standing { account, role: if appointed { "admin-appointed" } else { "employee" }, admin });
        }
        let person = self
            .control
            .principal(principal_id)?
            .is_some_and(|p| p.kind == gunnar_proto::PrincipalKind::User as i32);
        if !person || self.control.namespace(principal_id)?.is_none() {
            return Ok(Standing::default());
        }
        // Seats first (one key), then "has it enrolled anyone" — never the whole listing of its
        // members: this runs on every `Access.WhoAmI` (wiring tracer B 2026-10-08).
        let (seats, _) = self.control.seats(principal_id)?;
        let admin = seats > 0
            || match self.control.has_employees(principal_id) {
                Ok(any) => any,
                Err(crate::seam::ControlError::Unimplemented(_)) => false,
                Err(err) => return Err(err.into()),
            };
        Ok(Standing {
            account: principal_id.to_owned(),
            role: if admin { "admin" } else { "individual" },
            admin: self.display_name(principal_id)?,
        })
    }

    /// A principal's display name, its id when it has none.
    fn display_name(&self, principal_id: &str) -> Result<String, Status> {
        Ok(self
            .control
            .principal(principal_id)?
            .map(|p| p.name)
            .filter(|name| !name.trim().is_empty())
            .unwrap_or_else(|| principal_id.to_owned()))
    }

    /// The account `principal_id` belongs to (its employer, or its own).
    pub fn account_of_principal(&self, principal_id: &str) -> Option<String> {
        self.control.enterprise()?.account_of_principal(principal_id).ok().flatten()
    }

    /// **Is `namespace` closed to the caller's server-scope grants?** Asked
    /// per row by every server-wide listing, so an operator's listing shows
    /// what is not inside a customer's account and nothing that is.
    pub fn shielded(&self, caller: &Caller, namespace: &str) -> Result<bool, Status> {
        if self.config.auth == AuthMode::AllowAllInsecure {
            return Ok(false);
        }
        let Some(plane) = self.control.enterprise() else {
            return Ok(false);
        };
        plane
            .shielded(&caller.principal_id, namespace)
            .map_err(|err| Status::unavailable(format!("the identity table could not be read: {err}")))
    }

    /// `rows` without those inside an account closed to the caller.
    pub fn retain_unshielded<T>(
        &self,
        caller: &Caller,
        rows: Vec<T>,
        namespace_of: impl Fn(&T) -> Option<String>,
    ) -> Result<Vec<T>, Status> {
        let mut kept = Vec::with_capacity(rows.len());
        let mut closed_ns: std::collections::HashMap<String, bool> = std::collections::HashMap::new();
        for row in rows {
            let Some(ns) = namespace_of(&row) else {
                kept.push(row);
                continue;
            };
            let closed = match closed_ns.get(&ns) {
                Some(known) => *known,
                None => {
                    let known = self.shielded(caller, &ns)?;
                    closed_ns.insert(ns, known);
                    known
                }
            };
            if !closed {
                kept.push(row);
            }
        }
        Ok(kept)
    }

    /// **The name filter, at a door that makes a name**: clean passes; a
    /// borderline name is held for the account admin (and refused for now);
    /// a blocked or reserved one is refused and noted for the account admin.
    /// `exempt_reserved`: a server admin naming a namespace or a machine.
    /// Every refusal is in the account's audit log.
    pub fn filter_name(
        &self,
        caller: &Caller,
        kind: &str,
        name: &str,
        account: Option<String>,
        exempt_reserved: bool,
    ) -> Result<(), Status> {
        let Some(plane) = self.control.enterprise() else {
            return Ok(());
        };
        let in_account = account.clone().unwrap_or_default();
        let verdict = plane.name_check(kind, name, &in_account)?;
        if verdict.is_clean() || (exempt_reserved && verdict.is_reserved()) {
            return Ok(());
        }
        // A borderline name with nobody to review it (a new account's own
        // name) is the operator's to allow: a server admin's passes.
        if verdict.is_borderline() && account.is_none() && exempt_reserved {
            return Ok(());
        }
        // Held only where there is an account admin to ask; a new account's
        // own borderline name has nobody, so it is refused.
        let hold = verdict.is_borderline() && account.is_some();
        let sentence = if verdict.is_borderline() && !hold {
            format!(
                "the {kind} name {name:?} needs a person's yes and there is no account admin to ask yet; \
                 choose another name"
            )
        } else {
            verdict.sentence.clone()
        };
        if let Some(account) = account.as_deref() {
            if let Err(err) = plane.name_review_add(kind, name, account, &caller.principal_id, hold) {
                tracing::warn!(
                    target: crate::auth::AUDIT_TARGET,
                    event = "name_filter.not_queued",
                    account,
                    reason = %err,
                    "a name the filter stopped could not be put in its account's queue"
                );
            }
        }
        let kind_word = if hold { "name.held" } else { "name.refused" };
        self.account_event(
            account,
            crate::enterprise::AccountAuditEvent::by(caller, kind_word, name)
                .detail(format!("a {kind} name, {} by the name filter", verdict.verdict))
                .refused(),
        );
        tracing::info!(
            target: crate::auth::AUDIT_TARGET,
            event = "name_filter.stopped",
            principal = %caller.principal_id,
            kind,
            verdict = %verdict.verdict,
            "the name filter stopped a name"
        );
        Err(if hold {
            Status::failed_precondition(sentence)
        } else {
            Status::invalid_argument(sentence)
        })
    }
}

macro_rules! service {
    ($name:ident, $doc:literal) => {
        #[doc = $doc]
        #[derive(Clone)]
        pub struct $name {
            pub(crate) ctx: ServiceContext,
        }

        impl $name {
            pub fn new(ctx: ServiceContext) -> Self {
                Self { ctx }
            }
        }

        impl From<ServiceContext> for $name {
            fn from(ctx: ServiceContext) -> Self {
                Self::new(ctx)
            }
        }
    };
}

service!(ReposService, "Namespaces and repositories: the catalogue.");
service!(
    AccountsService,
    "The per-USER verbs: the account gate, and the hard purge that empties an \
     account. `Repos` reclaims one repository at a time; this one takes a \
     user's whole namespace, and locks them out while it does."
);
service!(
    ClansService,
    "Clans (2026-10-05): ordinary accounts grouped in a clan with a namespace \
     of its own, each member giving it part of its own room; what is made in \
     it is administered by the clan's admin."
);
service!(
    RefsService,
    "Refs, commit log, object reads, diffs. Read-only."
);
service!(AccessService, "Principals, public keys and grants (the ACL model).");
service!(
    SsoService,
    "Company single sign-on: an account's own identity provider (OIDC, SAML 2.0), \
     SCIM 2.0 provisioning and the CSV import. Its people are EMPLOYEES of the \
     account, created new or provisioned, never an attached principal."
);
service!(PolicyService, "Branch protection and ref-name policy (the pre-receive gate).");
service!(
    EnterpriseService,
    "The owner's enterprise list: the account admin's audit log, the operator's \
     aggregates and the look the account admin grants, the name filter's lists \
     and the account admin's review queue. Every account has it."
);
service!(
    EventsService,
    "`Watch`, the server-streaming event feed (the event stream)."
);
service!(
    AdminService,
    "Health, stats, storage backend info, and `Capacity` — \"could you serve a \
     cap of N bytes?\", and how much storage the box has."
);
#[cfg(feature = "metered")]
service!(
    EntitlementService,
    "The paid state per namespace, written by monetize over `Set` and read by \
     `authorise_service`. The METERED appliance's service: compiled only with \
     the `metered` feature, and absent — not stubbed — without it."
);
#[cfg(feature = "metered")]
service!(
    SealedService,
    "A named sealed slot: bytes under a name, sealed under the KEK and logged \
     as a `Secret` frame (the twin, snapshots and a recharge carry it), read \
     and written only by the machine principal that owns it. The METERED \
     appliance's, absent and not stubbed without the feature."
);
#[cfg(feature = "metered")]
service!(
    BillingService,
    "The tenant's own renewal path: the appliance's word, signed, that ONE \
     human may act for ONE tenant — checked against the identity table here \
     rather than by the console, which holds a server-wide grant. The METERED \
     appliance's second service, absent and not stubbed without the feature; \
     see `service::billing` for why the scope check cannot live upstream."
);
service!(
    UsageService,
    "Per-namespace meters (`pack_bytes`, `cache_bytes`, `lfs_bytes`, …): the \
     numbers every storage report is made of, served to self-hosters and operators \
     alike. Open source, always on."
);
service!(
    UsageBatchService,
    "Many namespaces' meters in one call, optionally only the ones that changed \
     (`UsageBatch.ReadMany`, 2026-10-06): what a usage poller reads instead of \
     one `Usage.Read` per namespace. Open source, always on."
);
service!(
    ForgeService,
    "Issues, change proposals and review, as signed append-only events \
     (roadmap phase 1). The server never signs; it verifies and appends."
);
service!(MirrorService, "Pull-mirroring a remote forge (mirroring).");
service!(
    WebhooksService,
    "Outbound webhook delivery of event-stream events to external URLs, signed with an \
     sshsig the receiver verifies against the server's published key."
);
service!(
    PullsService,
    "Pull requests: records in the control store, merges fast-forward-only \
     through the store's own ref CAS."
);
service!(
    BootstrapService,
    "First-boot enrolment. **The one service served without a credential**, and \
     only while nothing has ever been enrolled. It is mounted OUTSIDE \
     [`crate::auth::AuthGate`]; see [`bootstrap`] for why that is safe and what \
     holds it to two methods."
);

/// ★ **f-principal-reach (money 2026-09-28): a NARROWED principal.** The
/// primary enrolled monetize as a server-wide `machine admin`; a CENTRAL
/// monetize serving other owners' estates must not hold that. A narrowed
/// principal has no grant at all and may call exactly the methods its list
/// names — the ones monetize's gunnar plugin calls ([`crate::BILLING_METHODS`])
/// — and nothing else, by name.
/// **The web sessions a passkey login enrolled here** (A34-10), by the session
/// key's fingerprint: when, and whether the authenticator verified the user.
/// In memory, on purpose: a restart asks for one fresh touch, which the console
/// answers in place (`[stepup]`), and nothing here outlives the step-up window.
#[derive(Debug, Default)]
pub struct WebSessions {
    minted: std::sync::Mutex<std::collections::HashMap<String, (bool, i64)>>,
}

impl WebSessions {
    /// A web login enrolled the session key `fingerprint` at `minted_unix_ms`.
    pub fn note(&self, fingerprint: &str, user_verified: bool, minted_unix_ms: i64) {
        let mut minted = self.minted.lock().unwrap_or_else(|p| p.into_inner());
        let floor = minted_unix_ms - nornir_identity_session::step_up::STEP_UP_WINDOW_SECS * 1000;
        minted.retain(|_, (_, at)| *at >= floor);
        minted.insert(fingerprint.to_owned(), (user_verified, minted_unix_ms));
    }

    /// **Is the session `fingerprint` a fresh step-up?** Minted by a
    /// user-verified passkey login within the step-up window (edda's rule).
    pub fn stepped_up(&self, fingerprint: &str, now_unix_ms: i64) -> bool {
        let minted = self.minted.lock().unwrap_or_else(|p| p.into_inner());
        minted.get(fingerprint).is_some_and(|(uv, at)| {
            nornir_identity_session::step_up::stepped_up(*uv, at.div_euclid(1000), now_unix_ms.div_euclid(1000))
        })
    }
}

/// The step-up as both proto messages carry it (`AccountStepUp`, `WebStepUp`).
macro_rules! presented {
    ($up:expr) => {
        nornir_identity_session::step_up::Presented {
            credential_id: &$up.credential_id,
            client_data_json: &$up.client_data_json,
            authenticator_data: &$up.authenticator_data,
            signature: &$up.signature,
            rp_id: &$up.rp_id,
            origin: &$up.origin,
        }
    };
}
pub(crate) use presented;

/// **A fresh, user-verified step-up of the caller's OWN passkey**, over a
/// challenge this listener issued: the one rule (edda `nornir-identity-session`,
/// A34-21: Enterprise's approval and Access's machine approval each had a copy).
/// The outer error is the store's; the inner one the step-up's refusal.
pub(crate) fn step_up_of_own_passkey(
    ctx: &ServiceContext,
    up: nornir_identity_session::step_up::Presented<'_>,
    caller: &str,
) -> Result<Result<nornir_identity_session::step_up::SteppedUp, nornir_identity_session::step_up::Refused>, Status> {
    nornir_identity_session::step_up::verify_own_passkey(
        &up,
        caller,
        &ctx.web_challenges,
        &ctx.config.web_relying_parties,
        crate::bus::now_unix_ms(),
        |id| {
            Ok::<_, Status>(ctx.control.credential(id)?.map(|stored| nornir_identity_session::step_up::Stored {
                principal_id: stored.credential.principal_id,
                public_key: stored.public_key,
                sign_count: stored.credential.sign_count,
            }))
        },
    )
}

#[cfg(test)]
mod narrowed_tests {
    use super::*;
    use crate::seam::MemoryControlPlane;

    fn caller(id: &str) -> Caller {
        Caller { principal_id: id.into(), fingerprint: "SHA256:x".into(), machine: true, authenticated: true }
    }

    #[test]
    fn a_narrowed_principal_may_call_its_list_and_nothing_else() {
        let plane = Arc::new(MemoryControlPlane::new());
        plane.narrow("monetize-central", crate::BILLING_METHODS);
        let ctx = ServiceContext::new(plane.clone(), Arc::new(crate::bus::EventBus::new(8)), crate::ServerConfig::default());
        let c = caller("monetize-central");
        for m in ["Entitlement.Set", "Admin.Capacity", "Admin.GrowStart", "Usage.ReadAll"] {
            let m: &'static str = crate::BILLING_METHODS.iter().copied().find(|x| *x == m).expect("on the list");
            ctx.require(&c, Access::Server, GrantLevel::Admin, m).unwrap_or_else(|e| panic!("{m}: {e:?}"));
        }
        for m in ["Admin.SetTunnelAddress", "Access.AddKey", "Mirror.SetRole", "Admin.IssueTwinCertificate"] {
            let e = ctx.require(&c, Access::Server, GrantLevel::Admin, m).unwrap_err();
            assert_eq!(e.code(), tonic::Code::PermissionDenied, "{m}");
            assert!(e.message().contains("narrowed"), "{m}: {}", e.message());
        }
        // A principal that is NOT narrowed is judged by its grants as before.
        crate::seam::ControlPlane::add_principal(plane.as_ref(), "rickard", "rickard", gunnar_proto::PrincipalKind::User).unwrap();
        plane.grant_server("rickard", GrantLevel::Admin);
        ctx.require(&caller("rickard"), Access::Server, GrantLevel::Admin, "Admin.SetTunnelAddress").expect("an admin is an admin");
    }
}