New Wallet command

Cargo.toml

@@ -23,8 +23,8 @@ cmds-pdg = { path = "crates/cmds-pdg", version = "0.0.0" }
 cmds-solana = { path = "crates/cmds-solana", version = "0.0.0" }
 
 # lib/
-value = { path = "lib/value", version = "0.0.1" }
-flow-lib = { path = "lib/flow-lib", version = "0.0.1" }
+value = { path = "lib/flow-value", version = "0.1.0", package = "flow-value" }
+flow-lib = { path = "lib/flow-lib", version = "0.1.0" }
 space-lib = { path = "lib/space-lib", version = "0.5.0" }
 space-macro = { path = "lib/space-macro", version = "0.2.1" }
 

crates/cmds-solana/Cargo.toml

@@ -118,4 +118,4 @@ rev = "b209022b85d8e6cbf4e37b059bfe3ce7fa11c6e1"
 
 [dev-dependencies]
 rust_decimal_macros = "1.26"
-inventory = "0.3"
+inventory = "0.3"

crates/cmds-solana/src/wallet.rs

@@ -8,12 +8,8 @@ pub struct Wallet {
 }
 
 #[derive(Deserialize)]
-#[serde(tag = "wallet_type")]
-enum FormData {
-    #[serde(rename = "HARDCODED")]
-    HardCoded { wallet_data: String },
-    #[serde(rename = "ADAPTER")]
-    Adapter { wallet_data: String },
+struct FormData {
+    public_key: String,
 }
 
 #[derive(ThisError, Debug)]
@@ -33,29 +29,11 @@ fn adapter_wallet(pubkey: Pubkey) -> Output {
 
 impl FormData {
     fn into_output(self) -> Result<Output, WalletError> {
-        match self {
-            FormData::Adapter { wallet_data } => {
-                let pubkey = wallet_data
-                    .parse::<Pubkey>()
-                    .map_err(|_| WalletError::InvalidBase58)?;
-                Ok(adapter_wallet(pubkey))
-            }
-            FormData::HardCoded { wallet_data } => {
-                let mut buf = [0u8; 64];
-                let size = bs58::decode(wallet_data.trim())
-                    .into(&mut buf)
-                    .map_err(|_| WalletError::InvalidBase58)?;
-                if size != buf.len() {
-                    return Err(WalletError::InvalidBase58);
-                }
-                let keypair = Keypair::from_bytes(&buf).expect("correct size, never fail");
-
-                Ok(Output {
-                    pubkey: keypair.pubkey(),
-                    keypair,
-                })
-            }
-        }
+        let pubkey = self
+            .public_key
+            .parse::<Pubkey>()
+            .map_err(|_| WalletError::InvalidBase58)?;
+        Ok(adapter_wallet(pubkey))
     }
 }
 
@@ -134,38 +112,12 @@ mod tests {
             targets: Vec::new(),
             targets_form: TargetsForm {
                 form_data: json!({
-                    // there is also "wallet_id", but it is not used
-                    "wallet_type": "ADAPTER",
-                    "wallet_data": PUBKEY_STR,
+                    "public_key": PUBKEY_STR,
                 }),
                 extra: Extra::default(),
                 wasm_bytes: None,
             },
         };
         assert_eq!(Wallet::new(&nd).form.unwrap().pubkey, PUBKEY);
     }
-
-    #[test]
-    fn hardcoded() {
-        const KEYPAIR: &str = "oLXLpXdGn6RjMHz3fvcPdGNUDQxXu91t7YAFbtRew3TFVPHAU1UrZJpgiHDLKDtrWZRQg6trQFFp6zEX2TQ1S3k";
-
-        let nd = NodeData {
-            r#type: flow_lib::CommandType::Native,
-            node_id: WALLET.into(),
-            sources: Vec::new(),
-            targets: Vec::new(),
-            targets_form: TargetsForm {
-                form_data: json!({
-                    // there is also "wallet_id", but it is not used
-                    "wallet_type": "HARDCODED",
-                    "wallet_data": KEYPAIR,
-                }),
-                extra: Extra::default(),
-                wasm_bytes: None,
-            },
-        };
-        let wallet = Wallet::new(&nd).form.unwrap();
-        assert_eq!(wallet.keypair.to_base58_string(), KEYPAIR);
-        assert_eq!(wallet.keypair.pubkey(), wallet.pubkey);
-    }
 }

crates/db/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 flow = { workspace = true }
 flow-lib = { workspace = true }
 value = { workspace = true }
+utils = { workspace = true }
 
 serde = "1"
 serde_json = { version = "1", features = ["raw_value"] }

crates/db/src/connection.rs

