From decbbd837f8a01cfcfdb9a7405abdf1af6f781d7 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Tue, 11 Nov 2025 09:49:58 -0600 Subject: [PATCH 1/9] WIP add persistence --- server/src/api.rs | 8 +++- server/src/api/v1/routes.rs | 81 +++++++++++++++++++++++++++++++++++-- 2 files changed, 85 insertions(+), 4 deletions(-) diff --git a/server/src/api.rs b/server/src/api.rs index b671062..e9f4034 100644 --- a/server/src/api.rs +++ b/server/src/api.rs @@ -3,7 +3,11 @@ pub mod v1; use std::sync::Arc; use anyhow::Result; -use axum::{response::Redirect, routing::get, Router}; +use axum::{ + response::Redirect, + routing::{get, post, put}, + Router, +}; use log::info; use tokio::net::TcpListener; use utoipa::OpenApi; @@ -55,7 +59,9 @@ pub async fn start(ctx: Arc) -> Result<()> { ) .route("/v1/routes", get(v1::routes::list_routes)) .route("/v1/routes/{cid}", get(v1::routes::get_routes)) + .route("/v1/data", post(v1::routes::create_data)) .route("/v1/data/{cid}", get(v1::routes::get_data)) + .route("/v1/data/{cid}", put(v1::routes::put_data)) .with_state(ctx); let listener = TcpListener::bind(addr).await?; diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index 9607f3e..38118c1 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -9,15 +9,20 @@ use axum::{ Json, }; use axum_extra::extract::TypedHeader; +use bao_tree::blake3; +use bytes::BytesMut; use cid::Cid; -use cid_router_core::db::{Direction, OrderBy}; +use cid_router_core::{ + cid::blake3_hash_to_cid, + db::{Direction, OrderBy}, +}; use futures::StreamExt; -use headers::Authorization; +use headers::{Authorization, ContentType}; use http_body::Frame; use http_body_util::StreamBody; use log::info; use serde::{Deserialize, Serialize}; -use utoipa::{IntoParams, ToSchema}; +use utoipa::{openapi::content, IntoParams, ToSchema}; use crate::context::Context; @@ -180,3 +185,73 @@ pub async fn get_data( .body(Body::empty()) .unwrap()) } + +/// Create data for a CID +pub async fn create_data( + auth: Option>>, + content_type: Option>, + State(ctx): State>, + body: Body, +) -> ApiResult { + let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); + ctx.auth.service().await.authenticate(token).await?; + + let content_type = content_type.map(|TypedHeader(mime)| mime.to_string()); + let cid_type = match content_type.as_ref().map(|ct| ct.as_str()) { + None => cid_router_core::cid::Codec::Raw, + Some("application/x-www-form-urlencoded") => cid_router_core::cid::Codec::Raw, + Some("application/octet-stream") => cid_router_core::cid::Codec::Raw, + Some("application/vnd.ipld.dag-cbor") => cid_router_core::cid::Codec::DagCbor, + _ => { + return Ok(Response::builder() + .status(StatusCode::UNSUPPORTED_MEDIA_TYPE) + .body(Body::empty())?); + } + }; + + let mut buffer = BytesMut::new(); + let mut stream = body.into_data_stream(); + while let Some(chunk) = stream.next().await { + let Ok(chunk) = chunk else { + return Ok(Response::builder() + .status(StatusCode::BAD_REQUEST) + .body(Body::empty())?); + }; + buffer.extend_from_slice(&chunk); + } + + let data = buffer.freeze(); + let hash = blake3::hash(&data); + let cid = blake3_hash_to_cid(hash.into(), cid_type); + + // === 4. TODO: Store `data` using `cid` as key === + // ctx.storage.put(&cid, data).await?; // ← implement this + + // === 5. Respond === + let json = serde_json::json!({ + "cid": cid.to_string(), + "size": data.len(), + "location": format!("/v1/data/{}", cid) + }); + + Ok(Response::builder() + .status(StatusCode::CREATED) + .header(header::LOCATION, format!("/v1/data/{}", cid)) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(json.to_string())) + .unwrap()) +} + +/// Put data for a CID +/// +/// This assumes that the hash is known by the sender, it will be verified. +pub async fn put_data( + Path(cid): Path, + auth: Option>>, + State(ctx): State>, +) -> ApiResult { + // TODO: + // - put data to a remote iroh-blobs store via blobs proto + // - create route in db + todo!(); +} From f43c44338d762f3d66d8c323ceb101dc5e7771b3 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Wed, 12 Nov 2025 08:43:20 -0600 Subject: [PATCH 2/9] WIP stub write --- Cargo.lock | 12 ++++++++ Cargo.toml | 2 +- core/src/crp.rs | 46 ++++++++++++++++++++++++++++-- core/src/db.rs | 2 ++ crps/azure/src/container.rs | 1 + crps/iroh/src/iroh.rs | 34 ++++++++++++++++++---- server/src/api.rs | 3 +- server/src/api/v1/routes.rs | 56 ++++++++++++++++++++++++++----------- 8 files changed, 129 insertions(+), 27 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c5dfa16..b1061c7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -252,6 +252,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a18ed336352031311f4e0b4dd2ff392d4fbb370777c9d18d7fc9d7359f73871" dependencies = [ "axum-core", + "axum-macros", "bytes", "form_urlencoded", "futures-util", @@ -320,6 +321,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-macros" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "azure_core" version = "0.19.0" diff --git a/Cargo.toml b/Cargo.toml index 4f76d8e..f184940 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ resolver = "2" [workspace.dependencies] anyhow = "1" async-trait = "0.1" -axum = "0.8" +axum = { version = "0.8", features = ["macros"] } axum-extra = { version = "0.10", features = ["typed-header"] } azure_core = "0.19" azure_identity = "0.19" diff --git a/core/src/crp.rs b/core/src/crp.rs index 2e5b357..3dc940b 100644 --- a/core/src/crp.rs +++ b/core/src/crp.rs @@ -1,4 +1,4 @@ -use std::pin::Pin; +use std::{fmt::Debug, pin::Pin, sync::Arc}; use anyhow::Result; use async_trait::async_trait; @@ -39,7 +39,7 @@ impl std::fmt::Display for ProviderType { /// CID Route Provider (CRP) Trait #[async_trait] -pub trait Crp: Send + Sync { +pub trait Crp: Send + Sync + Debug { fn provider_id(&self) -> String; fn provider_type(&self) -> ProviderType; async fn reindex(&self, cx: &Context) -> Result<()>; @@ -53,10 +53,32 @@ pub trait Crp: Send + Sync { } } +#[async_trait] +impl Crp for Arc { + fn provider_id(&self) -> String { + self.as_ref().provider_id() + } + fn provider_type(&self) -> ProviderType { + self.as_ref().provider_type() + } + async fn reindex(&self, cx: &Context) -> Result<()> { + self.as_ref().reindex(cx).await + } + + fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { + self.as_ref().capabilities() + } + + fn cid_filter(&self) -> CidFilter { + self.as_ref().cid_filter() + } +} + /// All capabilities a CRP may have represented as self-referential trait objects. pub struct CrpCapabilities<'a> { pub route_resolver: Option<&'a dyn RouteResolver>, pub size_resolver: Option<&'a dyn SizeResolver>, + pub blob_writer: Option<&'a dyn BlobWriter>, } /// A RouteResolver can dereference a route, turning it into a stream of bytes, accepting @@ -89,3 +111,23 @@ pub trait SizeResolver { auth: Vec, ) -> Result>; } + +/// A RouteResolver can dereference a route, turning it into a stream of bytes, accepting +/// authentication data. +#[async_trait] +pub trait BlobWriter: Send + Sync { + async fn put_blob( + &self, + auth: Option, + cid: &Cid, + data: Pin< + Box< + dyn Stream>> + + Send, + >, + > + ) -> Result< + (), + Box, + >; +} \ No newline at end of file diff --git a/core/src/db.rs b/core/src/db.rs index da67c24..9a7e001 100644 --- a/core/src/db.rs +++ b/core/src/db.rs @@ -362,6 +362,7 @@ mod tests { Context, }; + #[derive(Debug)] struct StubAzureProvider {} #[async_trait] @@ -382,6 +383,7 @@ mod tests { CrpCapabilities { route_resolver: None, size_resolver: None, + blob_writer: None, } } diff --git a/crps/azure/src/container.rs b/crps/azure/src/container.rs index 2a3c06c..d9e5f66 100644 --- a/crps/azure/src/container.rs +++ b/crps/azure/src/container.rs @@ -49,6 +49,7 @@ impl Crp for Container { CrpCapabilities { route_resolver: Some(self), size_resolver: None, // TODO + blob_writer: None, // TODO } } diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index e8c3f92..4048239 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -4,11 +4,9 @@ use anyhow::Result; use async_trait::async_trait; use bao_tree::io::BaoContentItem; use bytes::Bytes; +use cid::Cid; use cid_router_core::{ - cid_filter::{CidFilter, CodeFilter}, - crp::{Crp, CrpCapabilities, ProviderType, RouteResolver}, - routes::Route, - Context, + Context, cid_filter::{CidFilter, CodeFilter}, crp::{BlobWriter, Crp, CrpCapabilities, ProviderType, RouteResolver}, routes::Route }; use futures::{Stream, StreamExt}; use iroh::{Endpoint, NodeAddr, NodeId}; @@ -20,6 +18,7 @@ use serde_json::Value; pub struct IrohCrp { node_addr: NodeAddr, endpoint: Endpoint, + allow_put: bool, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -59,6 +58,7 @@ impl IrohCrp { Ok(Self { node_addr, endpoint, + allow_put: true, }) } } @@ -75,13 +75,14 @@ impl Crp for IrohCrp { async fn reindex(&self, _cx: &Context) -> anyhow::Result<()> { // TODO: Implement reindexing logic - todo!(); + Ok(()) } fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { CrpCapabilities { route_resolver: Some(self), size_resolver: None, // TODO + blob_writer: if self.allow_put { Some(self) } else { None }, } } @@ -90,6 +91,29 @@ impl Crp for IrohCrp { } } + +#[async_trait] +impl BlobWriter for IrohCrp { + async fn put_blob( + &self, + _auth: Option, + cid: &Cid, + data: Pin< + Box< + dyn Stream>> + + Send, + >, + > + ) -> Result< + (), + Box, + > { + // TODO + println!("Putting blob with cid: {}", cid); + Ok(()) + } +} + #[async_trait] impl RouteResolver for IrohCrp { async fn get_bytes( diff --git a/server/src/api.rs b/server/src/api.rs index e9f4034..43271a1 100644 --- a/server/src/api.rs +++ b/server/src/api.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use anyhow::Result; use axum::{ response::Redirect, - routing::{get, post, put}, + routing::{get, post}, Router, }; use log::info; @@ -61,7 +61,6 @@ pub async fn start(ctx: Arc) -> Result<()> { .route("/v1/routes/{cid}", get(v1::routes::get_routes)) .route("/v1/data", post(v1::routes::create_data)) .route("/v1/data/{cid}", get(v1::routes::get_data)) - .route("/v1/data/{cid}", put(v1::routes::put_data)) .with_state(ctx); let listener = TcpListener::bind(addr).await?; diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index 38118c1..1347c20 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -22,7 +22,7 @@ use http_body::Frame; use http_body_util::StreamBody; use log::info; use serde::{Deserialize, Serialize}; -use utoipa::{openapi::content, IntoParams, ToSchema}; +use utoipa::{IntoParams, ToSchema}; use crate::context::Context; @@ -187,6 +187,7 @@ pub async fn get_data( } /// Create data for a CID +#[axum::debug_handler] pub async fn create_data( auth: Option>>, content_type: Option>, @@ -196,6 +197,7 @@ pub async fn create_data( let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); ctx.auth.service().await.authenticate(token).await?; + // Check if content-type is supported and translate to cid type let content_type = content_type.map(|TypedHeader(mime)| mime.to_string()); let cid_type = match content_type.as_ref().map(|ct| ct.as_str()) { None => cid_router_core::cid::Codec::Raw, @@ -209,6 +211,7 @@ pub async fn create_data( } }; + // Read data - we assume this to be small enough to fit into memory for now let mut buffer = BytesMut::new(); let mut stream = body.into_data_stream(); while let Some(chunk) = stream.next().await { @@ -220,14 +223,46 @@ pub async fn create_data( buffer.extend_from_slice(&chunk); } + // compute CID let data = buffer.freeze(); let hash = blake3::hash(&data); let cid = blake3_hash_to_cid(hash.into(), cid_type); - // === 4. TODO: Store `data` using `cid` as key === - // ctx.storage.put(&cid, data).await?; // ← implement this + // Find writers + let writers = ctx.providers.iter() + .filter(|p| p.provider_is_eligible_for_cid(&cid)) + .filter_map(|p| p.capabilities().blob_writer.map(|w| (p, w))).collect::>(); + if writers.is_empty() { + return Ok(Response::builder() + .status(StatusCode::SERVICE_UNAVAILABLE) + .body(Body::empty())?); + } + + let mut outcome = Vec::new(); + for (id, writer) in writers { + let data = data.clone(); + let res = writer.put_blob( + None, + &cid, + Box::pin(futures::stream::once(async move { Ok(data) })), + ).await; + outcome.push((id, res)); + } + + for (provider, res) in &outcome { + if res.is_ok() { + let route = cid_router_core::routes::Route::builder(*provider) + .cid(cid) + .multicodec(cid_router_core::cid::Codec::Raw) + .size(data.len() as u64) + .url(cid.to_string()) + .build(&ctx.core)?; + ctx.core.db().insert_route( + &route + ).await?; + } + } - // === 5. Respond === let json = serde_json::json!({ "cid": cid.to_string(), "size": data.len(), @@ -242,16 +277,3 @@ pub async fn create_data( .unwrap()) } -/// Put data for a CID -/// -/// This assumes that the hash is known by the sender, it will be verified. -pub async fn put_data( - Path(cid): Path, - auth: Option>>, - State(ctx): State>, -) -> ApiResult { - // TODO: - // - put data to a remote iroh-blobs store via blobs proto - // - create route in db - todo!(); -} From 23952ba8cdcf53bc3b4330786ba0e2d48df27c42 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Wed, 12 Nov 2025 09:49:01 -0600 Subject: [PATCH 3/9] Hacked in writing via push request This desperately needs clean up, but it does work. --- Cargo.lock | 107 +++++++++++++++++++++++++++++++++++- Cargo.toml | 1 + core/src/crp.rs | 14 +---- crps/azure/src/container.rs | 2 +- crps/iroh/Cargo.toml | 2 + crps/iroh/src/iroh.rs | 104 ++++++++++++++++++++++++++++------- server/src/api.rs | 2 + server/src/api/v1/routes.rs | 20 +++---- server/src/context.rs | 13 +++-- 9 files changed, 214 insertions(+), 51 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b1061c7..868da0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1010,6 +1010,8 @@ dependencies = [ "iroh", "iroh-base", "iroh-blobs", + "irpc 0.11.0", + "log", "n0-future 0.2.0", "serde", "serde_json", @@ -2488,7 +2490,7 @@ dependencies = [ "iroh-io", "iroh-metrics", "iroh-quinn", - "irpc", + "irpc 0.7.0", "n0-future 0.1.3", "n0-snafu", "nested_enum_utils", @@ -2661,10 +2663,10 @@ dependencies = [ "futures-buffered", "futures-util", "iroh-quinn", - "irpc-derive", + "irpc-derive 0.5.0", "n0-future 0.1.3", "postcard", - "rcgen", + "rcgen 0.13.2", "rustls", "serde", "smallvec", @@ -2674,6 +2676,28 @@ dependencies = [ "tracing", ] +[[package]] +name = "irpc" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bee97aaa18387c4f0aae61058195dc9f9dea3e41c0e272973fe3e9bf611563d" +dependencies = [ + "futures-buffered", + "futures-util", + "iroh-quinn", + "irpc-derive 0.9.0", + "n0-error", + "n0-future 0.3.1", + "postcard", + "rcgen 0.14.5", + "rustls", + "serde", + "smallvec", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "irpc-derive" version = "0.5.0" @@ -2685,6 +2709,17 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "irpc-derive" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "58148196d2230183c9679431ac99b57e172000326d664e8456fa2cd27af6505a" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.1" @@ -2976,6 +3011,27 @@ dependencies = [ "unsigned-varint", ] +[[package]] +name = "n0-error" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7d5969a2f40e9d9ed121a789c415f4114ac2b28e5731c080bdefee217d3b3fb" +dependencies = [ + "n0-error-macros", + "spez", +] + +[[package]] +name = "n0-error-macros" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a6908df844696d9af91c7c3950d50e52d67df327d02a95367f95bbf177d6556" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "n0-future" version = "0.1.3" @@ -3018,6 +3074,27 @@ dependencies = [ "web-time", ] +[[package]] +name = "n0-future" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c0709ac8235ce13b82bc4d180ee3c42364b90c1a8a628c3422d991d75a728b5" +dependencies = [ + "cfg_aliases", + "derive_more 1.0.0", + "futures-buffered", + "futures-lite 2.6.1", + "futures-util", + "js-sys", + "pin-project", + "send_wrapper", + "tokio", + "tokio-util", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-time", +] + [[package]] name = "n0-snafu" version = "0.2.1" @@ -4050,6 +4127,19 @@ dependencies = [ "yasna", ] +[[package]] +name = "rcgen" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fae430c6b28f1ad601274e78b7dffa0546de0b73b4cd32f46723c0c2a16f7a5" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "yasna", +] + [[package]] name = "redb" version = "2.4.0" @@ -4812,6 +4902,17 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "spez" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c87e960f4dca2788eeb86bbdde8dd246be8948790b7618d656e68f9b720a86e8" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "spin" version = "0.9.8" diff --git a/Cargo.toml b/Cargo.toml index f184940..04eaee5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,3 +52,4 @@ utoipa = { version = "5", features = ["axum_extras"] } utoipa-axum = "0.2" utoipa-swagger-ui = { version = "9", features = ["axum"] } uuid = { version = "1.18.1", features = ["serde", "v4"] } +irpc = "0.11.0" diff --git a/core/src/crp.rs b/core/src/crp.rs index 3dc940b..ef9ca48 100644 --- a/core/src/crp.rs +++ b/core/src/crp.rs @@ -120,14 +120,6 @@ pub trait BlobWriter: Send + Sync { &self, auth: Option, cid: &Cid, - data: Pin< - Box< - dyn Stream>> - + Send, - >, - > - ) -> Result< - (), - Box, - >; -} \ No newline at end of file + data: &[u8], + ) -> Result<(), Box>; +} diff --git a/crps/azure/src/container.rs b/crps/azure/src/container.rs index d9e5f66..fc8b59f 100644 --- a/crps/azure/src/container.rs +++ b/crps/azure/src/container.rs @@ -49,7 +49,7 @@ impl Crp for Container { CrpCapabilities { route_resolver: Some(self), size_resolver: None, // TODO - blob_writer: None, // TODO + blob_writer: None, // TODO } } diff --git a/crps/iroh/Cargo.toml b/crps/iroh/Cargo.toml index 6493296..8180a6f 100644 --- a/crps/iroh/Cargo.toml +++ b/crps/iroh/Cargo.toml @@ -21,3 +21,5 @@ n0-future.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true +irpc.workspace = true +log.workspace = true diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index 4048239..0760add 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -2,15 +2,29 @@ use std::{pin::Pin, str::FromStr}; use anyhow::Result; use async_trait::async_trait; -use bao_tree::io::BaoContentItem; +use bao_tree::{ + io::{outboard::PreOrderMemOutboard, BaoContentItem}, + ChunkRanges, +}; use bytes::Bytes; use cid::Cid; use cid_router_core::{ - Context, cid_filter::{CidFilter, CodeFilter}, crp::{BlobWriter, Crp, CrpCapabilities, ProviderType, RouteResolver}, routes::Route + cid_filter::{CidFilter, CodeFilter}, + crp::{BlobWriter, Crp, CrpCapabilities, ProviderType, RouteResolver}, + routes::Route, + Context, }; use futures::{Stream, StreamExt}; -use iroh::{Endpoint, NodeAddr, NodeId}; -use iroh_blobs::{get::request::GetBlobItem, ticket::BlobTicket, Hash}; +use iroh::{endpoint::SendStream, Endpoint, NodeAddr, NodeId, SecretKey}; +use iroh_blobs::{ + get::request::GetBlobItem, + protocol::{ChunkRangesSeq, PushRequest, RequestType}, + store::IROH_BLOCK_SIZE, + ticket::BlobTicket, + Hash, +}; +use irpc::util::WriteVarintExt; +use log::error; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -35,7 +49,7 @@ pub enum IrohNodeAddrRef { } impl IrohCrp { - pub async fn new_from_config(config: Value) -> Result { + pub async fn new_from_config(config: Value, secret_key: SecretKey) -> Result { let IrohCrpConfig { node_addr_ref } = serde_json::from_value(config)?; let node_addr = match node_addr_ref { @@ -53,7 +67,11 @@ impl IrohCrp { } }; - let endpoint = Endpoint::builder().discovery_n0().bind().await?; + let endpoint = Endpoint::builder() + .discovery_n0() + .secret_key(secret_key) + .bind() + .await?; Ok(Self { node_addr, @@ -91,29 +109,77 @@ impl Crp for IrohCrp { } } - #[async_trait] impl BlobWriter for IrohCrp { async fn put_blob( &self, _auth: Option, cid: &Cid, - data: Pin< - Box< - dyn Stream>> - + Send, - >, - > - ) -> Result< - (), - Box, - > { - // TODO - println!("Putting blob with cid: {}", cid); + data: &[u8], + ) -> Result<(), Box> { + error!("Putting blob with cid: {}", cid); + error!("I am: {:?}", self.endpoint.node_id()); + error!("Target: {:?}", self.node_addr.node_id); + if !self.allow_put { + return Err("Put operations are not allowed on this CRP".into()); + } + if cid.hash().code() != 0x1e { + return Err("Unsupported CID hash code; only blake3 is supported".into()); + } + let hash: [u8; 32] = cid + .hash() + .digest() + .try_into() + .expect("blake3 hash must be 32 bytes"); + let hash = iroh_blobs::Hash::from_bytes(hash); + error!("Connecting to node..."); + let conn = self + .endpoint + .connect(self.node_addr.clone(), iroh_blobs::ALPN) + .await?; + error!("Connected. Opening stream..."); + let (mut writer, mut reader) = conn.open_bi().await?; + error!("Opened stream. Writing push request..."); + let request = PushRequest::new(hash, ChunkRangesSeq::root()); + let request = write_push_request(request, &mut writer).await?; + let (hash, bao) = create_n0_bao(data, &ChunkRanges::all())?; + if hash != request.hash { + return Err("Computed hash does not match requested hash".into()); + } + writer.write_all(&bao).await?; + writer.finish()?; + let res = reader.read_to_end(1024).await; + if let Err(e) = res { + error!("Error reading response: {}", e); + return Err(Box::new(e)); + } + error!("Blob put completed for cid: {}", cid); Ok(()) } } +/// TODO: make this available in iroh-blobs +async fn write_push_request( + request: PushRequest, + stream: &mut SendStream, +) -> anyhow::Result { + let mut request_bytes = Vec::new(); + request_bytes.push(RequestType::Push as u8); + request_bytes.write_length_prefixed(&request).unwrap(); + stream.write_all(&request_bytes).await?; + Ok(request) +} + +/// TODO: move this to iroh-blobs +pub fn create_n0_bao(data: &[u8], ranges: &ChunkRanges) -> anyhow::Result<(Hash, Vec)> { + let outboard = PreOrderMemOutboard::create(data, IROH_BLOCK_SIZE); + let mut encoded = Vec::new(); + let size = data.len() as u64; + encoded.extend_from_slice(&size.to_le_bytes()); + bao_tree::io::sync::encode_ranges_validated(data, &outboard, ranges, &mut encoded)?; + Ok((outboard.root.into(), encoded)) +} + #[async_trait] impl RouteResolver for IrohCrp { async fn get_bytes( diff --git a/server/src/api.rs b/server/src/api.rs index 43271a1..df188c1 100644 --- a/server/src/api.rs +++ b/server/src/api.rs @@ -8,6 +8,7 @@ use axum::{ routing::{get, post}, Router, }; +use cid_router_core::context::Signer; use log::info; use tokio::net::TcpListener; use utoipa::OpenApi; @@ -46,6 +47,7 @@ pub async fn start(ctx: Arc) -> Result<()> { info!("🚀 Starting CID Router"); info!("🚀 HTTP API = {addr}"); + info!("🚀 ID = {}", ctx.core.public_key()); let router = Router::new() .merge( diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index 1347c20..24de4a8 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -229,9 +229,12 @@ pub async fn create_data( let cid = blake3_hash_to_cid(hash.into(), cid_type); // Find writers - let writers = ctx.providers.iter() + let writers = ctx + .providers + .iter() .filter(|p| p.provider_is_eligible_for_cid(&cid)) - .filter_map(|p| p.capabilities().blob_writer.map(|w| (p, w))).collect::>(); + .filter_map(|p| p.capabilities().blob_writer.map(|w| (p, w))) + .collect::>(); if writers.is_empty() { return Ok(Response::builder() .status(StatusCode::SERVICE_UNAVAILABLE) @@ -241,14 +244,10 @@ pub async fn create_data( let mut outcome = Vec::new(); for (id, writer) in writers { let data = data.clone(); - let res = writer.put_blob( - None, - &cid, - Box::pin(futures::stream::once(async move { Ok(data) })), - ).await; + let res = writer.put_blob(None, &cid, &data).await; outcome.push((id, res)); } - + for (provider, res) in &outcome { if res.is_ok() { let route = cid_router_core::routes::Route::builder(*provider) @@ -257,9 +256,7 @@ pub async fn create_data( .size(data.len() as u64) .url(cid.to_string()) .build(&ctx.core)?; - ctx.core.db().insert_route( - &route - ).await?; + ctx.core.db().insert_route(&route).await?; } } @@ -276,4 +273,3 @@ pub async fn create_data( .body(Body::from(json.to_string())) .unwrap()) } - diff --git a/server/src/context.rs b/server/src/context.rs index fb8d459..3138971 100644 --- a/server/src/context.rs +++ b/server/src/context.rs @@ -27,20 +27,23 @@ impl Context { let auth = config.auth.clone(); + let secret = repo.secret_key().await?; let core = cid_router_core::context::Context::from_repo(repo).await?; - let providers = future::join_all(config.providers.into_iter().map( - |provider_config| async move { + let providers = future::join_all(config.providers.into_iter().map(|provider_config| { + let secret = secret.clone(); + async move { match provider_config { ProviderConfig::Iroh(iroh_config) => Ok(Arc::new( - IrohCrp::new_from_config(serde_json::to_value(iroh_config)?).await?, + IrohCrp::new_from_config(serde_json::to_value(iroh_config)?, secret) + .await?, ) as Arc), ProviderConfig::Azure(azure_config) => { Ok(Arc::new(AzureContainer::new(azure_config)) as Arc) } } - }, - )) + } + })) .await .into_iter() .collect::>>()?; From 209fed35daeb1e7a151f9657716f9478628e64f4 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Wed, 12 Nov 2025 10:29:14 -0600 Subject: [PATCH 4/9] Don't store twice. --- core/src/crp.rs | 6 ++++++ server/src/api/v1/routes.rs | 18 ++++++++++++++---- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/core/src/crp.rs b/core/src/crp.rs index ef9ca48..d644b1b 100644 --- a/core/src/crp.rs +++ b/core/src/crp.rs @@ -116,6 +116,12 @@ pub trait SizeResolver { /// authentication data. #[async_trait] pub trait BlobWriter: Send + Sync { + + /// Puts a blob into the CRP, given optional authentication data, a CID, and the data bytes. + /// + /// Note that this assumes that the data fits in memory, which is probably the case for most + /// data that eqty wants to write. If this becomes a problem, we will add a second method that + /// takes a stream of bytes instead. async fn put_blob( &self, auth: Option, diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index 24de4a8..a189bbf 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -1,4 +1,4 @@ -use std::{str::FromStr, sync::Arc}; +use std::{collections::HashSet, str::FromStr, sync::Arc}; use api_utils::ApiResult; use axum::{ @@ -20,7 +20,7 @@ use futures::StreamExt; use headers::{Authorization, ContentType}; use http_body::Frame; use http_body_util::StreamBody; -use log::info; +use log::{error, info}; use serde::{Deserialize, Serialize}; use utoipa::{IntoParams, ToSchema}; @@ -241,11 +241,21 @@ pub async fn create_data( .body(Body::empty())?); } + let existing = ctx.core.db().routes_for_cid(cid).await?; + let existing_ids = existing + .iter() + .map(|r| r.provider_id.clone()) + .collect::>(); + let mut outcome = Vec::new(); - for (id, writer) in writers { + for (crp, writer) in writers { + if existing_ids.contains(&crp.provider_id()) { + error!("Skipping put to provider {} as route already exists", crp.provider_id()); + continue; + } let data = data.clone(); let res = writer.put_blob(None, &cid, &data).await; - outcome.push((id, res)); + outcome.push((crp, res)); } for (provider, res) in &outcome { From 3b88bb9d78d5266c61e59ed818657c0d2a0c20a4 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 5 Dec 2025 11:59:56 +0200 Subject: [PATCH 5/9] Implement inline CRP --- core/src/crp.rs | 3 +- crps/iroh/src/iroh.rs | 106 ++++++++++++++++++++++++++++++++++-- crps/iroh/src/lib.rs | 2 +- server/config.example.toml | 8 +-- server/src/api/v1/routes.rs | 5 +- server/src/context.rs | 18 +++--- 6 files changed, 116 insertions(+), 26 deletions(-) diff --git a/core/src/crp.rs b/core/src/crp.rs index d644b1b..6d37af8 100644 --- a/core/src/crp.rs +++ b/core/src/crp.rs @@ -116,9 +116,8 @@ pub trait SizeResolver { /// authentication data. #[async_trait] pub trait BlobWriter: Send + Sync { - /// Puts a blob into the CRP, given optional authentication data, a CID, and the data bytes. - /// + /// /// Note that this assumes that the data fits in memory, which is probably the case for most /// data that eqty wants to write. If this becomes a problem, we will add a second method that /// takes a stream of bytes instead. diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index 0760add..084dca6 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -1,4 +1,4 @@ -use std::{pin::Pin, str::FromStr}; +use std::{io, path::PathBuf, pin::Pin, str::FromStr}; use anyhow::Result; use async_trait::async_trait; @@ -29,14 +29,25 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; #[derive(Debug)] -pub struct IrohCrp { +pub struct IrohRemoteCrp { node_addr: NodeAddr, endpoint: Endpoint, allow_put: bool, } +#[derive(Debug, Clone)] +pub struct IrohCrp { + store: iroh_blobs::store::fs::FsStore, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct IrohCrpConfig { + /// Path to the directory where blobs are stored + pub path: PathBuf, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct IrohRemoteCrpConfig { pub node_addr_ref: IrohNodeAddrRef, } @@ -49,8 +60,22 @@ pub enum IrohNodeAddrRef { } impl IrohCrp { + pub async fn new_from_config(config: IrohCrpConfig) -> io::Result { + let path = if config.path.is_absolute() { + config.path + } else { + std::env::current_dir()?.join(config.path) + }; + let store = iroh_blobs::store::fs::FsStore::load(path) + .await + .map_err(|e| io::Error::other(e))?; + Ok(Self { store }) + } +} + +impl IrohRemoteCrp { pub async fn new_from_config(config: Value, secret_key: SecretKey) -> Result { - let IrohCrpConfig { node_addr_ref } = serde_json::from_value(config)?; + let IrohRemoteCrpConfig { node_addr_ref } = serde_json::from_value(config)?; let node_addr = match node_addr_ref { IrohNodeAddrRef::NodeId(node_id) => { @@ -82,7 +107,7 @@ impl IrohCrp { } #[async_trait] -impl Crp for IrohCrp { +impl Crp for IrohRemoteCrp { fn provider_id(&self) -> String { "iroh".to_string() } @@ -109,8 +134,79 @@ impl Crp for IrohCrp { } } +#[async_trait] +impl Crp for IrohCrp { + fn provider_id(&self) -> String { + "iroh_inline".to_string() + } + + fn provider_type(&self) -> ProviderType { + ProviderType::Iroh + } + + async fn reindex(&self, _cx: &Context) -> anyhow::Result<()> { + // TODO: Implement reindexing logic + Ok(()) + } + + fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { + CrpCapabilities { + route_resolver: Some(self), + size_resolver: None, // TODO + blob_writer: Some(self), + } + } + + fn cid_filter(&self) -> CidFilter { + CidFilter::MultihashCodeFilter(CodeFilter::Eq(0x1e)) // blake3 + } +} + #[async_trait] impl BlobWriter for IrohCrp { + async fn put_blob( + &self, + _auth: Option, + cid: &Cid, + data: &[u8], + ) -> Result<(), Box> { + let blobs = self.store.blobs().clone(); + let data = Bytes::copy_from_slice(data); + if cid.hash().code() != 0x1e { + return Err("Unsupported CID hash code; only blake3 is supported".into()); + } + blobs.add_bytes(data).with_tag().await.map_err(Box::new)?; + Ok(()) + } +} + +#[async_trait] +impl RouteResolver for IrohCrp { + async fn get_bytes( + &self, + route: &Route, + _auth: Option, + ) -> Result< + Pin< + Box< + dyn Stream>> + + Send, + >, + >, + Box, + > { + let cid = route.cid; + let hash = cid.hash().digest(); + let hash: [u8; 32] = hash.try_into()?; + let hash = Hash::from_bytes(hash); + let data = self.store.blobs().get_bytes(hash).await.map_err(Box::new)?; + let stream = futures::stream::once(async move { Ok(data) }); + Ok(Box::pin(stream)) + } +} + +#[async_trait] +impl BlobWriter for IrohRemoteCrp { async fn put_blob( &self, _auth: Option, @@ -181,7 +277,7 @@ pub fn create_n0_bao(data: &[u8], ranges: &ChunkRanges) -> anyhow::Result<(Hash, } #[async_trait] -impl RouteResolver for IrohCrp { +impl RouteResolver for IrohRemoteCrp { async fn get_bytes( &self, route: &Route, diff --git a/crps/iroh/src/lib.rs b/crps/iroh/src/lib.rs index d1b3697..6efc5a9 100644 --- a/crps/iroh/src/lib.rs +++ b/crps/iroh/src/lib.rs @@ -1,3 +1,3 @@ mod iroh; -pub use iroh::{IrohCrp, IrohCrpConfig, IrohNodeAddrRef}; +pub use iroh::{IrohCrp, IrohCrpConfig, IrohNodeAddrRef, IrohRemoteCrp, IrohRemoteCrpConfig}; diff --git a/server/config.example.toml b/server/config.example.toml index 24423d6..6f95322 100644 --- a/server/config.example.toml +++ b/server/config.example.toml @@ -1,10 +1,6 @@ port = 3080 - -# [[providers]] -# type = "ipfs" -# gateway_url = "http://localhost:8080" +auth = "none" [[providers]] type = "iroh" -# node_addr_ref = { node_id = "w36hbmld67hrocfllnfca4ahzae2ibrom2moj2lovjguye3gkmiq" } -node_addr_ref = { ticket = "blobaccbd3d6iyowiix4ixt5btbxndo5mamzbhcbfksn55krurogsrgbwajdnb2hi4dthixs65ltmuys2mjoojswyylzfzuxe33ifzxgk5dxn5zgwlrpauaesa732pf6aaqavqiqaaol4abablataaa4xyacacwboaabzpqaeagavaafbs7aaiax3vlpwtrmwr4owttczv6g4pglwz26xxj4bgovjfcmvus7awi6dda" } +path = "/Users/rklaehn/projects_git/cid_router/blobs" diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index a189bbf..60598d4 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -250,7 +250,10 @@ pub async fn create_data( let mut outcome = Vec::new(); for (crp, writer) in writers { if existing_ids.contains(&crp.provider_id()) { - error!("Skipping put to provider {} as route already exists", crp.provider_id()); + error!( + "Skipping put to provider {} as route already exists", + crp.provider_id() + ); continue; } let data = data.clone(); diff --git a/server/src/context.rs b/server/src/context.rs index 3138971..c7ccf99 100644 --- a/server/src/context.rs +++ b/server/src/context.rs @@ -26,24 +26,20 @@ impl Context { let port = config.port; let auth = config.auth.clone(); - - let secret = repo.secret_key().await?; let core = cid_router_core::context::Context::from_repo(repo).await?; - let providers = future::join_all(config.providers.into_iter().map(|provider_config| { - let secret = secret.clone(); - async move { + let providers = future::join_all(config.providers.into_iter().map( + |provider_config| async move { match provider_config { - ProviderConfig::Iroh(iroh_config) => Ok(Arc::new( - IrohCrp::new_from_config(serde_json::to_value(iroh_config)?, secret) - .await?, - ) as Arc), + ProviderConfig::Iroh(iroh_config) => { + Ok(Arc::new(IrohCrp::new_from_config(iroh_config).await?) as Arc) + } ProviderConfig::Azure(azure_config) => { Ok(Arc::new(AzureContainer::new(azure_config)) as Arc) } } - } - })) + }, + )) .await .into_iter() .collect::>>()?; From 05eca3d5fbea488275a357e0041cdb1bd2efc31b Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 5 Dec 2025 17:32:45 +0200 Subject: [PATCH 6/9] Move remote crp into separate file Also remove direct bao-tree dependency. --- Cargo.lock | 4 +- Cargo.toml | 1 - core/Cargo.toml | 1 - crps/iroh/Cargo.toml | 1 - crps/iroh/src/iroh.rs | 224 +----------------------------------- crps/iroh/src/lib.rs | 2 +- server/Cargo.toml | 2 +- server/src/api/v1/routes.rs | 1 - 8 files changed, 6 insertions(+), 230 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 868da0d..0591b83 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -673,7 +673,6 @@ dependencies = [ "api-utils", "async-trait", "axum", - "bao-tree", "bytes", "chrono", "cid", @@ -715,7 +714,7 @@ dependencies = [ "async-trait", "axum", "axum-extra", - "bao-tree", + "blake3", "bytes", "chrono", "cid", @@ -1002,7 +1001,6 @@ version = "0.0.0" dependencies = [ "anyhow", "async-trait", - "bao-tree", "bytes", "cid", "cid-router-core", diff --git a/Cargo.toml b/Cargo.toml index 04eaee5..3228ade 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,6 @@ azure_core = "0.19" azure_identity = "0.19" azure_storage = "0.19" azure_storage_blobs = "0.19" -bao-tree = "0.15.1" blake3 = "1.5" bytes = "1.10.1" chrono = "0.4" diff --git a/core/Cargo.toml b/core/Cargo.toml index 425f9c7..7284ee7 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -11,7 +11,6 @@ anyhow.workspace = true api-utils = { path = "../crates/api-utils" } async-trait.workspace = true axum.workspace = true -bao-tree.workspace = true bytes.workspace = true chrono.workspace = true cid.workspace = true diff --git a/crps/iroh/Cargo.toml b/crps/iroh/Cargo.toml index 8180a6f..d28df7b 100644 --- a/crps/iroh/Cargo.toml +++ b/crps/iroh/Cargo.toml @@ -9,7 +9,6 @@ edition = "2021" anyhow.workspace = true async-trait.workspace = true -bao-tree.workspace = true bytes.workspace = true cid.workspace = true cid-router-core = { path = "../../core" } diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index 084dca6..6b55cfc 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -1,11 +1,7 @@ -use std::{io, path::PathBuf, pin::Pin, str::FromStr}; +use std::{io, path::PathBuf, pin::Pin}; use anyhow::Result; use async_trait::async_trait; -use bao_tree::{ - io::{outboard::PreOrderMemOutboard, BaoContentItem}, - ChunkRanges, -}; use bytes::Bytes; use cid::Cid; use cid_router_core::{ @@ -14,26 +10,9 @@ use cid_router_core::{ routes::Route, Context, }; -use futures::{Stream, StreamExt}; -use iroh::{endpoint::SendStream, Endpoint, NodeAddr, NodeId, SecretKey}; -use iroh_blobs::{ - get::request::GetBlobItem, - protocol::{ChunkRangesSeq, PushRequest, RequestType}, - store::IROH_BLOCK_SIZE, - ticket::BlobTicket, - Hash, -}; -use irpc::util::WriteVarintExt; -use log::error; +use futures::Stream; +use iroh_blobs::Hash; use serde::{Deserialize, Serialize}; -use serde_json::Value; - -#[derive(Debug)] -pub struct IrohRemoteCrp { - node_addr: NodeAddr, - endpoint: Endpoint, - allow_put: bool, -} #[derive(Debug, Clone)] pub struct IrohCrp { @@ -46,19 +25,6 @@ pub struct IrohCrpConfig { pub path: PathBuf, } -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct IrohRemoteCrpConfig { - pub node_addr_ref: IrohNodeAddrRef, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum IrohNodeAddrRef { - NodeId(String), - NodeTicket(String), - Ticket(String), -} - impl IrohCrp { pub async fn new_from_config(config: IrohCrpConfig) -> io::Result { let path = if config.path.is_absolute() { @@ -73,67 +39,6 @@ impl IrohCrp { } } -impl IrohRemoteCrp { - pub async fn new_from_config(config: Value, secret_key: SecretKey) -> Result { - let IrohRemoteCrpConfig { node_addr_ref } = serde_json::from_value(config)?; - - let node_addr = match node_addr_ref { - IrohNodeAddrRef::NodeId(node_id) => { - let node_id = NodeId::from_str(&node_id)?; - NodeAddr::from(node_id) - } - IrohNodeAddrRef::NodeTicket(ticket) => { - let ticket = iroh_base::ticket::NodeTicket::from_str(&ticket)?; - ticket.node_addr().to_owned() - } - IrohNodeAddrRef::Ticket(ticket) => { - let ticket = BlobTicket::from_str(&ticket)?; - ticket.node_addr().clone() - } - }; - - let endpoint = Endpoint::builder() - .discovery_n0() - .secret_key(secret_key) - .bind() - .await?; - - Ok(Self { - node_addr, - endpoint, - allow_put: true, - }) - } -} - -#[async_trait] -impl Crp for IrohRemoteCrp { - fn provider_id(&self) -> String { - "iroh".to_string() - } - - fn provider_type(&self) -> ProviderType { - ProviderType::Iroh - } - - async fn reindex(&self, _cx: &Context) -> anyhow::Result<()> { - // TODO: Implement reindexing logic - Ok(()) - } - - fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { - CrpCapabilities { - route_resolver: Some(self), - size_resolver: None, // TODO - blob_writer: if self.allow_put { Some(self) } else { None }, - } - } - - fn cid_filter(&self) -> CidFilter { - CidFilter::MultihashCodeFilter(CodeFilter::Eq(0x1e)) // blake3 - } -} - #[async_trait] impl Crp for IrohCrp { fn provider_id(&self) -> String { @@ -204,126 +109,3 @@ impl RouteResolver for IrohCrp { Ok(Box::pin(stream)) } } - -#[async_trait] -impl BlobWriter for IrohRemoteCrp { - async fn put_blob( - &self, - _auth: Option, - cid: &Cid, - data: &[u8], - ) -> Result<(), Box> { - error!("Putting blob with cid: {}", cid); - error!("I am: {:?}", self.endpoint.node_id()); - error!("Target: {:?}", self.node_addr.node_id); - if !self.allow_put { - return Err("Put operations are not allowed on this CRP".into()); - } - if cid.hash().code() != 0x1e { - return Err("Unsupported CID hash code; only blake3 is supported".into()); - } - let hash: [u8; 32] = cid - .hash() - .digest() - .try_into() - .expect("blake3 hash must be 32 bytes"); - let hash = iroh_blobs::Hash::from_bytes(hash); - error!("Connecting to node..."); - let conn = self - .endpoint - .connect(self.node_addr.clone(), iroh_blobs::ALPN) - .await?; - error!("Connected. Opening stream..."); - let (mut writer, mut reader) = conn.open_bi().await?; - error!("Opened stream. Writing push request..."); - let request = PushRequest::new(hash, ChunkRangesSeq::root()); - let request = write_push_request(request, &mut writer).await?; - let (hash, bao) = create_n0_bao(data, &ChunkRanges::all())?; - if hash != request.hash { - return Err("Computed hash does not match requested hash".into()); - } - writer.write_all(&bao).await?; - writer.finish()?; - let res = reader.read_to_end(1024).await; - if let Err(e) = res { - error!("Error reading response: {}", e); - return Err(Box::new(e)); - } - error!("Blob put completed for cid: {}", cid); - Ok(()) - } -} - -/// TODO: make this available in iroh-blobs -async fn write_push_request( - request: PushRequest, - stream: &mut SendStream, -) -> anyhow::Result { - let mut request_bytes = Vec::new(); - request_bytes.push(RequestType::Push as u8); - request_bytes.write_length_prefixed(&request).unwrap(); - stream.write_all(&request_bytes).await?; - Ok(request) -} - -/// TODO: move this to iroh-blobs -pub fn create_n0_bao(data: &[u8], ranges: &ChunkRanges) -> anyhow::Result<(Hash, Vec)> { - let outboard = PreOrderMemOutboard::create(data, IROH_BLOCK_SIZE); - let mut encoded = Vec::new(); - let size = data.len() as u64; - encoded.extend_from_slice(&size.to_le_bytes()); - bao_tree::io::sync::encode_ranges_validated(data, &outboard, ranges, &mut encoded)?; - Ok((outboard.root.into(), encoded)) -} - -#[async_trait] -impl RouteResolver for IrohRemoteCrp { - async fn get_bytes( - &self, - route: &Route, - _auth: Option, - ) -> Result< - Pin< - Box< - dyn Stream>> - + Send, - >, - >, - Box, - > { - let Self { node_addr, .. } = self; - let cid = route.cid; - - let hash = cid.hash().digest(); - let hash: [u8; 32] = hash.try_into()?; - let hash = Hash::from_bytes(hash); - - let conn = self - .endpoint - .connect(node_addr.clone(), iroh_blobs::ALPN) - .await?; - - println!("get {:?} from {}", hash, node_addr.node_id.fmt_short()); - - let res = iroh_blobs::get::request::get_blob(conn, hash); - let res = res - .take_while(|item| n0_future::future::ready(!matches!(item, GetBlobItem::Done(_)))) - .filter_map(|item| { - n0_future::future::ready(match item { - GetBlobItem::Item(item) => match item { - BaoContentItem::Leaf(leaf) => Some(Ok(leaf.data)), - // TODO - I don't think this is right. returning None here - // will likely end the stream prematurely - BaoContentItem::Parent(_parent) => None, - }, - // This is filtered out, only for compiler happiness - GetBlobItem::Done(_stats) => None, - GetBlobItem::Error(err) => Some(Err( - Box::new(err) as Box - )), - }) - }); - - Ok(Box::pin(res)) - } -} diff --git a/crps/iroh/src/lib.rs b/crps/iroh/src/lib.rs index 6efc5a9..5fc44df 100644 --- a/crps/iroh/src/lib.rs +++ b/crps/iroh/src/lib.rs @@ -1,3 +1,3 @@ mod iroh; -pub use iroh::{IrohCrp, IrohCrpConfig, IrohNodeAddrRef, IrohRemoteCrp, IrohRemoteCrpConfig}; +pub use iroh::{IrohCrp, IrohCrpConfig}; diff --git a/server/Cargo.toml b/server/Cargo.toml index aa5838e..53058fe 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -12,10 +12,10 @@ api-utils = { path = "../crates/api-utils" } async-trait.workspace = true axum.workspace = true axum-extra.workspace = true -bao-tree.workspace = true bytes.workspace = true chrono.workspace = true cid.workspace = true +blake3.workspace = true cid-router-core = { path = "../core" } clap.workspace = true crp-azure = { path = "../crps/azure" } diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index 60598d4..d62080a 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -9,7 +9,6 @@ use axum::{ Json, }; use axum_extra::extract::TypedHeader; -use bao_tree::blake3; use bytes::BytesMut; use cid::Cid; use cid_router_core::{ From 0bb4b7aa7c67a82365896202178f4a6c62101601 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 5 Dec 2025 17:44:32 +0200 Subject: [PATCH 7/9] Remove size resolver --- core/src/crp.rs | 13 ------------- core/src/db.rs | 1 - crps/azure/src/container.rs | 1 - crps/iroh/src/iroh.rs | 1 - 4 files changed, 16 deletions(-) diff --git a/core/src/crp.rs b/core/src/crp.rs index 6d37af8..fe530a3 100644 --- a/core/src/crp.rs +++ b/core/src/crp.rs @@ -77,7 +77,6 @@ impl Crp for Arc { /// All capabilities a CRP may have represented as self-referential trait objects. pub struct CrpCapabilities<'a> { pub route_resolver: Option<&'a dyn RouteResolver>, - pub size_resolver: Option<&'a dyn SizeResolver>, pub blob_writer: Option<&'a dyn BlobWriter>, } @@ -100,18 +99,6 @@ pub trait RouteResolver { >; } -/// A SizeResolver can return the length in bytes of the blob a CID points at. -/// This is useful both as a preflight check before downloading a CID, -/// and as a fast means of checking if a CRP has the CID in the first place. -#[async_trait] -pub trait SizeResolver { - async fn get_size( - &self, - cid: &Cid, - auth: Vec, - ) -> Result>; -} - /// A RouteResolver can dereference a route, turning it into a stream of bytes, accepting /// authentication data. #[async_trait] diff --git a/core/src/db.rs b/core/src/db.rs index 9a7e001..2af3ceb 100644 --- a/core/src/db.rs +++ b/core/src/db.rs @@ -382,7 +382,6 @@ mod tests { fn capabilities(&self) -> CrpCapabilities<'_> { CrpCapabilities { route_resolver: None, - size_resolver: None, blob_writer: None, } } diff --git a/crps/azure/src/container.rs b/crps/azure/src/container.rs index fc8b59f..397de55 100644 --- a/crps/azure/src/container.rs +++ b/crps/azure/src/container.rs @@ -48,7 +48,6 @@ impl Crp for Container { fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { CrpCapabilities { route_resolver: Some(self), - size_resolver: None, // TODO blob_writer: None, // TODO } } diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index 6b55cfc..bec61cf 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -57,7 +57,6 @@ impl Crp for IrohCrp { fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { CrpCapabilities { route_resolver: Some(self), - size_resolver: None, // TODO blob_writer: Some(self), } } From f43c8e8337eac2cfc030fd9ea08c6e459b38fdf0 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Tue, 9 Dec 2025 12:21:28 +0200 Subject: [PATCH 8/9] Update the swagger API with the new routes --- crps/azure/src/container.rs | 2 +- crps/iroh/src/iroh.rs | 4 +- server/config.example.toml | 2 +- server/src/api.rs | 6 +- server/src/api/v1.rs | 1 + server/src/api/v1/data.rs | 193 ++++++++++++++++++++++++++++++++++++ server/src/api/v1/routes.rs | 160 +----------------------------- 7 files changed, 205 insertions(+), 163 deletions(-) create mode 100644 server/src/api/v1/data.rs diff --git a/crps/azure/src/container.rs b/crps/azure/src/container.rs index 397de55..5bf978a 100644 --- a/crps/azure/src/container.rs +++ b/crps/azure/src/container.rs @@ -48,7 +48,7 @@ impl Crp for Container { fn capabilities<'a>(&'a self) -> CrpCapabilities<'a> { CrpCapabilities { route_resolver: Some(self), - blob_writer: None, // TODO + blob_writer: None, // TODO } } diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index bec61cf..da7e555 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -12,6 +12,7 @@ use cid_router_core::{ }; use futures::Stream; use iroh_blobs::Hash; +use log::error; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone)] @@ -42,7 +43,7 @@ impl IrohCrp { #[async_trait] impl Crp for IrohCrp { fn provider_id(&self) -> String { - "iroh_inline".to_string() + "iroh".to_string() } fn provider_type(&self) -> ProviderType { @@ -99,6 +100,7 @@ impl RouteResolver for IrohCrp { >, Box, > { + error!("IrohCrp::get_bytes for route: {:?}", route); let cid = route.cid; let hash = cid.hash().digest(); let hash: [u8; 32] = hash.try_into()?; diff --git a/server/config.example.toml b/server/config.example.toml index 6f95322..75eb72d 100644 --- a/server/config.example.toml +++ b/server/config.example.toml @@ -3,4 +3,4 @@ auth = "none" [[providers]] type = "iroh" -path = "/Users/rklaehn/projects_git/cid_router/blobs" +path = "/Users/rklaehn/projects_git/cid-router/blobs" diff --git a/server/src/api.rs b/server/src/api.rs index df188c1..efe84dd 100644 --- a/server/src/api.rs +++ b/server/src/api.rs @@ -21,6 +21,8 @@ use crate::context::Context; paths( v1::routes::get_routes, v1::status::get_status, + v1::data::create_data, + v1::data::get_data, ), components( schemas( @@ -61,8 +63,8 @@ pub async fn start(ctx: Arc) -> Result<()> { ) .route("/v1/routes", get(v1::routes::list_routes)) .route("/v1/routes/{cid}", get(v1::routes::get_routes)) - .route("/v1/data", post(v1::routes::create_data)) - .route("/v1/data/{cid}", get(v1::routes::get_data)) + .route("/v1/data", post(v1::data::create_data)) + .route("/v1/data/{cid}", get(v1::data::get_data)) .with_state(ctx); let listener = TcpListener::bind(addr).await?; diff --git a/server/src/api/v1.rs b/server/src/api/v1.rs index 951945e..6fd63b4 100644 --- a/server/src/api/v1.rs +++ b/server/src/api/v1.rs @@ -1,2 +1,3 @@ +pub mod data; pub mod routes; pub mod status; diff --git a/server/src/api/v1/data.rs b/server/src/api/v1/data.rs new file mode 100644 index 0000000..e702cfb --- /dev/null +++ b/server/src/api/v1/data.rs @@ -0,0 +1,193 @@ +use std::{collections::HashSet, str::FromStr, sync::Arc}; + +use api_utils::{ApiError, ApiResult}; +use axum::{ + body::Body, + extract::{Path, State}, + http::{header, StatusCode}, + response::Response, + Json, +}; +use axum_extra::extract::TypedHeader; +use bytes::BytesMut; +use cid::Cid; +use cid_router_core::cid::blake3_hash_to_cid; +use futures::StreamExt; +use headers::{Authorization, ContentType}; +use http_body::Frame; +use http_body_util::StreamBody; +use log::info; +use serde::Serialize; +use utoipa::{IntoParams, ToSchema}; + +use crate::context::Context; + +/// Get a data stream for a CID +#[utoipa::path( + get, + path = "/v1/data/{cid}", + tag = "/v1/data/{cid}", + params( + ("authorization" = Option, Header, description = "Bearer token for authentication") + ), + responses( + (status = 200, description = "Get data for a CID") + ) +)] +pub async fn get_data( + Path(cid): Path, + auth: Option>>, + State(ctx): State>, +) -> ApiResult { + // TODO - remove unwraps + let cid = Cid::from_str(&cid).unwrap(); + let routes = ctx.core.db().routes_for_cid(cid).await.unwrap(); + let routes: Vec = routes.into_iter().collect(); + let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); + ctx.auth.service().await.authenticate(token).await?; + + for route in routes { + // iterate through providers until you find a match on provider_id and provider_type + let provider_id: String = route.provider_id.clone(); + if let Some(provider) = ctx + .providers + .iter() + .find(|p| provider_id == p.provider_id() && route.provider_type == p.provider_type()) + { + if let Some(route_resolver) = provider.capabilities().route_resolver { + let stream = route_resolver.get_bytes(&route, None).await.unwrap(); + + // Convert Stream into a response body + let body = StreamBody::new( + stream.map(|result| result.map(Frame::data).map_err(std::io::Error::other)), + ); + + return Ok(Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/octet-stream") + .body(Body::new(body)) + .unwrap()); + } + } + } + + Err(ApiError::new( + StatusCode::NOT_FOUND, + "No route found for CID", + )) +} + +#[derive(Serialize, ToSchema)] +pub struct CreateDataResponse { + pub cid: String, + pub size: usize, + pub location: String, +} + +/// Create data for a CID +#[utoipa::path( + post, + path = "/v1/data", + tag = "/v1/data", + params( + ("authorization" = Option, Header, description = "Bearer token for authentication") + ), + responses( + (status = 200, description = "Get data for a CID", body = CreateDataResponse) + ) +)] +#[axum::debug_handler] +pub async fn create_data( + auth: Option>>, + content_type: Option>, + State(ctx): State>, + body: Body, +) -> ApiResult> { + let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); + ctx.auth.service().await.authenticate(token).await?; + + // Check if content-type is supported and translate to cid type + let content_type = content_type.map(|TypedHeader(mime)| mime.to_string()); + let cid_type = match content_type.as_ref().map(|ct| ct.as_str()) { + None => cid_router_core::cid::Codec::Raw, + Some("application/x-www-form-urlencoded") => cid_router_core::cid::Codec::Raw, + Some("application/octet-stream") => cid_router_core::cid::Codec::Raw, + Some("application/vnd.ipld.dag-cbor") => cid_router_core::cid::Codec::DagCbor, + _ => { + return Err(ApiError::new( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "Unsupported content-type", + )) + } + }; + + // Read data - we assume this to be small enough to fit into memory for now + let mut buffer = BytesMut::new(); + let mut stream = body.into_data_stream(); + while let Some(chunk) = stream.next().await { + let Ok(chunk) = chunk else { + return Err(ApiError::new( + StatusCode::BAD_REQUEST, + "Failed to read request body", + )); + }; + buffer.extend_from_slice(&chunk); + } + + // compute CID + let data = buffer.freeze(); + let hash = blake3::hash(&data); + let cid = blake3_hash_to_cid(hash.into(), cid_type); + + // Find writers + let writers = ctx + .providers + .iter() + .filter(|p| p.provider_is_eligible_for_cid(&cid)) + .filter_map(|p| p.capabilities().blob_writer.map(|w| (p, w))) + .collect::>(); + if writers.is_empty() { + return Err(ApiError::new( + StatusCode::SERVICE_UNAVAILABLE, + "No eligible writers found for CID", + )); + } + + let existing = ctx.core.db().routes_for_cid(cid).await?; + let existing_ids = existing + .iter() + .map(|r| r.provider_id.clone()) + .collect::>(); + + let mut outcome = Vec::new(); + for (crp, writer) in writers { + if existing_ids.contains(&crp.provider_id()) { + info!( + "Skipping put to provider {} as route already exists", + crp.provider_id() + ); + continue; + } + let data = data.clone(); + let res = writer.put_blob(None, &cid, &data).await; + outcome.push((crp, res)); + } + + for (provider, res) in &outcome { + if res.is_ok() { + let route = cid_router_core::routes::Route::builder(*provider) + .cid(cid) + .multicodec(cid_router_core::cid::Codec::Raw) + .size(data.len() as u64) + .url(cid.to_string()) + .build(&ctx.core)?; + ctx.core.db().insert_route(&route).await?; + } + } + + Ok(Json(CreateDataResponse { + cid: cid.to_string(), + size: data.len(), + location: format!("/v1/data/{}", cid), + })) +} diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index d62080a..c2acc73 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -1,6 +1,6 @@ use std::{collections::HashSet, str::FromStr, sync::Arc}; -use api_utils::ApiResult; +use api_utils::{ApiError, ApiResult}; use axum::{ body::Body, extract::{Path, Query, State}, @@ -19,7 +19,7 @@ use futures::StreamExt; use headers::{Authorization, ContentType}; use http_body::Frame; use http_body_util::StreamBody; -use log::{error, info}; +use log::info; use serde::{Deserialize, Serialize}; use utoipa::{IntoParams, ToSchema}; @@ -129,159 +129,3 @@ pub async fn get_routes( Ok(Json(RoutesResponse { routes })) } - -/// Get a data stream for a CID -#[utoipa::path( - get, - path = "/v1/data/{cid}", - tag = "/v1/data/{cid}", - params( - ("authorization" = Option, Header, description = "Bearer token for authentication") - ), - responses( - (status = 200, description = "Get data for a CID", body = RoutesResponse) - ) -)] -pub async fn get_data( - Path(cid): Path, - auth: Option>>, - State(ctx): State>, -) -> ApiResult { - // TODO - remove unwraps - let cid = Cid::from_str(&cid).unwrap(); - let routes = ctx.core.db().routes_for_cid(cid).await.unwrap(); - let routes: Vec = routes.into_iter().collect(); - let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); - ctx.auth.service().await.authenticate(token).await?; - - for route in routes { - // iterate through providers until you find a match on provider_id and provider_type - let provider_id: String = route.provider_id.clone(); - if let Some(provider) = ctx - .providers - .iter() - .find(|p| provider_id == p.provider_id() && route.provider_type == p.provider_type()) - { - if let Some(route_resolver) = provider.capabilities().route_resolver { - let stream = route_resolver.get_bytes(&route, None).await.unwrap(); - - // Convert Stream into a response body - let body = StreamBody::new( - stream.map(|result| result.map(Frame::data).map_err(std::io::Error::other)), - ); - - return Ok(Response::builder() - .status(StatusCode::OK) - .header(header::CONTENT_TYPE, "application/octet-stream") - .body(Body::new(body)) - .unwrap()); - } - } - } - - Ok(Response::builder() - .status(StatusCode::NOT_FOUND) - .body(Body::empty()) - .unwrap()) -} - -/// Create data for a CID -#[axum::debug_handler] -pub async fn create_data( - auth: Option>>, - content_type: Option>, - State(ctx): State>, - body: Body, -) -> ApiResult { - let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); - ctx.auth.service().await.authenticate(token).await?; - - // Check if content-type is supported and translate to cid type - let content_type = content_type.map(|TypedHeader(mime)| mime.to_string()); - let cid_type = match content_type.as_ref().map(|ct| ct.as_str()) { - None => cid_router_core::cid::Codec::Raw, - Some("application/x-www-form-urlencoded") => cid_router_core::cid::Codec::Raw, - Some("application/octet-stream") => cid_router_core::cid::Codec::Raw, - Some("application/vnd.ipld.dag-cbor") => cid_router_core::cid::Codec::DagCbor, - _ => { - return Ok(Response::builder() - .status(StatusCode::UNSUPPORTED_MEDIA_TYPE) - .body(Body::empty())?); - } - }; - - // Read data - we assume this to be small enough to fit into memory for now - let mut buffer = BytesMut::new(); - let mut stream = body.into_data_stream(); - while let Some(chunk) = stream.next().await { - let Ok(chunk) = chunk else { - return Ok(Response::builder() - .status(StatusCode::BAD_REQUEST) - .body(Body::empty())?); - }; - buffer.extend_from_slice(&chunk); - } - - // compute CID - let data = buffer.freeze(); - let hash = blake3::hash(&data); - let cid = blake3_hash_to_cid(hash.into(), cid_type); - - // Find writers - let writers = ctx - .providers - .iter() - .filter(|p| p.provider_is_eligible_for_cid(&cid)) - .filter_map(|p| p.capabilities().blob_writer.map(|w| (p, w))) - .collect::>(); - if writers.is_empty() { - return Ok(Response::builder() - .status(StatusCode::SERVICE_UNAVAILABLE) - .body(Body::empty())?); - } - - let existing = ctx.core.db().routes_for_cid(cid).await?; - let existing_ids = existing - .iter() - .map(|r| r.provider_id.clone()) - .collect::>(); - - let mut outcome = Vec::new(); - for (crp, writer) in writers { - if existing_ids.contains(&crp.provider_id()) { - error!( - "Skipping put to provider {} as route already exists", - crp.provider_id() - ); - continue; - } - let data = data.clone(); - let res = writer.put_blob(None, &cid, &data).await; - outcome.push((crp, res)); - } - - for (provider, res) in &outcome { - if res.is_ok() { - let route = cid_router_core::routes::Route::builder(*provider) - .cid(cid) - .multicodec(cid_router_core::cid::Codec::Raw) - .size(data.len() as u64) - .url(cid.to_string()) - .build(&ctx.core)?; - ctx.core.db().insert_route(&route).await?; - } - } - - let json = serde_json::json!({ - "cid": cid.to_string(), - "size": data.len(), - "location": format!("/v1/data/{}", cid) - }); - - Ok(Response::builder() - .status(StatusCode::CREATED) - .header(header::LOCATION, format!("/v1/data/{}", cid)) - .header(header::CONTENT_TYPE, "application/json") - .body(Body::from(json.to_string())) - .unwrap()) -} From 70304f97aa1b962c2e54b7d1e7e48d1982445b46 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Tue, 9 Dec 2025 13:16:16 +0200 Subject: [PATCH 9/9] Handle a few unwraps Add schema to swagger Fix imports --- crps/iroh/src/iroh.rs | 4 ++-- server/src/api.rs | 1 + server/src/api/v1/data.rs | 39 +++++++++++++++++++++++++------------ server/src/api/v1/routes.rs | 14 +++---------- 4 files changed, 33 insertions(+), 25 deletions(-) diff --git a/crps/iroh/src/iroh.rs b/crps/iroh/src/iroh.rs index da7e555..67b438d 100644 --- a/crps/iroh/src/iroh.rs +++ b/crps/iroh/src/iroh.rs @@ -12,7 +12,7 @@ use cid_router_core::{ }; use futures::Stream; use iroh_blobs::Hash; -use log::error; +use log::info; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone)] @@ -100,7 +100,7 @@ impl RouteResolver for IrohCrp { >, Box, > { - error!("IrohCrp::get_bytes for route: {:?}", route); + info!("get_bytes for route: {:?}", route); let cid = route.cid; let hash = cid.hash().digest(); let hash: [u8; 32] = hash.try_into()?; diff --git a/server/src/api.rs b/server/src/api.rs index efe84dd..b0c5f5c 100644 --- a/server/src/api.rs +++ b/server/src/api.rs @@ -29,6 +29,7 @@ use crate::context::Context; v1::routes::RoutesResponse, v1::routes::Route, v1::status::StatusResponse, + v1::data::CreateDataResponse, // routes::AzureBlobStorageRouteMethod, // routes::UrlRouteMethod, // routes::IpfsRouteMethod, diff --git a/server/src/api/v1/data.rs b/server/src/api/v1/data.rs index e702cfb..c741b28 100644 --- a/server/src/api/v1/data.rs +++ b/server/src/api/v1/data.rs @@ -11,14 +11,14 @@ use axum::{ use axum_extra::extract::TypedHeader; use bytes::BytesMut; use cid::Cid; -use cid_router_core::cid::blake3_hash_to_cid; +use cid_router_core::cid::{Codec, blake3_hash_to_cid}; use futures::StreamExt; use headers::{Authorization, ContentType}; use http_body::Frame; use http_body_util::StreamBody; use log::info; use serde::Serialize; -use utoipa::{IntoParams, ToSchema}; +use utoipa::ToSchema; use crate::context::Context; @@ -31,7 +31,8 @@ use crate::context::Context; ("authorization" = Option, Header, description = "Bearer token for authentication") ), responses( - (status = 200, description = "Get data for a CID") + (status = 200, description = "Get raw data for a CID", content_type = "application/octet-stream"), + (status = 404, description = "No route found for CID") ) )] pub async fn get_data( @@ -39,9 +40,13 @@ pub async fn get_data( auth: Option>>, State(ctx): State>, ) -> ApiResult { - // TODO - remove unwraps - let cid = Cid::from_str(&cid).unwrap(); - let routes = ctx.core.db().routes_for_cid(cid).await.unwrap(); + let cid = Cid::from_str(&cid).map_err(|e| ApiError::new(StatusCode::BAD_REQUEST, e.to_string()))?; + let routes = ctx.core.db().routes_for_cid(cid).await.map_err(|e| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to fetch routes for cid {}: {}", cid, e), + ) + })?; let routes: Vec = routes.into_iter().collect(); let token = auth.map(|TypedHeader(Authorization(bearer))| bearer.token().to_string()); ctx.auth.service().await.authenticate(token).await?; @@ -55,7 +60,15 @@ pub async fn get_data( .find(|p| provider_id == p.provider_id() && route.provider_type == p.provider_type()) { if let Some(route_resolver) = provider.capabilities().route_resolver { - let stream = route_resolver.get_bytes(&route, None).await.unwrap(); + let stream = route_resolver.get_bytes(&route, None).await.map_err(|e| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!( + "Failed to get bytes for cid {} from provider {}: {}", + cid, provider_id, e + ), + ) + })?; // Convert Stream into a response body let body = StreamBody::new( @@ -93,7 +106,9 @@ pub struct CreateDataResponse { ("authorization" = Option, Header, description = "Bearer token for authentication") ), responses( - (status = 200, description = "Get data for a CID", body = CreateDataResponse) + (status = 200, description = "Get data for a CID", body = CreateDataResponse), + (status = 415, description = "Unsupported content-type"), + (status = 503, description = "No eligible writers found for CID"), ) )] #[axum::debug_handler] @@ -109,10 +124,10 @@ pub async fn create_data( // Check if content-type is supported and translate to cid type let content_type = content_type.map(|TypedHeader(mime)| mime.to_string()); let cid_type = match content_type.as_ref().map(|ct| ct.as_str()) { - None => cid_router_core::cid::Codec::Raw, - Some("application/x-www-form-urlencoded") => cid_router_core::cid::Codec::Raw, - Some("application/octet-stream") => cid_router_core::cid::Codec::Raw, - Some("application/vnd.ipld.dag-cbor") => cid_router_core::cid::Codec::DagCbor, + None => Codec::Raw, + Some("application/x-www-form-urlencoded") => Codec::Raw, + Some("application/octet-stream") => Codec::Raw, + Some("application/vnd.ipld.dag-cbor") => Codec::DagCbor, _ => { return Err(ApiError::new( StatusCode::UNSUPPORTED_MEDIA_TYPE, diff --git a/server/src/api/v1/routes.rs b/server/src/api/v1/routes.rs index c2acc73..651d5ce 100644 --- a/server/src/api/v1/routes.rs +++ b/server/src/api/v1/routes.rs @@ -1,24 +1,16 @@ -use std::{collections::HashSet, str::FromStr, sync::Arc}; +use std::{str::FromStr, sync::Arc}; -use api_utils::{ApiError, ApiResult}; +use api_utils::ApiResult; use axum::{ - body::Body, extract::{Path, Query, State}, - http::{header, StatusCode}, - response::Response, Json, }; use axum_extra::extract::TypedHeader; -use bytes::BytesMut; use cid::Cid; use cid_router_core::{ - cid::blake3_hash_to_cid, db::{Direction, OrderBy}, }; -use futures::StreamExt; -use headers::{Authorization, ContentType}; -use http_body::Frame; -use http_body_util::StreamBody; +use headers::Authorization; use log::info; use serde::{Deserialize, Serialize}; use utoipa::{IntoParams, ToSchema};