Feat/675 on chain ouis devaddrs

Cargo.toml

@@ -23,3 +23,4 @@ helium-proto = { git = "https://github.com/helium/proto", branch = "master", fea
     "services",
 ] }
 clap = { version = "4", features = ["derive"] }
+tokio = { version = "1.0", features = ["full"] }

helium-lib/Cargo.toml

@@ -48,4 +48,4 @@ clap = {workspace = true, optional = true}
 helium-mnemonic = { path = "../helium-mnemonic", optional = true }
 
 [dev-dependencies]
-rand = "0.8"
+rand = "0.8"

helium-lib/src/client.rs

@@ -4,14 +4,32 @@ use crate::{
     error::{DecodeError, Error},
     is_zero,
     keypair::{self, Pubkey},
-    solana_client,
+    solana_client::{
+        self,
+        nonblocking::tpu_client::TpuClient,
+        send_and_confirm_transactions_in_parallel::{
+            send_and_confirm_transactions_in_parallel, SendAndConfirmConfig,
+        },
+        tpu_client::TpuClientConfig,
+    },
+    solana_sdk::{
+        commitment_config::CommitmentConfig, instruction::Instruction, message::Message,
+        signature::Keypair, signer::Signer,
+    },
+    solana_transaction_utils::{
+        pack::pack_instructions_into_transactions, priority_fee::auto_compute_limit_and_price,
+    },
 };
+
 use futures::{stream, StreamExt, TryStreamExt};
 use itertools::Itertools;
 use jsonrpc_client::{JsonRpcError, SendRequest};
-use std::{marker::Send, sync::Arc};
+use solana_sdk::signer::EncodableKey;
+use std::{marker::Send, path::PathBuf, sync::Arc};
 use tracing::instrument;
 
+pub use solana_client::nonblocking::rpc_client::RpcClient as SolanaRpcClient;
+
 pub static ONBOARDING_URL_MAINNET: &str = "https://onboarding.dewi.org/api/v3";
 pub static ONBOARDING_URL_DEVNET: &str = "https://onboarding.web.test-helium.com/api/v3";
 
@@ -23,17 +41,200 @@ pub static SOLANA_URL_DEVNET: &str = "https://solana-rpc.web.test-helium.com?ses
 pub static SOLANA_URL_MAINNET_ENV: &str = "SOLANA_MAINNET_URL";
 pub static SOLANA_URL_DEVNET_ENV: &str = "SOLANA_DEVNET_URL";
 
