1pub mod proto_gen {
46 #![allow(unreachable_pub)]
47 pub mod lore {
48 pub mod model {
49 pub mod v1 {
50 tonic::include_proto!("lore.model.v1");
51 }
52 }
53 pub mod revision {
54 pub mod v1 {
55 tonic::include_proto!("lore.revision.v1");
56 }
57 }
58 pub mod repository {
59 pub mod v1 {
60 tonic::include_proto!("lore.repository.v1");
61 }
62 }
63 pub mod storage {
64 pub mod v1 {
65 tonic::include_proto!("lore.storage.v1");
66 }
67 }
68 pub mod thin_client {
69 pub mod v1 {
70 tonic::include_proto!("lore.thin_client.v1");
71 }
72 }
73 }
74}
75
76pub use proto_gen::lore::model::v1::Branch;
78pub use proto_gen::lore::repository::v1::repository_service_client::RepositoryServiceClient;
79pub use proto_gen::lore::repository::v1::{RepositoryGetRequest, RepositoryListRequest};
80pub use proto_gen::lore::revision::v1::branch_get_request;
81pub use proto_gen::lore::revision::v1::revision_service_client::RevisionServiceClient;
82pub use proto_gen::lore::revision::v1::{BranchGetRequest, BranchPushRequest};
83pub use proto_gen::lore::revision::v1::{BranchListRequest, RevisionListRequest};
84pub use proto_gen::lore::storage::v1::storage_service_client::StorageServiceClient;
85pub use proto_gen::lore::thin_client::v1::thin_client_service_client::ThinClientServiceClient;
86pub use proto_gen::lore::thin_client::v1::{RevisionInfoRequest, RevisionTreeRequest};
87
88use std::future::Future;
89use std::sync::LazyLock;
90use std::thread;
91use std::time::Duration;
92
93use tonic::codegen::InterceptedService;
94use tonic::metadata::{BinaryMetadataValue, MetadataValue};
95use tonic::service::Interceptor;
96use tonic::transport::{Channel, Endpoint};
97
98use crate::error::NapError;
99
100#[derive(Clone)]
113struct GrpcAuthInterceptor {
114 token: Option<String>,
115 repository_id_bytes: Vec<u8>,
116}
117
118impl Interceptor for GrpcAuthInterceptor {
119 fn call(
120 &mut self,
121 mut request: tonic::Request<()>,
122 ) -> Result<tonic::Request<()>, tonic::Status> {
123 if let Some(ref token) = self.token
125 && !token.is_empty()
126 {
127 let mut value: MetadataValue<_> = format!("Bearer {token}")
128 .parse()
129 .map_err(|e| tonic::Status::invalid_argument(format!("bad token metadata: {e}")))?;
130 value.set_sensitive(true);
131 request.metadata_mut().insert("authorization", value);
132 }
133
134 if !self.repository_id_bytes.is_empty() {
136 let bin_val = BinaryMetadataValue::from_bytes(&self.repository_id_bytes);
137 request
138 .metadata_mut()
139 .insert_bin("lore-partition", bin_val.clone());
140 request
141 .metadata_mut()
142 .insert_bin("urc-repository-id", bin_val);
143 }
144
145 Ok(request)
146 }
147}
148
149#[derive(Debug, Clone)]
166pub struct LoreGrpcClient {
167 channel: Channel,
168 token: Option<String>,
169 repository_id_bytes: Vec<u8>,
170}
171
172impl LoreGrpcClient {
173 pub fn builder() -> Builder {
175 Builder::default()
176 }
177
178 pub fn for_repository_id(&self, id: impl Into<Vec<u8>>) -> Self {
180 Self {
181 channel: self.channel.clone(),
182 token: self.token.clone(),
183 repository_id_bytes: id.into(),
184 }
185 }
186
187 pub async fn get_branch_by_name(&self, name: &str) -> Result<Branch, NapError> {
194 let mut client = self.make_client();
195 let response = client
196 .branch_get(BranchGetRequest {
197 query: Some(branch_get_request::Query::Name(name.to_string())),
198 })
199 .await
200 .map_err(|status| map_grpc_status("BranchGet", status))?;
201
202 response.into_inner().branch.ok_or_else(|| {
203 NapError::GrpcError(format!("BranchGet({name}) returned empty branch record"))
204 })
205 }
206
207 pub async fn list_branches(&self) -> Result<Vec<Branch>, NapError> {
208 let mut client = self.make_client();
209 let mut stream = client
210 .branch_list(BranchListRequest {
211 creator: None,
212 include_deleted: false,
213 })
214 .await
215 .map_err(|status| map_grpc_status("BranchList", status))?
216 .into_inner();
217 let mut branches = Vec::new();
218 while let Some(item) = stream
219 .message()
220 .await
221 .map_err(|status| map_grpc_status("BranchList", status))?
222 {
223 if let Some(branch) = item.branch {
224 branches.push(branch);
225 }
226 }
227 Ok(branches)
228 }
229
230 pub async fn list_revisions(
231 &self,
232 identifier: proto_gen::lore::model::v1::RevisionIdentifier,
233 ) -> Result<Vec<proto_gen::lore::model::v1::RevisionItem>, NapError> {
234 let mut client = self.make_client();
235 Ok(client
236 .revision_list(RevisionListRequest {
237 start: Some(
238 proto_gen::lore::revision::v1::revision_list_request::Start::Identifier(
239 identifier,
240 ),
241 ),
242 })
243 .await
244 .map_err(|status| map_grpc_status("RevisionList", status))?
245 .into_inner()
246 .items)
247 }
248
249 pub async fn push_branch(
259 &self,
260 branch_id: bytes::Bytes,
261 revision_signature: bytes::Bytes,
262 force: bool,
263 ) -> Result<(), NapError> {
264 let mut client = self.make_client();
265 client
266 .branch_push(BranchPushRequest {
267 id: branch_id,
268 revision_signature,
269 force,
270 fast_forward_merge: !force,
271 })
272 .await
273 .map_err(|status| map_grpc_status("BranchPush", status))?;
274 Ok(())
275 }
276
277 pub fn builder_from_env() -> Result<Option<Self>, NapError> {
285 Builder::from_env()
286 }
287
288 fn make_client(
290 &self,
291 ) -> RevisionServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
292 RevisionServiceClient::with_interceptor(
293 self.channel.clone(),
294 GrpcAuthInterceptor {
295 token: self.token.clone(),
296 repository_id_bytes: self.repository_id_bytes.clone(),
297 },
298 )
299 }
300
301 fn make_repository_client(
302 &self,
303 ) -> RepositoryServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
304 RepositoryServiceClient::with_interceptor(
305 self.channel.clone(),
306 GrpcAuthInterceptor {
307 token: self.token.clone(),
308 repository_id_bytes: Vec::new(),
309 },
310 )
311 }
312
313 fn make_storage_client(
314 &self,
315 ) -> StorageServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
316 StorageServiceClient::with_interceptor(
317 self.channel.clone(),
318 GrpcAuthInterceptor {
319 token: self.token.clone(),
320 repository_id_bytes: self.repository_id_bytes.clone(),
321 },
322 )
323 }
324
325 fn make_thin_client(
326 &self,
327 ) -> ThinClientServiceClient<InterceptedService<Channel, GrpcAuthInterceptor>> {
328 ThinClientServiceClient::with_interceptor(
329 self.channel.clone(),
330 GrpcAuthInterceptor {
331 token: self.token.clone(),
332 repository_id_bytes: self.repository_id_bytes.clone(),
333 },
334 )
335 }
336
337 pub async fn get_repository_by_name(
339 &self,
340 name: &str,
341 ) -> Result<proto_gen::lore::model::v1::Repository, NapError> {
342 let mut client = self.make_repository_client();
343 client
344 .repository_get(RepositoryGetRequest {
345 query: Some(
346 proto_gen::lore::repository::v1::repository_get_request::Query::Name(
347 name.to_string(),
348 ),
349 ),
350 })
351 .await
352 .map_err(|status| map_grpc_status("RepositoryGet", status))?
353 .into_inner()
354 .repository
355 .ok_or_else(|| {
356 NapError::GrpcError(format!("RepositoryGet({name}) returned no repository"))
357 })
358 }
359
360 pub async fn list_repositories(&self) -> Result<Vec<String>, NapError> {
362 let mut client = self.make_repository_client();
363 let mut stream = client
364 .repository_list(RepositoryListRequest { creator: None })
365 .await
366 .map_err(|status| map_grpc_status("RepositoryList", status))?
367 .into_inner();
368 let mut names = Vec::new();
369 while let Some(item) = stream
370 .message()
371 .await
372 .map_err(|status| map_grpc_status("RepositoryList", status))?
373 {
374 if let Some(repository) = item.repository {
375 names.push(repository.name);
376 }
377 }
378 Ok(names)
379 }
380
381 pub async fn read_file_at_revision(
384 &self,
385 identifier: proto_gen::lore::model::v1::RevisionIdentifier,
386 path: String,
387 ) -> Result<(Vec<u8>, Vec<u8>), NapError> {
388 let mut tree = self.make_thin_client();
389 let mut stream = tree
390 .revision_tree(RevisionTreeRequest {
391 query: Some(
392 proto_gen::lore::thin_client::v1::revision_tree_request::Query::Identifier(
393 identifier,
394 ),
395 ),
396 path_prefix: Some(path.clone()),
397 max_depth: Some(1),
398 })
399 .await
400 .map_err(|status| map_grpc_status("RevisionTree", status))?
401 .into_inner();
402 let mut signature = Vec::new();
403 let mut address = None;
404 while let Some(item) = stream
405 .message()
406 .await
407 .map_err(|status| map_grpc_status("RevisionTree", status))?
408 {
409 match item.payload {
410 Some(
411 proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Header(
412 header,
413 ),
414 ) => signature = header.signature.to_vec(),
415 Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
416 node,
417 )) if node.path == path => address = node.address,
418 _ => {}
419 }
420 }
421 let address = address.ok_or_else(|| NapError::ManifestNotFound(path.clone()))?;
422 let mut storage = self.make_storage_client();
423 let outgoing = tokio_stream::iter([address]);
424 let mut bytes = Vec::new();
425 let mut content = storage
426 .get(outgoing)
427 .await
428 .map_err(|status| map_grpc_status("StorageGet", status))?
429 .into_inner();
430 while let Some(chunk) = content
431 .message()
432 .await
433 .map_err(|status| map_grpc_status("StorageGet", status))?
434 {
435 bytes.extend_from_slice(&chunk.payload);
436 }
437 Ok((bytes, signature))
438 }
439
440 pub async fn read_file_at_signature(
442 &self,
443 signature: Vec<u8>,
444 path: String,
445 ) -> Result<(Vec<u8>, Vec<u8>), NapError> {
446 let mut tree = self.make_thin_client();
447 let mut stream = tree
448 .revision_tree(RevisionTreeRequest {
449 query: Some(
450 proto_gen::lore::thin_client::v1::revision_tree_request::Query::Signature(
451 signature.into(),
452 ),
453 ),
454 path_prefix: Some(path.clone()),
455 max_depth: Some(1),
456 })
457 .await
458 .map_err(|status| map_grpc_status("RevisionTree", status))?
459 .into_inner();
460 let mut resolved_signature = Vec::new();
461 let mut address = None;
462 while let Some(item) = stream
463 .message()
464 .await
465 .map_err(|status| map_grpc_status("RevisionTree", status))?
466 {
467 match item.payload {
468 Some(
469 proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Header(
470 header,
471 ),
472 ) => resolved_signature = header.signature.to_vec(),
473 Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
474 node,
475 )) if node.path == path => address = node.address,
476 _ => {}
477 }
478 }
479 let address = address.ok_or(NapError::ManifestNotFound(path))?;
480 let mut storage = self.make_storage_client();
481 let mut bytes = Vec::new();
482 let mut content = storage
483 .get(tokio_stream::iter([address]))
484 .await
485 .map_err(|status| map_grpc_status("StorageGet", status))?
486 .into_inner();
487 while let Some(chunk) = content
488 .message()
489 .await
490 .map_err(|status| map_grpc_status("StorageGet", status))?
491 {
492 bytes.extend_from_slice(&chunk.payload);
493 }
494 Ok((bytes, resolved_signature))
495 }
496
497 pub async fn list_paths_at_revision(
499 &self,
500 identifier: proto_gen::lore::model::v1::RevisionIdentifier,
501 prefix: String,
502 ) -> Result<Vec<String>, NapError> {
503 let mut tree = self.make_thin_client();
504 let mut stream = tree
505 .revision_tree(RevisionTreeRequest {
506 query: Some(
507 proto_gen::lore::thin_client::v1::revision_tree_request::Query::Identifier(
508 identifier,
509 ),
510 ),
511 path_prefix: Some(prefix),
512 max_depth: None,
513 })
514 .await
515 .map_err(|status| map_grpc_status("RevisionTree", status))?
516 .into_inner();
517 let mut paths = Vec::new();
518 while let Some(item) = stream
519 .message()
520 .await
521 .map_err(|status| map_grpc_status("RevisionTree", status))?
522 {
523 if let Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
524 node,
525 )) = item.payload
526 && node.node_type == proto_gen::lore::thin_client::v1::NodeType::File as i32
527 {
528 paths.push(node.path);
529 }
530 }
531 Ok(paths)
532 }
533
534 pub async fn file_address_at_revision(
536 &self,
537 identifier: proto_gen::lore::model::v1::RevisionIdentifier,
538 path: String,
539 ) -> Result<(proto_gen::lore::model::v1::Address, Vec<u8>), NapError> {
540 let mut tree = self.make_thin_client();
541 let mut stream = tree
542 .revision_tree(RevisionTreeRequest {
543 query: Some(
544 proto_gen::lore::thin_client::v1::revision_tree_request::Query::Identifier(
545 identifier,
546 ),
547 ),
548 path_prefix: Some(path.clone()),
549 max_depth: Some(1),
550 })
551 .await
552 .map_err(|status| map_grpc_status("RevisionTree", status))?
553 .into_inner();
554 let mut signature = Vec::new();
555 let mut address = None;
556 while let Some(item) = stream
557 .message()
558 .await
559 .map_err(|status| map_grpc_status("RevisionTree", status))?
560 {
561 match item.payload {
562 Some(
563 proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Header(
564 header,
565 ),
566 ) => signature = header.signature.to_vec(),
567 Some(proto_gen::lore::thin_client::v1::revision_tree_response::Payload::Node(
568 node,
569 )) if node.path == path => address = node.address,
570 _ => {}
571 }
572 }
573 address
574 .map(|address| (address, signature))
575 .ok_or(NapError::ManifestNotFound(path))
576 }
577
578 pub async fn revision_info_at_signature(
579 &self,
580 signature: Vec<u8>,
581 ) -> Result<proto_gen::lore::thin_client::v1::Revision, NapError> {
582 let mut client = self.make_thin_client();
583 client
584 .revision_info(RevisionInfoRequest {
585 query: Some(
586 proto_gen::lore::thin_client::v1::revision_info_request::Query::Signature(
587 signature.into(),
588 ),
589 ),
590 })
591 .await
592 .map_err(|status| map_grpc_status("RevisionInfo", status))?
593 .into_inner()
594 .revision
595 .ok_or_else(|| NapError::GrpcError("RevisionInfo returned no revision".to_string()))
596 }
597}
598
599#[derive(Default)]
614pub struct Builder {
615 endpoint: Option<String>,
616 token: Option<String>,
617 repository_id_bytes: Vec<u8>,
618 insecure: bool,
619}
620
621impl Builder {
622 pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
626 self.endpoint = Some(endpoint.into());
627 self
628 }
629
630 pub fn token(mut self, token: impl Into<String>) -> Self {
632 self.token = Some(token.into());
633 self
634 }
635
636 pub fn repository_id(mut self, id: impl Into<Vec<u8>>) -> Self {
641 self.repository_id_bytes = id.into();
642 self
643 }
644
645 pub fn insecure(mut self, insecure: bool) -> Self {
650 self.insecure = insecure;
651 self
652 }
653
654 pub fn build(self) -> Result<LoreGrpcClient, NapError> {
659 let endpoint_str = self.endpoint.ok_or_else(|| {
660 NapError::GrpcError(
661 "gRPC endpoint is required — set via .endpoint() or NAP_LORE_GRPC_ENDPOINT"
662 .to_string(),
663 )
664 })?;
665
666 let effective_url = if self.insecure {
671 endpoint_str
672 .strip_prefix("https://")
673 .map(|rest| format!("http://{rest}"))
674 .unwrap_or_else(|| endpoint_str.clone())
675 } else {
676 endpoint_str.clone()
677 };
678
679 let channel = Endpoint::from_shared(effective_url)
680 .map_err(|e| {
681 NapError::GrpcError(format!("invalid gRPC endpoint '{endpoint_str}': {e}"))
682 })?
683 .http2_keep_alive_interval(Duration::from_secs(30))
684 .keep_alive_timeout(Duration::from_secs(20))
685 .user_agent(concat!("nap-core/", env!("CARGO_PKG_VERSION")))
686 .map_err(|e| NapError::GrpcError(format!("user-agent configuration error: {e}")))?
687 .connect_lazy();
688
689 Ok(LoreGrpcClient {
690 channel,
691 token: self.token,
692 repository_id_bytes: self.repository_id_bytes,
693 })
694 }
695
696 pub fn from_env() -> Result<Option<LoreGrpcClient>, NapError> {
701 let endpoint = match std::env::var("NAP_LORE_GRPC_ENDPOINT") {
702 Ok(v) => v,
703 Err(_) => return Ok(None),
704 };
705
706 let token = std::env::var("NAP_LORE_GRPC_TOKEN").ok();
707 let insecure = std::env::var("NAP_LORE_GRPC_INSECURE")
708 .ok()
709 .is_some_and(|v| v == "1" || v == "true" || v == "yes");
710
711 let repository_id_bytes = std::env::var("NAP_LORE_GRPC_RID")
712 .ok()
713 .map(|hex| {
714 hex::decode(&hex).map_err(|e| {
715 NapError::GrpcError(format!("invalid NAP_LORE_GRPC_RID hex '{hex}': {e}"))
716 })
717 })
718 .transpose()?
719 .unwrap_or_default();
720
721 let mut builder = Builder::default().endpoint(endpoint).insecure(insecure);
722
723 if let Some(t) = token {
724 builder = builder.token(t);
725 }
726 if !repository_id_bytes.is_empty() {
727 builder = builder.repository_id(repository_id_bytes);
728 }
729
730 builder.build().map(Some)
731 }
732}
733
734pub fn block_on_grpc<F, T>(f: F) -> Result<T, NapError>
761where
762 F: Future<Output = Result<T, NapError>> + Send + 'static,
763 T: Send + 'static,
764{
765 static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
766 tokio::runtime::Builder::new_current_thread()
767 .enable_all()
768 .build()
769 .expect("failed to build gRPC tokio runtime")
770 });
771
772 let rt: &'static tokio::runtime::Runtime = &RUNTIME;
775
776 thread::Builder::new()
777 .name("nap-grpc".into())
778 .spawn(move || rt.block_on(f))
779 .expect("failed to spawn gRPC worker thread")
780 .join()
781 .map_err(|panic_payload| {
782 NapError::GrpcError(format!("gRPC thread panicked: {panic_payload:?}"))
783 })?
784}
785
786fn map_grpc_status(context: &str, status: tonic::Status) -> NapError {
792 let code = status.code();
793 let message = status.message();
794 match code {
795 tonic::Code::NotFound => NapError::RefNotFound(format!("{context}: {message}")),
796 tonic::Code::Unauthenticated | tonic::Code::PermissionDenied => {
797 NapError::PermissionDenied(format!("{context}: {message}"))
798 }
799 _ => NapError::GrpcError(format!("{context} ({code}): {message}")),
800 }
801}