holger-server-lib 0.6.9

//! The tonic gRPC surface — holger's only network API (no REST; the OCI/HTTP
//! gateway in `exposed/http.rs` is a separate shim). [`HolgerGrpc`] is the shared
//! state behind three services impl'd on `Arc<HolgerGrpc>`: `RepositoryService`
//! (fetch/list/put/stream artifacts), `ArchiveService` (list/stat the underlying
//! znippy archive), and `AdminService` (health, repo catalog, `ServerProfile`).
//! Every RPC resolves the target through [`FastRoutes`] to a
//! [`RepositoryBackendTrait`] and appends a best-effort audit line.
//!
//! Trust model, and the gotcha: READS ARE OPEN (anonymous), only writes go
//! through [`HolgerGrpc::authorize_write`] (Bearer/mTLS-CN authN, then RBAC
//! write-gate that fails closed once role policy is configured). Because reads
//! are unauthenticated, backend errors must be laundered through [`internal_err`]
//! — raw errors carry archive/file paths and never reach the client; the detail
//! lives only in the server log and audit sink. `ServerProfile` reports
//! server-side truth (read-only iff no repo is writable, unless pinned via
//! [`HolgerGrpc::with_read_only_profile`]). The generated proto is committed
//! (`generated/holger.v1.rs`, `cargo xtask gen-proto`), not built at compile time.

pub mod holger_proto {
    // The proto Rust is generated AHEAD OF TIME by `cargo xtask gen-proto`
    // (protox + tonic-build, no `protoc`, no shell-out) and committed under
    // `src/generated/holger.v1.rs`. We `include!` the checked-in file instead of
    // the old `tonic::include_proto!("holger.v1")`, which read from `OUT_DIR` and
    // depended on a build.rs that shelled out to `protoc` at every build.
    // Regenerate after editing `proto/holger.proto`: `cargo xtask gen-proto`.
    include!("generated/holger.v1.rs");
}

use std::sync::Arc;
use std::time::Instant;
use tonic::{Request, Status};

use holger_proto::*;

use crate::audit::{AuditEvent, AuditLog, NoopAuditLog};
use crate::auth::{self, AuthConfig};
use crate::exposed::fast_routes::FastRoutes;

/// Default ingest body cap (1 GiB) — matches the HTTP gateway's default
/// (`lib.rs`), so both surfaces share one configurable knob (`ron_max_body_bytes`).
pub const DEFAULT_MAX_BODY_BYTES: usize = 1024 * 1024 * 1024;

/// Log a backend error server-side and return a GENERIC gRPC status. Read paths are
/// unauthenticated in this model, so the raw error (which can carry archive/file
/// paths + internal corruption detail) must never reach the client — operators get
/// the detail in the server log, clients get an opaque "internal error".
fn internal_err(e: impl std::fmt::Display) -> Status {
    log::warn!("holger grpc: backend error: {e}");
    Status::internal("internal error")
}

/// **Introspection / emit marker** — record one functional-status row for the
/// nornir test matrix. Wraps `nornir_testmatrix::functional_status` behind the
/// `testmatrix` feature (a compiled-out `#[inline]` no-op otherwise, with no
/// nornir dep in a release build). `component` is the reporting surface (e.g.
/// `"holger-grpc/fetch_artifact"`), `check` what it verified, `ok` the verdict,
/// `detail` a short human note. Every gRPC handler + the RBAC write gate emit one
/// so `nornir test --features testmatrix` SEES each live surface. Mirrors the
/// reference emitter in `korp-collectors/src/lib.rs`.
#[inline]
pub(crate) fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
    #[cfg(feature = "testmatrix")]
    nornir_testmatrix::functional_status(component, check, ok, detail);
    #[cfg(not(feature = "testmatrix"))]
    {
        let _ = (component, check, ok, detail);
    }
}

/// The three gRPC service impls (Repository / Archive / Admin) on `Arc<HolgerGrpc>`
/// — a child module so it reaches this module's private helpers + `HolgerGrpc`'s
/// private state, keeping the root focused on shared state + wiring.
mod services;

