feat(UTXO): balance event streaming for Electrum clients

mm2src/coins/coin_balance.rs

@@ -10,7 +10,7 @@ use futures::compat::Future01CompatExt;
 use mm2_err_handle::prelude::*;
 use mm2_number::BigDecimal;
 #[cfg(test)] use mocktopus::macros::*;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::ops::Range;
 use std::{fmt, iter};
 
@@ -321,6 +321,12 @@ pub trait HDWalletBalanceOps: HDWalletCoinOps {
         let balance = self.known_address_balance(address).await?;
         Ok(AddressBalanceStatus::Used(balance))
     }
+
+    /// Prepares addresses for real time balance streaming if coin balance event is enabled.
+    async fn prepare_addresses_for_balance_stream_if_enabled(
+        &self,
+        addresses: HashSet<Self::Address>,
+    ) -> MmResult<(), String>;
 }
 
 #[async_trait]

mm2src/coins/rpc_command/get_new_address.rs

@@ -54,6 +54,8 @@ pub enum GetNewAddressRpcError {
     RpcInvalidResponse(String),
     #[display(fmt = "HD wallet storage error: {_0}")]
     WalletStorageError(String),
+    #[display(fmt = "Failed scripthash subscription. Error: {_0}")]
+    FailedScripthashSubscription(String),
     #[from_trait(WithTimeout::timeout)]
     #[display(fmt = "RPC timed out {_0:?}")]
     Timeout(Duration),
@@ -183,6 +185,7 @@ impl HttpStatusCode for GetNewAddressRpcError {
             GetNewAddressRpcError::Transport(_)
             | GetNewAddressRpcError::RpcInvalidResponse(_)
             | GetNewAddressRpcError::WalletStorageError(_)
+            | GetNewAddressRpcError::FailedScripthashSubscription(_)
             | GetNewAddressRpcError::HwError(_)
             | GetNewAddressRpcError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
             GetNewAddressRpcError::Timeout(_) => StatusCode::REQUEST_TIMEOUT,
@@ -380,8 +383,10 @@ pub(crate) mod common_impl {
     use super::*;
     use crate::coin_balance::{HDAddressBalanceScanner, HDWalletBalanceOps};
     use crate::hd_wallet::{HDAccountOps, HDWalletCoinOps, HDWalletOps};
+    use crate::utxo::UtxoCommonOps;
     use crate::{CoinWithDerivationMethod, HDAddress};
     use crypto::RpcDerivationPath;
+    use std::collections::HashSet;
     use std::fmt;
     use std::ops::DerefMut;
 
@@ -435,9 +440,12 @@ pub(crate) mod common_impl {
     ) -> MmResult<GetNewAddressResponse, GetNewAddressRpcError>
     where
         ConfirmAddress: HDConfirmAddress,
-        Coin:
-            HDWalletBalanceOps + CoinWithDerivationMethod<HDWallet = <Coin as HDWalletCoinOps>::HDWallet> + Send + Sync,
-        <Coin as HDWalletCoinOps>::Address: fmt::Display,
+        Coin: UtxoCommonOps
+            + HDWalletBalanceOps
+            + CoinWithDerivationMethod<HDWallet = <Coin as HDWalletCoinOps>::HDWallet>
+            + Send
+            + Sync,
+        <Coin as HDWalletCoinOps>::Address: fmt::Display + Into<keys::Address> + std::hash::Hash + std::cmp::Eq,
     {
         let hd_wallet = coin.derivation_method().hd_wallet_or_err()?;
 
@@ -462,9 +470,16 @@ pub(crate) mod common_impl {
             .await?;
 
         let balance = coin.known_address_balance(&address).await?;
+
+        let address_as_string = address.to_string();
+
+        coin.prepare_addresses_for_balance_stream_if_enabled(HashSet::from([address]))
+            .await
+            .map_err(|e| GetNewAddressRpcError::FailedScripthashSubscription(e.to_string()))?;
+
         Ok(GetNewAddressResponse {
             new_address: HDAddressBalance {
-                address: address.to_string(),
+                address: address_as_string,
                 derivation_path: RpcDerivationPath(derivation_path),
                 chain,
                 balance,

mm2src/coins/rpc_command/hd_account_balance_rpc_error.rs

@@ -26,6 +26,8 @@ pub enum HDAccountBalanceRpcError {
     WalletStorageError(String),
     #[display(fmt = "Electrum/Native RPC invalid response: {}", _0)]
     RpcInvalidResponse(String),
+    #[display(fmt = "Failed scripthash subscription. Error: {_0}")]
+    FailedScripthashSubscription(String),
     #[display(fmt = "Transport: {}", _0)]
     Transport(String),
     #[display(fmt = "Internal: {}", _0)]
@@ -44,6 +46,7 @@ impl HttpStatusCode for HDAccountBalanceRpcError {
             HDAccountBalanceRpcError::Transport(_)
             | HDAccountBalanceRpcError::WalletStorageError(_)
             | HDAccountBalanceRpcError::RpcInvalidResponse(_)
+            | HDAccountBalanceRpcError::FailedScripthashSubscription(_)
             | HDAccountBalanceRpcError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }

mm2src/coins/rpc_command/init_scan_for_new_addresses.rs

@@ -130,17 +130,23 @@ pub mod common_impl {
     use super::*;
     use crate::coin_balance::HDWalletBalanceOps;
     use crate::hd_wallet::{HDAccountOps, HDWalletCoinOps, HDWalletOps};
+    use crate::utxo::UtxoCommonOps;
     use crate::CoinWithDerivationMethod;
-    use std::fmt;
+    use keys::Address;
+    use std::collections::HashSet;
     use std::ops::DerefMut;
+    use std::str::FromStr;
 
     pub async fn scan_for_new_addresses_rpc<Coin>(
         coin: &Coin,
         params: ScanAddressesParams,
     ) -> MmResult<ScanAddressesResponse, HDAccountBalanceRpcError>
     where
-        Coin: CoinWithDerivationMethod<HDWallet = <Coin as HDWalletCoinOps>::HDWallet> + HDWalletBalanceOps + Sync,
-        <Coin as HDWalletCoinOps>::Address: fmt::Display,
+        Coin: UtxoCommonOps
+            + CoinWithDerivationMethod<HDWallet = <Coin as HDWalletCoinOps>::HDWallet>
+            + HDWalletBalanceOps
+            + Sync,
+        HashSet<<Coin as HDWalletCoinOps>::Address>: From<HashSet<keys::Address>>,
     {
         let hd_wallet = coin.derivation_method().hd_wallet_or_err()?;
 
@@ -157,6 +163,15 @@ pub mod common_impl {
             .scan_for_new_addresses(hd_wallet, hd_account.deref_mut(), &address_scanner, gap_limit)
             .await?;
 
+        let addresses: HashSet<_> = new_addresses
+            .iter()
+            .map(|address_balance| Address::from_str(&address_balance.address).expect("Valid address"))
+            .collect();
+
+        coin.prepare_addresses_for_balance_stream_if_enabled(addresses.into())
+            .await
+            .map_err(|e| HDAccountBalanceRpcError::FailedScripthashSubscription(e.to_string()))?;
+
         Ok(ScanAddressesResponse {
             account_index: account_id,
             derivation_path: RpcDerivationPath(account_derivation_path),

mm2src/coins/tendermint/tendermint_balance_events.rs

@@ -114,6 +114,7 @@ impl EventBehaviour for TendermintCoin {
                         })
                         .collect();
 
+                    let mut balance_updates = vec![];
                     for denom in denoms {
                         if let Some((ticker, decimals)) = self.active_ticker_and_decimals_from_denom(&denom) {
                             let balance_denom = match self.account_balance_for_denom(&self.account_id, denom).await {
@@ -139,17 +140,22 @@ impl EventBehaviour for TendermintCoin {
                             }
 
                             if broadcast {
-                                let payload = json!({
+                                balance_updates.push(json!({
                                     "ticker": ticker,
                                     "balance": { "spendable": balance_decimal, "unspendable": BigDecimal::default() }
-                                });
-
-                                ctx.stream_channel_controller
-                                    .broadcast(Event::new(Self::EVENT_NAME.to_string(), payload.to_string()))
-                                    .await;
+                                }));
                             }
                         }
                     }
+
+                    if !balance_updates.is_empty() {
+                        ctx.stream_channel_controller
+                            .broadcast(Event::new(
+                                Self::EVENT_NAME.to_string(),
+                                json!(balance_updates).to_string(),
+                            ))
+                            .await;
+                    }
                 }
             }
         }

mm2src/coins/utxo.rs

@@ -32,6 +32,7 @@ pub mod rpc_clients;
 pub mod slp;
 pub mod spv;
 pub mod swap_proto_v2_scripts;
+pub mod utxo_balance_events;
 pub mod utxo_block_header_storage;
 pub mod utxo_builder;
 pub mod utxo_common;
@@ -55,7 +56,7 @@ use crypto::{Bip32DerPathOps, Bip32Error, Bip44Chain, ChildNumber, DerivationPat
              StandardHDCoinAddress, StandardHDPathError, StandardHDPathToAccount, StandardHDPathToCoin};
 use derive_more::Display;
 #[cfg(not(target_arch = "wasm32"))] use dirs::home_dir;
-use futures::channel::mpsc::{Receiver as AsyncReceiver, Sender as AsyncSender, UnboundedSender};
+use futures::channel::mpsc::{Receiver as AsyncReceiver, Sender as AsyncSender, UnboundedReceiver, UnboundedSender};
 use futures::compat::Future01CompatExt;
 use futures::lock::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
 use futures01::Future;
@@ -65,7 +66,7 @@ pub use keys::{Address, AddressFormat as UtxoAddressFormat, AddressHashEnum, Key
                Type as ScriptType};
 #[cfg(not(target_arch = "wasm32"))]
 use lightning_invoice::Currency as LightningCurrency;
-use mm2_core::mm_ctx::MmArc;
+use mm2_core::mm_ctx::{MmArc, MmWeak};
 use mm2_err_handle::prelude::*;
 use mm2_metrics::MetricsArc;
 use mm2_number::BigDecimal;
@@ -142,6 +143,15 @@ pub type MatureUnspentMap = HashMap<Address, MatureUnspentList>;
 pub type RecentlySpentOutPointsGuard<'a> = AsyncMutexGuard<'a, RecentlySpentOutPoints>;
 pub type UtxoHDAddress = HDAddress<Address, Public>;
 
+pub enum ScripthashNotification {
+    Triggered(String),
+    SubscribeToAddresses(HashSet<Address>),
+    RefreshSubscriptions,
+}
+
+pub type ScripthashNotificationSender = Option<UnboundedSender<ScripthashNotification>>;
+type ScripthashNotificationHandler = Option<Arc<AsyncMutex<UnboundedReceiver<ScripthashNotification>>>>;
+
 #[cfg(windows)]
 #[cfg(not(target_arch = "wasm32"))]
 fn get_special_folder_path() -> PathBuf {
@@ -610,6 +620,11 @@ pub struct UtxoCoinFields {
     /// This abortable system is used to spawn coin's related futures that should be aborted on coin deactivation
     /// and on [`MmArc::stop`].
     pub abortable_system: AbortableQueue,
+    pub(crate) ctx: MmWeak,
+    /// This is used for balance event streaming implementation for UTXOs.
+    /// If balance event streaming isn't enabled, this value will always be `None`; otherwise,
+    /// it will be used for receiving scripthash notifications to re-fetch balances.
+    scripthash_notification_handler: ScripthashNotificationHandler,
 }
 
 #[derive(Debug, Display)]

mm2src/coins/utxo/qtum.rs

@@ -1,3 +1,4 @@
+use super::utxo_common::utxo_prepare_addresses_for_balance_stream_if_enabled;
 use super::*;
 use crate::coin_balance::{self, EnableCoinBalanceError, EnabledCoinBalanceParams, HDAccountBalance, HDAddressBalance,
                           HDWalletBalance, HDWalletBalanceOps};
@@ -1149,6 +1150,13 @@ impl HDWalletBalanceOps for QtumCoin {
     ) -> BalanceResult<Vec<(Self::Address, CoinBalance)>> {
         utxo_common::addresses_balances(self, addresses).await
     }
+
+    async fn prepare_addresses_for_balance_stream_if_enabled(
+        &self,
+        addresses: HashSet<Self::Address>,
+    ) -> MmResult<(), String> {
+        utxo_prepare_addresses_for_balance_stream_if_enabled(self, addresses).await
+    }
 }
 
 impl HDWalletCoinWithStorageOps for QtumCoin {