holger-server-lib 0.6.9

//! The three tonic gRPC service impls on `Arc<HolgerGrpc>` — `RepositoryService`
//! (fetch/list/put/stream), `ArchiveService` (list/stat the znippy archive), and
//! `AdminService` (health/catalog/profile). Split out of the `grpc` module root to
//! keep it focused on the shared state + helpers; as a child module this still
//! reaches `HolgerGrpc`'s private fields/methods and the parent's free helpers
//! (`internal_err` / `peer_addr_string` / `artifact_label`).

use std::sync::Arc;
use tonic::{Request, Response, Status};
use tokio_stream::wrappers::ReceiverStream;

use super::holger_proto::admin_service_server::AdminService;
use super::holger_proto::archive_service_server::ArchiveService;
use super::holger_proto::promotion_service_server::PromotionService;
use super::holger_proto::repository_service_server::RepositoryService;
use super::holger_proto::search_service_server::SearchService;
use super::holger_proto::*;
use super::{artifact_label, internal_err, peer_addr_string, HolgerGrpc};
use crate::audit::{AuditAction, AuditEvent};
use crate::promote::PromoteError;
use crate::search;

/// Turn an empty proto string into `None` (proto3 has no optional string here),
/// anything else into `Some`. The search axes all use "" = "not set".
fn opt(s: String) -> Option<String> {
    if s.is_empty() {
        None
    } else {
        Some(s)
    }
}

/// Build the engine's [`search::SearchQuery`] from the proto request.
pub(crate) fn search_query_from_proto(req: SearchRequest) -> search::SearchQuery {
    search::SearchQuery {
        name: opt(req.name),
        namespace: opt(req.namespace),
        version: opt(req.version),
        checksum: opt(req.checksum),
        path: opt(req.path),
        properties: req
            .properties
            .into_iter()
            .filter_map(|p| {
                if p.key.is_empty() { None } else { Some((p.key, p.value)) }
            })
            .collect(),
        repositories: req.repositories.into_iter().filter(|r| !r.is_empty()).collect(),
        limit: req.limit.max(0) as usize,
    }
}

/// Convert a proto [`ArtifactId`] into the internal [`traits::ArtifactId`] (an
/// empty namespace string ⇒ `None`). Shared by the SBOM RPCs.
pub(crate) fn trait_id_from_proto(p: &ArtifactId) -> traits::ArtifactId {
    traits::ArtifactId {
        namespace: if p.namespace.is_empty() { None } else { Some(p.namespace.clone()) },
        name: p.name.clone(),
        version: p.version.clone(),
    }
}

/// Map engine [`search::SearchResults`] onto the proto response.
pub(crate) fn search_results_to_proto(r: search::SearchResults) -> SearchResponse {
    SearchResponse {
        artifacts: r
            .artifacts
            .into_iter()
            .map(|h| ArtifactHit {
                repository: h.repository,
                id: Some(ArtifactId {
                    namespace: h.id.namespace.unwrap_or_default(),
                    name: h.id.name,
                    version: h.id.version,
                }),
                size_bytes: h.size_bytes,
                content_type: h.content_type,
                checksum: h.checksum.unwrap_or_default(),
            })
            .collect(),
        paths: r
            .paths
            .into_iter()
            .map(|p| PathHit { repository: p.repository, path: p.path })
            .collect(),
        truncated: r.truncated,
    }
}

