use std::str::FromStr; use std::time::Duration; use async_executor::Executor; use async_native_tls::TlsConnector; use async_std::sync::{Arc, Mutex}; use async_trait::async_trait; use futures::{SinkExt, StreamExt}; use log::{debug, error, info, warn}; use serde::Serialize; use serde_json::{json, Value}; use solana_client::{blockhash_query::BlockhashQuery, rpc_client::RpcClient}; use solana_sdk::{ native_token::lamports_to_sol, program_pack::Pack, pubkey::Pubkey, signature::{Signature, Signer}, signer::keypair::Keypair, system_instruction, transaction::Transaction, }; use spl_associated_token_account::{create_associated_token_account, get_associated_token_address}; use tungstenite::Message; use super::bridge::{NetworkClient, TokenNotification, TokenSubscribtion}; use crate::rpc::{jsonrpc, jsonrpc::JsonResult, websockets, websockets::WsStream}; use crate::serial::{deserialize, serialize, Decodable, Encodable}; use crate::util::{generate_id, parse::truncate, NetworkName}; use crate::{types::*, Error, Result}; pub const SOL_NATIVE_TOKEN_ID: &str = "So11111111111111111111111111111111111111112"; #[derive(Serialize)] struct SubscribeParams { encoding: Value, commitment: Value, } pub struct SolClient { main_keypair: Keypair, // Subscriptions vector of pubkey subscriptions: Arc>>, notify_channel: ( async_channel::Sender, async_channel::Receiver, ), rpc_server: &'static str, wss_server: &'static str, } impl SolClient { pub async fn new(main_keypair: Keypair, network: &str) -> Result> { let notify_channel = async_channel::unbounded(); info!(target: "SOL BRIDGE", "Main SOL wallet pubkey: {:?}", &main_keypair.pubkey()); let (rpc_server, wss_server) = match network { "mainnet" => ( "https://api.mainnet-beta.solana.com", "wss://api.devnet.solana.com", ), "devnet" => ( "https://api.devnet.solana.com", "wss://api.devnet.solana.com", ), "testnet" => ( "https://api.testnet.solana.com", "wss://api.testnet.solana.com", ), "localhost" => ("http://localhost:8899", "ws://localhost:8900"), _ => return Err(Error::NotSupportedNetwork), }; Ok(Arc::new(Self { main_keypair, subscriptions: Arc::new(Mutex::new(Vec::new())), notify_channel, rpc_server, wss_server, })) } fn check_main_account_balance(&self, rpc: &RpcClient) -> SolResult { let main_sol_balance = rpc .get_balance(&self.main_keypair.pubkey()) .map_err(SolFailed::from)?; let fees = rpc.get_fees()?; let lamports_per_signature = fees.fee_calculator.lamports_per_signature; let required_funds = lamports_per_signature * 3; Ok(main_sol_balance > required_funds) } async fn handle_subscribe_request( self: Arc, keypair: Keypair, drk_pub_key: DrkPublicKey, mint: Option, ) -> SolResult<()> { debug!(target: "SOL BRIDGE", "handle_subscribe_request()"); // Derive token pubkey if mint was provided. let pubkey = if mint.is_some() { get_associated_token_address(&keypair.pubkey(), &mint.unwrap()) } else { keypair.pubkey() }; if mint.is_some() { debug!(target: "SOL BRIDGE", "Got subscribe request for SPL token"); debug!(target: "SOL BRIDGE", "Main wallet: {}", keypair.pubkey()); debug!(target: "SOL BRIDGE", "Associated token address: {}", pubkey); } else { debug!(target: "SOL BRIDGE", "Got subscribe request for native SOL"); debug!(target: "SOL BRIDGE", "Main wallet: {}", keypair.pubkey()); } // Check if we're already subscribed if self.subscriptions.lock().await.contains(&pubkey) { return Ok(()); } let rpc = RpcClient::new(self.rpc_server.to_string()); // Fetch the current balance. let (prev_balance, decimals) = if mint.is_none() { (rpc.get_balance(&pubkey).map_err(SolFailed::from)?, 9) } else { let mint = mint.unwrap(); match get_account_token_balance(&rpc, &pubkey, &mint) { Ok(v) => v, Err(_) => { let (exists, decimals) = account_is_initialized_mint(&rpc, &mint); if !exists { debug!("Could not figure out the number of decimals in SPL token"); return Err(SolFailed::MintIsNotValid(mint.to_string())); } (0, decimals) } } }; // WebSocket connection let builder = native_tls::TlsConnector::builder(); let tls = TlsConnector::from(builder); let (stream, _) = websockets::connect(self.wss_server, tls).await?; let (mut write, mut read) = stream.split(); // Subscription request build let sub_params = SubscribeParams { encoding: json!("jsonParsed"), commitment: json!("finalized"), }; let subscription = jsonrpc::request( json!("accountSubscribe"), json!([json!(pubkey.to_string()), json!(sub_params)]), ); debug!(target: "SOLANA RPC", "--> {}", serde_json::to_string(&subscription)?); write .send(Message::text(serde_json::to_string(&subscription)?)) .await?; // Subscription ID used for unsubscribing later. let mut sub_id: i64 = 0; // The balance we are going to receive from the JSONRPC notification let cur_balance: u64; let ping_payload: Vec = vec![42, 33, 31, 42]; let iter_interval = 1; let mut sub_iter = 0; loop { let message = read .next() .await .ok_or_else(|| Error::TungsteniteError)?; let message = message?; if let Message::Pong(_) = message.clone() { if sub_iter > 60 * 10 { // 10 minutes self.unsubscribe(&mut write, &pubkey, &sub_id).await?; return Err(SolFailed::RpcError(format!( "Deposit for {:?} expired", pubkey ))); } sub_iter += iter_interval; async_std::task::sleep(Duration::from_secs(iter_interval)).await; write.send(Message::Ping(ping_payload.clone())).await?; continue; }; match serde_json::from_slice(&message.into_data())? { JsonResult::Resp(r) => { // ACK debug!(target: "SOLANA RPC", "<-- {}", serde_json::to_string(&r)?); self.subscriptions.lock().await.push(pubkey); sub_id = r.result.as_i64().unwrap(); // Start sending pings write.send(Message::Ping(ping_payload.clone())).await?; } JsonResult::Err(e) => { debug!(target: "SOLANA RPC", "<-- {}", serde_json::to_string(&e)?); self.unsubscribe(&mut write, &pubkey, &sub_id).await?; return Err(SolFailed::RpcError(e.error.message.to_string())); } JsonResult::Notif(n) => { // Account updated debug!(target: "SOLANA RPC", "Got WebSocket notification"); let params = n.params["result"]["value"].clone(); if mint.is_some() { cur_balance = params["data"]["parsed"]["info"]["tokenAmount"]["amount"] .as_str() .unwrap() .parse() .map_err(Error::from)?; } else { cur_balance = params["lamports"].as_u64().unwrap(); } break; } } } let send_notification = self.notify_channel.0.clone(); let self2 = self.clone(); self2.unsubscribe(&mut write, &pubkey, &sub_id).await?; if cur_balance < prev_balance { return Err(SolFailed::Notification( "New balance is less than previous balance".into(), )); } let amnt = cur_balance - prev_balance; if mint.is_some() { let ui_amnt = amnt / u64::pow(10, decimals as u32); send_notification .send(TokenNotification { network: NetworkName::Solana, token_id: generate_id(&mint.unwrap().to_string(), &NetworkName::Solana)?, drk_pub_key, received_balance: amnt, decimals: decimals as u16, }) .await .map_err(Error::from)?; debug!(target: "SOL BRIDGE", "Received {} {:?} tokens", ui_amnt, mint.unwrap()); let _ = self.send_tok_to_main_wallet(&rpc, &mint.unwrap(), amnt, decimals, &keypair)?; } else { let ui_amnt = lamports_to_sol(amnt); send_notification .send(TokenNotification { network: NetworkName::Solana, token_id: generate_id(SOL_NATIVE_TOKEN_ID, &NetworkName::Solana)?, drk_pub_key, received_balance: amnt, decimals: decimals as u16, }) .await .map_err(Error::from)?; debug!(target: "SOL BRIDGE", "Received {} SOL", ui_amnt); let _ = self.send_sol_to_main_wallet(&rpc, amnt, &keypair)?; } Ok(()) } async fn unsubscribe( self: Arc, write: &mut futures::stream::SplitSink, pubkey: &Pubkey, sub_id: &i64, ) -> Result<()> { { let mut subscriptions = self.subscriptions.lock().await; let index = subscriptions.iter().position(|p| p == pubkey); if let Some(ind) = index { debug!(target: "SOL BRIDGE", "Removing subscription from list"); subscriptions.remove(ind); } } let unsubscription = jsonrpc::request(json!("accountUnsubscribe"), json!([sub_id])); write .send(Message::text(serde_json::to_string(&unsubscription)?)) .await?; Ok(()) } fn send_tok_to_main_wallet( self: Arc, rpc: &RpcClient, mint: &Pubkey, amount: u64, decimals: u64, keypair: &Keypair, ) -> SolResult { debug!(target: "SOL BRIDGE", "Sending {} {:?} tokens to main wallet", amount / u64::pow(10, decimals as u32), mint); // The token account from our main wallet let main_tok_pk = get_associated_token_address(&self.main_keypair.pubkey(), mint); // The token account from the deposit wallet let temp_tok_pk = get_associated_token_address(&keypair.pubkey(), mint); let mut instructions = vec![]; match rpc.get_account_data(&main_tok_pk) { Ok(v) => { // This will fail in the event of unexpected data // otherwise it's valid token data, and we consider account initialized. spl_token::state::Account::unpack_from_slice(&v)?; } Err(_) => { // Unitinialized, so we add a creation instruction debug!("Main wallet token account is uninitialized. Adding init instruction."); let init_ix = create_associated_token_account( &self.main_keypair.pubkey(), // fee payer &self.main_keypair.pubkey(), // wallet mint, ); instructions.push(init_ix); } } // Transfer tokens from the deposit wallet to the main wallet let transfer_ix = spl_token::instruction::transfer_checked( &spl_token::id(), &temp_tok_pk, mint, &main_tok_pk, &keypair.pubkey(), &[], amount, decimals as u8, )?; instructions.push(transfer_ix); // Close the account and reap the rent if there's no more tokens on it. let (tok_balance, _) = get_account_token_balance(rpc, &temp_tok_pk, mint)?; if tok_balance - amount == 0 { debug!(target: "SOL BRIDGE", "Adding account close instruction because resulting balance is 0"); let close_ix = spl_token::instruction::close_account( &spl_token::id(), &temp_tok_pk, &self.main_keypair.pubkey(), &keypair.pubkey(), &[], )?; instructions.push(close_ix); } let tx = Transaction::new_with_payer(&instructions, Some(&self.main_keypair.pubkey())); let signature = sign_and_send_transaction(rpc, tx, vec![&self.main_keypair, keypair])?; debug!(target: "SOL BRIDGE", "Sent tokens to main wallet: {}", signature); Ok(signature) } fn send_sol_to_main_wallet( self: Arc, rpc: &RpcClient, amount: u64, keypair: &Keypair, ) -> SolResult { debug!(target: "SOL BRIDGE", "Sending {} SOL to main wallet", lamports_to_sol(amount)); let ix = system_instruction::transfer(&keypair.pubkey(), &self.main_keypair.pubkey(), amount); let tx = Transaction::new_with_payer(&[ix], Some(&self.main_keypair.pubkey())); let signature = sign_and_send_transaction(rpc, tx, vec![&self.main_keypair, keypair])?; debug!(target: "SOL BRIDGE", "Sent {} SOL to main wallet: {}", lamports_to_sol(amount), signature); Ok(signature) } fn check_mint_address(&self, mint_address: Option) -> SolResult> { if let Some(mint_addr) = mint_address { let pubkey = match Pubkey::from_str(&mint_addr) { Ok(v) => v, Err(e) => return Err(SolFailed::BadSolAddress(e.to_string())), }; let rpc = RpcClient::new(self.rpc_server.to_string()); if !account_is_initialized_mint(&rpc, &pubkey).0 { return Err(SolFailed::MintIsNotValid(mint_addr)); } Ok(Some(pubkey)) } else { Ok(None) } } } #[async_trait] impl NetworkClient for SolClient { async fn subscribe( self: Arc, drk_pub_key: DrkPublicKey, mint_address: Option, executor: Arc>, ) -> Result { let keypair = Keypair::new(); let public_key = keypair.pubkey().to_string(); let private_key = serialize(&keypair); let mint = self.check_mint_address(mint_address)?; let rpc = RpcClient::new(self.rpc_server.to_string()); if !self.check_main_account_balance(&rpc)? { warn!(target: "SOL BRIDGE", "Main account has no enough funds"); return Err(Error::from(SolFailed::MainAccountNotEnoughValue)); } executor .spawn(async move { let result = self .handle_subscribe_request(keypair, drk_pub_key, mint) .await; if let Err(e) = result { error!(target: "SOL BRIDGE SUBSCRIPTION","{}", e.to_string()); } }) .detach(); Ok(TokenSubscribtion { private_key, public_key, }) } // in solana case private key it's the same as keypair async fn subscribe_with_keypair( self: Arc, private_key: Vec, _public_key: Vec, drk_pub_key: DrkPublicKey, mint_address: Option, executor: Arc>, ) -> Result { let keypair: Keypair = deserialize(&private_key)?; let public_key = keypair.pubkey().to_string(); let mint = self.check_mint_address(mint_address)?; let rpc = RpcClient::new(self.rpc_server.to_string()); if !self.check_main_account_balance(&rpc)? { return Err(Error::from(SolFailed::MainAccountNotEnoughValue)); } executor .spawn(async move { let result = self .handle_subscribe_request(keypair, drk_pub_key, mint) .await; if let Err(e) = result { error!(target: "SOL BRIDGE SUBSCRIPTION","{}", e.to_string()); } }) .detach(); Ok(public_key) } async fn get_notifier(self: Arc) -> Result> { Ok(self.notify_channel.1.clone()) } async fn send( self: Arc, address: Vec, mint: Option, amount: u64, ) -> Result<()> { debug!(target: "SOL BRIDGE", "start sending {} sol", lamports_to_sol(amount) ); let rpc = RpcClient::new(self.rpc_server.to_string()); let address: Pubkey = deserialize(&address)?; let mut decimals = 9; if mint.is_some() { let mint_address: Option = self.check_mint_address(mint)?; if let Some(mint_addr) = mint_address { let tkn = rpc.get_token_supply(&mint_addr).map_err(SolFailed::from)?; decimals = tkn.decimals; }; } // reverse truncate let amount = truncate(amount, decimals as u16, 8)?; let instruction = system_instruction::transfer(&self.main_keypair.pubkey(), &address, amount); let mut tx = Transaction::new_with_payer(&[instruction], Some(&self.main_keypair.pubkey())); let bhq = BlockhashQuery::default(); match bhq.get_blockhash_and_fee_calculator(&rpc, rpc.commitment()) { Err(_) => panic!("Couldn't connect to RPC"), Ok(v) => tx.sign(&[&self.main_keypair], v.0), } let _signature = rpc .send_and_confirm_transaction(&tx) .map_err(SolFailed::from)?; Ok(()) } } /// Gets account token balance for given mint. /// Returns: (amount, decimals) pub fn get_account_token_balance( rpc: &RpcClient, address: &Pubkey, mint: &Pubkey, ) -> SolResult<(u64, u64)> { let mint_account = rpc.get_account(mint)?; let token_account = rpc.get_account(address)?; let mint_data = spl_token::state::Mint::unpack_from_slice(&mint_account.data)?; let token_data = spl_token::state::Account::unpack_from_slice(&token_account.data)?; Ok((token_data.amount, mint_data.decimals as u64)) } /// Check if given account is a valid token mint pub fn account_is_initialized_mint(rpc: &RpcClient, mint: &Pubkey) -> (bool, u64) { match rpc.get_token_supply(mint) { Ok(v) => (true, v.decimals as u64), Err(_) => (false, 0), } } pub fn sign_and_send_transaction( rpc: &RpcClient, mut tx: Transaction, signers: Vec<&Keypair>, ) -> SolResult { let bhq = BlockhashQuery::default(); match bhq.get_blockhash_and_fee_calculator(rpc, rpc.commitment()) { Err(_) => return Err(SolFailed::RpcError("Couldn't connect to RPC".into())), Ok(v) => tx.sign(&signers, v.0), } match rpc.send_and_confirm_transaction(&tx) { Ok(s) => Ok(s), Err(_) => Err(SolFailed::RpcError("Failed to send transaction".into())), } } impl Encodable for Keypair { fn encode(&self, s: S) -> Result { let key: Vec = self.to_bytes().to_vec(); let len = key.encode(s)?; Ok(len) } } impl Decodable for Keypair { fn decode(mut d: D) -> Result { let key: Vec = Decodable::decode(&mut d)?; let key = Keypair::from_bytes(key.as_slice()).map_err(|_| { crate::Error::from(SolFailed::DecodeAndEncodeError( "load keypair from slice".into(), )) })?; Ok(key) } } impl Encodable for Pubkey { fn encode(&self, s: S) -> Result { let key = self.to_string(); let len = key.encode(s)?; Ok(len) } } impl Decodable for Pubkey { fn decode(mut d: D) -> Result { let key: String = Decodable::decode(&mut d)?; let key = Pubkey::try_from(key.as_str()).map_err(|_| { crate::Error::from(SolFailed::DecodeAndEncodeError( "load public key from slice".into(), )) })?; Ok(key) } } #[derive(Debug)] pub enum SolFailed { NotEnoughValue(u64), MainAccountNotEnoughValue, BadSolAddress(String), DecodeAndEncodeError(String), WebSocketError(String), RpcError(String), SolClientError(String), Notification(String), ProgramError(String), MintIsNotValid(String), JsonError(String), ParseError(String), } impl std::error::Error for SolFailed {} impl std::fmt::Display for SolFailed { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { match self { SolFailed::NotEnoughValue(i) => { write!(f, "There is no enough value {}", i) } SolFailed::MainAccountNotEnoughValue => { write!(f, "Main Account Has no enough value") } SolFailed::BadSolAddress(ref err) => { write!(f, "Bad Sol Address: {}", err) } SolFailed::DecodeAndEncodeError(ref err) => { write!(f, "Decode and decode keys error: {}", err) } SolFailed::WebSocketError(i) => { write!(f, "WebSocket Error: {}", i) } SolFailed::RpcError(i) => { write!(f, "Rpc Error: {}", i) } SolFailed::ParseError(i) => { write!(f, "Parse Error: {}", i) } SolFailed::SolClientError(i) => { write!(f, "Solana Client Error: {}", i) } SolFailed::Notification(i) => { write!(f, "Received Notification Error: {}", i) } SolFailed::ProgramError(i) => { write!(f, "ProgramError Error: {}", i) } SolFailed::MintIsNotValid(i) => { write!(f, "Given mint is not valid: {}", i) } SolFailed::JsonError(i) => { write!(f, "JsonError: {}", i) } } } } impl From for SolFailed { fn from(err: solana_sdk::pubkey::ParsePubkeyError) -> SolFailed { SolFailed::ParseError(err.to_string()) } } impl From for SolFailed { fn from(err: tungstenite::Error) -> SolFailed { SolFailed::WebSocketError(err.to_string()) } } impl From for SolFailed { fn from(err: solana_client::client_error::ClientError) -> SolFailed { SolFailed::SolClientError(err.to_string()) } } impl From for SolFailed { fn from(err: solana_sdk::program_error::ProgramError) -> SolFailed { SolFailed::ProgramError(err.to_string()) } } impl From for SolFailed { fn from(err: crate::error::Error) -> SolFailed { SolFailed::SolClientError(err.to_string()) } } impl From for SolFailed { fn from(err: serde_json::Error) -> SolFailed { SolFailed::JsonError(err.to_string()) } } pub type SolResult = std::result::Result;