/// Shared state for all gRPC services
pub struct HolgerGrpc {
    pub routes: FastRoutes,
    pub start_time: Instant,
    pub auth_config: Arc<AuthConfig>,
    /// Append-only audit sink. Defaults to [`NoopAuditLog`]; install a real
    /// backend (e.g. `audit::default_audit_log(dir)`) via [`HolgerGrpc::with_audit`].
    pub audit: Arc<dyn AuditLog>,
    /// Maximum accepted `put_artifact` payload size in bytes. The streamed HTTP
    /// gateway caps uploads (`http.rs`); without this the gRPC write path was
    /// unbounded. Same default + config source as the gateway.
    pub max_body_bytes: usize,
    /// Explicit sealed/read-only profile override. `None` (default) ⇒ the
    /// `ServerProfile` RPC *derives* the profile from the live route table (the
    /// server is read-only iff no repository accepts writes — server truth).
    /// `Some(true)` forces the static "rigged for silent running" profile even
    /// when writable repos exist (a sealed static-holger declaring itself);
    /// `Some(false)` forces dynamic.
    pub read_only_profile: Option<bool>,
    /// Immutable, content-addressed PROD store for the DEV→PROD promote workflow
    /// (`PromotionService`). `None` (default) ⇒ the `PromoteArtifact` RPC returns
    /// `failed_precondition` (promotion not configured on this server). Installed
    /// via [`HolgerGrpc::with_promotion`].
    pub promotion: Option<Arc<crate::promote::PromotionEngine>>,
    /// Hosted-SBOM store for the `SbomService` (attach/fetch SBOM documents
    /// keyed by artifact coordinate). `None` (default) ⇒ `AttachSbom` returns
    /// `failed_precondition` and `FetchSbom` returns `not_found` (SBOM hosting
    /// not configured on this server). Installed via [`HolgerGrpc::with_sboms`].
    pub sboms: Option<crate::sbom::SharedSbomStore>,
    /// Custom-property store powering the `Search` property axis. `None` (default)
    /// ⇒ a property filter on a `Search` matches nothing (fail-closed — an
    /// unconfigured store never yields false hits). Installed via
    /// [`HolgerGrpc::with_properties`].
    pub properties: Option<crate::properties::SharedPropertyStore>,
    /// Serve-time **quarantine** predicate `(key, value)`. `Some` ⇒ an artifact
    /// whose properties match (e.g. `quarantine=true`) is REFUSED at the fetch
    /// boundary (served as `not_found`) even though it is stored — a bad/unverified
    /// blob is blocked without deleting it. `None` (default) ⇒ no serve gate (zero
    /// hot-path cost). Installed via [`HolgerGrpc::with_quarantine`].
    pub quarantine: Option<(String, String)>,
}

impl HolgerGrpc {
    pub fn new(routes: FastRoutes) -> Self {
        Self::with_auth(routes, Arc::new(AuthConfig::default()))
    }

    pub fn with_auth(routes: FastRoutes, auth_config: Arc<AuthConfig>) -> Self {
        Self {
            routes,
            start_time: Instant::now(),
            auth_config,
            audit: Arc::new(NoopAuditLog),
            max_body_bytes: DEFAULT_MAX_BODY_BYTES,
            read_only_profile: None,
            promotion: None,
            sboms: None,
            properties: None,
            quarantine: None,
        }
    }

    /// Install the immutable PROD store for the promote workflow (builder style).
    pub fn with_promotion(mut self, engine: Arc<crate::promote::PromotionEngine>) -> Self {
        self.promotion = Some(engine);
        self
    }

    /// Install the hosted-SBOM store for the `SbomService` (builder style). The
    /// server wires `Holger::sbom_store()` here so gRPC, HTTP, and the CLI share
    /// one on-disk store.
    pub fn with_sboms(mut self, store: crate::sbom::SharedSbomStore) -> Self {
        self.sboms = Some(store);
        self
    }

    /// Install the custom-property store powering the `Search` property axis
    /// (builder style). The server wires `Holger::property_store()` here so gRPC,
    /// HTTP, and the CLI share one on-disk store.
    pub fn with_properties(mut self, store: crate::properties::SharedPropertyStore) -> Self {
        self.properties = Some(store);
        self
    }

    /// Install the serve-time quarantine predicate `(key, value)` (builder style).
    /// The server wires it from `HOLGER_QUARANTINE_PROPERTY`; a matching artifact is
    /// refused at the fetch boundary. Requires a property store to bite.
    pub fn with_quarantine(mut self, key: String, value: String) -> Self {
        self.quarantine = Some((key, value));
        self
    }

    /// Whether `(repo, id)`'s properties match the configured quarantine predicate
    /// (so it must NOT be served). `false` when no quarantine or no store is
    /// configured — the shared [`properties::map_matches`](crate::properties::map_matches)
    /// rule, so `quarantine=true` means what it does everywhere else.
    pub(crate) fn is_quarantined(&self, repo: &str, id: &traits::ArtifactId) -> bool {
        match (&self.quarantine, &self.properties) {
            (Some((key, value)), Some(store)) => {
                crate::properties::map_matches(&store.get(repo, id), key, value)
            }
            _ => false,
        }
    }

    /// Install an audit backend (builder style). The server wires the default
    /// Arrow-IPC log here when an audit directory is configured.
    pub fn with_audit(mut self, audit: Arc<dyn AuditLog>) -> Self {
        self.audit = audit;
        self
    }

    /// Override the ingest body cap (builder style). The server wires the
    /// endpoint's `ron_max_body_bytes` here so gRPC and HTTP share one limit.
    pub fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self {
        self.max_body_bytes = max_body_bytes;
        self
    }

    /// Declare the server's profile explicitly (builder style). `true` pins the
    /// static "rigged for silent running" / read-only profile, `false` pins
    /// dynamic. Without this the `ServerProfile` RPC derives the profile from the
    /// live route table. A sealed static-holger sets `true` here.
    pub fn with_read_only_profile(mut self, read_only: bool) -> Self {
        self.read_only_profile = Some(read_only);
        self
    }

    /// The number of configured repositories that accept writes.
    fn writable_repo_count(&self) -> usize {
        self.routes.all_repos().iter().filter(|(_, repo)| repo.is_writable()).count()
    }

    /// The server's effective read-only status: the explicit override if set,
    /// else derived from the route table (read-only iff *no* repo is writable).
    /// An empty server (no repos) is read-only — it can serve nothing writable.
    fn is_read_only(&self) -> bool {
        self.read_only_profile.unwrap_or_else(|| self.writable_repo_count() == 0)
    }

