Da blob api

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

@@ -1,5 +1,6 @@
 use std::{fmt::Debug, hash::Hash, net::SocketAddr, sync::Arc};
 
+
 use axum::{
     extract::State,
     response::{IntoResponse, Response},
@@ -23,6 +24,7 @@ use crate::{
 #[derive(Clone)]
 pub struct AxumBackendSettings {
     pub addr: SocketAddr,
+    pub da_mempool: OverwatchHandle,
     pub da: OverwatchHandle,
     pub cl: OverwatchHandle,
     pub carnot: OverwatchHandle,
@@ -88,6 +90,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>))
@@ -109,7 +112,7 @@ where
     )
 )]
 async fn da_metrics(State(store): State<Store>) -> Response {
-    match da::da_mempool_metrics(&store.da).await {
+    match da::da_mempool_metrics(&store.da_mempool).await {
         Ok(metrics) => (StatusCode::OK, Json(metrics)).into_response(),
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
@@ -127,7 +130,25 @@ async fn da_status(
     State(store): State<Store>,
     Json(items): Json<Vec<<Blob as blob::Blob>::Hash>>,
 ) -> Response {
-    match da::da_mempool_status(&store.da, items).await {
+    match da::da_mempool_status(&store.da_mempool, items).await {
+        Ok(status) => (StatusCode::OK, Json(status)).into_response(),
+        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    }
+}
+
+#[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 {
+    match da::da_blob(&store.da, items).await {
         Ok(status) => (StatusCode::OK, Json(status)).into_response(),
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }

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,13 +12,19 @@
 };
 use tokio::sync::oneshot;
 
 pub(crate) 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(crate) async fn da_mempool_metrics(
     handle: &overwatch_rs::overwatch::handle::OverwatchHandle,
 ) -> Result<MempoolMetrics, super::DynError> {
     let relay = handle.relay::<DaMempoolService>().connect().await?;
@@ -29,7 +39,7 @@
     Ok(receiver.await.unwrap())
 }
 
 pub(crate) async fn da_mempool_status(
     handle: &overwatch_rs::overwatch::handle::OverwatchHandle,
     items: Vec<<Blob as blob::Blob>::Hash>,
 ) -> Result<Vec<Status>, super::DynError> {
@@ -45,3 +55,20 @@
 
     Ok(receiver.await.unwrap())
 }
+
+pub(crate) async fn da_blob(
+    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,