feat: Implement u5c submit interface

balius-runtime/src/lib.rs

@@ -3,6 +3,7 @@ use logging::LoggerHost;
 use router::Router;
 use sign::SignerHost;
 use std::{collections::HashMap, io::Read, path::Path, sync::Arc};
+use submit::SubmitHost;
 use thiserror::Error;
 use tokio::sync::Mutex;
 use tracing::{debug, info, warn};
@@ -60,6 +61,9 @@ pub enum Error {
     #[error("ledger error: {0}")]
     Ledger(String),
 
+    #[error("submit error: {0}")]
+    Submit(String),
+
     #[error("config error: {0}")]
     Config(String),
 
@@ -290,7 +294,7 @@ struct WorkerState {
     pub logging: Option<logging::LoggerHost>,
     pub kv: Option<kv::KvHost>,
     pub sign: Option<sign::SignerHost>,
-    pub submit: Option<submit::Submit>,
+    pub submit: Option<submit::SubmitHost>,
     pub http: Option<http::Http>,
 }
 
@@ -528,7 +532,10 @@ impl Runtime {
                     .as_ref()
                     .map(|kv| KvHost::new(id, kv, &self.metrics)),
                 sign: self.sign.as_ref().map(|s| SignerHost::new(id, s)),
-                submit: self.submit.clone(),
+                submit: self
+                    .submit
+                    .as_ref()
+                    .map(|s| SubmitHost::new(id, s, &self.metrics)),
                 http: self.http.clone(),
             },
         );

balius-runtime/src/metrics.rs

@@ -13,6 +13,7 @@ pub struct Metrics {
     tx_handled: Counter<u64>,
     undo_utxo_handled: Counter<u64>,
     undo_tx_handled: Counter<u64>,
+    submit_tx: Counter<u64>,
 }
 
 impl Metrics {
@@ -64,6 +65,11 @@ impl Metrics {
             .with_description("Amount of undo Tx event handled per worker.")
             .build();
 
+        let submit_tx = meter
+            .u64_counter("submit_tx")
+            .with_description("Amount of submit_tx calls per worker.")
+            .build();
+
         Metrics {
             requests,
             kv_get,
@@ -74,6 +80,7 @@ impl Metrics {
             tx_handled,
             undo_utxo_handled,
             undo_tx_handled,
+            submit_tx,
         }
     }
 
@@ -132,6 +139,11 @@ impl Metrics {
         self.undo_tx_handled
             .add(1, &[KeyValue::new("worker", worker_id.to_owned())]);
     }
+
+    pub fn submit_tx(&self, worker_id: &str) {
+        self.submit_tx
+            .add(1, &[KeyValue::new("worker", worker_id.to_owned())]);
+    }
 }
 
 impl Default for Metrics {

balius-runtime/src/submit/mod.rs

@@ -1,15 +1,48 @@
-use crate::wit::balius::app::submit as wit;
+use std::sync::Arc;
+
+use tokio::sync::Mutex;
+
+use crate::{metrics::Metrics, wit::balius::app::submit as wit};
+
+pub mod u5c;
 
 #[derive(Clone)]
+#[allow(clippy::large_enum_variant)]
 pub enum Submit {
     Mock,
+    U5C(u5c::Submit),
+    Custom(Arc<Mutex<dyn wit::Host + Send + Sync>>),
+}
+
+pub struct SubmitHost {
+    worker_id: String,
+    submit: Submit,
+    metrics: Arc<Metrics>,
+}
+impl SubmitHost {
+    pub fn new(worker_id: &str, submit: &Submit, metrics: &Arc<Metrics>) -> Self {
+        Self {
+            worker_id: worker_id.to_string(),
+            submit: submit.clone(),
+            metrics: metrics.clone(),
+        }
+    }
 }
 
 #[async_trait::async_trait]
-impl wit::Host for Submit {
+impl wit::Host for SubmitHost {
     async fn submit_tx(&mut self, tx: wit::Cbor) -> Result<(), wit::SubmitError> {
-        println!("{}", hex::encode(tx));
-
-        Ok(())
+        self.metrics.submit_tx(&self.worker_id);
+        match &mut self.submit {
+            Submit::Mock => {
+                println!("{}", hex::encode(tx));
+                Ok(())
+            }
+            Submit::U5C(x) => x.submit_tx(tx).await,
+            Submit::Custom(x) => {
+                let mut lock = x.lock().await;
+                lock.submit_tx(tx).await
+            }
+        }
     }
 }

balius-runtime/src/submit/u5c.rs

@@ -0,0 +1,49 @@
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::wit::balius::app::submit as wit;
+
+#[derive(Clone, Serialize, Deserialize, Debug)]
+pub struct Config {
+    pub endpoint_url: String,
+    pub headers: Option<HashMap<String, String>>,
+}
+
+#[derive(Clone)]
+pub struct Submit {
+    client: utxorpc::CardanoSubmitClient,
+}
+
+impl Submit {
+    pub async fn new(config: &Config) -> Result<Self, crate::Error> {
+        let mut builder = utxorpc::ClientBuilder::new().uri(&config.endpoint_url)?;
+        if let Some(headers) = &config.headers {
+            for (k, v) in headers.iter() {
+                builder = builder.metadata(k, v)?;
+            }
+        }
+
+        Ok(Self {
+            client: builder.build().await,
+        })
+    }
+
+    pub async fn submit_tx(&mut self, tx: wit::Cbor) -> Result<(), wit::SubmitError> {
+        self.client
+            .submit_tx(vec![tx])
+            .await
+            .map_err(|err| match err {
+                utxorpc::Error::GrpcError(status) => {
+                    let code: i32 = status.code().into();
+                    if code == 3 {
+                        wit::SubmitError::Invalid(status.to_string())
+                    } else {
+                        wit::SubmitError::Internal(status.to_string())
+                    }
+                }
+                utxorpc::Error::TransportError(err) => wit::SubmitError::Internal(err.to_string()),
+            })?;
+        Ok(())
+    }
+}

balius-sdk/src/qol.rs

@@ -27,6 +27,8 @@ pub enum Error {
     Ledger(wit::balius::app::ledger::LedgerError),
     #[error("sign error: {0}")]
     Sign(wit::balius::app::sign::SignError),
+    #[error("submit error: {0}")]
+    Submit(wit::balius::app::submit::SubmitError),
     #[error("http error: {0}")]
     Http(wit::balius::app::http::ErrorCode),
 }
@@ -74,6 +76,10 @@ impl From<Error> for wit::HandleError {
                 code: 9,
                 message: format!("event mismatch, expected {}", x),
             },
+            Error::Submit(err) => wit::HandleError {
+                code: 10,
+                message: err.to_string(),
+            },
         }
     }
 }