#[tonic::async_trait]
impl RepositoryService for Arc<HolgerGrpc> {
    async fn fetch_artifact(
        &self,
        request: Request<FetchArtifactRequest>,
    ) -> Result<Response<FetchArtifactResponse>, Status> {
        // Reads are open in this model (no auth), so the principal is anonymous.
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo_name = req.repository.clone();
        let repo = self.get_repo(&repo_name)?;
        let id = req.id.ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let artifact = artifact_label(&id);

        let trait_id = traits::ArtifactId {
            namespace: if id.namespace.is_empty() { None } else { Some(id.namespace) },
            name: id.name,
            version: id.version,
        };

        // Serve-time quarantine gate (property-policy §): a blob tagged with the
        // configured quarantine property (e.g. quarantine=true) is REFUSED here —
        // served as not_found so its existence isn't leaked — even though it is
        // stored. Off unless configured, so the default read path is unchanged.
        if self.is_quarantined(&repo_name, &trait_id) {
            self.record_audit(
                AuditEvent::new(
                    "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 404, 0,
                )
                .with_detail("quarantined — refused at serve boundary"),
            );
            super::functional_status(
                "holger-grpc/fetch_artifact", "quarantine_blocked", true, &artifact,
            );
            return Err(Status::not_found("Artifact not found"));
        }

        match repo.fetch(&trait_id) {
            Ok(Some(data)) => {
                let n = data.len() as u64;
                self.record_audit(AuditEvent::new(
                    "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 200, n,
                ));
                super::functional_status(
                    "holger-grpc/fetch_artifact", "artifact_resolved", true, &repo_name,
                );
                Ok(Response::new(FetchArtifactResponse {
                    size_bytes: data.len() as i64,
                    content_type: "application/octet-stream".into(),
                    data,
                }))
            }
            Ok(None) => {
                self.record_audit(AuditEvent::new(
                    "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 404, 0,
                ));
                super::functional_status(
                    "holger-grpc/fetch_artifact", "artifact_resolved", false, "not found",
                );
                Err(Status::not_found("Artifact not found"))
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 500, 0,
                    )
                    .with_detail(e.to_string()),
                );
                super::functional_status(
                    "holger-grpc/fetch_artifact", "artifact_resolved", false, "backend error",
                );
                Err(internal_err(&e))
            }
        }
    }

    async fn list_artifacts(
        &self,
        request: Request<ListArtifactsRequest>,
    ) -> Result<Response<ListArtifactsResponse>, Status> {
        // Listings are open reads (no auth) → anonymous principal. The audit
        // artifact column carries the name filter (empty = whole repo).
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo_name = req.repository.clone();
        let filter_label = req.name_filter.clone();
        let repo = match self.get_repo(&repo_name) {
            Ok(r) => r,
            Err(status) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::List, &repo_name, &filter_label, &source_ip, 404, 0,
                    )
                    .with_detail("repository not found"),
                );
                return Err(status);
            }
        };

        // Empty proto string means "no filter" (proto3 has no optional string
        // here); anything else is a real substring filter.
        let name_filter = if req.name_filter.is_empty() {
            None
        } else {
            Some(req.name_filter.as_str())
        };
        // proto `limit` is int32 (the page size); the backend takes a usize.
        let limit = req.limit.max(0) as usize;
        // `page_token` is the opaque offset cursor; empty = first page.
        let page_token = if req.page_token.is_empty() {
            None
        } else {
            Some(req.page_token.as_str())
        };

        let (entries, next_page_token) =
            match crate::object::paginate_artifacts(repo.as_ref(), name_filter, limit, page_token) {
                Ok(v) => v,
                Err(e) => {
                    self.record_audit(
                        AuditEvent::new(
                            "anonymous", AuditAction::List, &repo_name, &filter_label, &source_ip, 500, 0,
                        )
                        .with_detail(e.to_string()),
                    );
                    return Err(internal_err(&e));
                }
            };

        self.record_audit(AuditEvent::new(
            "anonymous", AuditAction::List, &repo_name, &filter_label, &source_ip, 200, entries.len() as u64,
        ));
        super::functional_status(
            "holger-grpc/list_artifacts", "listing_served", true, &repo_name,
        );

        let artifacts = entries
            .into_iter()
            .map(|e| ArtifactEntry {
                id: Some(ArtifactId {
                    // None namespace (Rust-style, no namespace) → empty proto
                    // string, the wire convention this server uses elsewhere.
                    namespace: e.id.namespace.unwrap_or_default(),
                    name: e.id.name,
                    version: e.id.version,
                }),
                size_bytes: e.size_bytes,
                content_type: e.content_type,
            })
            .collect();

        Ok(Response::new(ListArtifactsResponse {
            artifacts,
            next_page_token,
        }))
    }

    async fn put_artifact(
        &self,
        request: Request<PutArtifactRequest>,
    ) -> Result<Response<PutArtifactResponse>, Status> {
        let source_ip = peer_addr_string(&request);
        // Target repo is known from the request body — authorize the write SCOPED
        // to it, so a per-repo ACL (`repo_roles`) can grant/deny per repository.
        let repo_name = request.get_ref().repository.clone();
        // The resolved write principal (OIDC sub / mTLS CN), or anonymous when
        // auth is open. Captured for the audit trail.
        let ident = self
            .authorize_write(&request, &repo_name)
            .await?
            .map(|id| id.subject)
            .unwrap_or_else(|| "anonymous".to_string());
        let req = request.into_inner();
        let repo = self.get_repo(&repo_name)?;

        if !repo.is_writable() {
            self.record_audit(
                AuditEvent::new(&ident, AuditAction::Upload, &repo_name, "", &source_ip, 403, 0)
                    .with_detail("repository is read-only"),
            );
            return Err(Status::permission_denied("Repository is read-only"));
        }

        // Ingest body cap (matches the HTTP gateway). An oversized gRPC upload
        // must be rejected rather than buffered/stored unbounded.
        if req.data.len() > self.max_body_bytes {
            self.record_audit(
                AuditEvent::new(
                    &ident,
                    AuditAction::Upload,
                    &repo_name,
                    "",
                    &source_ip,
                    413,
                    req.data.len() as u64,
                )
                .with_detail("payload too large"),
            );
            return Err(Status::resource_exhausted(format!(
                "Payload too large: {} bytes exceeds limit of {} bytes",
                req.data.len(),
                self.max_body_bytes
            )));
        }

        let id = req.id.ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let artifact = artifact_label(&id);
        let trait_id = traits::ArtifactId {
            namespace: if id.namespace.is_empty() { None } else { Some(id.namespace) },
            name: id.name,
            version: id.version,
        };
        let nbytes = req.data.len() as u64;

        match repo.put(&trait_id, &req.data) {
            Ok(()) => {
                self.record_audit(AuditEvent::new(
                    &ident, AuditAction::Upload, &repo_name, &artifact, &source_ip, 200, nbytes,
                ));
                super::functional_status(
                    "holger-grpc/put_artifact", "artifact_stored", true, &repo_name,
                );
                Ok(Response::new(PutArtifactResponse {
                    success: true,
                    message: "Artifact stored".into(),
                }))
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident, AuditAction::Upload, &repo_name, &artifact, &source_ip, 500, 0,
                    )
                    .with_detail(e.to_string()),
                );
                super::functional_status(
                    "holger-grpc/put_artifact", "artifact_stored", false, "backend error",
                );
                Err(internal_err(&e))
            }
        }
    }

    type StreamArtifactStream = ReceiverStream<Result<ArtifactChunk, Status>>;

    async fn stream_artifact(
        &self,
        request: Request<FetchArtifactRequest>,
    ) -> Result<Response<Self::StreamArtifactStream>, Status> {
        // Streamed reads are open (no auth) → anonymous principal.
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo_name = req.repository.clone();
        let repo = self.get_repo(&repo_name)?;
        let id = req.id.ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let artifact = artifact_label(&id);

        let trait_id = traits::ArtifactId {
            namespace: if id.namespace.is_empty() { None } else { Some(id.namespace) },
            name: id.name,
            version: id.version,
        };

        let data = match repo.fetch(&trait_id) {
            Ok(Some(data)) => {
                self.record_audit(AuditEvent::new(
                    "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 200, data.len() as u64,
                ));
                data
            }
            Ok(None) => {
                self.record_audit(AuditEvent::new(
                    "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 404, 0,
                ));
                return Err(Status::not_found("Artifact not found"));
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::Download, &repo_name, &artifact, &source_ip, 500, 0,
                    )
                    .with_detail(e.to_string()),
                );
                return Err(internal_err(&e));
            }
        };

        super::functional_status(
            "holger-grpc/stream_artifact", "stream_opened", true, &repo_name,
        );

        let (tx, rx) = tokio::sync::mpsc::channel(16);
        const CHUNK_SIZE: usize = 64 * 1024; // 64KB chunks

        // Take ownership of the artifact once as a refcounted `Bytes` buffer; each
        // chunk frame is then a cheap `Bytes::slice` (refcount bump, no per-chunk
        // allocation/copy) rather than a fresh `Vec<u8>` per 64 KiB chunk.
        let buf = prost::bytes::Bytes::from(data);
        tokio::spawn(async move {
            let mut offset = 0;
            while offset < buf.len() {
                let end = (offset + CHUNK_SIZE).min(buf.len());
                if tx.send(Ok(ArtifactChunk { data: buf.slice(offset..end) })).await.is_err() {
                    break;
                }
                offset = end;
            }
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }
}

// ─── ArchiveService ──────────────────────────────────────────────────

