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");
}
}
}
}
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;
#[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> {
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);
}
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_val.clone());
request
.metadata_mut()
.insert_bin("urc-repository-id", bin_val);
}
Ok(request)
}
}
#[derive(Debug, Clone)]
pub struct LoreGrpcClient {
channel: Channel,
token: Option<String>,
repository_id_bytes: Vec<u8>,
}
impl LoreGrpcClient {
pub fn builder() -> Builder {
Builder::default()
}
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(),
}
}
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)
}
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(())
}
pub fn builder_from_env() -> Result<Option<Self>, NapError> {
Builder::from_env()
}
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(),
},
)
}
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"))
})
}
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)
}
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))
}
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))
}
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 mut stream = 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
.map_err(|status| map_grpc_status("RevisionTree", status))?
.into_inner();
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)
}
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()))
}
}
#[derive(Default)]
pub struct Builder {
endpoint: Option<String>,
token: Option<String>,
repository_id_bytes: Vec<u8>,
insecure: bool,
}
impl Builder {
pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
self.endpoint = Some(endpoint.into());
self
}
pub fn token(mut self, token: impl Into<String>) -> Self {
self.token = Some(token.into());
self
}
pub fn repository_id(mut self, id: impl Into<Vec<u8>>) -> Self {
self.repository_id_bytes = id.into();
self
}
pub fn insecure(mut self, insecure: bool) -> Self {
self.insecure = insecure;
self
}
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(),
)
})?;
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,
})
}
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)
}
}
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")
});
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:?}"))
})?
}
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}")),
}
}