@@ -1,4 +1,4 @@
-use crate::{Error, WasmStorage};
+use crate::{Error, Wallet, WasmStorage};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use deadpool_postgres::Object as Connection;
@@ -29,6 +29,8 @@ pub struct UserConnection {
 pub trait UserConnectionTrait: Any + 'static {
     async fn clone_flow(&mut self, flow_id: FlowId) -> crate::Result<HashMap<FlowId, FlowId>>;
 
+    async fn get_wallets(&self) -> crate::Result<Vec<Wallet>>;
+
     async fn new_flow_run(
         &self,
         config: &ClientConfig,
@@ -118,6 +120,10 @@ pub trait UserConnectionTrait: Any + 'static {
 
 #[async_trait]
 impl UserConnectionTrait for UserConnection {
+    async fn get_wallets(&self) -> crate::Result<Vec<Wallet>> {
+        self.get_wallets().await
+    }
+
     async fn clone_flow(&mut self, flow_id: FlowId) -> crate::Result<HashMap<FlowId, FlowId>> {
         self.clone_flow(flow_id).await
     }

crates/db/src/connection/conn_impl.rs

@@ -1,4 +1,5 @@
 use flow_lib::config::client::NodeDataSkipWasm;
+use utils::bs58_decode;
 
 use super::*;
 
@@ -11,6 +12,37 @@ impl UserConnection {
         }
     }
 
+    pub async fn get_wallets(&self) -> crate::Result<Vec<Wallet>> {
+        let stmt = self
+            .conn
+            .prepare_cached("SELECT public_key, keypair FROM wallets WHERE user_id = $1")
+            .await
+            .map_err(Error::exec("prepare get_wallets"))?;
+        self.conn
+            .query(&stmt, &[&self.user_id])
+            .await
+            .map_err(Error::exec("get wallets"))?
+            .into_iter()
+            .map(|r| {
+                let pubkey_str = r
+                    .try_get::<_, String>(0)
+                    .map_err(Error::data("wallets.public_key"))?;
+                let pubkey =
+                    bs58_decode(&pubkey_str).map_err(Error::parsing("wallets.public_key"))?;
+
+                let keypair_str = r
+                    .try_get::<_, Option<String>>(1)
+                    .map_err(Error::data("wallets.keypair"))?;
+                let keypair = keypair_str
+                    .map(|s| utils::bs58_decode(&s))
+                    .transpose()
+                    .map_err(Error::parsing("wallets.keypair"))?;
+
+                Ok(Wallet { pubkey, keypair })
+            })
+            .collect()
+    }
+
     pub async fn clone_flow(&mut self, flow_id: FlowId) -> crate::Result<HashMap<FlowId, FlowId>> {
         let tx = self
             .conn

crates/db/src/connection/proxied_user_conn.rs

@@ -85,6 +85,10 @@
 
 #[async_trait::async_trait]
 impl UserConnectionTrait for ProxiedUserConn {
+    async fn get_wallets(&self) -> crate::Result<Vec<Wallet>> {
+        self.send::<[(); 0], _>("get_wallets", &[]).await
+    }
+
     async fn clone_flow(&mut self, flow_id: FlowId) -> crate::Result<HashMap<FlowId, FlowId>> {
         self.send("clone_flow", &(flow_id,)).await
     }
@@ -230,6 +234,10 @@
     pub async fn process_rpc(&mut self, req_json: &str) -> Result<Box<RawValue>, BoxError> {
         let req: RpcRequest<'_, &'_ RawValue> = serde_json::from_str(req_json)?;
         match req.method {
+            "get_wallets" => {
+                let res = self.get_wallets().await?;
+                Ok(serde_json::value::to_raw_value(&res)?)
+            }
             "clone_flow" => {
                 let (flow_id,) = serde_json::from_str(req.params.get())?;
                 let res = self.clone_flow(flow_id).await?;
@@ -252,51 +260,51 @@
             }
             "set_start_time" => {
                 let (id, time) = serde_json::from_str(req.params.get())?;
                 let res = self.set_start_time(&id, &time).await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "push_flow_error" => {
                 let (id, error): (_, String) = serde_json::from_str(req.params.get())?;
                 let res = self.push_flow_error(&id, &error).await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "push_flow_log" => {
                 let (id, index, time, level, module, content): (_, _, _, String, _, String) =
                     serde_json::from_str(req.params.get())?;
                 let res = self
                     .push_flow_log(&id, &index, &time, &level, &module, &content)
                     .await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "set_run_result" => {
                 let (id, time, not_run, output): (_, _, Vec<NodeId>, _) =
                     serde_json::from_str(req.params.get())?;
                 let res = self.set_run_result(&id, &time, &not_run, &output).await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "new_node_run" => {
                 let (id, node_id, times, time, input) = serde_json::from_str(req.params.get())?;
                 let res = self
                     .new_node_run(&id, &node_id, &times, &time, &input)
                     .await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "save_node_output" => {
                 let (id, node_id, times, output) = serde_json::from_str(req.params.get())?;
                 let res = self
                     .save_node_output(&id, &node_id, &times, &output)
                     .await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "push_node_error" => {
                 let (id, node_id, times, error): (_, _, _, String) =
                     serde_json::from_str(req.params.get())?;
                 let res = self.push_node_error(&id, &node_id, &times, &error).await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "set_node_finish" => {
                 let (id, node_id, times, time) = serde_json::from_str(req.params.get())?;
                 let res = self.set_node_finish(&id, &node_id, &times, &time).await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "new_signature_request" => {
@@ -309,7 +317,7 @@
             "save_signature" => {
                 let (id, signature): (_, Value) = serde_json::from_str(req.params.get())?;
                 let signature = value::from_value::<ConstBytes<64>>(signature)?.0;
                 let res = self.save_signature(&id, &signature).await?;
                 Ok(serde_json::value::to_raw_value(&res)?)
             }
             "read_item" => {