#[tonic::async_trait]
impl ArchiveService for Arc<HolgerGrpc> {
    async fn list_archive_files(
        &self,
        request: Request<ListArchiveFilesRequest>,
    ) -> Result<Response<ListArchiveFilesResponse>, Status> {
        // Open read → anonymous principal. The audit artifact column carries
        // the path-prefix filter (empty = whole archive).
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo_name = req.repository.clone();
        let prefix_label = req.prefix.clone();
        let repo = match self.get_repo(&repo_name) {
            Ok(r) => r,
            Err(status) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::List, &repo_name, &prefix_label, &source_ip, 404, 0,
                    )
                    .with_detail("repository not found"),
                );
                return Err(status);
            }
        };

        // Empty proto string means "no prefix filter" (proto3 has no optional
        // string here); anything else is a real path-prefix filter.
        let prefix = if req.prefix.is_empty() {
            None
        } else {
            Some(req.prefix.as_str())
        };

        let paths = match repo.archive_files(prefix) {
            Ok(p) => p,
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::List, &repo_name, &prefix_label, &source_ip, 500, 0,
                    )
                    .with_detail(e.to_string()),
                );
                return Err(internal_err(&e));
            }
        };

        self.record_audit(AuditEvent::new(
            "anonymous", AuditAction::List, &repo_name, &prefix_label, &source_ip, 200, paths.len() as u64,
        ));
        super::functional_status(
            "holger-grpc/list_archive_files", "archive_listed", true, &repo_name,
        );

        Ok(Response::new(ListArchiveFilesResponse {
            // proto `total_files` is int64.
            total_files: paths.len() as i64,
            paths,
        }))
    }

    async fn archive_info(
        &self,
        request: Request<ArchiveInfoRequest>,
    ) -> Result<Response<ArchiveInfoResponse>, Status> {
        // Open read of archive metadata → anonymous principal, no specific
        // artifact.
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo_name = req.repository.clone();
        let repo = match self.get_repo(&repo_name) {
            Ok(r) => r,
            Err(status) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::List, &repo_name, "", &source_ip, 404, 0,
                    )
                    .with_detail("repository not found"),
                );
                return Err(status);
            }
        };

        let info = match repo.archive_info() {
            Ok(i) => i,
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        "anonymous", AuditAction::List, &repo_name, "", &source_ip, 500, 0,
                    )
                    .with_detail(e.to_string()),
                );
                return Err(internal_err(&e));
            }
        };

        self.record_audit(AuditEvent::new(
            "anonymous", AuditAction::List, &repo_name, "", &source_ip, 200, info.file_count,
        ));
        super::functional_status(
            "holger-grpc/archive_info", "archive_stat_served", true, &repo_name,
        );

        Ok(Response::new(ArchiveInfoResponse {
            // proto counters are int64; the trait carries u64.
            file_count: info.file_count as i64,
            total_uncompressed_bytes: info.total_uncompressed_bytes as i64,
            archive_path: info.archive_path,
        }))
    }
}

// ─── AdminService ────────────────────────────────────────────────────

#[tonic::async_trait]
impl AdminService for Arc<HolgerGrpc> {
    async fn health(
        &self,
        _request: Request<HealthRequest>,
    ) -> Result<Response<HealthResponse>, Status> {
        super::functional_status(
            "holger-grpc/health", "health_reported", true, env!("CARGO_PKG_VERSION"),
        );
        Ok(Response::new(HealthResponse {
            status: "ok".into(),
            version: env!("CARGO_PKG_VERSION").into(),
            uptime_seconds: self.start_time.elapsed().as_secs() as i64,
        }))
    }

    async fn list_repositories(
        &self,
        request: Request<ListRepositoriesRequest>,
    ) -> Result<Response<ListRepositoriesResponse>, Status> {
        // Open read of the repository catalog → anonymous principal, no repo or
        // artifact targeted.
        let source_ip = peer_addr_string(&request);
        let repos = self.routes.all_repos();
        self.record_audit(AuditEvent::new(
            "anonymous", AuditAction::List, "", "", &source_ip, 200, repos.len() as u64,
        ));
        super::functional_status(
            "holger-grpc/list_repositories",
            "catalog_served",
            true,
            &format!("{} repos", repos.len()),
        );
        let infos = repos
            .iter()
            .map(|(name, repo)| RepositoryInfo {
                name: name.clone(),
                repo_type: format!("{:?}", repo.format()),
                writable: repo.is_writable(),
                has_archive: repo.has_archive(),
            })
            .collect();

        Ok(Response::new(ListRepositoriesResponse {
            repositories: infos,
        }))
    }

    /// Report the server's operating profile — server-reported truth so the
    /// operator UI reflects the server, not a local toggle. Read-only is the
    /// explicit sealed-mode override if set, else derived from the route table
    /// (read-only iff no repository accepts writes). Maps to the demo's static
    /// "rigged for silent running" chrome.
    async fn server_profile(
        &self,
        _request: Request<ServerProfileRequest>,
    ) -> Result<Response<ServerProfileResponse>, Status> {
        let read_only = self.is_read_only();
        let writable = self.writable_repo_count() as i32;
        let (profile, label) = if read_only {
            ("static", "RIGGED FOR SILENT RUNNING / READ-ONLY")
        } else {
            ("dynamic", "DYNAMIC / WRITABLE")
        };
        super::functional_status(
            "holger-grpc/server_profile", "profile_reported", true, profile,
        );
        Ok(Response::new(ServerProfileResponse {
            profile: profile.into(),
            read_only,
            label: label.into(),
            writable_repo_count: writable,
        }))
    }
}

// ─── PromotionService (DEV → PROD) ───────────────────────────────────
//
// The one write that is ADMIN-gated (fail-closed) rather than writer-gated:
// promotion moves an artifact from a mutable DEV repo into the immutable,
// content-addressed PROD store. Bytes are read from the source backend,
// content-addressed (blake3), and SEALED into the store — where a promoted
// coordinate is immutable (a re-promote is refused with `already_exists`). Every
// promotion (success or refusal) is audited under `AuditAction::Promote`.

