nap-core 0.8.10

//! gRPC client for lore-server's revision service (branch ref sync).
//!
//! The lore-server exposes its mutable state — branch pointers, revision
//! pointers — exclusively over gRPC, whereas content-addressed blob data
//! is transferred via HTTP / the `lore` CLI.  This module implements the
//! gRPC half of the push/pull protocol.
//!
//! # Architecture
//!
//! ```text
//! LoreBackend::push / pull   (sync, on tokio runtime)
//!     │
//!     ▼
//! block_on_grpc(…)           (spawns dedicated OS thread)
//!     │
//!     ▼
//! LoreGrpcClient             (tonic RevisionServiceClient wrapper)
//!     │
//!     ▼
//! lore-server gRPC endpoint
//!     ├── RevisionService.BranchGet   → fetch remote tip
//!     └── RevisionService.BranchPush  → advance remote tip
//! ```
//!
//! # Sync/Async Bridge
//!
//! The [`VcsBackend`] trait is synchronous.  gRPC is inherently async.
//! Rather than changing the trait (which would break every implementation),
//! we bridge via [`block_on_grpc`]: a dedicated OS thread hosts a shared
//! single-threaded tokio runtime that executes the async gRPC call.  This
//! avoids the "Cannot start a runtime from within a runtime" panic that
//! would occur if we called `Runtime::block_on` directly inside axum
//! request handlers.

// ---------------------------------------------------------------------------
// Generated proto modules — must nest exactly as prost expects for
// cross-package references (lore.revision.v1 → lore.model.v1)
// ---------------------------------------------------------------------------

/// Generated gRPC service and message types.
///
/// Two packages are compiled:
/// - `lore.model.v1`      — Branch, BranchPoint, etc.
/// - `lore.revision.v1`   — RevisionService, BranchGetRequest, etc.
pub mod proto_gen {
    #![allow(unreachable_pub)]
    pub mod lore {
        pub mod model {
            pub mod v1 {
                tonic::include_proto!("lore.model.v1");
            }
        }
        pub mod revision {
            pub mod v1 {
                tonic::include_proto!("lore.revision.v1");
            }
        }
        pub mod repository {
            pub mod v1 {
                tonic::include_proto!("lore.repository.v1");
            }
        }
        pub mod storage {
            pub mod v1 {
                tonic::include_proto!("lore.storage.v1");
            }
        }
        pub mod thin_client {
            pub mod v1 {
                tonic::include_proto!("lore.thin_client.v1");
            }
        }
    }
}

// Re-export the types callers need most frequently.
pub use proto_gen::lore::model::v1::Branch;
pub use proto_gen::lore::repository::v1::repository_service_client::RepositoryServiceClient;
pub use proto_gen::lore::repository::v1::{RepositoryGetRequest, RepositoryListRequest};
pub use proto_gen::lore::revision::v1::branch_get_request;
pub use proto_gen::lore::revision::v1::revision_service_client::RevisionServiceClient;
pub use proto_gen::lore::revision::v1::{BranchGetRequest, BranchPushRequest};
pub use proto_gen::lore::revision::v1::{BranchListRequest, RevisionListRequest};
pub use proto_gen::lore::storage::v1::storage_service_client::StorageServiceClient;
pub use proto_gen::lore::thin_client::v1::thin_client_service_client::ThinClientServiceClient;
pub use proto_gen::lore::thin_client::v1::{RevisionInfoRequest, RevisionTreeRequest};

use std::future::Future;
use std::sync::LazyLock;
use std::thread;
use std::time::Duration;

use tonic::codegen::InterceptedService;
use tonic::metadata::{BinaryMetadataValue, MetadataValue};
use tonic::service::Interceptor;
use tonic::transport::{Channel, Endpoint};

use crate::error::NapError;

// ===========================================================================
// Auth interceptor
// ===========================================================================

/// Injects JWT bearer token and repository-scope metadata into every
/// outgoing gRPC request.
///
/// The token is sent as `Authorization: Bearer <token>` with
/// `set_sensitive(true)` so proxy logs do not leak it.
///
/// The repository ID is sent as binary metadata (keys with `-bin` suffix)
/// matching the lore-client's `inject_repository()` protocol.
#[derive(Clone)]
struct GrpcAuthInterceptor {
    token: Option<String>,
    repository_id_bytes: Vec<u8>,
}