-pub use solana_client::nonblocking::rpc_client::RpcClient as SolanaRpcClient;
+static USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
+
+#[derive(Clone)]
+pub struct SolanaClient {
+    pub inner: Arc<SolanaRpcClient>,
+    pub base_url: String,
+    pub wallet: Option<Arc<Keypair>>,
+}
+
+impl Default for SolanaClient {
+    fn default() -> Self {
+        Self::new(SOLANA_URL_MAINNET, None).unwrap()
+    }
+}
+
+impl SolanaClient {
+    pub fn new(url: &str, wallet_path: Option<PathBuf>) -> Result<Self, Error> {
+        let client = Arc::new(
+            solana_client::nonblocking::rpc_client::RpcClient::new_with_commitment(
+                url.to_string(),
+                CommitmentConfig::finalized(),
+            ),
+        );
+
+        let wallet = if let Some(path) = wallet_path {
+            Some(Arc::new(
+                Keypair::read_from_file(&path).map_err(Error::from)?,
+            ))
+        } else {
+            None
+        };
+
+        Ok(Self {
+            inner: client,
+            base_url: url.to_string(),
+            wallet,
+        })
+    }
+
+    pub fn solana_rpc_client(&self) -> Arc<SolanaRpcClient> {
+        self.inner.clone()
+    }
+
+    pub fn ws_url(&self) -> String {
+        self.base_url
+            .replace("https", "wss")
+            .replace("http", "ws")
+            .replace("127.0.0.1:8899", "127.0.0.1:8900")
+    }
+
+    pub fn wallet(&self) -> Result<Pubkey, Error> {
+        self.wallet
+            .as_ref()
+            .map(|wallet| wallet.pubkey())
+            .ok_or_else(|| Error::Other("Wallet not configured".to_string()))
+    }
+
+    pub async fn send_instructions(
+        &self,
+        ixs: Vec<Instruction>,
+        extra_signers: &[Keypair],
+        sequentially: bool,
+    ) -> Result<(), Error> {
+        let wallet = self
+            .wallet
+            .as_ref()
+            .ok_or_else(|| Error::Other("Wallet not configured".to_string()))?;
+
+        let (blockhash, _) = self
+            .inner
+            .as_ref()
+            .get_latest_blockhash_with_commitment(CommitmentConfig::finalized())
+            .await
+            .expect("Failed to get latest blockhash");
+
+        let txs = pack_instructions_into_transactions(vec![ixs], &wallet);
+        let mut with_auto_compute: Vec<Message> = Vec::new();
+        let keys: Vec<&dyn Signer> = std::iter::once(&wallet as &dyn Signer)
+            .chain(extra_signers.iter().map(|k| k as &dyn Signer))
+            .collect();
+
+        for (tx, _) in &txs {
+            // This is just a tx with compute ixs. Skip it
+            if tx.len() == 2 {
+                continue;
+            }
+
+            let computed = auto_compute_limit_and_price(
+                &self.solana_rpc_client(),
+                tx.clone(),
+                &keys,
+                1.2,
+                Some(&wallet.pubkey()),
+                Some(blockhash),
+            )
+            .await
+            .unwrap();
+
+            with_auto_compute.push(Message::new(&computed, Some(&wallet.pubkey())));
+        }
+
+        if with_auto_compute.is_empty() {
+            return Ok(());
+        }
+
+        let results;
+        let tpu_client = TpuClient::new(
+            "helium-lib",
+            self.solana_rpc_client(),
+            &self.ws_url(),
+            TpuClientConfig::default(),
+        )
+        .await?;
+
+        match sequentially {
+            true => {
+                results = tpu_client
+                    .send_and_confirm_messages_with_spinner(&with_auto_compute, &keys)
+                    .await?;
+            }
+            false => {
+                results = send_and_confirm_transactions_in_parallel(
+                    self.solana_rpc_client(),
+                    Some(tpu_client),
+                    &with_auto_compute,
+                    &keys,
+                    SendAndConfirmConfig {
+                        with_spinner: true,
+                        resign_txs_count: Some(5),
+                    },
+                )
+                .await?;
+            }
+        }
+
+        if let Some(err) = results.into_iter().flatten().next() {
+            return Err(Error::from(err));
+        }
+
+        Ok(())
+    }
+}
 
 #[derive(Clone)]
 pub struct Client {
-    pub solana_client: Arc<SolanaRpcClient>,
+    pub solana_client: Arc<SolanaClient>,
     pub das_client: Arc<DasClient>,
 }
 
