server/lib/src/object.rs
//! `HolgerObject` adapters — two transports behind one interface.
//!
//! An external partner holds an `Arc<dyn HolgerObject>` (see `holger-traits`)
//! and never cares what's behind the vtable, exactly like a Java interface
//! reference. Two concrete implementations live here:
//!
//! * [`LocalHolger`] — in-process: calls the local engine directly (no
//! network, no serialization). Built from the same [`FastRoutes`] table the
//! gRPC server uses, so it is `Send + Sync`.
//! * [`RemoteHolger`] — remote: forwards every call over gRPC using the
//! tonic-generated client. The language-neutral contract is `holger.proto`.
//!
//! Pick the transport once at startup; the CLI / UI-backend code is identical
//! either way.
use std::sync::Arc;
use std::time::Instant;
use anyhow::{anyhow, Result};
use async_trait::async_trait;
use traits::{
ArchiveInfo, ArtifactEntry, ArtifactId, Health, HolgerObject, PropertyKv,
RepositoryBackendTrait, RepositoryInfo, SbomDoc, SearchArtifactHit, SearchPathHit, SearchQuery,
SearchResults, ServerProfile,
};
use crate::exposed::fast_routes::FastRoutes;
/// Paginate a backend listing with an opaque offset cursor.
///
/// `RepositoryBackendTrait::list` makes no stable-order guarantee across calls
/// (a `HashMap` / `read_dir` may reorder), so offset paging can't slice the raw
/// listing directly. We fetch the full matching set once, sort it into a stable
/// total order (namespace, name, version), then return the requested window.
/// `page_token` is the integer offset (empty/None = start); the returned token
/// is the next offset, or empty when the listing is exhausted. `page_size == 0`
/// means "everything from the offset" (no further paging).
///
/// NB: fetching the whole set per page is O(n) — fine for the archive sizes
/// holger serves today; a cursor pushed into the backend would scale further.
pub(crate) fn paginate_artifacts(
repo: &dyn RepositoryBackendTrait,
name_filter: Option<&str>,
page_size: usize,
page_token: Option<&str>,
) -> Result<(Vec<ArtifactEntry>, String)> {
// A backend that does not list has no page; its catalog row says
// `supports_listing: false`, so the empty page is not a count of zero.
if !repo.caps().listing {
return Ok((Vec::new(), String::new()));
}
let offset: usize = page_token.and_then(|t| t.parse().ok()).unwrap_or(0);
let mut all = repo.list(name_filter, 0)?;
all.sort_by(|a, b| {
a.id.namespace
.cmp(&b.id.namespace)
.then_with(|| a.id.name.cmp(&b.id.name))
.then_with(|| a.id.version.cmp(&b.id.version))
});
let total = all.len();
if offset >= total {
return Ok((Vec::new(), String::new()));
}
let end = if page_size == 0 {
total
} else {
offset.saturating_add(page_size).min(total)
};
let page = all[offset..end].to_vec();
let next = if end < total {
end.to_string()
} else {
String::new()
};
Ok((page, next))
}
// ─── LocalHolger: direct in-process transport ────────────────────────
/// In-process `HolgerObject`. Resolves repositories through a [`FastRoutes`]
/// table and calls the backends directly.
pub struct LocalHolger {
routes: FastRoutes,
start: Instant,
#[cfg(feature = "intake")]
sboms: Option<crate::sbom::SharedSbomStore>,
#[cfg(feature = "intake")]
properties: Option<crate::properties::SharedPropertyStore>,
}
impl LocalHolger {
pub fn new(routes: FastRoutes) -> Self {
Self {
routes,
start: Instant::now(),
#[cfg(feature = "intake")]
sboms: None,
#[cfg(feature = "intake")]
properties: None,
}
}
#[cfg(feature = "intake")]
/// Serve `fetch_sbom` from `store` (the one the gRPC door and CLI share).
pub fn with_sboms(mut self, store: crate::sbom::SharedSbomStore) -> Self {
self.sboms = Some(store);
self
}
#[cfg(feature = "intake")]
/// Serve `get_properties` and the search property axis from `store`.
pub fn with_properties(mut self, store: crate::properties::SharedPropertyStore) -> Self {
self.properties = Some(store);
self
}
fn repo(&self, name: &str) -> Result<Arc<dyn RepositoryBackendTrait>> {
self.routes
.lookup(name)
.ok_or_else(|| anyhow!("Repository '{}' not found", name))
}
}
#[async_trait]
impl HolgerObject for LocalHolger {
async fn fetch(&self, repository: &str, id: &ArtifactId) -> Result<Option<traits::Bytes>> {
self.repo(repository)?.fetch(id)
}
async fn put(&self, repository: &str, id: &ArtifactId, data: &[u8]) -> Result<()> {
let repo = self.repo(repository)?;
if !repo.is_writable() {
anyhow::bail!("Repository '{}' is read-only", repository);
}
repo.put(id, data)
}
async fn list_repositories(&self) -> Result<Vec<RepositoryInfo>> {
Ok(self
.routes
.all_repos()
.into_iter()
.map(|(name, repo)| RepositoryInfo::of(&name, repo.as_ref()))
.collect())
}
/// List artifacts straight off the backend (no network). Mirrors the gRPC
/// transport's `ListArtifacts`, minus serialization: resolve the repo and
/// page through its listing with [`paginate_artifacts`] (offset cursor in
/// `page_token`, `limit` as the page size).
async fn list_artifacts(
&self,
repository: &str,
name_filter: Option<String>,
limit: u32,
page_token: Option<String>,
) -> Result<(Vec<ArtifactEntry>, String)> {
let repo = self.repo(repository)?;
paginate_artifacts(
repo.as_ref(),
name_filter.as_deref(),
limit as usize,
page_token.as_deref(),
)
}
/// Browse the backing archive's raw file paths straight off the backend (no
/// network). Resolves the repo and delegates to its `archive_files`.
async fn list_archive_files(
&self,
repository: &str,
prefix: Option<String>,
) -> Result<Vec<String>> {
let repo = self.repo(repository)?;
if !repo.caps().archive {
return Ok(Vec::new());
}
Ok(repo.archive()?.paths(prefix.as_deref()))
}
/// Archive stats straight off the backend (no network).
async fn archive_info(&self, repository: &str) -> Result<ArchiveInfo> {
let repo = self.repo(repository)?;
if !repo.caps().archive {
return Ok(ArchiveInfo {
archive_path: repository.to_string(),
..Default::default()
});
}
Ok(repo.archive()?.info())
}
async fn health(&self) -> Result<Health> {
Ok(Health {
status: "ok".into(),
// Single-source identity: `holger v<ver> (<git-hash>)` — matches the
// server bins' `--version` and the gRPC health surface exactly.
version: crate::VERSION_LINE.into(),
uptime_seconds: self.start.elapsed().as_secs() as i64,
})
}
async fn server_profile(&self) -> Result<ServerProfile> {
// Server truth derived from the live route table: read-only iff no
// repository accepts writes (an empty server serves nothing writable).
// Mirrors HolgerGrpc::server_profile (grpc.rs).
let writable = self
.routes
.all_repos()
.iter()
.filter(|(_, repo)| repo.is_writable())
.count() as i32;
let read_only = writable == 0;
let (profile, label) = if read_only {
("static", "RIGGED FOR SILENT RUNNING / READ-ONLY")
} else {
("dynamic", "DYNAMIC / WRITABLE")
};
Ok(ServerProfile {
profile: profile.into(),
read_only,
label: label.into(),
writable_repo_count: writable,
})
}
/// The same engine the gRPC `Search` runs, over every served repository.
async fn search(&self, query: &SearchQuery) -> Result<SearchResults> {
let opt = |s: &str| (!s.is_empty()).then(|| s.to_string());
let q = crate::search::SearchQuery {
name: opt(&query.name),
namespace: opt(&query.namespace),
version: opt(&query.version),
checksum: None,
path: opt(&query.path),
properties: Vec::new(),
repositories: query.repositories.clone(),
limit: query.limit.max(0) as usize,
};
let repos = self.routes.all_repos();
#[cfg(feature = "intake")]
let props = self.properties.clone();
let r = tokio::task::spawn_blocking(move || {
#[cfg(feature = "intake")]
let lookup = props
.as_deref()
.map(|p| p as &dyn crate::search::PropertyLookup);
#[cfg(not(feature = "intake"))]
let lookup = None;
crate::search::search_repos_with_properties(&repos, &q, lookup)
})
.await?;
Ok(SearchResults {
artifacts: r
.artifacts
.into_iter()
.map(|h| SearchArtifactHit {
repository: h.repository,
id: h.id,
size_bytes: h.size_bytes,
content_type: h.content_type,
})
.collect(),
paths: r
.paths
.into_iter()
.map(|p| SearchPathHit {
repository: p.repository,
path: p.path,
})
.collect(),
truncated: r.truncated,
})
}
/// `Ok(None)` when none is attached — and when no SBOM store is configured,
/// as the gRPC door answers `NotFound` for both.
async fn fetch_sbom(&self, repository: &str, id: &ArtifactId) -> Result<Option<SbomDoc>> {
#[cfg(feature = "intake")]
{
let Some(store) = &self.sboms else {
return Ok(None);
};
Ok(store.fetch(repository, id)?.map(|document| SbomDoc {
size_bytes: document.len() as i64,
content_address: blake3::hash(&document).to_hex().to_string(),
document,
}))
}
#[cfg(not(feature = "intake"))]
{
let _ = (repository, id);
anyhow::bail!("SBOM hosting is not built into this holger (feature `supply-chain`)")
}
}
/// Empty when none are set or no property store is configured (as the gRPC door).
async fn get_properties(&self, repository: &str, id: &ArtifactId) -> Result<Vec<PropertyKv>> {
#[cfg(feature = "intake")]
{
let Some(store) = &self.properties else {
return Ok(Vec::new());
};
Ok(store
.get(repository, id)
.into_iter()
.map(|(key, values)| PropertyKv { key, values })
.collect())
}
#[cfg(not(feature = "intake"))]
{
let _ = (repository, id);
anyhow::bail!(
"custom properties are not built into this holger (feature `supply-chain`)"
)
}
}
}
// ─── RemoteHolger: gRPC transport ────────────────────────────────────
/// The gRPC client of another Holger: compiled only with `fetch` (lane HOL-CDS; the one-way
/// downstream dials no Holger, so it carries no client).
#[cfg(feature = "fetch")]
mod remote {
use super::*;
use crate::grpc::holger_proto::{
admin_service_client::AdminServiceClient, archive_service_client::ArchiveServiceClient,
property_service_client::PropertyServiceClient,
repository_service_client::RepositoryServiceClient, sbom_service_client::SbomServiceClient,
search_service_client::SearchServiceClient, ArchiveInfoRequest,
ArtifactId as ProtoArtifactId, FetchArtifactRequest, FetchSbomRequest,
GetPropertiesRequest, HealthRequest, ListArchiveFilesRequest, ListArtifactsRequest,
ListRepositoriesRequest, PutArtifactRequest, SearchRequest, ServerProfileRequest,
};
use tonic::metadata::{Ascii, MetadataValue};
use tonic::service::interceptor::InterceptedService;
use tonic::service::Interceptor;
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
use tonic::Code;
use tonic::{Request, Status};
/// A tonic interceptor that stamps `authorization: Bearer <token>` onto every
/// request when a token is configured. A `None` token is a no-op, so the same
/// client construction path serves both authenticated and open servers.
#[derive(Clone)]
pub struct BearerAuth(Option<MetadataValue<Ascii>>);
impl Interceptor for BearerAuth {
fn call(&mut self, mut req: Request<()>) -> Result<Request<()>, Status> {
if let Some(v) = &self.0 {
req.metadata_mut().insert("authorization", v.clone());
}
Ok(req)
}
}
/// Remote `HolgerObject`. Forwards every call to a Holger gRPC server using the
/// tonic-generated client. Cheap to clone (the channel is reference-counted).
pub struct RemoteHolger {
channel: Channel,
auth: BearerAuth,
}
impl RemoteHolger {
/// Connect to a Holger gRPC endpoint, e.g. `http://127.0.0.1:50051`, with no
/// credentials (open servers / read-only browsing).
pub async fn connect(endpoint: impl Into<String>) -> Result<Self> {
let endpoint = endpoint.into();
crate::egress::check("remote holger", &endpoint)?;
let channel = Channel::from_shared(endpoint)?.connect().await?;
Ok(Self {
channel,
auth: BearerAuth(None),
})
}
/// Connect with an OIDC bearer token injected on every request — the auth
/// path a UI uses for write access (validated server-side by
/// `auth::validate_request`).
pub async fn connect_with_token(
endpoint: impl Into<String>,
token: impl AsRef<str>,
) -> Result<Self> {
let endpoint = endpoint.into();
crate::egress::check("remote holger", &endpoint)?;
let channel = Channel::from_shared(endpoint)?.connect().await?;
Ok(Self {
channel,
auth: bearer(Some(token.as_ref()))?,
})
}
/// Connect over TLS (and, when a `client_identity` is supplied, mTLS).
///
/// * `ca_pem` — PEM bytes of the CA to verify the server certificate
/// against. `None` falls back to the platform/webpki roots tonic uses by
/// default.
/// * `client_identity` — `(cert_pem, key_pem)` for the **client**
/// certificate that authenticates *us* to the server. `Some(..)` is what
/// turns plain TLS into mTLS (the holger CN → role auth path).
/// * `token` — an optional bearer token stamped on every request, exactly as
/// in [`RemoteHolger::connect_with_token`]; orthogonal to TLS so a server
/// can require either or both.
///
/// NOTE: this only establishes the client side. There is no end-to-end mTLS
/// integration test here — exercising it needs generated certs and a TLS
/// server (a manual / mannequin step), so this path is verified to *compile*
/// only. See the task brief (feature B).
pub async fn connect_with_tls(
endpoint: impl Into<String>,
ca_pem: Option<Vec<u8>>,
client_identity: Option<(Vec<u8>, Vec<u8>)>,
token: Option<String>,
) -> Result<Self> {
Self::connect_with_tls_named(endpoint, ca_pem, client_identity, token, None, None).await
}
/// [`Self::connect_with_tls`] with **the server's name** (lane OX-WGDS3): the TLS name its
/// certificate is checked against AND the requests' `:authority`, when the endpoint is an
/// address (a downstream reaching its estate's front at `https://10.0.2.2:<port>`, whose
/// certificate and repository route are `repo.<estate>`'s). `None`: the endpoint's host.
pub async fn connect_with_tls_named(
endpoint: impl Into<String>,
ca_pem: Option<Vec<u8>>,
client_identity: Option<(Vec<u8>, Vec<u8>)>,
token: Option<String>,
server_name: Option<String>,
authority: Option<String>,
) -> Result<Self> {
let mut tls = ClientTlsConfig::new();
if let Some(name) = &server_name {
tls = tls.domain_name(name.clone());
}
if let Some(ca) = ca_pem {
tls = tls.ca_certificate(Certificate::from_pem(ca));
}
if let Some((cert, key)) = client_identity {
tls = tls.identity(Identity::from_pem(cert, key));
}
let endpoint = endpoint.into();
crate::egress::check("remote holger", &endpoint)?;
let mut ep = Channel::from_shared(endpoint)?.tls_config(tls)?;
// The calls' `:authority`: the one named apart (lane OX-WGDS3), else the TLS name.
if let Some(name) = authority.as_ref().or(server_name.as_ref()) {
ep = ep.origin(format!("https://{name}").parse()?);
}
let channel = ep.connect().await?;
Ok(Self {
channel,
auth: bearer(token.as_deref())?,
})
}
/// Build from an already-established channel (no credentials).
pub fn from_channel(channel: Channel) -> Self {
Self {
channel,
auth: BearerAuth(None),
}
}
/// The channel this client speaks over — for another service on the same
/// connection (the mirror's sync report rides the mTLS channel it pulled on).
pub fn channel(&self) -> Channel {
self.channel.clone()
}
fn repo_client(&self) -> RepositoryServiceClient<InterceptedService<Channel, BearerAuth>> {
RepositoryServiceClient::with_interceptor(self.channel.clone(), self.auth.clone())
.max_decoding_message_size(crate::grpc::DEFAULT_MAX_BODY_BYTES)
.max_encoding_message_size(crate::grpc::DEFAULT_MAX_BODY_BYTES)
}
fn admin_client(&self) -> AdminServiceClient<InterceptedService<Channel, BearerAuth>> {
AdminServiceClient::with_interceptor(self.channel.clone(), self.auth.clone())
.max_decoding_message_size(crate::grpc::DEFAULT_MAX_BODY_BYTES)
.max_encoding_message_size(crate::grpc::DEFAULT_MAX_BODY_BYTES)
}
fn search_client(&self) -> SearchServiceClient<InterceptedService<Channel, BearerAuth>> {
SearchServiceClient::with_interceptor(self.channel.clone(), self.auth.clone())
}
fn sbom_client(&self) -> SbomServiceClient<InterceptedService<Channel, BearerAuth>> {
SbomServiceClient::with_interceptor(self.channel.clone(), self.auth.clone())
}
fn property_client(
&self,
) -> PropertyServiceClient<InterceptedService<Channel, BearerAuth>> {
PropertyServiceClient::with_interceptor(self.channel.clone(), self.auth.clone())
}
fn archive_client(&self) -> ArchiveServiceClient<InterceptedService<Channel, BearerAuth>> {
ArchiveServiceClient::with_interceptor(self.channel.clone(), self.auth.clone())
.max_decoding_message_size(crate::grpc::DEFAULT_MAX_BODY_BYTES)
.max_encoding_message_size(crate::grpc::DEFAULT_MAX_BODY_BYTES)
}
}
/// Build a [`BearerAuth`] interceptor from an optional token. `None` yields the
/// no-op interceptor (open servers); `Some` parses `Bearer <token>` once so the
/// per-request `call` is just a clone+insert. Shared by every token-bearing
/// constructor so the parse/validation logic lives in one place.
fn bearer(token: Option<&str>) -> Result<BearerAuth> {
match token {
None => Ok(BearerAuth(None)),
Some(t) => {
let value: MetadataValue<Ascii> = format!("Bearer {t}")
.parse()
.map_err(|e| anyhow!("invalid bearer token: {e}"))?;
Ok(BearerAuth(Some(value)))
}
}
}
fn to_proto_id(id: &ArtifactId) -> ProtoArtifactId {
ProtoArtifactId {
namespace: id.namespace.clone().unwrap_or_default(),
name: id.name.clone(),
version: id.version.clone(),
}
}
#[async_trait]
impl HolgerObject for RemoteHolger {
async fn fetch(&self, repository: &str, id: &ArtifactId) -> Result<Option<traits::Bytes>> {
let mut client = self.repo_client();
let request = FetchArtifactRequest {
repository: repository.to_string(),
id: Some(to_proto_id(id)),
};
match client.fetch_artifact(request).await {
Ok(resp) => Ok(Some(resp.into_inner().data)),
Err(status) if status.code() == Code::NotFound => Ok(None),
Err(status) => Err(anyhow!("gRPC fetch failed: {}", status)),
}
}
async fn put(&self, repository: &str, id: &ArtifactId, data: &[u8]) -> Result<()> {
let mut client = self.repo_client();
let request = PutArtifactRequest {
repository: repository.to_string(),
id: Some(to_proto_id(id)),
// One copy, and it is the only one: the caller hands a borrowed
// slice, so the owned buffer has to be minted here. Everything
// downstream of it moves by refcount.
data: traits::Bytes::copy_from_slice(data),
async_job: false,
};
let resp = client
.put_artifact(request)
.await
.map_err(|s| anyhow!("gRPC put failed: {}", s))?
.into_inner();
if resp.success {
Ok(())
} else {
Err(anyhow!("put rejected: {}", resp.message))
}
}
async fn list_repositories(&self) -> Result<Vec<RepositoryInfo>> {
let mut client = self.admin_client();
let resp = client
.list_repositories(ListRepositoriesRequest {})
.await
.map_err(|s| anyhow!("gRPC list_repositories failed: {}", s))?
.into_inner();
Ok(resp
.repositories
.into_iter()
.map(|r| RepositoryInfo {
name: r.name,
repo_type: r.repo_type,
writable: r.writable,
has_archive: r.has_archive,
supports_listing: r.supports_listing,
is_group: r.is_group,
members: r.members,
})
.collect())
}
async fn list_artifacts(
&self,
repository: &str,
name_filter: Option<String>,
limit: u32,
page_token: Option<String>,
) -> Result<(Vec<ArtifactEntry>, String)> {
let mut client = self.repo_client();
let request = ListArtifactsRequest {
repository: repository.to_string(),
name_filter: name_filter.unwrap_or_default(),
// proto `limit` is int32; the trait takes u32 (no negative page size).
limit: limit as i32,
page_token: page_token.unwrap_or_default(),
};
let resp = client
.list_artifacts(request)
.await
.map_err(|s| anyhow!("gRPC list_artifacts failed: {}", s))?
.into_inner();
let entries = resp
.artifacts
.into_iter()
.map(|a| {
let id = a.id.unwrap_or_default();
ArtifactEntry {
id: ArtifactId {
// proto carries an empty namespace string for the
// namespace-less ecosystems (Rust); map that back to None.
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
},
size_bytes: a.size_bytes,
content_type: a.content_type,
// ── Reading the two optional fields off the wire ──
//
// They are `optional` in the proto, so absent and zero are
// different states here and an OLD server — one built
// before these fields existed — sends neither and is read
// as `None`, which is the truth about it.
//
// The normalizers are the belt to that braces: a peer that
// is not this codebase can still put `0` or `""` on the
// wire, and neither is allowed to become a value. Anything
// that is not strictly-positive milliseconds, and anything
// that is not 64 hex characters, lands as `None` = unknown.
// A wrong digest that was carried would be worse than no
// digest: a mirror compares against it.
published_unix_ms: traits::normalize_published_unix_ms(a.published_unix_ms),
sha256: a.sha256.as_deref().and_then(traits::normalize_sha256),
// The third one, same rule: an empty or control-bearing
// token from a hand-rolled peer lands as None, because an
// empty token compares EQUAL to another empty token.
change_token: a
.change_token
.as_deref()
.and_then(traits::normalize_change_token),
}
})
.collect();
Ok((entries, resp.next_page_token))
}
/// Browse the backing archive's raw file paths over gRPC
/// (`ArchiveService::ListArchiveFiles`). An empty `prefix` Option maps to the
/// empty proto string (proto3 has no optional string here) = "no filter".
async fn list_archive_files(
&self,
repository: &str,
prefix: Option<String>,
) -> Result<Vec<String>> {
let mut client = self.archive_client();
let request = ListArchiveFilesRequest {
repository: repository.to_string(),
prefix: prefix.unwrap_or_default(),
};
let resp = client
.list_archive_files(request)
.await
.map_err(|s| anyhow!("gRPC list_archive_files failed: {}", s))?
.into_inner();
Ok(resp.paths)
}
/// Archive stats over gRPC (`ArchiveService::ArchiveInfo`). The proto carries
/// int64 counters; cast back to the trait's u64.
async fn archive_info(&self, repository: &str) -> Result<ArchiveInfo> {
let mut client = self.archive_client();
let request = ArchiveInfoRequest {
repository: repository.to_string(),
};
let resp = client
.archive_info(request)
.await
.map_err(|s| anyhow!("gRPC archive_info failed: {}", s))?
.into_inner();
Ok(ArchiveInfo {
file_count: resp.file_count as u64,
total_uncompressed_bytes: resp.total_uncompressed_bytes as u64,
archive_path: resp.archive_path,
})
}
async fn health(&self) -> Result<Health> {
let mut client = self.admin_client();
let resp = client
.health(HealthRequest {})
.await
.map_err(|s| anyhow!("gRPC health failed: {}", s))?
.into_inner();
Ok(Health {
status: resp.status,
version: resp.version,
uptime_seconds: resp.uptime_seconds,
})
}
async fn server_profile(&self) -> Result<ServerProfile> {
let mut client = self.admin_client();
let resp = client
.server_profile(ServerProfileRequest {})
.await
.map_err(|s| anyhow!("gRPC server_profile failed: {}", s))?
.into_inner();
Ok(ServerProfile {
profile: resp.profile,
read_only: resp.read_only,
label: resp.label,
writable_repo_count: resp.writable_repo_count,
})
}
/// Cross-repo search over gRPC (`SearchService::Search`) — an open read.
async fn search(&self, query: &SearchQuery) -> Result<SearchResults> {
let mut client = self.search_client();
let request = SearchRequest {
name: query.name.clone(),
namespace: query.namespace.clone(),
version: query.version.clone(),
checksum: String::new(),
path: query.path.clone(),
repositories: query.repositories.clone(),
limit: query.limit,
properties: Vec::new(),
};
let resp = client
.search(request)
.await
.map_err(|s| anyhow!("gRPC search failed: {}", s))?
.into_inner();
Ok(SearchResults {
artifacts: resp
.artifacts
.into_iter()
.map(|h| {
let id = h.id.unwrap_or_default();
SearchArtifactHit {
repository: h.repository,
id: ArtifactId {
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
},
size_bytes: h.size_bytes,
content_type: h.content_type,
}
})
.collect(),
paths: resp
.paths
.into_iter()
.map(|p| SearchPathHit {
repository: p.repository,
path: p.path,
})
.collect(),
truncated: resp.truncated,
})
}
/// Hosted-SBOM fetch over gRPC (`SbomService::FetchSbom`); NOT_FOUND ⇒ `None`.
async fn fetch_sbom(&self, repository: &str, id: &ArtifactId) -> Result<Option<SbomDoc>> {
let mut client = self.sbom_client();
let request = FetchSbomRequest {
repository: repository.to_string(),
artifact: Some(to_proto_id(id)),
};
match client.fetch_sbom(request).await {
Ok(resp) => {
let r = resp.into_inner();
Ok(Some(SbomDoc {
document: r.document,
size_bytes: r.size_bytes,
content_address: r.content_address,
}))
}
Err(status) if status.code() == Code::NotFound => Ok(None),
Err(status) => Err(anyhow!("gRPC fetch_sbom failed: {}", status)),
}
}
/// Property read over gRPC (`PropertyService::GetProperties`) — empty when none.
async fn get_properties(
&self,
repository: &str,
id: &ArtifactId,
) -> Result<Vec<PropertyKv>> {
let mut client = self.property_client();
let request = GetPropertiesRequest {
repository: repository.to_string(),
artifact: Some(to_proto_id(id)),
};
let resp = client
.get_properties(request)
.await
.map_err(|s| anyhow!("gRPC get_properties failed: {}", s))?
.into_inner();
Ok(resp
.properties
.into_iter()
.map(|p| PropertyKv {
key: p.key,
values: p.values,
})
.collect())
}
}
}
#[cfg(feature = "fetch")]
pub use remote::*;
#[cfg(test)]
mod tests {
use super::*;
use traits::{ArtifactFormat, ArtifactId};
/// A minimal in-memory backend for exercising [`paginate_artifacts`] purely.
/// It holds a fixed set of entries and returns them **in a deliberately
/// SCRAMBLED order** on every `list` call, so the test proves the paginator
/// imposes its own stable total order (namespace, name, version) rather than
/// leaning on backend order — the exact contract the doc-comment promises.
struct FakeRepo {
entries: Vec<ArtifactEntry>,
}
impl FakeRepo {
fn new(entries: Vec<ArtifactEntry>) -> Self {
Self { entries }
}
}
#[async_trait]
impl RepositoryBackendTrait for FakeRepo {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"fake"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _id: &ArtifactId) -> Result<Option<traits::Bytes>> {
Ok(None)
}
fn put(&self, _id: &ArtifactId, _data: &[u8]) -> Result<()> {
anyhow::bail!("read-only fake")
}
fn list(&self, name_filter: Option<&str>, _limit: usize) -> Result<Vec<ArtifactEntry>> {
// Return matching entries in a SCRAMBLED order (reversed) so the test
// can prove the paginator re-sorts into a stable total order.
let mut v: Vec<ArtifactEntry> = self
.entries
.iter()
.filter(|e| match name_filter {
Some(f) => e.id.name.contains(f),
None => true,
})
.cloned()
.collect();
v.reverse();
Ok(v)
}
fn handle_http2_request(
&self,
_method: &str,
_suburl: &str,
_body: &[u8],
) -> Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, Vec::new(), Vec::new()))
}
}
fn entry(namespace: Option<&str>, name: &str, version: &str) -> ArtifactEntry {
ArtifactEntry {
id: ArtifactId {
namespace: namespace.map(|s| s.to_string()),
name: name.into(),
version: version.into(),
},
size_bytes: 0,
content_type: "application/octet-stream".into(),
// A test double knows neither, and says so.
..ArtifactEntry::UNKNOWN_PROVENANCE
}
}
/// A 5-entry corpus intentionally NOT in sorted order, spanning two namespaces
/// so the (namespace, name, version) sort key is exercised on every axis.
fn corpus() -> Vec<ArtifactEntry> {
vec![
entry(Some("b"), "zeta", "1.0.0"),
entry(Some("a"), "alpha", "2.0.0"),
entry(Some("a"), "alpha", "1.0.0"),
entry(None, "raw-thing", "0.1.0"),
entry(Some("a"), "beta", "1.0.0"),
]
}
/// The stable total order the paginator MUST impose: `None` namespace sorts
/// before `Some(_)` (Option's derived Ord), then by name, then by version.
fn sorted_ids() -> Vec<(Option<String>, String, String)> {
vec![
(None, "raw-thing".into(), "0.1.0".into()),
(Some("a".into()), "alpha".into(), "1.0.0".into()),
(Some("a".into()), "alpha".into(), "2.0.0".into()),
(Some("a".into()), "beta".into(), "1.0.0".into()),
(Some("b".into()), "zeta".into(), "1.0.0".into()),
]
}
fn id_tuple(e: &ArtifactEntry) -> (Option<String>, String, String) {
(
e.id.namespace.clone(),
e.id.name.clone(),
e.id.version.clone(),
)
}
#[test]
fn page_size_zero_returns_everything_sorted_no_next_token() {
let repo = FakeRepo::new(corpus());
// page_size == 0 means "everything from the offset" — one full page.
let (page, next) = paginate_artifacts(&repo, None, 0, None).unwrap();
assert_eq!(page.len(), 5, "all entries returned on a single page");
assert!(
next.is_empty(),
"an exhausted listing yields an empty next token"
);
let got: Vec<_> = page.iter().map(id_tuple).collect();
assert_eq!(
got,
sorted_ids(),
"paginator imposes its OWN stable total order"
);
}
#[test]
fn walks_every_page_exactly_once_across_the_full_listing() {
// Page size 2 across 5 entries → windows [0,2), [2,4), [4,5), then done.
let repo = FakeRepo::new(corpus());
let mut collected: Vec<(Option<String>, String, String)> = Vec::new();
let mut token: Option<String> = None;
let mut pages = 0;
loop {
let (page, next) = paginate_artifacts(&repo, None, 2, token.as_deref()).unwrap();
pages += 1;
collected.extend(page.iter().map(id_tuple));
if next.is_empty() {
break;
}
token = Some(next);
assert!(pages < 10, "cursor must terminate, not loop forever");
}
assert_eq!(pages, 3, "5 entries at page size 2 = 3 pages (2+2+1)");
assert_eq!(
collected,
sorted_ids(),
"every entry seen exactly once, in order"
);
}
#[test]
fn next_token_is_the_end_offset_and_slices_the_right_window() {
let repo = FakeRepo::new(corpus());
// First page of 2 → entries 0,1 and a next token of "2".
let (page0, next0) = paginate_artifacts(&repo, None, 2, None).unwrap();
assert_eq!(page0.len(), 2);
assert_eq!(
next0, "2",
"next token is the end offset as a decimal string"
);
assert_eq!(id_tuple(&page0[0]), sorted_ids()[0]);
assert_eq!(id_tuple(&page0[1]), sorted_ids()[1]);
// Resuming at offset "2" yields entries 2,3 and a next token of "4".
let (page1, next1) = paginate_artifacts(&repo, None, 2, Some("2")).unwrap();
assert_eq!(id_tuple(&page1[0]), sorted_ids()[2]);
assert_eq!(id_tuple(&page1[1]), sorted_ids()[3]);
assert_eq!(next1, "4");
// The final partial page (1 of 2 remaining) exhausts the listing.
let (page2, next2) = paginate_artifacts(&repo, None, 2, Some("4")).unwrap();
assert_eq!(page2.len(), 1, "last page is the single trailing entry");
assert_eq!(id_tuple(&page2[0]), sorted_ids()[4]);
assert!(next2.is_empty(), "exhausted listing → empty next token");
}
#[test]
fn offset_past_the_end_returns_empty_page_and_no_token() {
let repo = FakeRepo::new(corpus());
// Any offset >= total is out of range → empty window, no next token
// (never an out-of-bounds slice panic).
let (page, next) = paginate_artifacts(&repo, None, 2, Some("5")).unwrap();
assert!(page.is_empty());
assert!(next.is_empty());
let (page_far, next_far) = paginate_artifacts(&repo, None, 2, Some("999")).unwrap();
assert!(page_far.is_empty());
assert!(next_far.is_empty());
}
#[test]
fn garbage_and_missing_token_both_start_from_offset_zero() {
let repo = FakeRepo::new(corpus());
// A non-numeric token parses to offset 0 (start), same as no token.
let (from_none, _) = paginate_artifacts(&repo, None, 2, None).unwrap();
let (from_garbage, _) = paginate_artifacts(&repo, None, 2, Some("not-a-number")).unwrap();
let a: Vec<_> = from_none.iter().map(id_tuple).collect();
let b: Vec<_> = from_garbage.iter().map(id_tuple).collect();
assert_eq!(a, b, "unparseable page_token falls back to the start");
assert_eq!(a[0], sorted_ids()[0]);
}
#[test]
fn name_filter_narrows_the_set_before_paging() {
let repo = FakeRepo::new(corpus());
// Only the two `alpha` entries match; both fit one page, no next token.
let (page, next) = paginate_artifacts(&repo, Some("alpha"), 10, None).unwrap();
assert_eq!(
page.len(),
2,
"filter applied to the whole set before windowing"
);
assert!(next.is_empty());
assert!(page.iter().all(|e| e.id.name == "alpha"));
// The filtered page is still stably ordered by version.
assert_eq!(page[0].id.version, "1.0.0");
assert_eq!(page[1].id.version, "2.0.0");
}
#[test]
fn empty_backend_lists_nothing() {
let repo = FakeRepo::new(Vec::new());
let (page, next) = paginate_artifacts(&repo, None, 5, None).unwrap();
assert!(page.is_empty());
assert!(next.is_empty(), "an empty backend never emits a next token");
}
/// RED drift bug: `list_repositories` hardcoded `has_archive: true`, so a
/// repository with no archive told every client it had one.
#[tokio::test]
async fn local_list_repositories_reports_the_backend_not_a_constant() {
let routes = FastRoutes::new(vec![(
"fake".to_string(),
Arc::new(FakeRepo::new(corpus())) as Arc<dyn RepositoryBackendTrait>,
)]);
let infos = LocalHolger::new(routes).list_repositories().await.unwrap();
assert!(!infos[0].has_archive, "FakeRepo has no archive");
}
/// RED drift bug: LocalHolger inherited the empty `search` default, so an
/// in-process caller found nothing in a repository that held the artifact.
#[tokio::test]
async fn local_search_finds_what_the_repository_holds() {
let routes = FastRoutes::new(vec![(
"fake".to_string(),
Arc::new(FakeRepo::new(corpus())) as Arc<dyn RepositoryBackendTrait>,
)]);
let q = SearchQuery {
name: "alpha".into(),
..Default::default()
};
let r = LocalHolger::new(routes).search(&q).await.unwrap();
assert_eq!(r.artifacts.len(), 2, "two alpha versions are held");
}
}