impl Interceptor for GrpcAuthInterceptor {
    fn call(
        &mut self,
        mut request: tonic::Request<()>,
    ) -> Result<tonic::Request<()>, tonic::Status> {
        // ── Authorization header ──────────────────────────────────────
        if let Some(ref token) = self.token
            && !token.is_empty()
        {
            let mut value: MetadataValue<_> = format!("Bearer {token}")
                .parse()
                .map_err(|e| tonic::Status::invalid_argument(format!("bad token metadata: {e}")))?;
            value.set_sensitive(true);
            request.metadata_mut().insert("authorization", value);
        }

        // ── Repository-scope binary metadata ──────────────────────────
        if !self.repository_id_bytes.is_empty() {
            let bin_val = BinaryMetadataValue::from_bytes(&self.repository_id_bytes);
            request
                .metadata_mut()
                .insert_bin("lore-partition-bin", bin_val.clone());
            request
                .metadata_mut()
                .insert_bin("urc-repository-id-bin", bin_val);
        }

        Ok(request)
    }
}

// ===========================================================================
// LoreGrpcClient
// ===========================================================================

/// A gRPC client for lore-server's [`RevisionService`].
///
/// This client handles **only** lightweight metadata operations:
///
/// | Operation | RPC | Purpose |
/// |-----------|-----|---------|
/// | `get_branch_by_name` | `BranchGet` | Fetch remote branch tip before pull |
/// | `push_branch` | `BranchPush` | Advance remote branch tip after push |
///
/// Blob transfer (the heavy payload) remains on the `lore` CLI / HTTP.
///
/// [`RevisionService`]: proto_gen::lore::revision::v1::revision_service_client::RevisionServiceClient
#[derive(Debug, Clone)]
pub struct LoreGrpcClient {
    channel: Channel,
    token: Option<String>,
    repository_id_bytes: Vec<u8>,
}

impl LoreGrpcClient {
    /// Return a builder for fine-grained configuration.
    pub fn builder() -> Builder {
        Builder::default()
    }

    /// Return a clone scoped to a repository returned by `RepositoryGet`.
    pub fn for_repository_id(&self, id: impl Into<Vec<u8>>) -> Self {
        Self {
            channel: self.channel.clone(),
            token: self.token.clone(),
            repository_id_bytes: id.into(),
        }
    }

    // ── Public RPC methods ───────────────────────────────────────────

    /// Look up a branch by its human-readable name.
    ///
    /// Returns the [`Branch`] record containing `id` (binary UUID),
    /// `name`, `latest` (tip signature), and other metadata.
    pub async fn get_branch_by_name(&self, name: &str) -> Result<Branch, NapError> {
        let mut client = self.make_client();
        let response = client
            .branch_get(BranchGetRequest {
                query: Some(branch_get_request::Query::Name(name.to_string())),
            })
            .await
            .map_err(|status| map_grpc_status("BranchGet", status))?;

        response.into_inner().branch.ok_or_else(|| {
            NapError::GrpcError(format!("BranchGet({name}) returned empty branch record"))
        })
    }

    pub async fn list_branches(&self) -> Result<Vec<Branch>, NapError> {
        let mut client = self.make_client();
        let mut stream = client
            .branch_list(BranchListRequest {
                creator: None,
                include_deleted: false,
            })
            .await
            .map_err(|status| map_grpc_status("BranchList", status))?
            .into_inner();
        let mut branches = Vec::new();
        while let Some(item) = stream
            .message()
            .await
            .map_err(|status| map_grpc_status("BranchList", status))?
        {
            if let Some(branch) = item.branch {
                branches.push(branch);
            }
        }
        Ok(branches)
    }

    pub async fn list_revisions(
        &self,
        identifier: proto_gen::lore::model::v1::RevisionIdentifier,
    ) -> Result<Vec<proto_gen::lore::model::v1::RevisionItem>, NapError> {
        let mut client = self.make_client();
        Ok(client
            .revision_list(RevisionListRequest {
                start: Some(
                    proto_gen::lore::revision::v1::revision_list_request::Start::Identifier(
                        identifier,
                    ),
                ),
            })
            .await
            .map_err(|status| map_grpc_status("RevisionList", status))?
            .into_inner()
            .items)
    }