@@ -102,6 +108,12 @@ impl From<wit::balius::app::sign::SignError> for Error {
     }
 }
 
+impl From<wit::balius::app::submit::SubmitError> for Error {
+    fn from(error: wit::balius::app::submit::SubmitError) -> Self {
+        Error::Submit(error)
+    }
+}
+
 impl From<wit::balius::app::http::ErrorCode> for Error {
     fn from(error: wit::balius::app::http::ErrorCode) -> Self {
         Error::Http(error)

baliusd/src/main.rs

@@ -68,6 +68,13 @@ pub enum SignerConfig {
     Memory,
 }
 
+#[derive(Deserialize, Serialize, Clone, Debug)]
+#[serde(tag = "type")]
+#[serde(rename_all = "lowercase")]
+pub enum SubmitConfig {
+    U5c(balius_runtime::submit::u5c::Config),
+}
+
 #[derive(Deserialize, Serialize, Clone, Debug)]
 pub struct Config {
     pub rpc: drivers::jsonrpc::Config,
@@ -79,6 +86,7 @@ pub struct Config {
     pub logger: Option<LoggerConfig>,
     pub metrics: Option<MetricsConfig>,
     pub sign: Option<SignerConfig>,
+    pub submit: Option<SubmitConfig>,
     pub store: Option<StoreConfig>,
 }
 
@@ -104,13 +112,27 @@ impl From<&Config> for balius_runtime::logging::Logger {
         }
     }
 }
+
 impl From<&Config> for balius_runtime::sign::Signer {
     fn from(_value: &Config) -> Self {
         // Only one option for now
         balius_runtime::sign::Signer::InMemory(balius_runtime::sign::in_memory::Signer::default())
     }
 }
 
+impl Config {
+    pub async fn into_submit(&self) -> balius_runtime::submit::Submit {
+        match &self.submit {
+            Some(SubmitConfig::U5c(cfg)) => balius_runtime::submit::Submit::U5C(
+                balius_runtime::submit::u5c::Submit::new(cfg)
+                    .await
+                    .expect("Failed to convert config into submit interface"),
+            ),
+            None => balius_runtime::submit::Submit::Mock,
+        }
+    }
+}
+
 fn load_worker_config(config_path: Option<PathBuf>) -> miette::Result<serde_json::Value> {
     match config_path {
         Some(path) => {
@@ -155,6 +177,7 @@ async fn main() -> miette::Result<()> {
         .with_kv((&config).into())
         .with_logger((&config).into())
         .with_signer((&config).into())
+        .with_submit(config.into_submit().await)
         .build()
         .into_diagnostic()
         .context("setting up runtime")?;

examples/comprehensive/src/lib.rs

@@ -157,6 +157,19 @@ fn signer_sign_payload(
     }))
 }
 
+#[serde_as]
+#[derive(Serialize, Deserialize)]
+struct SubmitTxParams {
+    cbor: String,
+}
+
+fn submit_tx(_: Config<MyConfig>, request: Params<SubmitTxParams>) -> WorkerResult<()> {
+    let cbor = hex::decode(&request.cbor).map_err(|_| Error::BadParams)?;
+    balius_sdk::wit::balius::app::submit::submit_tx(&cbor)?;
+
+    Ok(())
+}
+
 #[derive(Serialize, Deserialize, Clone)]
 struct Datum {}
 
@@ -201,6 +214,7 @@ fn main() -> balius_sdk::Worker {
             FnHandler::from(signer_get_public_key),
         )
         .with_request_handler("signer-sign-payload", FnHandler::from(signer_sign_payload))
+        .with_request_handler("submit-tx", FnHandler::from(submit_tx))
         .with_signer("alice", "ed25519")
         .with_signer("bob", "ed25519")
 }

wit/balius.wit

@@ -105,7 +105,10 @@ interface sign {
 
 interface submit {
     type cbor = list<u8>;
-    type submit-error = u32;
+    variant submit-error {
+      internal(string),
+      invalid(string)
+    }
 
     submit-tx: func(tx: cbor) -> result<_, submit-error>;
 }