#[tonic::async_trait]
impl PromotionService for Arc<HolgerGrpc> {
    async fn promote_artifact(
        &self,
        request: Request<PromoteArtifactRequest>,
    ) -> Result<Response<PromoteArtifactResponse>, Status> {
        let source_ip = peer_addr_string(&request);
        // ADMIN gate, fail-closed, SCOPED to the promotion DESTINATION repo (a
        // per-repo ACL can name who may promote INTO a given PROD shelf). A
        // non-admin (or unauthenticated, when policy is set) caller is refused
        // here, before any store access.
        let to_repo = request.get_ref().to_repo.clone();
        let ident = self
            .authorize_admin(&request, &to_repo)
            .await?
            .map(|id| id.subject)
            .unwrap_or_else(|| "anonymous".to_string());

        let engine = match &self.promotion {
            Some(e) => e.clone(),
            None => {
                return Err(Status::failed_precondition(
                    "promotion is not configured on this server",
                ));
            }
        };

        let req = request.into_inner();
        let id_proto = req
            .artifact
            .ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let label = artifact_label(&id_proto);
        let trait_id = traits::ArtifactId {
            namespace: if id_proto.namespace.is_empty() {
                None
            } else {
                Some(id_proto.namespace.clone())
            },
            name: id_proto.name.clone(),
            version: id_proto.version.clone(),
        };

        // Resolve the source (DEV) backend and read the bytes to promote.
        let source = self.get_repo(&req.from_repo)?;
        let data = match source.fetch(&trait_id) {
            Ok(Some(d)) => d,
            Ok(None) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident,
                        AuditAction::Promote,
                        &req.to_repo,
                        &label,
                        &source_ip,
                        404,
                        0,
                    )
                    .with_detail(format!("artifact not found in source repo '{}'", req.from_repo)),
                );
                return Err(Status::not_found(format!(
                    "artifact '{}' not found in source repo '{}'",
                    label, req.from_repo
                )));
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident,
                        AuditAction::Promote,
                        &req.to_repo,
                        &label,
                        &source_ip,
                        500,
                        0,
                    )
                    .with_detail(e.to_string()),
                );
                return Err(internal_err(&e));
            }
        };

        // Policy gate (design §4) — run the REAL CVE/license/provenance scan at
        // the boundary BEFORE sealing. Pass-through when no gate is configured
        // (the default path is unchanged); fail-closed when a gate is configured
        // and the verdict Blocks. The source repo's `format()` keys the ecosystem
        // the scanner scores against.
        if engine.has_gate() {
            let ecosystem = crate::security::ecosystem_for_format(source.format());
            if let Err(e) = engine.gate_check(&ecosystem, &trait_id) {
                let detail = e.to_string();
                self.record_audit(
                    AuditEvent::new(
                        &ident,
                        AuditAction::Promote,
                        &req.to_repo,
                        &label,
                        &source_ip,
                        412,
                        0,
                    )
                    .with_detail(detail.clone()),
                );
                super::functional_status(
                    "holger-grpc/promote_artifact",
                    "artifact_promoted",
                    false,
                    "blocked by policy gate",
                );
                return Err(Status::failed_precondition(detail));
            }
        }

        // Property-requirement gate (property-policy §): only promote when the
        // SOURCE coordinate carries the required custom property (e.g. signed=true).
        // Fail-closed — a requirement configured but no property store means we
        // cannot verify it, so the promotion is refused rather than waved through.
        if let Some((key, value)) = engine.required_property() {
            let props = match &self.properties {
                Some(store) => store.get(&req.from_repo, &trait_id),
                None => crate::properties::PropertyMap::new(),
            };
            if !engine.property_requirement_met(&props) {
                let detail = format!(
                    "promotion requires property '{key}{}' on the source artifact",
                    if value.is_empty() { String::new() } else { format!("={value}") }
                );
                self.record_audit(
                    AuditEvent::new(&ident, AuditAction::Promote, &req.to_repo, &label, &source_ip, 412, 0)
                        .with_detail(detail.clone()),
                );
                super::functional_status(
                    "holger-grpc/promote_artifact",
                    "artifact_promoted",
                    false,
                    "blocked by property requirement",
                );
                return Err(Status::failed_precondition(detail));
            }
        }

        // Timestamp from the lib's existing wall-clock source (audit already uses
        // it); seconds since the epoch for the record.
        let promoted_at = crate::audit::now_nanos() / 1_000_000_000;

        match engine.seal(&req.from_repo, &req.to_repo, &trait_id, &data, &ident, promoted_at) {
            Ok(record) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident,
                        AuditAction::Promote,
                        &req.to_repo,
                        &label,
                        &source_ip,
                        200,
                        record.size,
                    )
                    .with_detail(format!(
                        "promoted from '{}' — content {}",
                        req.from_repo, record.content_address
                    )),
                );
                super::functional_status(
                    "holger-grpc/promote_artifact",
                    "artifact_promoted",
                    true,
                    &req.to_repo,
                );
                Ok(Response::new(PromoteArtifactResponse {
                    success: true,
                    message: format!("promoted '{}' to '{}'", label, req.to_repo),
                    content_address: record.content_address,
                    promoted_by: ident,
                }))
            }
            // Immutable: the coordinate is already sealed — refuse the overwrite.
            Err(PromoteError::AlreadyPromoted) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident,
                        AuditAction::Promote,
                        &req.to_repo,
                        &label,
                        &source_ip,
                        409,
                        0,
                    )
                    .with_detail("already promoted — immutable, overwrite refused"),
                );
                super::functional_status(
                    "holger-grpc/promote_artifact",
                    "artifact_promoted",
                    false,
                    "already promoted (immutable)",
                );
                Err(Status::already_exists(format!(
                    "'{}' is already promoted to '{}' (immutable)",
                    label, req.to_repo
                )))
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident,
                        AuditAction::Promote,
                        &req.to_repo,
                        &label,
                        &source_ip,
                        500,
                        0,
                    )
                    .with_detail(e.to_string()),
                );
                Err(internal_err(&e))
            }
        }
    }
}

// ─── SearchService (cross-repo artifact discovery) ───────────────────
//
// A read-only aggregate over every configured backend's `list()` +
// `archive_files()`. Reads are open (anonymous) like the other read RPCs; the
// pure `search::search_repos` engine does the matching, so the same behaviour
// backs the HTTP `/-/search` door and the `holger-server search` CLI verb.