    /// Best-effort audit append. An audit-backend failure must NOT fail the
    /// request it describes (availability over a single lost line), but it is
    /// surfaced via `log::warn!` so a broken sink is never silent.
    fn record_audit(&self, event: AuditEvent) {
        // Count the request on the process-global `/metrics` registry BEFORE the
        // (best-effort) audit append — this is the single gRPC-plane choke point
        // every RPC exits through, so the whole gRPC surface is observable in
        // `/metrics` without threading a handle into each handler. The HTTP door
        // bumps at its own exit (`exposed::http`); the two never double-count
        // because HTTP does not route through this method. `Other` actions aren't
        // request traffic and map to `None` (not counted).
        if let Some(verb) = crate::metrics::Verb::from_audit_action(event.action) {
            crate::metrics::global().record_request(verb, event.status, event.bytes);
        }
        if let Err(e) = self.audit.record(event) {
            log::warn!("audit record failed: {e}");
        }
    }

    fn get_repo(&self, name: &str) -> Result<Arc<dyn traits::RepositoryBackendTrait>, Status> {
        self.routes
            .lookup(name)
            .cloned()
            .ok_or_else(|| Status::not_found(format!("Repository '{}' not found", name)))
    }

    /// Validate write-request credentials against the configured auth methods.
    /// Accepts a Bearer token or an mTLS client-cert CN. An empty config means
    /// open access.
    async fn authorize_write<T>(
        &self,
        request: &Request<T>,
        repo: &str,
    ) -> Result<Option<auth::AuthIdentity>, Status> {
        let bearer = request
            .metadata()
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "));

        // mTLS identity from the TLS handshake, if the server runs with a
        // client CA configured.
        let client_cn = request
            .peer_certs()
            .and_then(|certs| {
                let ders: Vec<rustls::pki_types::CertificateDer<'_>> = certs
                    .iter()
                    .map(|c| rustls::pki_types::CertificateDer::from(c.as_ref().to_vec()))
                    .collect();
                crate::exposed::tls::leaf_common_name(&ders)
            });

        let identity = match auth::validate_request(&self.auth_config, bearer, client_cn.as_deref())
            .await
        {
            Ok(id) => id,
            Err(_) => {
                functional_status(
                    "holger-grpc/authorize_write",
                    "credentials_valid",
                    false,
                    "invalid or missing write credentials",
                );
                return Err(Status::unauthenticated(
                    "Valid credentials required for write access",
                ));
            }
        };

        // Authorization (RBAC). Inactive unless role policy is configured, in
        // which case the writer's resolved role must permit writes; `None`
        // identity here means auth is open (no methods) — leave writes open to
        // preserve the authN-only behaviour, EXCEPT when role policy is set
        // (fail closed: policy configured but nobody authenticated).
        if self.auth_config.rbac_enabled() {
            match &identity {
                Some(id) => {
                    let role = self.auth_config.role_for_repo(repo, &id.subject);
                    if !role.can_write() {
                        functional_status(
                            "holger-grpc/authorize_write",
                            "rbac_write_permitted",
                            false,
                            "identity lacks writer/admin role",
                        );
                        return Err(Status::permission_denied(format!(
                            "identity '{}' has role {:?}; writer or admin required for write access",
                            id.subject, role
                        )));
                    }
                    functional_status(
                        "holger-grpc/authorize_write",
                        "rbac_write_permitted",
                        true,
                        "writer/admin role granted",
                    );
                }
                None => {
                    functional_status(
                        "holger-grpc/authorize_write",
                        "rbac_write_permitted",
                        false,
                        "rbac configured but no authenticated identity",
                    );
                    return Err(Status::unauthenticated(
                        "authenticated identity required for write access",
                    ));
                }
            }
        }
        Ok(identity)
    }

    /// Authorize a PRIVILEGED (admin) request — promotion/delete/manage. Mirrors
    /// [`HolgerGrpc::authorize_write`] but requires the `Admin` role, and is
    /// **fail-closed**: with RBAC on, the identity must resolve to `Admin`
    /// (`can_admin`) or the request is denied; a configured-but-unauthenticated
    /// request is denied. With no auth methods at all (fully open) it preserves
    /// the dev/test open behaviour, exactly as writes do.
    async fn authorize_admin<T>(
        &self,
        request: &Request<T>,
        repo: &str,
    ) -> Result<Option<auth::AuthIdentity>, Status> {
        let bearer = request
            .metadata()
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "));

        let client_cn = request.peer_certs().and_then(|certs| {
            let ders: Vec<rustls::pki_types::CertificateDer<'_>> = certs
                .iter()
                .map(|c| rustls::pki_types::CertificateDer::from(c.as_ref().to_vec()))
                .collect();
            crate::exposed::tls::leaf_common_name(&ders)
        });

        let identity =
            match auth::validate_request(&self.auth_config, bearer, client_cn.as_deref()).await {
                Ok(id) => id,
                Err(_) => {
                    return Err(Status::unauthenticated(
                        "Valid credentials required for promotion",
                    ));
                }
            };

        if self.auth_config.rbac_enabled() {
            match &identity {
                Some(id) => {
                    let role = self.auth_config.role_for_repo(repo, &id.subject);
                    if !role.can_admin() {
                        functional_status(
                            "holger-grpc/authorize_admin",
                            "rbac_admin_permitted",
                            false,
                            "identity lacks admin role",
                        );
                        return Err(Status::permission_denied(format!(
                            "identity '{}' has role {:?}; admin required to promote",
                            id.subject, role
                        )));
                    }
                    functional_status(
                        "holger-grpc/authorize_admin",
                        "rbac_admin_permitted",
                        true,
                        "admin role granted",
                    );
                }
                None => {
                    return Err(Status::unauthenticated(
                        "authenticated admin identity required to promote",
                    ));
                }
            }
        }
        Ok(identity)
    }
}