    /// Push a revision as the new tip of a branch.
    ///
    /// * `branch_id` — binary branch UUID (obtained from
    ///   [`get_branch_by_name`]).
    /// * `revision_signature` — raw content hash of the revision to set as
    ///   the new tip.
    /// * `force` — if `true`, bypasses fast-forward checks on the server.
    ///   When `false`, the server requires the new tip to descend from the
    ///   current tip (or performs a fast-forward merge).
    pub async fn push_branch(
        &self,
        branch_id: bytes::Bytes,
        revision_signature: bytes::Bytes,
        force: bool,
    ) -> Result<(), NapError> {
        let mut client = self.make_client();
        client
            .branch_push(BranchPushRequest {
                id: branch_id,
                revision_signature,
                force,
                fast_forward_merge: !force,
            })
            .await
            .map_err(|status| map_grpc_status("BranchPush", status))?;
        Ok(())
    }

    // ── Internal helpers ─────────────────────────────────────────────

    /// Convenience constructor that reads all configuration from environment
    /// variables.  Returns `Ok(None)` when `NAP_LORE_GRPC_ENDPOINT` is not
    /// set, allowing callers to gracefully skip gRPC integration.
    ///
    /// See [`Builder::from_env`] for the list of recognised variables.
    pub fn builder_from_env() -> Result<Option<Self>, NapError> {
        Builder::from_env()
    }