#[tonic::async_trait]
impl SearchService for Arc<HolgerGrpc> {
    async fn search(
        &self,
        request: Request<SearchRequest>,
    ) -> Result<Response<SearchResponse>, Status> {
        // Open read → anonymous principal. The audit artifact column carries a
        // compact rendering of the query terms.
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let query = search_query_from_proto(req);
        let query_label = format!(
            "name={} ns={} ver={} sum={} path={} props={}",
            query.name.as_deref().unwrap_or("-"),
            query.namespace.as_deref().unwrap_or("-"),
            query.version.as_deref().unwrap_or("-"),
            if query.checksum.is_some() { "yes" } else { "-" },
            query.path.as_deref().unwrap_or("-"),
            query.properties.len(),
        );

        // The engine is synchronous + may fetch bytes for checksum search, so run
        // it off the async reactor. The property store (when configured) powers the
        // `property` axis; without one, a property filter fails closed.
        let repos = self.routes.all_repos();
        let props = self.properties.clone();
        let results = tokio::task::spawn_blocking(move || {
            search::search_repos_with_properties(
                &repos,
                &query,
                props.as_deref().map(|p| p as &dyn search::PropertyLookup),
            )
        })
            .await
            .map_err(|e| {
                super::functional_status(
                    "holger-grpc/search", "search_served", false, "dispatch task failed",
                );
                internal_err(format!("search task failed: {e}"))
            })?;

        let hit_count = (results.artifacts.len() + results.paths.len()) as u64;
        self.record_audit(AuditEvent::new(
            "anonymous", AuditAction::List, "", &query_label, &source_ip, 200, hit_count,
        ));
        super::functional_status(
            "holger-grpc/search",
            "search_served",
            true,
            &format!("{hit_count} hits"),
        );

        Ok(Response::new(search_results_to_proto(results)))
    }
}

// ─── SbomService (host + serve SBOM documents) ───────────────────────
//
// Attach hosts a CycloneDX SBOM document against an artifact coordinate; Fetch
// serves it back byte-identical. Attach is a WRITE (writer/admin, fail-closed
// under a configured role policy), scoped to the artifact's repository; Fetch is
// an open read, mirrored by the read-only `GET /-/sbom` HTTP door. Both audit and
// emit a functional_status cell. The store is the shared `sbom::SbomStore`, so the
// exact same bytes back the HTTP door and the `holger-server sbom` CLI verb.

use super::holger_proto::sbom_service_server::SbomService;

#[tonic::async_trait]
impl SbomService for Arc<HolgerGrpc> {
    async fn attach_sbom(
        &self,
        request: Request<AttachSbomRequest>,
    ) -> Result<Response<AttachSbomResponse>, Status> {
        let source_ip = peer_addr_string(&request);
        let repo = request.get_ref().repository.clone();
        // WRITE gate (writer/admin), scoped to the artifact's repository —
        // fail-closed under a configured role policy, open when auth is unset.
        let ident = self
            .authorize_write(&request, &repo)
            .await?
            .map(|id| id.subject)
            .unwrap_or_else(|| "anonymous".to_string());

        let store = match &self.sboms {
            Some(s) => s.clone(),
            None => {
                return Err(Status::failed_precondition(
                    "SBOM hosting is not configured on this server",
                ))
            }
        };

        let req = request.into_inner();
        let id_proto = req
            .artifact
            .ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let label = artifact_label(&id_proto);
        let trait_id = trait_id_from_proto(&id_proto);
        if req.document.is_empty() {
            return Err(Status::invalid_argument("empty SBOM document"));
        }
        let size = req.document.len() as u64;

        match store.attach(&repo, &trait_id, &req.document) {
            Ok(content_address) => {
                self.record_audit(
                    AuditEvent::new(
                        &ident, AuditAction::Upload, &repo, &label, &source_ip, 200, size,
                    )
                    .with_detail(format!("sbom attached — content {content_address}")),
                );
                super::functional_status("holger-grpc/attach_sbom", "sbom_attached", true, &label);
                Ok(Response::new(AttachSbomResponse {
                    success: true,
                    message: format!("SBOM attached to '{label}' in '{repo}'"),
                    content_address,
                }))
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(&ident, AuditAction::Upload, &repo, &label, &source_ip, 500, 0)
                        .with_detail(e.to_string()),
                );
                super::functional_status(
                    "holger-grpc/attach_sbom",
                    "sbom_attached",
                    false,
                    "store write failed",
                );
                Err(internal_err(&e))
            }
        }
    }

    async fn fetch_sbom(
        &self,
        request: Request<FetchSbomRequest>,
    ) -> Result<Response<FetchSbomResponse>, Status> {
        // Open read (anonymous), like the other read RPCs + the HTTP door.
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo = req.repository;
        let id_proto = req
            .artifact
            .ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let label = artifact_label(&id_proto);
        let trait_id = trait_id_from_proto(&id_proto);

        let store = match &self.sboms {
            Some(s) => s.clone(),
            None => {
                return Err(Status::not_found(
                    "SBOM hosting is not configured on this server",
                ))
            }
        };

        match store.fetch(&repo, &trait_id) {
            Ok(Some(document)) => {
                let size = document.len() as i64;
                let content_address = blake3::hash(&document).to_hex().to_string();
                self.record_audit(AuditEvent::new(
                    "anonymous",
                    AuditAction::Download,
                    &repo,
                    &label,
                    &source_ip,
                    200,
                    size as u64,
                ));
                super::functional_status("holger-grpc/fetch_sbom", "sbom_served", true, &label);
                Ok(Response::new(FetchSbomResponse {
                    document,
                    size_bytes: size,
                    content_address,
                }))
            }
            // A missing SBOM is a not-found, never an error (RED-when-broken: the
            // HTTP/CLI twins assert this 404/None path independently).
            Ok(None) => {
                self.record_audit(AuditEvent::new(
                    "anonymous",
                    AuditAction::Download,
                    &repo,
                    &label,
                    &source_ip,
                    404,
                    0,
                ));
                super::functional_status(
                    "holger-grpc/fetch_sbom",
                    "sbom_served",
                    false,
                    "no SBOM for coordinate",
                );
                Err(Status::not_found(format!(
                    "no SBOM hosted for '{label}' in '{repo}'"
                )))
            }
            Err(e) => Err(internal_err(&e)),
        }
    }
}

// ─── PropertyService (custom artifact properties — write + read) ──────────
//
// SetProperty sets/removes a `key`'s values on a coordinate; GetProperties reads
// the full map back. SetProperty is a WRITE (writer/admin, fail-closed under a
// configured role policy, scoped to the artifact's repository) — the remote twin
// of the `holger-server properties set` CLI verb; GetProperties is an open read,
// mirrored by `GET /-/properties`. Both share the one `properties::PropertyStore`
// that also backs the `property` search axis, so a set here is immediately
// searchable and byte-consistent across every surface.

use super::holger_proto::property_service_server::PropertyService;

/// Convert a [`properties::PropertyMap`] into the proto entry list (stable key
/// order — the map is a `BTreeMap`).
fn property_map_to_proto(map: crate::properties::PropertyMap) -> Vec<PropertyEntry> {
    map.into_iter()
        .map(|(key, values)| PropertyEntry { key, values })
        .collect()
}

