Da blob api (#487)

* add da_blob for new http api

nodes/nomos-node-api/src/http/backend/axum.rs

@@ -80,6 +80,7 @@ where
             .merge(SwaggerUi::new("/swagger-ui").url("/api-docs/openapi.json", ApiDoc::openapi()))
             .route("/da/metrics", routing::get(da_metrics))
             .route("/da/status", routing::post(da_status))
+            .route("/da/blobs", routing::post(da_blobs))
             .route("/cl/metrics", routing::get(cl_metrics::<T>))
             .route("/cl/status", routing::post(cl_status::<T>))
             .route("/carnot/info", routing::get(carnot_info::<T, S, SIZE>))
@@ -128,6 +129,21 @@ async fn da_status(
     make_request_and_return_response!(da::da_mempool_status(store, items))
 }
 
+#[utoipa::path(
+    post,
+    path = "/da/blobs",
+    responses(
+        (status = 200, description = "Get pending blobs", body = Vec<Blob>),
+        (status = 500, description = "Internal server error", body = String),
+    )
+)]
+async fn da_blobs(
+    State(store): State<Store>,
+    Json(items): Json<Vec<<Blob as blob::Blob>::Hash>>,
+) -> Response {
+    make_request_and_return_response!(da::da_blobs(store, items))
+}
+
 #[utoipa::path(
     get,
     path = "/cl/metrics",

nodes/nomos-node-api/src/http/da.rs

@@ -1,5 +1,9 @@
-use full_replication::{Blob, Certificate};
+use full_replication::{AbsoluteNumber, Attestation, Blob, Certificate, FullReplication};
 use nomos_core::da::blob;
+use nomos_da::{
+    backend::memory_cache::BlobCache, network::adapters::libp2p::Libp2pAdapter as DaLibp2pAdapter,
+    DaMsg, DataAvailabilityService,
+};
 use nomos_mempool::{
     backend::mockpool::MockPool,
     network::adapters::libp2p::Libp2pAdapter,
@@ -8,12 +12,18 @@ use nomos_mempool::{
 };
 use tokio::sync::oneshot;
 
-type DaMempoolService = MempoolService<
+pub type DaMempoolService = MempoolService<
     Libp2pAdapter<Certificate, <Blob as blob::Blob>::Hash>,
     MockPool<Certificate, <Blob as blob::Blob>::Hash>,
     CertDiscriminant,
 >;
 
+pub type DataAvailability = DataAvailabilityService<
+    FullReplication<AbsoluteNumber<Attestation, Certificate>>,
+    BlobCache<<Blob as nomos_core::da::blob::Blob>::Hash, Blob>,
+    DaLibp2pAdapter<Blob, Attestation>,
+>;
+
 pub async fn da_mempool_metrics(
     handle: &overwatch_rs::overwatch::handle::OverwatchHandle,
 ) -> Result<MempoolMetrics, super::DynError> {
@@ -45,3 +55,20 @@ pub async fn da_mempool_status(
 
     Ok(receiver.await.unwrap())
 }
+
+pub async fn da_blobs(
+    handle: &overwatch_rs::overwatch::handle::OverwatchHandle,
+    ids: Vec<<Blob as blob::Blob>::Hash>,
+) -> Result<Vec<Blob>, super::DynError> {
+    let relay = handle.relay::<DataAvailability>().connect().await?;
+    let (reply_channel, receiver) = oneshot::channel();
+    relay
+        .send(DaMsg::Get {
+            ids: Box::new(ids.into_iter()),
+            reply_channel,
+        })
+        .await
+        .map_err(|(e, _)| e)?;
+
+    Ok(receiver.await?)
+}

nomos-da/full-replication/src/lib.rs

@@ -93,6 +93,7 @@ impl CertificateStrategy for AbsoluteNumber<Attestation, Certificate> {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Eq, Hash, PartialEq)]
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 
 pub struct Blob {
     data: Bytes,