+impl TryFrom<&str> for Client {
+    type Error = Error;
+    fn try_from(value: &str) -> Result<Self, Self::Error> {
+        fn env_or(key: &str, default: &str) -> String {
+            std::env::var(key).unwrap_or_else(|_| default.to_string())
+        }
+        let url = match value {
+            "m" | "mainnet-beta" => &env_or(SOLANA_URL_MAINNET_ENV, SOLANA_URL_MAINNET),
+            "d" | "devnet" => &env_or(SOLANA_URL_DEVNET_ENV, SOLANA_URL_DEVNET),
+            url => url,
+        };
+        let das_client = Arc::new(DasClient::with_base_url(url)?);
+        let solana_client = Arc::new(SolanaClient::new(url, None)?);
+
+        Ok(Self {
+            solana_client,
+            das_client,
+        })
+    }
+}
+
+impl AsRef<SolanaRpcClient> for SolanaClient {
+    fn as_ref(&self) -> &SolanaRpcClient {
+        &self.inner
+    }
+}
+
+impl AsRef<SolanaRpcClient> for Client {
+    fn as_ref(&self) -> &SolanaRpcClient {
+        &self.solana_client.inner
+    }
+}
+
+impl AsRef<DasClient> for Client {
+    fn as_ref(&self) -> &DasClient {
+        &self.das_client
+    }
+}
+
 #[async_trait::async_trait]
 pub trait GetAnchorAccount {
-    async fn anchor_account<T: AccountDeserialize>(&self, pubkey: &Pubkey) -> Result<T, Error>;
+    async fn anchor_account<T: AccountDeserialize>(
+        &self,
+        pubkey: &Pubkey,
+    ) -> Result<Option<T>, Error>;
     async fn anchor_accounts<T: AccountDeserialize + Send>(
         &self,
         pubkeys: &[Pubkey],
@@ -42,18 +243,21 @@ pub trait GetAnchorAccount {
 
 #[async_trait::async_trait]
 impl GetAnchorAccount for SolanaRpcClient {
-    async fn anchor_account<T: AccountDeserialize>(&self, pubkey: &Pubkey) -> Result<T, Error> {
+    async fn anchor_account<T: AccountDeserialize>(
+        &self,
+        pubkey: &Pubkey,
+    ) -> Result<Option<T>, Error> {
         let account = self.get_account(pubkey).await?;
         let decoded = T::try_deserialize(&mut account.data.as_ref())?;
-        Ok(decoded)
+        Ok(Some(decoded))
     }
 
     async fn anchor_accounts<T: AccountDeserialize + Send>(
         &self,
         pubkeys: &[Pubkey],
     ) -> Result<Vec<Option<T>>, Error> {
         async fn get_accounts<A: AccountDeserialize + Send>(
-            client: &SolanaRpcClient,
+            client: &solana_client::nonblocking::rpc_client::RpcClient,
             pubkeys: &[Pubkey],
         ) -> Result<Vec<Option<A>>, Error> {
             let accounts = client.get_multiple_accounts(pubkeys).await?;
@@ -86,46 +290,14 @@ impl GetAnchorAccount for Client {
     async fn anchor_account<T: AccountDeserialize>(
         &self,
         pubkey: &keypair::Pubkey,
-    ) -> Result<T, Error> {
-        self.solana_client.anchor_account(pubkey).await
+    ) -> Result<Option<T>, Error> {
+        self.solana_client.inner.anchor_account(pubkey).await
     }
     async fn anchor_accounts<T: AccountDeserialize + Send>(
         &self,
         pubkeys: &[Pubkey],
     ) -> Result<Vec<Option<T>>, Error> {
-        self.solana_client.anchor_accounts(pubkeys).await
-    }
-}
-
-impl TryFrom<&str> for Client {
-    type Error = Error;
-    fn try_from(value: &str) -> Result<Self, Self::Error> {
-        fn env_or(key: &str, default: &str) -> String {
-            std::env::var(key).unwrap_or_else(|_| default.to_string())
-        }
-        let url = match value {
-            "m" | "mainnet-beta" => &env_or(SOLANA_URL_MAINNET_ENV, SOLANA_URL_MAINNET),
-            "d" | "devnet" => &env_or(SOLANA_URL_DEVNET_ENV, SOLANA_URL_DEVNET),
-            url => url,
-        };
-        let das_client = Arc::new(DasClient::with_base_url(url)?);
-        let solana_client = Arc::new(SolanaRpcClient::new(url.to_string()));
-        Ok(Self {
-            solana_client,
-            das_client,
-        })
-    }
-}
-
-impl AsRef<SolanaRpcClient> for Client {
-    fn as_ref(&self) -> &SolanaRpcClient {
-        &self.solana_client
-    }
-}
-
-impl AsRef<DasClient> for Client {
-    fn as_ref(&self) -> &DasClient {
-        &self.das_client
+        self.solana_client.inner.anchor_accounts(pubkeys).await
     }
 }
 
@@ -199,8 +371,6 @@ impl DasClientError {
 #[jsonrpc_client::api]
 pub trait DAS {}
 
-static USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
-
 #[jsonrpc_client::implement(DAS)]
 #[derive(Debug, Clone)]
 pub struct DasClient {
@@ -210,7 +380,6 @@ pub struct DasClient {
 
 impl Default for DasClient {
     fn default() -> Self {
-        // safe to unwrap
         Self::with_base_url(SOLANA_URL_MAINNET).unwrap()
     }
 }
@@ -369,12 +538,12 @@ pub mod config {
     }
 
     #[derive(Clone)]
-    pub enum Client {
+    pub enum ClientType {
         Iot(iot::Client),
         Mobile(mobile::Client),
     }
 
-    impl Client {
+    impl ClientType {
         pub fn for_subdao(
             subdao: SubDao,
             config: &str,