#[tonic::async_trait]
impl PropertyService for Arc<HolgerGrpc> {
    async fn set_property(
        &self,
        request: Request<SetPropertyRequest>,
    ) -> Result<Response<SetPropertyResponse>, Status> {
        let source_ip = peer_addr_string(&request);
        let repo = request.get_ref().repository.clone();
        // WRITE gate (writer/admin), scoped to the artifact's repository —
        // fail-closed under a configured role policy, open when auth is unset.
        let ident = self
            .authorize_write(&request, &repo)
            .await?
            .map(|id| id.subject)
            .unwrap_or_else(|| "anonymous".to_string());

        let store = match &self.properties {
            Some(s) => s.clone(),
            None => {
                return Err(Status::failed_precondition(
                    "custom properties are not configured on this server",
                ))
            }
        };

        let req = request.into_inner();
        let id_proto = req
            .artifact
            .ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let label = artifact_label(&id_proto);
        let trait_id = trait_id_from_proto(&id_proto);
        if req.key.is_empty() {
            return Err(Status::invalid_argument("empty property key"));
        }
        let removing = req.values.is_empty();

        match store.set(&repo, &trait_id, &req.key, req.values.clone()) {
            Ok(map) => {
                self.record_audit(
                    AuditEvent::new(&ident, AuditAction::Upload, &repo, &label, &source_ip, 200, 0)
                        .with_detail(format!(
                            "property '{}' {} on {label}",
                            req.key,
                            if removing { "removed" } else { "set" }
                        )),
                );
                // RED-when-broken observability: `ok` reflects that the key is now
                // present iff we were setting (absent iff removing).
                super::functional_status(
                    "holger-grpc/set_property",
                    "property_set",
                    map.contains_key(&req.key) != removing,
                    &req.key,
                );
                Ok(Response::new(SetPropertyResponse {
                    success: true,
                    message: format!(
                        "property '{}' {} on '{label}' in '{repo}'",
                        req.key,
                        if removing { "removed" } else { "set" }
                    ),
                    properties: property_map_to_proto(map),
                }))
            }
            Err(e) => {
                self.record_audit(
                    AuditEvent::new(&ident, AuditAction::Upload, &repo, &label, &source_ip, 500, 0)
                        .with_detail(e.to_string()),
                );
                super::functional_status(
                    "holger-grpc/set_property",
                    "property_set",
                    false,
                    "store write failed",
                );
                Err(internal_err(&e))
            }
        }
    }

    async fn get_properties(
        &self,
        request: Request<GetPropertiesRequest>,
    ) -> Result<Response<GetPropertiesResponse>, Status> {
        // Open read (anonymous), like the other read RPCs + the HTTP door.
        let source_ip = peer_addr_string(&request);
        let req = request.into_inner();
        let repo = req.repository;
        let id_proto = req
            .artifact
            .ok_or_else(|| Status::invalid_argument("Missing artifact ID"))?;
        let label = artifact_label(&id_proto);
        let trait_id = trait_id_from_proto(&id_proto);

        // No store configured ⇒ an empty map (an unset coordinate has none either
        // way), never an error — consistent with the HTTP door.
        let map = match &self.properties {
            Some(s) => s.get(&repo, &trait_id),
            None => crate::properties::PropertyMap::new(),
        };
        self.record_audit(AuditEvent::new(
            "anonymous",
            AuditAction::List,
            &repo,
            &label,
            &source_ip,
            200,
            map.len() as u64,
        ));
        super::functional_status(
            "holger-grpc/get_properties",
            "properties_served",
            true,
            &format!("{} key(s) {label}", map.len()),
        );
        Ok(Response::new(GetPropertiesResponse {
            properties: property_map_to_proto(map),
        }))
    }
}

// ─── ReplicationService (copy a repo's artifacts source → destination) ────
//
// Replicate every artifact from `from_repo` into a writable `to_repo` on THIS
// server, content-id idempotent (an artifact already present in the destination
// by content id is skipped, so a re-run is a no-op). Reads only from the source,
// writes only into the destination. ADMIN-gated on the destination repo (like
// promotion), fail-closed. Dry-run unless `execute`. The same pure
// `replication::replicate_pairs` engine backs the `holger-server replicate
// --from --to` CLI verb across two separate stores.

use super::holger_proto::replication_service_server::ReplicationService;
use crate::replication;

#[tonic::async_trait]
impl ReplicationService for Arc<HolgerGrpc> {
    async fn replicate(
        &self,
        request: Request<ReplicateRequest>,
    ) -> Result<Response<ReplicateResponse>, Status> {
        let source_ip = peer_addr_string(&request);
        // ADMIN gate, fail-closed, SCOPED to the DESTINATION repo (who may write
        // into `to_repo`). Refused before any store access.
        let to_repo = request.get_ref().to_repo.clone();
        let ident = self
            .authorize_admin(&request, &to_repo)
            .await?
            .map(|id| id.subject)
            .unwrap_or_else(|| "anonymous".to_string());

        let req = request.into_inner();
        if req.from_repo == req.to_repo {
            return Err(Status::invalid_argument(
                "replication source and destination repositories must differ",
            ));
        }
        let source = self.get_repo(&req.from_repo)?;
        let dest = self.get_repo(&req.to_repo)?;

        let pairs = vec![replication::RepoPair {
            source_name: req.from_repo.clone(),
            source,
            dest_name: req.to_repo.clone(),
            dest,
        }];
        let opts = replication::ReplicationOptions {
            execute: req.execute,
            limit: req.limit.max(0) as usize,
        };

        // The engine is synchronous + does fetch/put I/O, so run it off the async
        // reactor.
        let report = tokio::task::spawn_blocking(move || {
            replication::replicate_pairs(&pairs, &opts)
        })
        .await
        .map_err(|e| {
            super::functional_status(
                "holger-grpc/replicate", "replication_served", false, "dispatch task failed",
            );
            internal_err(format!("replication task failed: {e}"))
        })?;

        let copied = report.copied() as u64;
        let status_code = if report.is_complete() { 200 } else { 500 };
        self.record_audit(
            AuditEvent::new(
                &ident,
                AuditAction::Upload,
                &req.to_repo,
                &req.from_repo,
                &source_ip,
                status_code,
                report.copied_bytes() as u64,
            )
            .with_detail(format!(
                "replicate {}→{}: copied {} skipped {} failed {} (execute={})",
                req.from_repo,
                req.to_repo,
                report.copied(),
                report.skipped(),
                report.failed(),
                req.execute,
            )),
        );
        super::functional_status(
            "holger-grpc/replicate",
            "replication_served",
            report.is_complete(),
            &format!("copied {copied}, {} failed", report.failed()),
        );

        Ok(Response::new(ReplicateResponse {
            examined: report.examined() as i64,
            copied: report.copied() as i64,
            skipped: report.skipped() as i64,
            would_copy: report.would_copy() as i64,
            failed: report.failed() as i64,
            copied_bytes: report.copied_bytes(),
            truncated: report.truncated,
            complete: report.is_complete(),
            message: format!(
                "replicate {}→{} {}: examined {}, copied {}, skipped {}, failed {}",
                req.from_repo,
                req.to_repo,
                if req.execute { "executed" } else { "dry-run" },
                report.examined(),
                report.copied(),
                report.skipped(),
                report.failed(),
            ),
        }))
    }
}

