diff --git a/mm2src/coins/z_coin.rs b/mm2src/coins/z_coin.rs index 4e125b0f3a..0b55cbf41d 100644 --- a/mm2src/coins/z_coin.rs +++ b/mm2src/coins/z_coin.rs @@ -1,3 +1,12 @@ +pub mod storage; +mod z_balance_streaming; +mod z_coin_errors; +#[cfg(all(test, feature = "zhtlc-native-tests"))] +mod z_coin_native_tests; +mod z_htlc; +mod z_rpc; +mod z_tx_history; + use crate::coin_errors::{MyAddressError, ValidatePaymentResult}; use crate::my_tx_history_v2::{MyTxHistoryErrorV2, MyTxHistoryRequestV2, MyTxHistoryResponseV2}; use crate::rpc_command::init_withdraw::{InitWithdrawCoin, WithdrawInProgressStatus, WithdrawTaskHandleShared}; @@ -14,6 +23,7 @@ use crate::utxo::{sat_from_big_decimal, utxo_common, ActualTxFee, AdditionalTxDa UtxoCommonOps, UtxoRpcMode, UtxoTxBroadcastOps, UtxoTxGenerationOps, VerboseTransactionFrom}; use crate::utxo::{UnsupportedAddr, UtxoFeeDetails}; use crate::z_coin::storage::{BlockDbImpl, WalletDbShared}; +use crate::z_coin::z_balance_streaming::ZBalanceEventHandler; use crate::z_coin::z_tx_history::{fetch_tx_history_from_db, ZCoinTxHistoryItem}; use crate::{BalanceError, BalanceFut, CheckIfMyPaymentSentArgs, CoinBalance, CoinFutSpawner, ConfirmPaymentInput, DexFee, FeeApproxStage, FoundSwapTxSpend, HistorySyncState, MakerSwapTakerCoin, MarketCoinOps, MmCoin, @@ -48,6 +58,7 @@ use keys::hash::H256; use keys::{KeyPair, Message, Public}; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; +use mm2_event_stream::behaviour::{EventBehaviour, EventInitStatus}; use mm2_number::{BigDecimal, MmNumber}; #[cfg(test)] use mocktopus::macros::*; use primitives::bytes::Bytes; @@ -60,8 +71,10 @@ use std::convert::TryInto; use std::iter; use std::path::PathBuf; use std::sync::Arc; -#[cfg(target_arch = "wasm32")] -use z_coin_errors::ZCoinBalanceError; +pub use z_coin_errors::*; +use z_htlc::{z_p2sh_spend, z_send_dex_fee, z_send_htlc}; +use z_rpc::init_light_client; +pub use z_rpc::{FirstSyncBlock, SyncStatus}; use z_rpc::{SaplingSyncConnector, SaplingSyncGuard}; use zcash_client_backend::encoding::{decode_payment_address, encode_extended_spending_key, encode_payment_address}; use zcash_client_backend::wallet::{AccountId, SpendableNote}; @@ -78,13 +91,6 @@ use zcash_primitives::{constants::mainnet as z_mainnet_constants, sapling::Payme zip32::ExtendedFullViewingKey, zip32::ExtendedSpendingKey}; use zcash_proofs::prover::LocalTxProver; -mod z_htlc; -use z_htlc::{z_p2sh_spend, z_send_dex_fee, z_send_htlc}; - -mod z_rpc; -use z_rpc::init_light_client; -pub use z_rpc::{FirstSyncBlock, SyncStatus}; - cfg_native!( use common::{async_blocking, sha256_digest}; use zcash_client_sqlite::error::SqliteClientError as ZcashClientError; @@ -94,22 +100,14 @@ cfg_native!( ); cfg_wasm32!( - use crate::z_coin::z_params::ZcashParamsWasmImpl; + use crate::z_coin::storage::ZcashParamsWasmImpl; use common::executor::AbortOnDropHandle; use futures::channel::oneshot; use rand::rngs::OsRng; use zcash_primitives::transaction::builder::TransactionMetadata; + use z_coin_errors::ZCoinBalanceError; ); -#[allow(unused)] mod z_coin_errors; -pub use z_coin_errors::*; - -pub mod storage; -#[cfg(all(test, feature = "zhtlc-native-tests"))] -mod z_coin_native_tests; -#[cfg(target_arch = "wasm32")] mod z_params; -mod z_tx_history; - /// `ZP2SHSpendError` compatible `TransactionErr` handling macro. macro_rules! try_ztx_s { ($e: expr) => { @@ -209,6 +207,7 @@ pub struct ZCoinFields { light_wallet_db: WalletDbShared, consensus_params: ZcoinConsensusParams, sync_state_connector: AsyncMutex, + z_balance_event_handler: Option, } impl Transaction for ZTransaction { @@ -654,6 +653,17 @@ impl ZCoin { paging_options: request.paging_options, }) } + + async fn spawn_balance_stream_if_enabled(&self, ctx: &MmArc) -> Result<(), String> { + let coin = self.clone(); + if let Some(stream_config) = &ctx.event_stream_configuration { + if let EventInitStatus::Failed(err) = EventBehaviour::spawn_if_active(coin, stream_config).await { + return ERR!("Failed spawning zcoin balance event with error: {}", err); + } + } + + Ok(()) + } } impl AsRef for ZCoin { @@ -875,10 +885,24 @@ impl<'a> UtxoCoinBuilder for ZCoinBuilder<'a> { ); let blocks_db = self.init_blocks_db().await?; + let (z_balance_event_sender, z_balance_event_handler) = if self.ctx.event_stream_configuration.is_some() { + let (sender, receiver) = futures::channel::mpsc::unbounded(); + (Some(sender), Some(Arc::new(AsyncMutex::new(receiver)))) + } else { + (None, None) + }; + let (sync_state_connector, light_wallet_db) = match &self.z_coin_params.mode { #[cfg(not(target_arch = "wasm32"))] ZcoinRpcMode::Native => { - init_native_client(&self, self.native_client()?, blocks_db, &z_spending_key).await? + init_native_client( + &self, + self.native_client()?, + blocks_db, + &z_spending_key, + z_balance_event_sender, + ) + .await? }, ZcoinRpcMode::Light { light_wallet_d_servers, @@ -893,11 +917,13 @@ impl<'a> UtxoCoinBuilder for ZCoinBuilder<'a> { sync_params, skip_sync_params.unwrap_or_default(), &z_spending_key, + z_balance_event_sender, ) .await? }, }; - let z_fields = ZCoinFields { + + let z_fields = Arc::new(ZCoinFields { dex_fee_addr, my_z_addr, my_z_addr_encoded, @@ -907,12 +933,16 @@ impl<'a> UtxoCoinBuilder for ZCoinBuilder<'a> { light_wallet_db, consensus_params: self.protocol_info.consensus_params, sync_state_connector, - }; + z_balance_event_handler, + }); - Ok(ZCoin { - utxo_arc, - z_fields: Arc::new(z_fields), - }) + let zcoin = ZCoin { utxo_arc, z_fields }; + zcoin + .spawn_balance_stream_if_enabled(self.ctx) + .await + .map_to_mm(ZCoinBuildError::FailedSpawningBalanceEvents)?; + + Ok(zcoin) } } diff --git a/mm2src/coins/z_coin/storage.rs b/mm2src/coins/z_coin/storage.rs index 5b1f3f1f00..08e478f27a 100644 --- a/mm2src/coins/z_coin/storage.rs +++ b/mm2src/coins/z_coin/storage.rs @@ -4,8 +4,13 @@ pub mod blockdb; pub use blockdb::*; pub mod walletdb; +#[cfg(target_arch = "wasm32")] mod z_params; +#[cfg(target_arch = "wasm32")] +pub(crate) use z_params::ZcashParamsWasmImpl; + pub use walletdb::*; +use crate::z_coin::z_balance_streaming::ZBalanceEventSender; use mm2_err_handle::mm_error::MmResult; #[cfg(target_arch = "wasm32")] use walletdb::wasm::storage::DataConnStmtCacheWasm; @@ -55,7 +60,7 @@ pub struct CompactBlockRow { #[derive(Clone)] pub enum BlockProcessingMode { Validate, - Scan(DataConnStmtCacheWrapper), + Scan(DataConnStmtCacheWrapper, Option), } /// Checks that the scanned blocks in the data database, when combined with the recent @@ -114,7 +119,7 @@ pub async fn scan_cached_block( params: &ZcoinConsensusParams, block: &CompactBlock, last_height: &mut BlockHeight, -) -> Result<(), ValidateBlocksError> { +) -> Result { let mut data_guard = data.inner().clone(); // Fetch the ExtendedFullViewingKeys we are tracking let extfvks = data_guard.get_extended_full_viewing_keys().await?; @@ -201,5 +206,6 @@ pub async fn scan_cached_block( *last_height = current_height; - Ok(()) + // If there are any transactions in the block, return the transaction count + Ok(txs.len()) } diff --git a/mm2src/coins/z_coin/storage/blockdb/blockdb_idb_storage.rs b/mm2src/coins/z_coin/storage/blockdb/blockdb_idb_storage.rs index 112fb67400..cccf8cc0a9 100644 --- a/mm2src/coins/z_coin/storage/blockdb/blockdb_idb_storage.rs +++ b/mm2src/coins/z_coin/storage/blockdb/blockdb_idb_storage.rs @@ -3,6 +3,7 @@ use crate::z_coin::storage::{scan_cached_block, validate_chain, BlockDbImpl, Blo use crate::z_coin::z_coin_errors::ZcoinStorageError; use async_trait::async_trait; +use futures_util::SinkExt; use mm2_core::mm_ctx::MmArc; use mm2_db::indexed_db::{BeBigUint, ConstructibleDb, DbIdentifier, DbInstance, DbLocked, DbUpgrader, IndexedDb, IndexedDbBuilder, InitDbResult, MultiIndex, OnUpgradeResult, TableSignature}; @@ -123,7 +124,7 @@ impl BlockDbImpl { } /// Asynchronously rewinds the storage to a specified block height, effectively - /// removing data beyond the specified height from the storage. + /// removing data beyond the specified height from the storage. pub async fn rewind_to_height(&self, height: BlockHeight) -> ZcoinStorageRes { let locked_db = self.lock_db().await?; let db_transaction = locked_db.get_inner().transaction().await?; @@ -224,7 +225,7 @@ impl BlockDbImpl { BlockProcessingMode::Validate => validate_from .map(|(height, _)| height) .unwrap_or(BlockHeight::from_u32(params.sapling_activation_height) - 1), - BlockProcessingMode::Scan(data) => data.inner().block_height_extrema().await.map(|opt| { + BlockProcessingMode::Scan(data, _) => data.inner().block_height_extrema().await.map(|opt| { opt.map(|(_, max)| max) .unwrap_or(BlockHeight::from_u32(params.sapling_activation_height) - 1) })?, @@ -250,8 +251,15 @@ impl BlockDbImpl { BlockProcessingMode::Validate => { validate_chain(block, &mut prev_height, &mut prev_hash).await?; }, - BlockProcessingMode::Scan(data) => { - scan_cached_block(data, ¶ms, &block, &mut from_height).await?; + BlockProcessingMode::Scan(data, z_balance_change_sender) => { + let tx_size = scan_cached_block(data, ¶ms, &block, &mut from_height).await?; + // If there is/are transactions present in the current scanned block(s), + // we trigger a `Triggered` event to update the balance change. + if tx_size > 0 { + if let Some(mut sender) = z_balance_change_sender.clone() { + sender.send(()).await.expect("No receiver is available/dropped"); + }; + }; }, } } diff --git a/mm2src/coins/z_coin/storage/blockdb/blockdb_sql_storage.rs b/mm2src/coins/z_coin/storage/blockdb/blockdb_sql_storage.rs index 74c790bf89..e8ce70d6dd 100644 --- a/mm2src/coins/z_coin/storage/blockdb/blockdb_sql_storage.rs +++ b/mm2src/coins/z_coin/storage/blockdb/blockdb_sql_storage.rs @@ -6,6 +6,7 @@ use crate::z_coin::ZcoinConsensusParams; use common::async_blocking; use db_common::sqlite::rusqlite::{params, Connection}; use db_common::sqlite::{query_single_row, run_optimization_pragmas, rusqlite}; +use futures_util::SinkExt; use itertools::Itertools; use mm2_core::mm_ctx::MmArc; use mm2_err_handle::prelude::*; @@ -193,7 +194,7 @@ impl BlockDbImpl { BlockProcessingMode::Validate => validate_from .map(|(height, _)| height) .unwrap_or(BlockHeight::from_u32(params.sapling_activation_height) - 1), - BlockProcessingMode::Scan(data) => { + BlockProcessingMode::Scan(data, _) => { let data = data.inner(); data.block_height_extrema().await.map(|opt| { opt.map(|(_, max)| max) @@ -224,8 +225,15 @@ impl BlockDbImpl { BlockProcessingMode::Validate => { validate_chain(block, &mut prev_height, &mut prev_hash).await?; }, - BlockProcessingMode::Scan(data) => { - scan_cached_block(data, ¶ms, &block, &mut from_height).await?; + BlockProcessingMode::Scan(data, z_balance_change_sender) => { + let tx_size = scan_cached_block(data, ¶ms, &block, &mut from_height).await?; + // If there are transactions present in the current scanned block, + // we send a `Triggered` event to update the balance change. + if tx_size > 0 { + if let Some(mut sender) = z_balance_change_sender.clone() { + sender.send(()).await.expect("No receiver is available/dropped"); + }; + }; }, } } diff --git a/mm2src/coins/z_coin/z_params/indexeddb.rs b/mm2src/coins/z_coin/storage/z_params/indexeddb.rs similarity index 100% rename from mm2src/coins/z_coin/z_params/indexeddb.rs rename to mm2src/coins/z_coin/storage/z_params/indexeddb.rs diff --git a/mm2src/coins/z_coin/z_params/mod.rs b/mm2src/coins/z_coin/storage/z_params/mod.rs similarity index 100% rename from mm2src/coins/z_coin/z_params/mod.rs rename to mm2src/coins/z_coin/storage/z_params/mod.rs diff --git a/mm2src/coins/z_coin/z_balance_streaming.rs b/mm2src/coins/z_coin/z_balance_streaming.rs new file mode 100644 index 0000000000..2e4c77df77 --- /dev/null +++ b/mm2src/coins/z_coin/z_balance_streaming.rs @@ -0,0 +1,110 @@ +use crate::common::Future01CompatExt; +use crate::hd_wallet::AsyncMutex; +use crate::z_coin::ZCoin; +use crate::{MarketCoinOps, MmCoin}; + +use async_trait::async_trait; +use common::executor::{AbortSettings, SpawnAbortable}; +use common::log::{error, info}; +use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; +use futures::channel::oneshot; +use futures::channel::oneshot::{Receiver, Sender}; +use futures_util::StreamExt; +use mm2_core::mm_ctx::MmArc; +use mm2_event_stream::behaviour::{EventBehaviour, EventInitStatus}; +use mm2_event_stream::{Event, EventStreamConfiguration}; +use std::sync::Arc; + +pub type ZBalanceEventSender = UnboundedSender<()>; +pub type ZBalanceEventHandler = Arc>>; + +#[async_trait] +impl EventBehaviour for ZCoin { + const EVENT_NAME: &'static str = "COIN_BALANCE"; + const ERROR_EVENT_NAME: &'static str = "COIN_BALANCE_ERROR"; + + async fn handle(self, _interval: f64, tx: Sender) { + const RECEIVER_DROPPED_MSG: &str = "Receiver is dropped, which should never happen."; + + macro_rules! send_status_on_err { + ($match: expr, $sender: tt, $msg: literal) => { + match $match { + Some(t) => t, + None => { + $sender + .send(EventInitStatus::Failed($msg.to_owned())) + .expect(RECEIVER_DROPPED_MSG); + panic!("{}", $msg); + }, + } + }; + } + + let ctx = send_status_on_err!( + MmArc::from_weak(&self.as_ref().ctx), + tx, + "MM context must have been initialized already." + ); + let z_balance_change_handler = send_status_on_err!( + self.z_fields.z_balance_event_handler.as_ref(), + tx, + "Z balance change receiver can not be empty." + ); + + tx.send(EventInitStatus::Success).expect(RECEIVER_DROPPED_MSG); + + // Locks the balance change handler, iterates through received events, and updates balance changes accordingly. + let mut bal = z_balance_change_handler.lock().await; + while (bal.next().await).is_some() { + match self.my_balance().compat().await { + Ok(balance) => { + let payload = json!({ + "ticker": self.ticker(), + "address": self.my_z_address_encoded(), + "balance": { "spendable": balance.spendable, "unspendable": balance.unspendable } + }); + + ctx.stream_channel_controller + .broadcast(Event::new(Self::EVENT_NAME.to_string(), payload.to_string())) + .await; + }, + Err(err) => { + let ticker = self.ticker(); + error!("Failed getting balance for '{ticker}'. Error: {err}"); + let e = serde_json::to_value(err).expect("Serialization should't fail."); + return ctx + .stream_channel_controller + .broadcast(Event::new( + format!("{}:{}", Self::ERROR_EVENT_NAME, ticker), + e.to_string(), + )) + .await; + }, + }; + } + } + + async fn spawn_if_active(self, config: &EventStreamConfiguration) -> EventInitStatus { + if let Some(event) = config.get_event(Self::EVENT_NAME) { + info!( + "{} event is activated for {} address {}. `stream_interval_seconds`({}) has no effect on this.", + Self::EVENT_NAME, + self.ticker(), + self.my_z_address_encoded(), + event.stream_interval_seconds + ); + + let (tx, rx): (Sender, Receiver) = oneshot::channel(); + let fut = self.clone().handle(event.stream_interval_seconds, tx); + let settings = + AbortSettings::info_on_abort(format!("{} event is stopped for {}.", Self::EVENT_NAME, self.ticker())); + self.spawner().spawn_with_settings(fut, settings); + + rx.await.unwrap_or_else(|e| { + EventInitStatus::Failed(format!("Event initialization status must be received: {}", e)) + }) + } else { + EventInitStatus::Inactive + } + } +} diff --git a/mm2src/coins/z_coin/z_coin_errors.rs b/mm2src/coins/z_coin/z_coin_errors.rs index 5b5992baae..7fcf06cb12 100644 --- a/mm2src/coins/z_coin/z_coin_errors.rs +++ b/mm2src/coins/z_coin/z_coin_errors.rs @@ -249,6 +249,7 @@ pub enum ZCoinBuildError { ZCashParamsError(String), ZDerivationPathNotSet, SaplingParamsInvalidChecksum, + FailedSpawningBalanceEvents(String), } #[cfg(not(target_arch = "wasm32"))] diff --git a/mm2src/coins/z_coin/z_rpc.rs b/mm2src/coins/z_coin/z_rpc.rs index 55af1ac36b..bfdc61695f 100644 --- a/mm2src/coins/z_coin/z_rpc.rs +++ b/mm2src/coins/z_coin/z_rpc.rs @@ -32,6 +32,7 @@ use zcash_primitives::zip32::ExtendedSpendingKey; pub(crate) mod z_coin_grpc { tonic::include_proto!("pirate.wallet.sdk.rpc"); } +use crate::z_coin::z_balance_streaming::ZBalanceEventSender; use z_coin_grpc::compact_tx_streamer_client::CompactTxStreamerClient; use z_coin_grpc::{ChainSpec, CompactBlock as TonicCompactBlock}; @@ -507,6 +508,7 @@ pub(super) async fn init_light_client<'a>( sync_params: &Option, skip_sync_params: bool, z_spending_key: &ExtendedSpendingKey, + z_balance_event_sender: Option, ) -> Result<(AsyncMutex, WalletDbShared), MmError> { let coin = builder.ticker.to_string(); let (sync_status_notifier, sync_watcher) = channel(1); @@ -543,7 +545,7 @@ pub(super) async fn init_light_client<'a>( WalletDbShared::new(builder, maybe_checkpoint_block, z_spending_key, continue_from_prev_sync).await?; // Check min_height in blocks_db and rewind blocks_db to 0 if sync_height != min_height if !continue_from_prev_sync && (sync_height != min_height) { - // let user know we're clearing cache and resyncing from new provided height. + // let user know we're clearing cache and re-syncing from new provided height. if min_height > 0 { info!("Older/Newer sync height detected!, rewinding blocks_db to new height: {sync_height:?}"); } @@ -566,6 +568,7 @@ pub(super) async fn init_light_client<'a>( is_pre_sapling: sync_height < sapling_activation_height, actual: sync_height.max(sapling_activation_height), }, + z_balance_event_sender, }; let abort_handle = spawn_abortable(light_wallet_db_sync_loop(sync_handle, Box::new(light_rpc_clients))); @@ -582,6 +585,7 @@ pub(super) async fn init_native_client<'a>( native_client: NativeClient, blocks_db: BlockDbImpl, z_spending_key: &ExtendedSpendingKey, + z_balance_event_sender: Option, ) -> Result<(AsyncMutex, WalletDbShared), MmError> { let coin = builder.ticker.to_string(); let (sync_status_notifier, sync_watcher) = channel(1); @@ -610,6 +614,7 @@ pub(super) async fn init_native_client<'a>( scan_blocks_per_iteration: builder.z_coin_params.scan_blocks_per_iteration, scan_interval_ms: builder.z_coin_params.scan_interval_ms, first_sync_block, + z_balance_event_sender, }; let abort_handle = spawn_abortable(light_wallet_db_sync_loop(sync_handle, Box::new(native_client))); @@ -708,6 +713,7 @@ pub struct SaplingSyncLoopHandle { scan_blocks_per_iteration: u32, scan_interval_ms: u64, first_sync_block: FirstSyncBlock, + z_balance_event_sender: Option, } impl SaplingSyncLoopHandle { @@ -804,8 +810,7 @@ impl SaplingSyncLoopHandle { } } - let latest_block_height = blocks_db.get_latest_block().await?; - let current_block = BlockHeight::from_u32(latest_block_height); + let current_block = BlockHeight::from_u32(blocks_db.get_latest_block().await?); loop { match wallet_ops.block_height_extrema().await? { Some((_, max_in_wallet)) => { @@ -822,7 +827,7 @@ impl SaplingSyncLoopHandle { blocks_db .process_blocks_with_mode( self.consensus_params.clone(), - BlockProcessingMode::Scan(scan), + BlockProcessingMode::Scan(scan, self.z_balance_event_sender.clone()), None, Some(self.scan_blocks_per_iteration), )