feat: adding certificate delivery subscriptions

Signed-off-by: Simon Paitrault <simon.paitrault@gmail.com>

Cargo.toml

@@ -58,8 +58,8 @@ tonic = { version = "0.9", default-features = false }
 
 # Axum server (GraphQL + Metrics)
 axum = "0.6"
-async-graphql = "5"
-async-graphql-axum = "5"
+async-graphql = "6"
+async-graphql-axum = "6"
 http = "0.2.9"
 tower-http = { version = "0.4", features = ["cors"] }
 

crates/topos-api/src/graphql/certificate.rs

@@ -15,38 +15,34 @@ pub struct Certificate {
     pub prev_id: String,
     pub proof: String,
     pub signature: String,
-    pub source_subnet_id: String,
+    pub source_subnet_id: SubnetId,
     pub state_root: String,
     pub target_subnets: Vec<SubnetId>,
     pub tx_root_hash: String,
     pub receipts_root_hash: String,
     pub verifier: u32,
 }
 
-impl From<topos_uci::Certificate> for Certificate {
-    fn from(uci_cert: topos_uci::Certificate) -> Self {
+impl From<&topos_uci::Certificate> for Certificate {
+    fn from(uci_cert: &topos_uci::Certificate) -> Self {
         Self {
             id: uci_cert.id.to_string(),
             prev_id: uci_cert.prev_id.to_string(),
             proof: hex::encode(&uci_cert.proof),
             signature: hex::encode(&uci_cert.signature),
-            source_subnet_id: uci_cert.source_subnet_id.to_string(),
+            source_subnet_id: SubnetId::from(&uci_cert.source_subnet_id),
             state_root: hex::encode(uci_cert.state_root),
-            target_subnets: uci_cert
-                .target_subnets
-                .into_iter()
-                .map(SubnetId::from)
-                .collect(),
+            target_subnets: uci_cert.target_subnets.iter().map(SubnetId::from).collect(),
             tx_root_hash: hex::encode(uci_cert.tx_root_hash),
             receipts_root_hash: format!("0x{}", hex::encode(uci_cert.receipts_root_hash)),
             verifier: uci_cert.verifier,
         }
     }
 }
 
-impl From<topos_uci::SubnetId> for SubnetId {
-    fn from(uci_id: topos_uci::SubnetId) -> Self {
-        SubnetId {
+impl From<&topos_uci::SubnetId> for SubnetId {
+    fn from(uci_id: &topos_uci::SubnetId) -> Self {
+        Self {
             value: uci_id.to_string(),
         }
     }

crates/topos-api/src/graphql/filter.rs

@@ -0,0 +1,13 @@
+use crate::graphql::subnet::SubnetId;
+
+#[derive(Debug, serde::Serialize, serde::Deserialize, async_graphql::OneofObject)]
+pub enum SubnetFilter {
+    Source(SubnetId),
+    Target(SubnetId),
+}
+
+impl PartialEq<topos_uci::SubnetId> for SubnetId {
+    fn eq(&self, other: &topos_uci::SubnetId) -> bool {
+        other.as_array().eq(&self.value.as_bytes())
+    }
+}

crates/topos-api/src/graphql/mod.rs

@@ -1,5 +1,6 @@
 pub mod certificate;
 pub mod checkpoint;
 pub mod errors;
+pub mod filter;
 pub mod query;
 pub mod subnet;

crates/topos-api/src/graphql/subnet.rs

@@ -4,7 +4,9 @@ use std::str::FromStr;
 
 use super::errors::GraphQLServerError;
 
-#[derive(Clone, Debug, Default, Serialize, Deserialize, SimpleObject, InputObject)]
+#[derive(
+    Clone, Debug, Default, Serialize, Deserialize, SimpleObject, InputObject, PartialEq, Eq,
+)]
 #[graphql(input_name = "SubnetIdInput")]
 pub struct SubnetId {
     pub value: String,

crates/topos-tce-api/src/graphql/builder.rs

@@ -1,6 +1,6 @@
 use std::{net::SocketAddr, sync::Arc};
 
-use async_graphql::{EmptyMutation, EmptySubscription, Schema};
+use async_graphql::{EmptyMutation, Schema};
 use async_graphql_axum::GraphQLSubscription;
 use axum::{extract::Extension, routing::get, Router, Server};
 use http::{header, Method};
@@ -11,12 +11,14 @@ use tower_http::cors::{Any, CorsLayer};
 use crate::{
     graphql::{
         query::{QueryRoot, ServiceSchema},
-        routes::{graphql_handler, graphql_playground, health},
+        routes::{graphql_playground, health},
     },
     runtime::InternalRuntimeCommand,
 };
 use topos_tce_storage::fullnode::FullNodeStore;
 
+use super::query::SubscriptionRoot;
+
 #[derive(Default)]
 pub struct ServerBuilder {
     store: Option<Arc<FullNodeStore>>,
@@ -61,12 +63,22 @@ impl ServerBuilder {
             .take()
             .expect("Cannot build GraphQL server without a FullNode store");
 
-        let schema: ServiceSchema = Schema::build(QueryRoot, EmptyMutation, EmptySubscription)
+        let runtime = self
+            .runtime
+            .take()
+            .expect("Cannot build GraphQL server without the internal runtime channel");
+
+        let schema: ServiceSchema = Schema::build(QueryRoot, EmptyMutation, SubscriptionRoot)
             .data(store)
+            .data(runtime)
             .finish();
 
         let app = Router::new()
-            .route("/", get(graphql_playground).post(graphql_handler))
+            .route(
+                "/",
+                get(graphql_playground)
+                    .post_service(async_graphql_axum::GraphQL::new(schema.clone())),
+            )
             .route_service("/ws", GraphQLSubscription::new(schema.clone()))
             .route("/health", get(health))
             .layer(cors)

crates/topos-tce-api/src/graphql/filter.rs

@@ -0,0 +1,4 @@
+pub(crate) enum FilterIs {
+    Source,
+    Target,
+}

crates/topos-tce-api/src/graphql/mod.rs

@@ -1,4 +1,5 @@
 pub mod builder;
+mod filter;
 mod query;
 mod routes;
 #[cfg(test)]