#[cfg(test)]
mod replication_tests {
    use super::*;
    use crate::exposed::fast_routes::FastRoutes;
    use std::sync::Mutex;
    use traits::{ArtifactEntry, ArtifactFormat, RepositoryBackendTrait};

    /// A mutable in-memory backend (put really stores, so a copy is fetchable).
    struct MemRepo {
        name: String,
        writable: bool,
        store: Mutex<Vec<(traits::ArtifactId, Vec<u8>)>>,
    }
    #[tonic::async_trait]
    impl RepositoryBackendTrait for MemRepo {
        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.store.lock().unwrap().iter().find(|(e, _)| e == id).map(|(_, b)| b.clone()))
        }
        fn put(&self, id: &traits::ArtifactId, data: &[u8]) -> anyhow::Result<()> {
            if !self.writable {
                anyhow::bail!("read-only");
            }
            self.store.lock().unwrap().push((id.clone(), data.to_vec()));
            Ok(())
        }
        fn list(&self, _n: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
            Ok(self
                .store
                .lock()
                .unwrap()
                .iter()
                .map(|(id, b)| ArtifactEntry {
                    id: id.clone(),
                    size_bytes: b.len() as i64,
                    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()))
        }
    }

    fn seeded_id(name: &str) -> traits::ArtifactId {
        traits::ArtifactId { namespace: None, name: name.into(), version: "1".into() }
    }

    /// A gRPC state with a seeded read-only `dev` repo and a writable empty `prod`.
    fn grpc_pair() -> Arc<HolgerGrpc> {
        let dev: Arc<dyn RepositoryBackendTrait> = Arc::new(MemRepo {
            name: "dev".into(),
            writable: false,
            store: Mutex::new(vec![
                (traits::ArtifactId { namespace: None, name: "a".into(), version: "1".into() }, b"aaa".to_vec()),
                (traits::ArtifactId { namespace: None, name: "b".into(), version: "1".into() }, b"bbb".to_vec()),
            ]),
        });
        let prod: Arc<dyn RepositoryBackendTrait> =
            Arc::new(MemRepo { name: "prod".into(), writable: true, store: Mutex::new(Vec::new()) });
        Arc::new(HolgerGrpc::new(FastRoutes::new(vec![
            ("dev".to_string(), dev),
            ("prod".to_string(), prod),
        ])))
    }

    #[tokio::test]
    async fn replicate_rpc_copies_then_is_idempotent() {
        let grpc = grpc_pair();
        // Execute run copies both artifacts.
        let run = ReplicationService::replicate(
            &grpc,
            Request::new(ReplicateRequest {
                from_repo: "dev".into(),
                to_repo: "prod".into(),
                execute: true,
                limit: 0,
            }),
        )
        .await
        .expect("replicate ok")
        .into_inner();
        assert_eq!(run.copied, 2, "both artifacts copied");
        assert!(run.complete, "no failures");

        // RED-when-broken: the destination now serves the copied artifact.
        let prod = grpc.routes.lookup("prod").unwrap();
        assert_eq!(prod.fetch(&seeded_id("a")).unwrap().as_deref(), Some(b"aaa".as_ref()));

        // Re-run is a no-op (idempotent by content id).
        let again = ReplicationService::replicate(
            &grpc,
            Request::new(ReplicateRequest {
                from_repo: "dev".into(),
                to_repo: "prod".into(),
                execute: true,
                limit: 0,
            }),
        )
        .await
        .expect("replicate ok")
        .into_inner();
        assert_eq!(again.copied, 0, "re-run copies nothing");
        assert_eq!(again.skipped, 2, "both already present by content id");
    }

    #[tokio::test]
    async fn replicate_rpc_rejects_same_source_and_dest() {
        let grpc = grpc_pair();
        let err = ReplicationService::replicate(
            &grpc,
            Request::new(ReplicateRequest {
                from_repo: "dev".into(),
                to_repo: "dev".into(),
                execute: true,
                limit: 0,
            }),
        )
        .await
        .unwrap_err();
        assert_eq!(err.code(), tonic::Code::InvalidArgument);
    }
}

#[cfg(test)]
mod sbom_tests {
    use super::*;
    use crate::exposed::fast_routes::FastRoutes;
    use crate::sbom::{build_cyclonedx, SbomComponent, SbomInput, SbomStore};

    /// A gRPC state with an SBOM store rooted at `dir` and no repos/auth.
    fn grpc_with_sboms(dir: &std::path::Path) -> Arc<HolgerGrpc> {
        Arc::new(
            HolgerGrpc::new(FastRoutes::new(Vec::new())).with_sboms(Arc::new(SbomStore::new(dir))),
        )
    }

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

    /// Attach over the RPC, then fetch over the RPC: the served document is
    /// byte-identical to the attached one and carries its content address.
    #[tokio::test]
    async fn attach_then_fetch_rpc_is_byte_identical() {
        let tmp = tempfile::tempdir().unwrap();
        let grpc = grpc_with_sboms(tmp.path());
        let doc = build_cyclonedx(&SbomInput {
            subject: "rust-dev".into(),
            components: vec![SbomComponent {
                ecosystem: "cargo".into(),
                name: "serde".into(),
                version: "1.0.0".into(),
                license: Some("MIT OR Apache-2.0".into()),
            }],
            vulns: vec![],
        });

        let attached = SbomService::attach_sbom(
            &grpc,
            Request::new(AttachSbomRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
                document: doc.clone(),
            }),
        )
        .await
        .expect("attach ok")
        .into_inner();
        assert!(attached.success);