/// Client source address (`ip:port`) as the server sees it, or empty when the
/// transport exposes none.
fn peer_addr_string<T>(request: &Request<T>) -> String {
    request
        .remote_addr()
        .map(|a| a.to_string())
        .unwrap_or_default()
}

/// Human-readable artifact label for the audit log: `namespace/name@version`
/// (the namespace segment dropped when empty, the Rust-style no-namespace case).
fn artifact_label(id: &ArtifactId) -> String {
    if id.namespace.is_empty() {
        format!("{}@{}", id.name, id.version)
    } else {
        format!("{}/{}@{}", id.namespace, id.name, id.version)
    }
}

// ─── RepositoryService ───────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::audit::MemoryAuditLog;
    use std::sync::Arc;
    use traits::{ArchiveInfo, ArtifactFormat, RepositoryBackendTrait};
    // The service traits + AuditAction used to reach the tests via super's glob;
    // the split moved the service impls out, so import them here directly.
    use super::holger_proto::admin_service_server::AdminService;
    use super::holger_proto::archive_service_server::ArchiveService;
    use super::holger_proto::repository_service_server::RepositoryService;
    use crate::audit::AuditAction;

    /// Minimal in-memory backend: holds one named blob, reports a fixed archive
    /// stat block. Read-only by default; `writable` flips `is_writable` so the
    /// `ServerProfile` derivation (read-only iff no writable repo) is testable.
    struct MockRepo {
        name: String,
        blob: Option<Vec<u8>>,
        writable: bool,
    }

    impl RepositoryBackendTrait for MockRepo {
        fn name(&self) -> &str {
            &self.name
        }
        fn format(&self) -> ArtifactFormat {
            ArtifactFormat::Rust
        }
        fn is_writable(&self) -> bool {
            self.writable
        }
        fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            Ok(self.blob.clone())
        }
        fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
            anyhow::bail!("read-only")
        }
        fn archive_files(&self, _prefix: Option<&str>) -> anyhow::Result<Vec<String>> {
            Ok(vec!["a.rs".into(), "b.rs".into()])
        }
        fn archive_info(&self) -> anyhow::Result<ArchiveInfo> {
            Ok(ArchiveInfo {
                file_count: 2,
                total_uncompressed_bytes: 99,
                archive_path: "mock".into(),
            })
        }
        fn handle_http2_request(
            &self,
            _method: &str,
            _suburl: &str,
            _body: &[u8],
        ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            Ok((200, vec![], Vec::new()))
        }
    }

    fn grpc_with(blob: Option<Vec<u8>>) -> (Arc<HolgerGrpc>, Arc<MemoryAuditLog>) {
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "crates".into(), blob, writable: false });
        let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
        let audit = Arc::new(MemoryAuditLog::new());
        let grpc = Arc::new(HolgerGrpc::new(routes).with_audit(audit.clone()));
        (grpc, audit)
    }

    /// Build a grpc state over one repo whose writability is `writable`.
    fn grpc_with_writable(writable: bool) -> Arc<HolgerGrpc> {
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "crates".into(), blob: None, writable });
        let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
        Arc::new(HolgerGrpc::new(routes))
    }

    fn artifact_id() -> ArtifactId {
        ArtifactId { namespace: String::new(), name: "serde".into(), version: "1.0".into() }
    }

    /// Serve-time quarantine gate (property-policy §): a coordinate tagged
    /// `quarantine=true` is REFUSED at `fetch_artifact` (NotFound + audited 404)
    /// even though the backend holds bytes; an UNtagged coordinate serves; clearing
    /// the tag restores serving. RED-when-broken — a missing gate would serve the
    /// quarantined blob.
    #[tokio::test]
    async fn fetch_artifact_quarantine_gate_refuses_tagged_serves_untagged() {
        use crate::properties::{PropertyStore, SharedPropertyStore};
        let tmp = tempfile::tempdir().unwrap();
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "crates".into(), blob: Some(b"bytes".to_vec()), writable: false });
        let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
        let store: SharedPropertyStore = Arc::new(PropertyStore::new(tmp.path()));
        let audit = Arc::new(MemoryAuditLog::new());
        let grpc = Arc::new(
            HolgerGrpc::new(routes)
                .with_audit(audit.clone())
                .with_properties(store.clone())
                .with_quarantine("quarantine".into(), "true".into()),
        );

        let bad = ArtifactId { namespace: String::new(), name: "bad".into(), version: "1.0".into() };
        let good = ArtifactId { namespace: String::new(), name: "good".into(), version: "1.0".into() };
        let tid = |i: &ArtifactId| traits::ArtifactId {
            namespace: None, name: i.name.clone(), version: i.version.clone(),
        };
        let fetch = |id: ArtifactId| {
            Request::new(FetchArtifactRequest { repository: "crates".into(), id: Some(id) })
        };

        // Tag "bad" quarantine=true.
        store.set("crates", &tid(&bad), "quarantine", vec!["true".into()]).unwrap();

        // Quarantined → refused (NotFound), audited 404.
        let err = RepositoryService::fetch_artifact(&grpc, fetch(bad.clone()))
            .await
            .expect_err("a quarantined blob must be refused");
        assert_eq!(err.code(), tonic::Code::NotFound);
        assert_eq!(audit.events().last().unwrap().status, 404);

        // Untagged sibling → served (RED-when-broken: proves it's the tag, not a
        // blanket block).
        let ok = RepositoryService::fetch_artifact(&grpc, fetch(good))
            .await
            .expect("an untagged blob serves")
            .into_inner();
        assert_eq!(ok.data, b"bytes");

        // Clear the tag → serving restored.
        store.set("crates", &tid(&bad), "quarantine", vec![]).unwrap();
        let restored = RepositoryService::fetch_artifact(&grpc, fetch(bad))
            .await
            .expect("a cleared blob serves again")
            .into_inner();
        assert_eq!(restored.data, b"bytes");
    }

    #[tokio::test]
    async fn stream_artifact_audits_download_hit_and_miss() {
        // Hit: a present blob records a 200 download with the byte count.
        let (grpc, audit) = grpc_with(Some(b"hello".to_vec()));
        let req = Request::new(FetchArtifactRequest {
            repository: "crates".into(),
            id: Some(artifact_id()),
        });
        grpc.stream_artifact(req).await.expect("stream ok");
        let ev = audit.events();
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, AuditAction::Download);
        assert_eq!(ev[0].ident, "anonymous");
        assert_eq!(ev[0].repo, "crates");
        assert_eq!(ev[0].artifact, "serde@1.0");
        assert_eq!(ev[0].status, 200);
        assert_eq!(ev[0].bytes, 5);

        // Miss: absent blob records a 404 download.
        let (grpc, audit) = grpc_with(None);
        let req = Request::new(FetchArtifactRequest {
            repository: "crates".into(),
            id: Some(artifact_id()),
        });
        assert!(grpc.stream_artifact(req).await.is_err());
        let ev = audit.events();
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, AuditAction::Download);
        assert_eq!(ev[0].status, 404);
    }

    #[tokio::test]
    async fn list_artifacts_audits_list() {
        let (grpc, audit) = grpc_with(None);
        let req = Request::new(ListArtifactsRequest {
            repository: "crates".into(),
            name_filter: "ser".into(),
            limit: 10,
            page_token: String::new(),
        });
        RepositoryService::list_artifacts(&grpc, req).await.expect("list ok");
        let ev = audit.events();
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, AuditAction::List);
        assert_eq!(ev[0].repo, "crates");
        assert_eq!(ev[0].artifact, "ser"); // filter recorded in the artifact column
        assert_eq!(ev[0].status, 200);
    }

    #[tokio::test]
    async fn list_artifacts_unknown_repo_audits_404() {
        let (grpc, audit) = grpc_with(None);
        let req = Request::new(ListArtifactsRequest {
            repository: "ghost".into(),
            name_filter: String::new(),
            limit: 10,
            page_token: String::new(),
        });
        assert!(RepositoryService::list_artifacts(&grpc, req).await.is_err());
        let ev = audit.events();
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, AuditAction::List);
        assert_eq!(ev[0].repo, "ghost");
        assert_eq!(ev[0].status, 404);
    }

    #[tokio::test]
    async fn archive_endpoints_audit_list() {
        let (grpc, audit) = grpc_with(None);

        let req = Request::new(ListArchiveFilesRequest {
            repository: "crates".into(),
            prefix: "src/".into(),
        });
        grpc.list_archive_files(req).await.expect("archive files ok");

        let req = Request::new(ArchiveInfoRequest { repository: "crates".into() });
        grpc.archive_info(req).await.expect("archive info ok");

        let ev = audit.events();
        assert_eq!(ev.len(), 2);
        assert!(ev.iter().all(|e| e.action == AuditAction::List && e.status == 200));
        assert_eq!(ev[0].artifact, "src/"); // prefix recorded
        assert_eq!(ev[1].bytes, 2); // archive_info logs the file count
    }

    #[tokio::test]
    async fn list_repositories_audits_list() {
        let (grpc, audit) = grpc_with(None);
        let req = Request::new(ListRepositoriesRequest {});
        AdminService::list_repositories(&grpc, req).await.expect("list repos ok");
        let ev = audit.events();
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, AuditAction::List);
        assert_eq!(ev[0].bytes, 1); // one configured repo
    }

    #[tokio::test]
    async fn server_profile_derived_static_when_all_repos_read_only() {
        // No writable repo ⇒ the server reports the static "silent running"
        // profile as read-only truth (derived from the route table, no override).
        let grpc = grpc_with_writable(false);
        let resp = AdminService::server_profile(&grpc, Request::new(ServerProfileRequest {}))
            .await
            .expect("profile ok")
            .into_inner();
        assert_eq!(resp.profile, "static");
        assert!(resp.read_only);
        assert_eq!(resp.writable_repo_count, 0);
        assert!(resp.label.contains("SILENT RUNNING"), "label = {}", resp.label);
    }

    #[tokio::test]
    async fn server_profile_derived_dynamic_when_a_repo_is_writable() {
        // A writable repo ⇒ derived dynamic profile (writable count reflected).
        let grpc = grpc_with_writable(true);
        let resp = AdminService::server_profile(&grpc, Request::new(ServerProfileRequest {}))
            .await
            .expect("profile ok")
            .into_inner();
        assert_eq!(resp.profile, "dynamic");
        assert!(!resp.read_only);
        assert_eq!(resp.writable_repo_count, 1);
    }

    #[tokio::test]
    async fn server_profile_override_forces_static_over_writable_repos() {
        // A sealed static-holger declares itself read-only even with writable
        // repos present — the explicit override wins over the derivation.
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "crates".into(), blob: None, writable: true });
        let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
        let grpc = Arc::new(HolgerGrpc::new(routes).with_read_only_profile(true));
        let resp = AdminService::server_profile(&grpc, Request::new(ServerProfileRequest {}))
            .await
            .expect("profile ok")
            .into_inner();
        assert_eq!(resp.profile, "static");
        assert!(resp.read_only);
        // The override pins read-only, but the honest writable count still shows
        // 1 so the UI can warn "sealed despite writable repos".
        assert_eq!(resp.writable_repo_count, 1);
    }

    /// Writable in-memory backend: accepts any put. Lets us exercise the put
    /// path (the read-only MockRepo short-circuits before the body cap).
    struct WritableMock {
        name: String,
    }
    impl RepositoryBackendTrait for WritableMock {
        fn name(&self) -> &str {
            &self.name
        }
        fn format(&self) -> ArtifactFormat {
            ArtifactFormat::Rust
        }
        fn is_writable(&self) -> bool {
            true
        }
        fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            Ok(None)
        }
        fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
            Ok(())
        }
        fn archive_files(&self, _prefix: Option<&str>) -> anyhow::Result<Vec<String>> {
            Ok(Vec::new())
        }
        fn archive_info(&self) -> anyhow::Result<ArchiveInfo> {
            Ok(ArchiveInfo::default())
        }
        fn handle_http2_request(
            &self,
            _method: &str,
            _suburl: &str,
            _body: &[u8],
        ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            Ok((200, vec![], Vec::new()))
        }
    }

    #[tokio::test]
    async fn put_artifact_enforces_body_cap() {
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(WritableMock { name: "crates".into() });
        let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
        let audit = Arc::new(MemoryAuditLog::new());
        let grpc = Arc::new(
            HolgerGrpc::new(routes)
                .with_audit(audit.clone())
                .with_max_body_bytes(16),
        );

        // Over the cap → rejected with ResourceExhausted, audited 413, put skipped.
        let req = Request::new(PutArtifactRequest {
            repository: "crates".into(),
            id: Some(artifact_id()),
            data: vec![0u8; 17],
        });
        let err = RepositoryService::put_artifact(&grpc, req)
            .await
            .expect_err("over-cap put must be rejected");
        assert_eq!(err.code(), tonic::Code::ResourceExhausted);
        assert_eq!(audit.events().last().unwrap().status, 413);

        // Under the cap → accepted and audited 200.
        let req = Request::new(PutArtifactRequest {
            repository: "crates".into(),
            id: Some(artifact_id()),
            data: vec![0u8; 8],
        });
        RepositoryService::put_artifact(&grpc, req)
            .await
            .expect("under-cap put must succeed");
        assert_eq!(audit.events().last().unwrap().status, 200);
    }

    // ─── PromotionService (DEV → PROD), admin-gated, offline OIDC auth ────────
    //
    // These exercise Part B's handler end-to-end AND reuse Part A's offline OIDC:
    // the admin/writer identities come from a signed JWT verified offline against
    // an inline test JWKS (no network), so the RBAC admin gate is real.
    use super::holger_proto::promotion_service_server::PromotionService;
    use crate::auth::{test_support, AuthConfig, AuthMethodConfig, Role};
    use crate::promote::PromotionEngine;

    /// Build a promotion-ready grpc state: a source repo `rust-dev` holding
    /// `blob`, OIDC offline auth for `issuer` with `admin@corp`=Admin +
    /// `bot@corp`=Writer, a fresh PROD store under `prod_root`, and a memory audit.
    fn promo_grpc(
        issuer: &str,
        blob: Vec<u8>,
        prod_root: std::path::PathBuf,
    ) -> (Arc<HolgerGrpc>, Arc<MemoryAuditLog>) {
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "rust-dev".into(), blob: Some(blob), writable: true });
        let routes = FastRoutes::new(vec![("rust-dev".to_string(), backend)]);
        let audit = Arc::new(MemoryAuditLog::new());

        let mut roles = std::collections::HashMap::new();
        roles.insert("admin@corp".to_string(), Role::Admin);
        roles.insert("bot@corp".to_string(), Role::Writer);
        let auth = AuthConfig {
            methods: vec![test_support::oidc_offline_method(issuer)],
            roles,
            default_role: Some(Role::Reader),
            ..Default::default()
        };

        let grpc = Arc::new(
            HolgerGrpc::with_auth(routes, Arc::new(auth))
                .with_audit(audit.clone())
                .with_promotion(Arc::new(PromotionEngine::new(prod_root))),
        );
        (grpc, audit)
    }

    fn promote_req(sub_token: &str) -> Request<PromoteArtifactRequest> {
        let mut req = Request::new(PromoteArtifactRequest {
            from_repo: "rust-dev".into(),
            to_repo: "rust-prod".into(),
            artifact: Some(ArtifactId {
                namespace: String::new(),
                name: "serde".into(),
                version: "1.0.0".into(),
            }),
        });
        req.metadata_mut()
            .insert("authorization", format!("Bearer {sub_token}").parse().unwrap());
        req
    }

    /// An ADMIN promotes: the artifact transitions + seals into the PROD store
    /// (content address returned), and a `Promote` audit row is written.
    #[tokio::test]
    async fn promote_seals_and_audits_for_admin() {
        let dir = tempfile::tempdir().unwrap();
        let issuer = "https://idp.example/promote-admin";
        let bytes = b"the-crate-bytes-to-promote".to_vec();
        let (grpc, audit) = promo_grpc(issuer, bytes.clone(), dir.path().to_path_buf());

        let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);
        let resp = PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect("admin promote ok")
            .into_inner();

        assert!(resp.success);
        assert_eq!(resp.promoted_by, "admin@corp");
        assert_eq!(resp.content_address, blake3::hash(&bytes).to_hex().to_string());

        let ev = audit.events();
        let last = ev.last().unwrap();
        assert_eq!(last.action, AuditAction::Promote);
        assert_eq!(last.ident, "admin@corp");
        assert_eq!(last.repo, "rust-prod");
        assert_eq!(last.artifact, "serde@1.0.0");
        assert_eq!(last.status, 200);
    }

    /// A NON-admin (writer) is refused — fail-closed at the admin gate, before any
    /// store access.
    #[tokio::test]
    async fn promote_refused_for_non_admin() {
        let dir = tempfile::tempdir().unwrap();
        let issuer = "https://idp.example/promote-writer";
        let (grpc, _audit) = promo_grpc(issuer, b"bytes".to_vec(), dir.path().to_path_buf());

        let token = test_support::token_for(issuer, "holger", "bot@corp", 3600);
        let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect_err("writer must be refused");
        assert_eq!(err.code(), tonic::Code::PermissionDenied);
    }

    /// A promoted artifact is IMMUTABLE: a second admin promote of the same
    /// coordinate is refused (`already_exists`) and audited 409.
    #[tokio::test]
    async fn promote_is_immutable_on_repromote() {
        let dir = tempfile::tempdir().unwrap();
        let issuer = "https://idp.example/promote-immut";
        let (grpc, audit) = promo_grpc(issuer, b"v1-bytes".to_vec(), dir.path().to_path_buf());
        let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);

        PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect("first promote ok");
        let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect_err("re-promote must be refused");
        assert_eq!(err.code(), tonic::Code::AlreadyExists);
        assert_eq!(audit.events().last().unwrap().status, 409);
    }

    /// With no promotion store configured, the RPC fails cleanly (still admin-gated
    /// first).
    #[tokio::test]
    async fn promote_without_store_fails_precondition() {
        let issuer = "https://idp.example/promote-nostore";
        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "rust-dev".into(), blob: Some(b"x".to_vec()), writable: true });
        let routes = FastRoutes::new(vec![("rust-dev".to_string(), backend)]);
        let mut roles = std::collections::HashMap::new();
        roles.insert("admin@corp".to_string(), Role::Admin);
        let auth = AuthConfig {
            methods: vec![test_support::oidc_offline_method(issuer)],
            roles,
            default_role: None,
            ..Default::default()
        };
        // No .with_promotion(...)
        let grpc = Arc::new(HolgerGrpc::with_auth(routes, Arc::new(auth)));
        let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);
        let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect_err("no store → error");
        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
    }

    /// Property-requirement gate (property-policy §): with a `signed=true`
    /// requirement, an UNSIGNED source artifact is refused promotion
    /// (RED-when-broken — a broken gate would seal it), and the SAME artifact
    /// promotes once the property is set on the source coordinate.
    #[tokio::test]
    async fn promote_requires_source_property() {
        use crate::properties::{PropertyStore, SharedPropertyStore};
        let dir = tempfile::tempdir().unwrap();
        let issuer = "https://idp.example/promote-prop";
        let bytes = b"signed-build-bytes".to_vec();

        let backend: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MockRepo { name: "rust-dev".into(), blob: Some(bytes), writable: true });
        let routes = FastRoutes::new(vec![("rust-dev".to_string(), backend)]);
        let mut roles = std::collections::HashMap::new();
        roles.insert("admin@corp".to_string(), Role::Admin);
        let auth = AuthConfig {
            methods: vec![test_support::oidc_offline_method(issuer)],
            roles,
            default_role: None,
            ..Default::default()
        };
        let props: SharedPropertyStore = Arc::new(PropertyStore::new(dir.path().join("props")));
        let engine = PromotionEngine::new(dir.path().join("prod"))
            .with_required_property("signed".into(), "true".into());
        let audit = Arc::new(MemoryAuditLog::new());
        let grpc = Arc::new(
            HolgerGrpc::with_auth(routes, Arc::new(auth))
                .with_audit(audit.clone())
                .with_promotion(Arc::new(engine))
                .with_properties(props.clone()),
        );
        let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);

        // Unsigned source → refused (FailedPrecondition), audited 412, nothing sealed.
        let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect_err("an unsigned artifact must be refused promotion");
        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
        assert_eq!(audit.events().last().unwrap().status, 412);

        // Tag the SOURCE coordinate signed=true → the same promote now succeeds.
        let id = traits::ArtifactId { namespace: None, name: "serde".into(), version: "1.0.0".into() };
        props.set("rust-dev", &id, "signed", vec!["true".into()]).unwrap();
        let resp = PromotionService::promote_artifact(&grpc, promote_req(&token))
            .await
            .expect("a signed artifact promotes")
            .into_inner();
        assert!(resp.success, "signed=true satisfies the requirement");
    }

    /// Sanity: the `AuthMethodConfig` import is exercised (offline method built).
    #[test]
    fn offline_method_is_oidc() {
        assert!(matches!(
            test_support::oidc_offline_method("https://x"),
            AuthMethodConfig::Oidc { .. }
        ));
    }

    // ─── SearchService (cross-repo discovery) ────────────────────────────────
    use super::holger_proto::search_service_server::SearchService;
    use traits::{ArtifactEntry, ArtifactId as TraitArtifactId};

    /// A list-capable backend for the search RPC test: holds a fixed entry set
    /// (name → size) so `list()` returns real artifacts to match against.
    struct SearchMock {
        name: String,
        entries: Vec<(String, i64)>,
    }
    impl RepositoryBackendTrait for SearchMock {
        fn name(&self) -> &str {
            &self.name
        }
        fn format(&self) -> ArtifactFormat {
            ArtifactFormat::Rust
        }
        fn is_writable(&self) -> bool {
            false
        }
        fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            Ok(None)
        }
        fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
            anyhow::bail!("read-only")
        }
        fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
            Ok(self
                .entries
                .iter()
                .map(|(n, sz)| ArtifactEntry {
                    id: TraitArtifactId { namespace: None, name: n.clone(), version: "1.0.0".into() },
                    size_bytes: *sz,
                    content_type: "application/octet-stream".into(),
                })
                .collect())
        }
        fn handle_http2_request(
            &self,
            _m: &str,
            _s: &str,
            _b: &[u8],
        ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            Ok((404, vec![], Vec::new()))
        }
    }

    /// The Search RPC finds the matching artifact across repos, EXCLUDES the
    /// non-matches, tags each hit with its repo, and audits a List. The exclusion
    /// assertion is the RED-when-broken guard.
    #[tokio::test]
    async fn search_rpc_finds_hits_excludes_non_matches_and_audits() {
        let a: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
            name: "rust-dev".into(),
            entries: vec![("serde".into(), 10), ("tokio".into(), 20)],
        });
        let b: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
            name: "extra".into(),
            entries: vec![("serde_json".into(), 30)],
        });
        let routes = FastRoutes::new(vec![("rust-dev".into(), a), ("extra".into(), b)]);
        let audit = Arc::new(MemoryAuditLog::new());
        let grpc = Arc::new(HolgerGrpc::new(routes).with_audit(audit.clone()));

        let resp = SearchService::search(
            &grpc,
            Request::new(SearchRequest {
                name: "serde".into(),
                limit: 0,
                ..Default::default()
            }),
        )
        .await
        .expect("search ok")
        .into_inner();

        // serde (rust-dev) + serde_json (extra) match; tokio must be EXCLUDED.
        assert_eq!(resp.artifacts.len(), 2, "both serde* names match");
        assert!(
            !resp.artifacts.iter().any(|h| h.id.as_ref().unwrap().name == "tokio"),
            "tokio must be EXCLUDED from a serde search"
        );
        // Hits carry the repo they were found in.
        let repos: std::collections::HashSet<_> =
            resp.artifacts.iter().map(|h| h.repository.clone()).collect();
        assert!(repos.contains("rust-dev") && repos.contains("extra"));

        // Audited as a List with the hit count.
        let ev = audit.events();
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, AuditAction::List);
        assert_eq!(ev[0].bytes, 2, "audit records the hit count");
    }

    /// The Search RPC's `property` axis (§8): with a property store installed
    /// (`with_properties`), a `PropertyMatch` filter returns ONLY the tagged
    /// coordinate and EXCLUDES the untagged one (RED-when-broken). Without the
    /// store the same filter matches nothing (fail-closed).
    #[tokio::test]
    async fn search_rpc_property_axis_filters_and_fails_closed() {
        use super::holger_proto::PropertyMatch;
        let a: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
            name: "rust-dev".into(),
            entries: vec![("serde".into(), 10), ("tokio".into(), 20)],
        });
        let routes = FastRoutes::new(vec![("rust-dev".into(), a)]);

        // Tag only serde 1.0.0 with env=prod.
        let tmp = tempfile::tempdir().unwrap();
        let store: crate::properties::SharedPropertyStore =
            Arc::new(crate::properties::PropertyStore::new(tmp.path()));
        let serde_id = traits::ArtifactId { namespace: None, name: "serde".into(), version: "1.0.0".into() };
        store.set("rust-dev", &serde_id, "env", vec!["prod".into()]).unwrap();

        let req = || SearchRequest {
            properties: vec![PropertyMatch { key: "env".into(), value: "prod".into() }],
            limit: 0,
            ..Default::default()
        };

        // With the store: only serde matches; tokio (untagged) is excluded.
        let grpc = Arc::new(HolgerGrpc::new(routes.clone()).with_properties(store));
        let resp = SearchService::search(&grpc, Request::new(req())).await.unwrap().into_inner();
        assert_eq!(resp.artifacts.len(), 1, "only the env=prod artifact");
        assert_eq!(resp.artifacts[0].id.as_ref().unwrap().name, "serde");

        // WITHOUT the store: the property filter fails closed (no hits).
        let grpc_nostore = Arc::new(HolgerGrpc::new(routes));
        let resp2 = SearchService::search(&grpc_nostore, Request::new(req())).await.unwrap().into_inner();
        assert_eq!(resp2.artifacts.len(), 0, "no property store ⇒ property filter matches nothing");
    }
}