    /// Build a fresh client with the interceptor wired in.
    fn make_client(
        &self,
    ) -> RevisionServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
        RevisionServiceClient::with_interceptor(
            self.channel.clone(),
            GrpcAuthInterceptor {
                token: self.token.clone(),
                repository_id_bytes: self.repository_id_bytes.clone(),
            },
        )
    }

    fn make_repository_client(
        &self,
    ) -> RepositoryServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
        RepositoryServiceClient::with_interceptor(
            self.channel.clone(),
            GrpcAuthInterceptor {
                token: self.token.clone(),
                repository_id_bytes: Vec::new(),
            },
        )
    }

    fn make_storage_client(
        &self,
    ) -> StorageServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
        StorageServiceClient::with_interceptor(
            self.channel.clone(),
            GrpcAuthInterceptor {
                token: self.token.clone(),
                repository_id_bytes: self.repository_id_bytes.clone(),
            },
        )
    }

    fn make_thin_client(
        &self,
    ) -> ThinClientServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
        ThinClientServiceClient::with_interceptor(
            self.channel.clone(),
            GrpcAuthInterceptor {
                token: self.token.clone(),
                repository_id_bytes: self.repository_id_bytes.clone(),
            },
        )
    }

    /// Look up a repository before constructing a repository-scoped client.
    pub async fn get_repository_by_name(
        &self,
        name: &str,
    ) -> Result<proto_gen::lore::model::v1::Repository, NapError> {
        let mut client = self.make_repository_client();
        client
            .repository_get(RepositoryGetRequest {
                query: Some(
                    proto_gen::lore::repository::v1::repository_get_request::Query::Name(
                        name.to_string(),
                    ),
                ),
            })
            .await
            .map_err(|status| map_grpc_status("RepositoryGet", status))?
            .into_inner()
            .repository
            .ok_or_else(|| {
                NapError::GrpcError(format!("RepositoryGet({name}) returned no repository"))
            })
    }

    /// Return names of repositories visible to the current identity.
    pub async fn list_repositories(&self) -> Result<Vec<String>, NapError> {
        let mut client = self.make_repository_client();
        let mut stream = client
            .repository_list(RepositoryListRequest { creator: None })
            .await
            .map_err(|status| map_grpc_status("RepositoryList", status))?
            .into_inner();
        let mut names = Vec::new();
        while let Some(item) = stream
            .message()
            .await
            .map_err(|status| map_grpc_status("RepositoryList", status))?
        {
            if let Some(repository) = item.repository {
                names.push(repository.name);
            }
        }
        Ok(names)
    }

    /// Read a single file at a revision tree path. The caller must scope this
    /// client with the repository id returned by `RepositoryGet`.
    pub async fn read_file_at_revision(
        &self,
        identifier: proto_gen::lore::model::v1::RevisionIdentifier,
        path: String,
    ) -> Result<(Vec<u8>, Vec<u8>), NapError> {
        let mut tree = self.make_thin_client();
        let mut stream = tree
            .revision_tree(RevisionTreeRequest {
                query: Some(
                    proto_gen::lore::thin_client::v1::revision_tree_request::Query::Identifier(
                        identifier,
                    ),
                ),
                path_prefix: Some(path.clone()),
                max_depth: Some(1),
            })
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
            .into_inner();
        let mut signature = Vec::new();
        let mut address = None;
        while let Some(item) = stream
            .message()
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
        {
            match item.payload {
                Some(
                    proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Header(
                        header,
                    ),
                ) => signature = header.signature.to_vec(),
                Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
                    node,
                )) if node.path == path => address = node.address,
                _ => {}
            }
        }
        let address = address.ok_or_else(|| NapError::ManifestNotFound(path.clone()))?;
        let mut storage = self.make_storage_client();
        let outgoing = tokio_stream::iter([address]);
        let mut bytes = Vec::new();
        let mut content = storage
            .get(outgoing)
            .await
            .map_err(|status| map_grpc_status("StorageGet", status))?
            .into_inner();
        while let Some(chunk) = content
            .message()
            .await
            .map_err(|status| map_grpc_status("StorageGet", status))?
        {
            bytes.extend_from_slice(&chunk.payload);
        }
        Ok((bytes, signature))
    }

    /// Read a file selected by its immutable Lore revision signature.
    pub async fn read_file_at_signature(
        &self,
        signature: Vec<u8>,
        path: String,
    ) -> Result<(Vec<u8>, Vec<u8>), NapError> {
        let mut tree = self.make_thin_client();
        let mut stream = tree
            .revision_tree(RevisionTreeRequest {
                query: Some(
                    proto_gen::lore::thin_client::v1::revision_tree_request::Query::Signature(
                        signature.into(),
                    ),
                ),
                path_prefix: Some(path.clone()),
                max_depth: Some(1),
            })
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
            .into_inner();
        let mut resolved_signature = Vec::new();
        let mut address = None;
        while let Some(item) = stream
            .message()
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
        {
            match item.payload {
                Some(
                    proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Header(
                        header,
                    ),
                ) => resolved_signature = header.signature.to_vec(),
                Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
                    node,
                )) if node.path == path => address = node.address,
                _ => {}
            }
        }
        let address = address.ok_or(NapError::ManifestNotFound(path))?;
        let mut storage = self.make_storage_client();
        let mut bytes = Vec::new();
        let mut content = storage
            .get(tokio_stream::iter([address]))
            .await
            .map_err(|status| map_grpc_status("StorageGet", status))?
            .into_inner();
        while let Some(chunk) = content
            .message()
            .await
            .map_err(|status| map_grpc_status("StorageGet", status))?
        {
            bytes.extend_from_slice(&chunk.payload);
        }
        Ok((bytes, resolved_signature))
    }

    /// List file paths below a revision tree prefix without downloading them.
    pub async fn list_paths_at_revision(
        &self,
        identifier: proto_gen::lore::model::v1::RevisionIdentifier,
        prefix: String,
    ) -> Result<Vec<String>, NapError> {
        let mut tree = self.make_thin_client();
        let response = tree
            .revision_tree(RevisionTreeRequest {
                query: Some(
                    proto_gen::lore::thin_client::v1::revision_tree_request::Query::Identifier(
                        identifier,
                    ),
                ),
                path_prefix: Some(prefix),
                max_depth: None,
            })
            .await;
        let mut stream = match response {
            Ok(response) => response.into_inner(),
            // Lore represents a repository with no committed revisions as a
            // zero signature. Listing an empty repository is still valid;
            // surface it as an empty list rather than leaking this server
            // implementation detail to NAP users.
            Err(status)
                if status.code() == tonic::Code::InvalidArgument
                    && status.message().contains("zeroed revision") =>
            {
                return Ok(Vec::new());
            }
            Err(status) => return Err(map_grpc_status("RevisionTree", status)),
        };
        let mut paths = Vec::new();
        while let Some(item) = stream
            .message()
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
        {
            if let Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
                node,
            )) = item.payload
                && node.node_type == proto_gen::lore::thin_client::v1::NodeType::File as i32
            {
                paths.push(node.path);
            }
        }
        Ok(paths)
    }

    /// Look up a file's content address without downloading its payload.
    pub async fn file_address_at_revision(
        &self,
        identifier: proto_gen::lore::model::v1::RevisionIdentifier,
        path: String,
    ) -> Result<(proto_gen::lore::model::v1::Address, Vec<u8>), NapError> {
        let mut tree = self.make_thin_client();
        let mut stream = tree
            .revision_tree(RevisionTreeRequest {
                query: Some(
                    proto_gen::lore::thin_client::v1::revision_tree_request::Query::Identifier(
                        identifier,
                    ),
                ),
                path_prefix: Some(path.clone()),
                max_depth: Some(1),
            })
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
            .into_inner();
        let mut signature = Vec::new();
        let mut address = None;
        while let Some(item) = stream
            .message()
            .await
            .map_err(|status| map_grpc_status("RevisionTree", status))?
        {
            match item.payload {
                Some(
                    proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Header(
                        header,
                    ),
                ) => signature = header.signature.to_vec(),
                Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
                    node,
                )) if node.path == path => address = node.address,
                _ => {}
            }
        }
        address
            .map(|address| (address, signature))
            .ok_or(NapError::ManifestNotFound(path))
    }

    pub async fn revision_info_at_signature(
        &self,
        signature: Vec<u8>,
    ) -> Result<proto_gen::lore::thin_client::v1::Revision, NapError> {
        let mut client = self.make_thin_client();
        client
            .revision_info(RevisionInfoRequest {
                query: Some(
                    proto_gen::lore::thin_client::v1::revision_info_request::Query::Signature(
                        signature.into(),
                    ),
                ),
            })
            .await
            .map_err(|status| map_grpc_status("RevisionInfo", status))?
            .into_inner()
            .revision
            .ok_or_else(|| NapError::GrpcError("RevisionInfo returned no revision".to_string()))
    }
}