        let fetched = SbomService::fetch_sbom(
            &grpc,
            Request::new(FetchSbomRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
            }),
        )
        .await
        .expect("fetch ok")
        .into_inner();
        assert_eq!(fetched.document, doc, "served SBOM is byte-identical to the attached one");
        assert_eq!(fetched.content_address, attached.content_address);
        assert_eq!(fetched.size_bytes, doc.len() as i64);
    }

    /// RED-when-broken: fetching a coordinate with no attached SBOM is NOT_FOUND,
    /// and a wrong coordinate (different version) does not match an attached one.
    #[tokio::test]
    async fn fetch_missing_or_wrong_coordinate_is_not_found() {
        let tmp = tempfile::tempdir().unwrap();
        let grpc = grpc_with_sboms(tmp.path());

        // Nothing attached yet → NotFound.
        let miss = SbomService::fetch_sbom(
            &grpc,
            Request::new(FetchSbomRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
            }),
        )
        .await;
        assert_eq!(miss.unwrap_err().code(), tonic::Code::NotFound);

        // Attach serde@1.0.0, then fetch serde@2.0.0 → still NotFound.
        SbomService::attach_sbom(
            &grpc,
            Request::new(AttachSbomRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
                document: b"doc".to_vec(),
            }),
        )
        .await
        .expect("attach ok");
        let wrong = SbomService::fetch_sbom(
            &grpc,
            Request::new(FetchSbomRequest {
                repository: "rust-dev".into(),
                artifact: Some(ArtifactId {
                    namespace: String::new(),
                    name: "serde".into(),
                    version: "2.0.0".into(),
                }),
            }),
        )
        .await;
        assert_eq!(wrong.unwrap_err().code(), tonic::Code::NotFound);
    }
}

#[cfg(test)]
mod property_tests {
    use super::*;
    use crate::auth::{AuthConfig, Role};
    use crate::exposed::fast_routes::FastRoutes;
    use crate::properties::PropertyStore;

    /// A gRPC state with a property store rooted at `dir`, no repos, auth open.
    fn grpc_with_props(dir: &std::path::Path) -> Arc<HolgerGrpc> {
        Arc::new(
            HolgerGrpc::new(FastRoutes::new(Vec::new()))
                .with_properties(Arc::new(PropertyStore::new(dir))),
        )
    }

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

    /// SetProperty then GetProperties over the RPC: the map round-trips, and a set
    /// on one coordinate is not visible on another (RED-when-broken exclusion).
    #[tokio::test]
    async fn set_then_get_rpc_round_trip() {
        let tmp = tempfile::tempdir().unwrap();
        let grpc = grpc_with_props(tmp.path());

        let set = PropertyService::set_property(
            &grpc,
            Request::new(SetPropertyRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
                key: "env".into(),
                values: vec!["prod".into(), "staging".into()],
            }),
        )
        .await
        .expect("set ok")
        .into_inner();
        assert!(set.success);
        assert!(set.properties.iter().any(|e| e.key == "env" && e.values == vec!["prod", "staging"]));

        let got = PropertyService::get_properties(
            &grpc,
            Request::new(GetPropertiesRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
            }),
        )
        .await
        .expect("get ok")
        .into_inner();
        assert_eq!(got.properties.len(), 1);
        assert_eq!(got.properties[0].key, "env");
        assert_eq!(got.properties[0].values, vec!["prod".to_string(), "staging".to_string()]);

        // A DIFFERENT coordinate must not see it.
        let other = PropertyService::get_properties(
            &grpc,
            Request::new(GetPropertiesRequest {
                repository: "rust-dev".into(),
                artifact: Some(ArtifactId { namespace: String::new(), name: "serde".into(), version: "2.0.0".into() }),
            }),
        )
        .await
        .expect("get ok")
        .into_inner();
        assert!(other.properties.is_empty(), "another coordinate must not see the set property");
    }

    /// SetProperty with EMPTY values removes the key (and leaves other keys).
    #[tokio::test]
    async fn set_empty_values_removes_key_over_rpc() {
        let tmp = tempfile::tempdir().unwrap();
        let grpc = grpc_with_props(tmp.path());
        for (k, v) in [("env", vec!["prod".to_string()]), ("team", vec!["core".to_string()])] {
            PropertyService::set_property(
                &grpc,
                Request::new(SetPropertyRequest {
                    repository: "r".into(), artifact: Some(coord()), key: k.into(), values: v,
                }),
            )
            .await
            .expect("set ok");
        }
        // Remove env (empty values).
        let removed = PropertyService::set_property(
            &grpc,
            Request::new(SetPropertyRequest {
                repository: "r".into(), artifact: Some(coord()), key: "env".into(), values: vec![],
            }),
        )
        .await
        .expect("remove ok")
        .into_inner();
        // env gone, team kept.
        assert!(!removed.properties.iter().any(|e| e.key == "env"), "env removed");
        assert!(removed.properties.iter().any(|e| e.key == "team"), "team preserved");
    }

    /// RED-when-broken: SetProperty is WRITE-gated. With a role policy configured
    /// but no authenticated identity, the write is refused (unauthenticated) and
    /// nothing is written — a broken gate would 200 and persist the property.
    #[tokio::test]
    async fn set_property_is_write_gated() {
        let tmp = tempfile::tempdir().unwrap();
        // RBAC on (a default_role makes rbac_enabled() true), no auth methods ⇒
        // an unauthenticated write fails closed.
        let auth = AuthConfig { default_role: Some(Role::Reader), ..Default::default() };
        let grpc = Arc::new(
            HolgerGrpc::with_auth(FastRoutes::new(Vec::new()), Arc::new(auth))
                .with_properties(Arc::new(PropertyStore::new(tmp.path()))),
        );

        let denied = PropertyService::set_property(
            &grpc,
            Request::new(SetPropertyRequest {
                repository: "rust-dev".into(),
                artifact: Some(coord()),
                key: "env".into(),
                values: vec!["prod".into()],
            }),
        )
        .await;
        assert_eq!(
            denied.unwrap_err().code(),
            tonic::Code::Unauthenticated,
            "write gate must refuse an unauthenticated set under a role policy"
        );
        // And nothing was written — a subsequent open GET returns an empty map.
        let got = PropertyService::get_properties(
            &grpc,
            Request::new(GetPropertiesRequest { repository: "rust-dev".into(), artifact: Some(coord()) }),
        )
        .await
        .expect("get ok")
        .into_inner();
        assert!(got.properties.is_empty(), "a denied write must persist nothing");
    }

    /// SetProperty with no store configured is a failed-precondition (not a panic /
    /// silent success).
    #[tokio::test]
    async fn set_property_without_store_fails_precondition() {
        let grpc = Arc::new(HolgerGrpc::new(FastRoutes::new(Vec::new()))); // no .with_properties
        let err = PropertyService::set_property(
            &grpc,
            Request::new(SetPropertyRequest {
                repository: "r".into(), artifact: Some(coord()), key: "env".into(), values: vec!["prod".into()],
            }),
        )
        .await
        .expect_err("no store → error");
        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
    }
}