holger-server-lib 0.6.9

//! `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, RepositoryBackendTrait,
    RepositoryInfo, 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)> {
    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,
}

impl LocalHolger {
    pub fn new(routes: FastRoutes) -> Self {
        Self { routes, start: Instant::now() }
    }

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

#[async_trait]
impl HolgerObject for LocalHolger {
    async fn fetch(&self, repository: &str, id: &ArtifactId) -> Result<Option<Vec<u8>>> {
        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 {
                name,
                repo_type: format!("{:?}", repo.format()),
                writable: repo.is_writable(),
                has_archive: true,
            })
            .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>> {
        self.repo(repository)?.archive_files(prefix.as_deref())
    }

    /// Archive stats straight off the backend (no network).
    async fn archive_info(&self, repository: &str) -> Result<ArchiveInfo> {
        self.repo(repository)?.archive_info()
    }

    async fn health(&self) -> Result<Health> {
        Ok(Health {
            status: "ok".into(),
            version: env!("CARGO_PKG_VERSION").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,
        })
    }
}

// ─── RemoteHolger: gRPC transport ────────────────────────────────────

use crate::grpc::holger_proto::{
    admin_service_client::AdminServiceClient,
    archive_service_client::ArchiveServiceClient,
    repository_service_client::RepositoryServiceClient,
    ArchiveInfoRequest, ArtifactId as ProtoArtifactId, FetchArtifactRequest, HealthRequest,
    ListArchiveFilesRequest, ListArtifactsRequest, ListRepositoriesRequest, PutArtifactRequest,
    ServerProfileRequest,
};
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
use tonic::Code;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::service::interceptor::InterceptedService;
use tonic::service::Interceptor;
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 channel = Channel::from_shared(endpoint.into())?.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 channel = Channel::from_shared(endpoint.into())?.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> {
        let mut tls = ClientTlsConfig::new();
        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 channel = Channel::from_shared(endpoint.into())?
            .tls_config(tls)?
            .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) }
    }

    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 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<Vec<u8>>> {
        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)),
            data: data.to_vec(),
        };
        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,
            })
            .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,
                }
            })
            .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,
        })
    }
}

#[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 name(&self) -> &str {
            "fake"
        }
        fn format(&self) -> ArtifactFormat {
            ArtifactFormat::Raw
        }
        fn is_writable(&self) -> bool {
            false
        }
        fn fetch(&self, _id: &ArtifactId) -> Result<Option<Vec<u8>>> {
            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 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");
    }
}