feat(node/web): expose transaction pool rpcs

Author: binier

node/common/src/service/rpc/mod.rs

@@ -3,6 +3,7 @@ pub use sender::RpcSender;
 
 pub mod state;
 pub mod stats;
+pub mod transaction_pool;
 
 use node::rpc::{
     RpcBestChainResponse, RpcBlockProducerStatsGetResponse, RpcConsensusConstantsGetResponse,

node/common/src/service/rpc/sender.rs

@@ -10,6 +10,7 @@ use node::rpc::*;
 
 use super::state::State;
 use super::stats::Stats;
+use super::transaction_pool::TransactionPool;
 use super::NodeRpcRequest;
 
 #[derive(Clone)]
@@ -76,6 +77,10 @@ impl RpcSender {
     pub fn stats(&self) -> Stats {
         Stats::new(self.clone())
     }
+
+    pub fn transaction_pool(&self) -> TransactionPool {
+        TransactionPool::new(self.clone())
+    }
 }
 
 #[cfg(target_family = "wasm")]

node/common/src/service/rpc/transaction_pool.rs

@@ -0,0 +1,102 @@
+#[cfg(target_family = "wasm")]
+use gloo_utils::format::JsValueSerdeExt;
+use mina_p2p_messages::v2;
+use node::rpc::*;
+#[cfg(target_family = "wasm")]
+use wasm_bindgen::prelude::*;
+
+use super::RpcSender;
+
+#[derive(Clone)]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+pub struct TransactionPool {
+    #[allow(unused)]
+    sender: RpcSender,
+}
+
+#[derive(Clone)]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+pub struct TransactionPoolInject {
+    #[allow(unused)]
+    sender: RpcSender,
+}
+
+impl TransactionPool {
+    pub fn new(sender: RpcSender) -> Self {
+        Self { sender }
+    }
+
+    pub fn inject(&self) -> TransactionPoolInject {
+        TransactionPoolInject {
+            sender: self.sender.clone(),
+        }
+    }
+
+    async fn _get(&self) -> Option<RpcTransactionPoolResponse> {
+        self.sender
+            .oneshot_request(RpcRequest::TransactionPoolGet)
+            .await
+    }
+}
+
+#[cfg(not(target_family = "wasm"))]
+impl TransactionPool {
+    pub async fn get(&self) -> Option<RpcTransactionPoolResponse> {
+        self._get().await
+    }
+}
+
+#[cfg(target_family = "wasm")]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+impl TransactionPool {
+    pub async fn get(&self) -> JsValue {
+        JsValue::from_serde(&self._get().await).unwrap_or_default()
+    }
+}
+
+impl TransactionPoolInject {
+    async fn _payment(
+        &self,
+        payments: Vec<RpcInjectPayment>,
+    ) -> Result<Option<RpcTransactionInjectResponse>, String> {
+        let res = self
+            .sender
+            .oneshot_request(RpcRequest::TransactionInject(
+                payments
+                    .into_iter()
+                    .map(|cmd| v2::MinaBaseUserCommandStableV2::try_from(cmd))
+                    .collect::<Result<_, _>>()
+                    .map_err(|err| err.to_string())?,
+            ))
+            .await;
+        Ok(res)
+    }
+}
+
+#[cfg(not(target_family = "wasm"))]
+impl TransactionPoolInject {
+    pub async fn payment(
+        &self,
+        payments: Vec<RpcInjectPayment>,
+    ) -> Result<Option<RpcTransactionInjectResponse>, String> {
+        self._payment(payments).await
+    }
+}
+
+#[cfg(target_family = "wasm")]
+#[cfg_attr(target_family = "wasm", wasm_bindgen)]
+impl TransactionPoolInject {
+    pub async fn payment(&self, payments: JsValue) -> Result<JsValue, JsValue> {
+        let payments: Vec<RpcInjectPayment> = if payments.is_array() {
+            payments.into_serde().map_err(|err| err.to_string())?
+        } else {
+            let payment = payments.into_serde().map_err(|err| err.to_string())?;
+            vec![payment]
+        };
+
+        self._payment(payments)
+            .await
+            .map(|res| JsValue::from_serde(&res).unwrap_or_default())
+            .map_err(Into::into)
+    }
+}

node/native/src/http_server.rs

@@ -453,14 +453,12 @@ pub async fn run(port: u16, rpc_sender: RpcSender) {
             let rpc_sender_clone = rpc_sender_clone.clone();
             async move {
                 rpc_sender_clone
-                    .oneshot_request(RpcRequest::TransactionPoolGet)
+                    .transaction_pool()
+                    .get()
                     .await
-                    .map_or_else(
-                        dropped_channel_response,
-                        |reply: node::rpc::RpcTransactionPoolResponse| {
-                            with_json_reply(&reply, StatusCode::OK)
-                        },
-                    )
+                    .map_or_else(dropped_channel_response, |reply| {
+                        with_json_reply(&reply, StatusCode::OK)
+                    })
             }
         });
 
@@ -489,19 +487,23 @@ pub async fn run(port: u16, rpc_sender: RpcSender) {
             let rpc_sender_clone = rpc_sender_clone.clone();
 
             async move {
-                rpc_sender_clone
-                    .oneshot_request(RpcRequest::TransactionInject(
-                        body.into_iter()
-                            .map(|cmd| cmd.try_into().unwrap())
-                            .collect(),
-                    ))
+                match rpc_sender_clone
+                    .transaction_pool()
+                    .inject()
+                    .payment(body)
                     .await
-                    .map_or_else(
+                {
+                    Err(err) => with_status(
+                        warp::reply::json(&serde_json::json!({"error": err})),
+                        StatusCode::INTERNAL_SERVER_ERROR,
+                    ),
+                    Ok(res) => res.map_or_else(
                         dropped_channel_response,
                         |reply: node::rpc::RpcTransactionInjectResponse| {
                             with_json_reply(&reply, StatusCode::OK)
                         },
-                    )
+                    ),
+                }
             }
         });