// ===========================================================================
// Builder
// ===========================================================================

/// Configuration builder for [`LoreGrpcClient`].
///
/// # Environment variables
///
/// | Variable | Required | Default | Description |
/// |----------|----------|---------|-------------|
/// | `NAP_LORE_GRPC_ENDPOINT` | Yes | — | gRPC endpoint URL |
/// | `NAP_LORE_GRPC_TOKEN` | No | — | JWT bearer token |
/// | `NAP_LORE_GRPC_RID` | No | — | Repository ID (hex-encoded binary) |
/// | `NAP_LORE_GRPC_INSECURE` | No | `0` | Skip TLS verification when `1` |
#[derive(Default)]
pub struct Builder {
    endpoint: Option<String>,
    token: Option<String>,
    repository_id_bytes: Vec<u8>,
    insecure: bool,
}

impl Builder {
    /// Set the gRPC endpoint URL.
    ///
    /// Format: `https://host:port` (TLS) or `http://host:port` (plain).
    pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
        self.endpoint = Some(endpoint.into());
        self
    }

    /// Set a JWT bearer token for authenticated requests.
    pub fn token(mut self, token: impl Into<String>) -> Self {
        self.token = Some(token.into());
        self
    }

    /// Set the repository ID to inject as binary metadata.
    ///
    /// This should match the repository / partition UUID the lore-server
    /// expects.  Pass the raw bytes (not hex-encoded).
    pub fn repository_id(mut self, id: impl Into<Vec<u8>>) -> Self {
        self.repository_id_bytes = id.into();
        self
    }

    /// When `true`, skip TLS certificate validation.
    ///
    /// Use this in development environments where the lore-server uses
    /// self-signed certificates.
    pub fn insecure(mut self, insecure: bool) -> Self {
        self.insecure = insecure;
        self
    }

    /// Build the [`LoreGrpcClient`].
    ///
    /// Connection is deferred via [`Endpoint::connect_lazy`]; the
    /// first RPC will establish the TCP + TLS handshake.
    pub fn build(self) -> Result<LoreGrpcClient, NapError> {
        let endpoint_str = self.endpoint.ok_or_else(|| {
            NapError::GrpcError(
                "gRPC endpoint is required — set via .endpoint() or NAP_LORE_GRPC_ENDPOINT"
                    .to_string(),
            )
        })?;

        // In insecure mode, downgrade https:// → http:// to skip TLS
        // verification entirely (self-signed certs in development).
        // In secure mode, Endpoint::from_shared auto-configures TLS with
        // native roots for https:// URLs — no explicit tls_config needed.
        let effective_url = if self.insecure {
            endpoint_str
                .strip_prefix("https://")
                .map(|rest| format!("http://{rest}"))
                .unwrap_or_else(|| endpoint_str.clone())
        } else {
            endpoint_str.clone()
        };

        let channel = Endpoint::from_shared(effective_url)
            .map_err(|e| {
                NapError::GrpcError(format!("invalid gRPC endpoint '{endpoint_str}': {e}"))
            })?
            .http2_keep_alive_interval(Duration::from_secs(30))
            .keep_alive_timeout(Duration::from_secs(20))
            .user_agent(concat!("nap-core/", env!("CARGO_PKG_VERSION")))
            .map_err(|e| NapError::GrpcError(format!("user-agent configuration error: {e}")))?
            .connect_lazy();

        Ok(LoreGrpcClient {
            channel,
            token: self.token,
            repository_id_bytes: self.repository_id_bytes,
        })
    }

    /// Build from environment variables.
    ///
    /// Returns `Ok(None)` when `NAP_LORE_GRPC_ENDPOINT` is not set
    /// (allowing the caller to skip gRPC integration gracefully).
    pub fn from_env() -> Result<Option<LoreGrpcClient>, NapError> {
        let endpoint = match std::env::var("NAP_LORE_GRPC_ENDPOINT") {
            Ok(v) => v,
            Err(_) => return Ok(None),
        };

        let token = std::env::var("NAP_LORE_GRPC_TOKEN").ok();
        let insecure = std::env::var("NAP_LORE_GRPC_INSECURE")
            .ok()
            .is_some_and(|v| v == "1" || v == "true" || v == "yes");

        let repository_id_bytes = std::env::var("NAP_LORE_GRPC_RID")
            .ok()
            .map(|hex| {
                hex::decode(&hex).map_err(|e| {
                    NapError::GrpcError(format!("invalid NAP_LORE_GRPC_RID hex '{hex}': {e}"))
                })
            })
            .transpose()?
            .unwrap_or_default();

        let mut builder = Builder::default().endpoint(endpoint).insecure(insecure);

        if let Some(t) = token {
            builder = builder.token(t);
        }
        if !repository_id_bytes.is_empty() {
            builder = builder.repository_id(repository_id_bytes);
        }

        builder.build().map(Some)
    }
}

// ===========================================================================
// Sync→async bridge
// ===========================================================================

/// Execute an async gRPC operation from a synchronous context.
///
/// # Why a dedicated thread?
///
/// The [`VcsBackend`] trait methods (`push`, `pull`) are synchronous.
/// gRPC client calls are async.  If we called `Runtime::block_on` directly
/// from within an axum HTTP handler (which already runs on a tokio runtime),
/// tokio would panic with "Cannot start a runtime from within a runtime".
///
/// This function spawns a **dedicated OS thread** that hosts the future
/// on a shared single-threaded tokio runtime.  The runtime is created once
/// and reused across all gRPC calls, preserving HTTP/2 keepalive state and
/// TLS session tickets.
///
/// # Type bounds
///
/// * `F` must be `Send + 'static` because it crosses a thread boundary.
/// * `T` must be `Send + 'static` for the same reason.
/// * The closure return type is `Result<T, NapError>` so that error
///   propagation through the thread join is straightforward.
///
/// [`VcsBackend`]: crate::vcs::VcsBackend
pub fn block_on_grpc<F, T>(f: F) -> Result<T, NapError>
where
    F: Future<Output = Result<T, NapError>> + Send + 'static,
    T: Send + 'static,
{
    static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("failed to build gRPC tokio runtime")
    });

    // `&'static Runtime` is both `Send` and `Sync` because the static
    // reference lives forever.  It is safe to pass to a spawned thread.
    let rt: &'static tokio::runtime::Runtime = &RUNTIME;

    thread::Builder::new()
        .name("nap-grpc".into())
        .spawn(move || rt.block_on(f))
        .expect("failed to spawn gRPC worker thread")
        .join()
        .map_err(|panic_payload| {
            NapError::GrpcError(format!("gRPC thread panicked: {panic_payload:?}"))
        })?
}

// ===========================================================================
// Error mapping
// ===========================================================================

/// Map a [`tonic::Status`] to a structured [`NapError`].
fn map_grpc_status(context: &str, status: tonic::Status) -> NapError {
    let code = status.code();
    let message = status.message();
    match code {
        tonic::Code::NotFound => NapError::RefNotFound(format!("{context}: {message}")),
        tonic::Code::Unauthenticated | tonic::Code::PermissionDenied => {
            NapError::PermissionDenied(format!("{context}: {message}"))
        }
        _ => NapError::GrpcError(format!("{context} ({code}): {message}")),
    }
}