Repository: farcaster-project/farcaster-node
Pull request #774 (2022-11-14 15:34:23)

Syncer: Implement logic for balance retrieval
Also adds a cli call to trigger the get balance procedure


Patch:
diff --git a/src/bus/ctl.rs b/src/bus/ctl.rs
index 9a8ccd20..cd4bd82f 100644
--- a/src/bus/ctl.rs
+++ b/src/bus/ctl.rs
@@ -125,6 +125,9 @@ pub enum CtlMsg {
     #[display("set_address_secret_key")]
     SetAddressSecretKey(AddressSecretKey),
 
+    #[display("get_balance")]
+    GetBalance(AddressSecretKey),
+
     #[display("funding_updated()")]
     FundingUpdated,
 
diff --git a/src/bus/info.rs b/src/bus/info.rs
index a2e87115..f24670b2 100644
--- a/src/bus/info.rs
+++ b/src/bus/info.rs
@@ -17,6 +17,7 @@ use crate::cli::OfferSelector;
 use crate::farcasterd::stats::Stats;
 use crate::swapd::StateReport;
 use crate::syncerd::runtime::SyncerdTask;
+use crate::Error;
 
 use super::ctl::FundingInfo;
 use super::StateTransition;
@@ -169,6 +170,9 @@ pub enum InfoMsg {
     // - End GetCheckpointEntry section
     #[display("{0}")]
     FundingInfos(FundingInfos),
+
+    #[display("{0}")]
+    AddressBalance(AddressBalance),
 }
 
 #[derive(Clone, PartialEq, Eq, Debug, Display, NetworkEncode, NetworkDecode)]
@@ -368,7 +372,13 @@ pub enum ProgressEvent {
     Failure(Failure),
 }
 
-#[derive(Eq, PartialEq, Clone, Debug, Display, NetworkDecode, NetworkEncode)]
+#[cfg_attr(feature = "serde", serde_as)]
+#[derive(Eq, PartialEq, Clone, Debug, Display, Hash, NetworkDecode, NetworkEncode)]
+#[cfg_attr(
+    feature = "serde",
+    derive(Serialize, Deserialize),
+    serde(crate = "serde_crate")
+)]
 pub enum Address {
     #[display("{0}")]
     Bitcoin(bitcoin::Address),
@@ -376,6 +386,31 @@ pub enum Address {
     Monero(monero::Address),
 }
 
+impl FromStr for Address {
+    type Err = Error;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        bitcoin::Address::from_str(s)
+            .map(|a| Address::Bitcoin(a))
+            .map_err(|e| Error::Farcaster(e.to_string()))
+            .or(monero::Address::from_str(s)
+                .map(|a| Address::Monero(a))
+                .map_err(|e| Error::Farcaster(e.to_string())))
+    }
+}
+
+#[cfg_attr(feature = "serde", serde_as)]
+#[derive(Eq, PartialEq, Clone, Debug, Display, Hash, NetworkDecode, NetworkEncode)]
+#[cfg_attr(
+    feature = "serde",
+    derive(Serialize, Deserialize),
+    serde(crate = "serde_crate")
+)]
+#[display(AddressBalance::to_yaml_string)]
+pub struct AddressBalance {
+    pub address: Address,
+    pub balance: u64,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, Display, NetworkEncode, NetworkDecode)]
 pub enum OfferStatusSelector {
     #[display("Open")]
@@ -450,3 +485,5 @@ impl ToYamlString for SyncerInfo {}
 impl ToYamlString for ProgressEvent {}
 #[cfg(feature = "serde")]
 impl ToYamlString for FundingInfos {}
+#[cfg(feature = "serde")]
+impl ToYamlString for AddressBalance {}
diff --git a/src/bus/types.rs b/src/bus/types.rs
index 509bf0d9..2920a936 100644
--- a/src/bus/types.rs
+++ b/src/bus/types.rs
@@ -12,6 +12,7 @@ use farcaster_core::{
 use amplify::{ToYamlString, Wrapper};
 use internet2::addr::NodeId;
 use microservices::rpc;
+use serde_with::DisplayFromStr;
 use strict_encoding::{NetworkDecode, NetworkEncode};
 
 use crate::swapd::StateReport;
@@ -32,7 +33,12 @@ pub struct CheckpointEntry {
     pub expected_counterparty_node_id: Option<NodeId>,
 }
 
-#[derive(Clone, Debug, Display, NetworkDecode, NetworkEncode)]
+#[derive(Clone, Debug, Display, Eq, PartialEq, Hash, NetworkDecode, NetworkEncode)]
+#[cfg_attr(
+    feature = "serde",
+    derive(Serialize, Deserialize),
+    serde(crate = "serde_crate")
+)]
 pub enum AddressSecretKey {
     #[display("addr_key({address}, ..)")]
     Bitcoin {
@@ -46,18 +52,31 @@ pub enum AddressSecretKey {
     },
 }
 
-#[derive(Clone, Debug, Display, Eq, PartialEq, NetworkDecode, NetworkEncode)]
+#[derive(Clone, Debug, Display, Eq, PartialEq, Hash, NetworkDecode, NetworkEncode)]
+#[cfg_attr(
+    feature = "serde",
+    derive(Serialize, Deserialize),
+    serde(crate = "serde_crate")
+)]
 #[display("bitcoin secret key info")]
 pub struct BitcoinSecretKeyInfo {
     pub swap_id: Option<SwapId>,
     pub secret_key: bitcoin::secp256k1::SecretKey,
 }
 
-#[derive(Clone, Debug, Display, Eq, PartialEq, NetworkDecode, NetworkEncode)]
+#[cfg_attr(feature = "serde", serde_as)]
+#[derive(Clone, Debug, Display, Eq, PartialEq, Hash, NetworkDecode, NetworkEncode)]
+#[cfg_attr(
+    feature = "serde",
+    derive(Serialize, Deserialize),
+    serde(crate = "serde_crate")
+)]
 #[display("monero secret key info")]
 pub struct MoneroSecretKeyInfo {
     pub swap_id: Option<SwapId>,
+    #[serde_as(as = "DisplayFromStr")]
     pub view: monero::PrivateKey,
+    #[serde_as(as = "DisplayFromStr")]
     pub spend: monero::PrivateKey,
     pub creation_height: Option<u64>,
 }
diff --git a/src/cli/command.rs b/src/cli/command.rs
index c36ebb92..36da195c 100644
--- a/src/cli/command.rs
+++ b/src/cli/command.rs
@@ -583,6 +583,24 @@ impl Exec for Command {
                 }
             }
 
+            Command::GetBalance { address } => {
+                runtime.request_info(ServiceId::Database, InfoMsg::GetAddressSecretKey(address))?;
+                if let BusMsg::Info(InfoMsg::AddressSecretKey(address_secret_key)) =
+                    runtime.report_failure()?
+                {
+                    runtime.request_ctl(
+                        ServiceId::Farcasterd,
+                        CtlMsg::GetBalance(address_secret_key),
+                    )?;
+                    runtime.report_response_or_fail()?;
+                } else {
+                    return Err(Error::Farcaster(
+                        "Can only get balance for old funding addresses. Address not found"
+                            .to_string(),
+                    ));
+                }
+            }
+
             Command::SweepMoneroAddress {
                 source_address,
                 destination_address,
diff --git a/src/cli/opts.rs b/src/cli/opts.rs
index 14c3279a..cd49ce1e 100644
--- a/src/cli/opts.rs
+++ b/src/cli/opts.rs
@@ -25,6 +25,7 @@ use farcaster_core::{
     swap::{btcxmr::PublicOffer, SwapId},
 };
 
+use crate::bus::info::Address;
 use crate::bus::HealthCheckSelector;
 
 /// Command-line tool for working with Farcaster node
@@ -294,6 +295,13 @@ pub enum Command {
         destination_address: XmrAddress,
     },
 
+    /// Returns the balance for a given address. The needs to be a previous funding address
+    #[display("get-balance<{address}>")]
+    GetBalance {
+        /// Address for which the balance should be retrieved.
+        address: Address,
+    },
+
     /// Output shell completion code for the specified shell (bash, zsh or fish)
     ///
     /// The shell code must be evaluated to provide interactive completion of swap-cli commands.
diff --git a/src/farcasterd/runtime.rs b/src/farcasterd/runtime.rs
index 07229a4c..585f12b0 100644
--- a/src/farcasterd/runtime.rs
+++ b/src/farcasterd/runtime.rs
@@ -23,6 +23,7 @@ use crate::farcasterd::stats::Stats;
 use crate::farcasterd::syncer_state_machine::{SyncerStateMachine, SyncerStateMachineExecutor};
 use crate::farcasterd::trade_state_machine::{TradeStateMachine, TradeStateMachineExecutor};
 use crate::farcasterd::Opts;
+use crate::syncerd::AddressBalance;
 use crate::syncerd::{Event as SyncerEvent, HealthResult, SweepSuccess, TaskId};
 use crate::{
     bus::ctl::{Keys, ProgressStack, Token},
@@ -902,15 +903,23 @@ impl Runtime {
         source: &ServiceId,
     ) -> Result<Option<SyncerStateMachine>, Error> {
         match (req, source) {
-            (BusMsg::Ctl(CtlMsg::SweepAddress(..)), _) => Ok(Some(SyncerStateMachine::Start)),
-            (BusMsg::Ctl(CtlMsg::HealthCheck(..)), _) => Ok(Some(SyncerStateMachine::Start)),
+            (BusMsg::Ctl(CtlMsg::SweepAddress(..)), _)
+            | (BusMsg::Ctl(CtlMsg::HealthCheck(..)), _)
+            | (BusMsg::Ctl(CtlMsg::GetBalance(..)), _) => Ok(Some(SyncerStateMachine::Start)),
             (
                 BusMsg::Sync(SyncMsg::Event(SyncerEvent::SweepSuccess(SweepSuccess {
                     id, ..
                 }))),
                 _,
-            ) => Ok(self.syncer_state_machines.remove(id)),
-            (
+            )
+            | (
+                BusMsg::Sync(SyncMsg::Event(SyncerEvent::AddressBalance(AddressBalance {
+                    id,
+                    ..
+                }))),
+                _,
+            )
+            | (
                 BusMsg::Sync(SyncMsg::Event(SyncerEvent::HealthResult(HealthResult {
                     id, ..
                 }))),
diff --git a/src/farcasterd/syncer_state_machine.rs b/src/farcasterd/syncer_state_machine.rs
index 60566e7e..bcc0bc9c 100644
--- a/src/farcasterd/syncer_state_machine.rs
+++ b/src/farcasterd/syncer_state_machine.rs
@@ -4,12 +4,13 @@ use farcaster_core::blockchain::{Blockchain, Network};
 use crate::{
     bus::ctl::CtlMsg,
     bus::info::InfoMsg,
-    bus::sync::SyncMsg,
     bus::BusMsg,
+    bus::{sync::SyncMsg, AddressSecretKey, Failure, FailureCode},
     error::Error,
     event::{Event, StateMachine, StateMachineExecutor},
     syncerd::{
-        Event as SyncerEvent, Health, HealthCheck, SweepAddress, SweepAddressAddendum, Task, TaskId,
+        Event as SyncerEvent, GetAddressBalance, Health, HealthCheck, SweepAddress,
+        SweepAddressAddendum, Task, TaskId,
     },
     ServiceId,
 };
@@ -180,6 +181,47 @@ fn attempt_transition_to_awaiting_syncer_or_awaiting_syncer_request(
             }
         }
 
+        BusMsg::Ctl(CtlMsg::GetBalance(address_secret_key)) => {
+            let syncer_task_id = TaskId(runtime.syncer_task_counter);
+            runtime.syncer_task_counter += 1;
+            let (blockchain, network) = match &address_secret_key {
+                AddressSecretKey::Bitcoin { address, .. } => {
+                    (Blockchain::Bitcoin, address.network.into())
+                }
+                AddressSecretKey::Monero { address, .. } => {
+                    (Blockchain::Monero, address.network.into())
+                }
+            };
+            let syncer_task = Task::GetAddressBalance(GetAddressBalance {
+                id: syncer_task_id,
+                address_secret_key,
+            });
+            // check if a monero syncer is up
+            if let Some(service_id) = syncer_up(
+                &mut runtime.spawning_services,
+                &mut runtime.registered_services,
+                blockchain,
+                network,
+                &runtime.config,
+            )? {
+                event.complete_sync_service(service_id, SyncMsg::Task(syncer_task))?;
+                Ok(Some(SyncerStateMachine::AwaitingSyncerRequest(
+                    AwaitingSyncerRequest {
+                        source,
+                        syncer_task_id,
+                        syncer: ServiceId::Syncer(blockchain, network),
+                    },
+                )))
+            } else {
+                Ok(Some(SyncerStateMachine::AwaitingSyncer(AwaitingSyncer {
+                    source,
+                    syncer: ServiceId::Syncer(blockchain, network),
+                    syncer_task: syncer_task,
+                    syncer_task_id,
+                })))
+            }
+        }
+
         BusMsg::Ctl(CtlMsg::HealthCheck(blockchain, network)) => {
             let syncer_task_id = TaskId(runtime.syncer_task_counter);
             runtime.syncer_task_counter += 1;
@@ -325,6 +367,48 @@ fn attempt_transition_to_end(
             Ok(None)
         }
 
+        (BusMsg::Sync(SyncMsg::Event(SyncerEvent::AddressBalance(res))), syncer_id)
+            if syncer == syncer_id && res.id == syncer_task_id =>
+        {
+            if let Some(err) = res.err {
+                event.send_client_ctl(
+                    source,
+                    CtlMsg::Failure(Failure {
+                        code: FailureCode::Unknown,
+                        info: format!("Failed to get adddress balance {}", err),
+                    }),
+                )?;
+            } else {
+                event.send_client_info(
+                    source,
+                    InfoMsg::AddressBalance(crate::bus::info::AddressBalance {
+                        address: res.address,
+                        balance: res.balance,
+                    }),
+                )?;
+            }
+            runtime.registered_services = runtime
+                .registered_services
+                .clone()
+                .drain()
+                .filter(|service| {
+                    if let ServiceId::Syncer(..) = service {
+                        if !runtime.syncer_has_client(service) {
+                            info!("Terminating {}", service);
+                            event
+                                .send_ctl_service(service.clone(), CtlMsg::Terminate)
+                                .is_err()
+                        } else {
+                            true
+                        }
+                    } else {
+                        true
+                    }
+                })
+                .collect();
+            Ok(None)
+        }
+
         (BusMsg::Sync(SyncMsg::Event(SyncerEvent::HealthResult(res))), syncer_id)
             if syncer == syncer_id && res.id == syncer_task_id =>
         {
diff --git a/src/swapd/runtime.rs b/src/swapd/runtime.rs
index b69f5748..cb09cfb5 100644
--- a/src/swapd/runtime.rs
+++ b/src/swapd/runtime.rs
@@ -2553,6 +2553,10 @@ impl Runtime {
                         debug!("{}", event)
                     }
 
+                    Event::AddressBalance(event) => {
+                        debug!("{}", event)
+                    }
+
                     Event::FeeEstimation(FeeEstimation {
                         fee_estimations:
                             FeeEstimations::BitcoinFeeEstimation {
diff --git a/src/syncerd/bitcoin_syncer.rs b/src/syncerd/bitcoin_syncer.rs
index 45687ff2..5c7cbfa0 100644
--- a/src/syncerd/bitcoin_syncer.rs
+++ b/src/syncerd/bitcoin_syncer.rs
@@ -1,5 +1,6 @@
+use crate::bus::info::Address;
 use crate::bus::sync::{BridgeEvent, SyncMsg};
-use crate::bus::BusMsg;
+use crate::bus::{AddressSecretKey, BusMsg};
 use crate::error::SyncerError;
 use crate::syncerd::opts::Opts;
 use crate::syncerd::runtime::SyncerdTask;
@@ -11,10 +12,11 @@ use crate::syncerd::BtcAddressAddendum;
 use crate::syncerd::Event;
 use crate::syncerd::FeeEstimations;
 use crate::syncerd::GetTx;
+use crate::syncerd::Health;
 use crate::syncerd::TaskTarget;
 use crate::syncerd::TransactionBroadcasted;
 use crate::syncerd::TransactionRetrieved;
-use crate::syncerd::{BroadcastTransaction, Health};
+use crate::syncerd::{AddressBalance, BroadcastTransaction};
 use crate::{error::Error, syncerd::syncer_state::create_set};
 use crate::{LogStyle, ServiceId};
 use bitcoin::hashes::{hex::ToHex, Hash};
@@ -26,6 +28,7 @@ use electrum_client::{
 use farcaster_core::bitcoin::segwitv0::signature_hash;
 use farcaster_core::bitcoin::transaction::TxInRef;
 use farcaster_core::blockchain::{Blockchain, Network};
+use hex;
 use internet2::session::LocalSession;
 use internet2::zeromq::ZmqSocketType;
 use internet2::SendRecvMessage;
@@ -38,8 +41,7 @@ use tokio::sync::mpsc::Receiver as TokioReceiver;
 use tokio::sync::mpsc::Sender as TokioSender;
 use tokio::sync::Mutex;
 
-use hex;
-
+use super::GetAddressBalance;
 use super::HealthCheck;
 
 const RETRY_TIMEOUT: u64 = 5;
@@ -530,6 +532,7 @@ async fn run_syncerd_task_receiver(
     state: Arc<Mutex<SyncerState>>,
     transaction_broadcast_tx: TokioSender<(BroadcastTransaction, ServiceId)>,
     transaction_get_tx: TokioSender<(GetTx, ServiceId)>,
+    balance_get_tx: TokioSender<(GetAddressBalance, ServiceId)>,
     terminate_tx: TokioSender<()>,
 ) {
     tokio::spawn(async move {
@@ -545,6 +548,12 @@ async fn run_syncerd_task_receiver(
                                 .await
                                 .expect("failed on transaction_get sender");
                         }
+                        Task::GetAddressBalance(task) => {
+                            balance_get_tx
+                                .send((task, syncerd_task.source))
+                                .await
+                                .expect("failed on balance_get sender");
+                        }
                         Task::WatchEstimateFee(task) => {
                             let mut state_guard = state.lock().await;
                             state_guard.estimate_fee(task, syncerd_task.source).await;
@@ -1099,7 +1108,79 @@ fn transaction_fetcher(
                         })
                         .await
                         .expect("error sending transaction retrieved event");
-                    debug!("failed to retrieved tx: {:?}", e);
+                    debug!("failed to retrieve tx: {:?}", e);
+                }
+            }
+        }
+    })
+}
+
+fn balance_fetcher(
+    electrum_server: String,
+    proxy_address: Option<String>,
+    mut balance_get_rx: TokioReceiver<(GetAddressBalance, ServiceId)>,
+    tx_event: TokioSender<BridgeEvent>,
+) -> tokio::task::JoinHandle<()> {
+    tokio::task::spawn(async move {
+        while let Some((get_balance, source)) = balance_get_rx.recv().await {
+            let address = match get_balance.address_secret_key {
+                AddressSecretKey::Monero { address, .. } => {
+                    tx_event
+                        .send(BridgeEvent {
+                            event: Event::AddressBalance(AddressBalance {
+                                address: Address::Monero(address),
+                                id: get_balance.id,
+                                balance: 0,
+                                err: Some(
+                                    "Sent monero address balance to bitcoin syncer".to_string(),
+                                ),
+                            }),
+                            source,
+                        })
+                        .await
+                        .expect("error sending address balance event");
+                    warn!("Received monero address balance task in bitcoin syncer");
+                    continue;
+                }
+                AddressSecretKey::Bitcoin { address, .. } => address,
+            };
+
+            debug!("creating balance fetcher electrum client");
+
+            match create_electrum_client(&electrum_server, proxy_address.clone()).and_then(
+                |transaction_client| {
+                    transaction_client.script_get_balance(&address.script_pubkey())
+                },
+            ) {
+                Ok(balance) => {
+                    tx_event
+                        .send(BridgeEvent {
+                            event: Event::AddressBalance(AddressBalance {
+                                id: get_balance.id,
+                                address: Address::Bitcoin(address),
+                                balance: balance.unconfirmed.unsigned_abs(),
+                                err: None,
+                            }),
+                            source,
+                        })
+                        .await
+                        .expect("error sending address balance event");
+                    debug!("successfully retrieved address balance: {:?}", balance);
+                }
+                Err(e) => {
+                    tx_event
+                        .send(BridgeEvent {
+                            event: Event::AddressBalance(AddressBalance {
+                                id: get_balance.id,
+                                address: Address::Bitcoin(address),
+                                balance: 0,
+                                err: Some(e.to_string()),
+                            }),
+                            source,
+                        })
+                        .await
+                        .expect("error sending address balance event");
+                    debug!("failed to retrieve address balance: {}", e);
                 }
             }
         }
@@ -1153,6 +1234,10 @@ impl Synclet for BitcoinSyncer {
                         TokioSender<(GetTx, ServiceId)>,
                         TokioReceiver<(GetTx, ServiceId)>,
                     ) = tokio::sync::mpsc::channel(200);
+                    let (balance_get_tx, balance_get_rx): (
+                        TokioSender<(GetAddressBalance, ServiceId)>,
+                        TokioReceiver<(GetAddressBalance, ServiceId)>,
+                    ) = tokio::sync::mpsc::channel(200);
                     let (terminate_tx, terminate_rx): (TokioSender<()>, TokioReceiver<()>) =
                         tokio::sync::mpsc::channel(1);
                     let state = Arc::new(Mutex::new(SyncerState::new(
@@ -1167,6 +1252,7 @@ impl Synclet for BitcoinSyncer {
                         Arc::clone(&state),
                         transaction_broadcast_tx.clone(),
                         transaction_get_tx,
+                        balance_get_tx,
                         terminate_tx,
                     )
                     .await;
@@ -1205,6 +1291,13 @@ impl Synclet for BitcoinSyncer {
                         event_tx.clone(),
                     );
 
+                    let balance_get_handle = balance_fetcher(
+                        electrum_server.clone(),
+                        proxy_address.clone(),
+                        balance_get_rx,
+                        event_tx.clone(),
+                    );
+
                     let estimate_fee_handle = estimate_fee_polling(
                         electrum_server.clone(),
                         proxy_address.clone(),
@@ -1226,6 +1319,7 @@ impl Synclet for BitcoinSyncer {
                         unseen_transaction_handle,
                         transaction_broadcast_handle,
                         transaction_get_handle,
+                        balance_get_handle,
                         estimate_fee_handle,
                         sweep_handle,
                         terminate_handle,
diff --git a/src/syncerd/monero_syncer.rs b/src/syncerd/monero_syncer.rs
index 5fb0b57e..1a24cef3 100644
--- a/src/syncerd/monero_syncer.rs
+++ b/src/syncerd/monero_syncer.rs
@@ -1,5 +1,6 @@
+use crate::bus::info::Address;
 use crate::bus::sync::{BridgeEvent, SyncMsg};
-use crate::bus::BusMsg;
+use crate::bus::{AddressSecretKey, BusMsg};
 use crate::error::{Error, SyncerError};
 use crate::service::LogStyle;
 use crate::syncerd::opts::Opts;
@@ -9,16 +10,19 @@ use crate::syncerd::syncer_state::create_set;
 use crate::syncerd::syncer_state::AddressTx;
 use crate::syncerd::syncer_state::SyncerState;
 use crate::syncerd::types::{AddressAddendum, Boolean, SweepAddressAddendum, Task};
+use crate::syncerd::AddressBalance;
 use crate::syncerd::TaskTarget;
 use crate::syncerd::TransactionBroadcasted;
 use crate::syncerd::XmrAddressAddendum;
 use crate::syncerd::{Event, Health};
+use crate::ServiceId;
 use farcaster_core::blockchain::{Blockchain, Network};
+use hex;
 use internet2::session::LocalSession;
 use internet2::zeromq::ZmqSocketType;
 use internet2::SendRecvMessage;
 use internet2::TypedEnum;
-use monero::Hash;
+use monero::{Hash, PrivateKey};
 use monero_rpc::{
     GenerateFromKeysArgs, GetBlockHeaderSelector, GetTransfersCategory, GetTransfersSelector,
     PrivateKeyType,
@@ -32,8 +36,7 @@ use tokio::sync::mpsc::Receiver as TokioReceiver;
 use tokio::sync::mpsc::Sender as TokioSender;
 use tokio::sync::Mutex;
 
-use hex;
-
+use super::GetAddressBalance;
 use super::HealthCheck;
 
 #[derive(Debug, Clone)]
@@ -385,6 +388,7 @@ async fn run_syncerd_task_receiver(
     syncer_servers: MoneroSyncerServers,
     receive_task_channel: Receiver<SyncerdTask>,
     state: Arc<Mutex<SyncerState>>,
+    balance_get_tx: TokioSender<(GetAddressBalance, ServiceId)>,
     tx_event: TokioSender<BridgeEvent>,
 ) {
     tokio::spawn(async move {
@@ -397,6 +401,12 @@ async fn run_syncerd_task_receiver(
                         Task::GetTx(_) => {
                             error!("get tx not implemented for monero syncer");
                         }
+                        Task::GetAddressBalance(task) => {
+                            balance_get_tx
+                                .send((task, syncerd_task.source))
+                                .await
+                                .expect("failed on balance_get sender");
+                        }
                         Task::WatchEstimateFee(_) => {
                             error!("estimate fee not implemented for monero syncer");
                         }
@@ -791,6 +801,133 @@ async fn run_syncerd_bridge_event_sender(
     });
 }
 
+async fn fetch_balance(
+    wallet_mutex: Arc<Mutex<monero_rpc::WalletClient>>,
+    wallet_dir_path: Option<PathBuf>,
+    address: monero::Address,
+    viewkey: PrivateKey,
+) -> Result<monero::Amount, Error> {
+    let wallet_filename = format!("balance:{}", address);
+    let password = s!(" ");
+    debug!("creating balance fetcher wallet client");
+    let wallet = wallet_mutex.lock().await;
+    trace!("taking balance wallet lock");
+
+    while let Err(err) = wallet
+        .open_wallet(wallet_filename.clone(), Some(password.clone()))
+        .await
+    {
+        debug!(
+            "error opening to be sweeped wallet: {:?}, falling back to generating a new wallet",
+            err,
+        );
+        wallet
+            .generate_from_keys(GenerateFromKeysArgs {
+                restore_height: Some(0), // fixme
+                filename: wallet_filename.clone(),
+                address,
+                spendkey: None,
+                viewkey,
+                password: password.clone(),
+                autosave_current: Some(true),
+            })
+            .await?;
+    }
+
+    let (account, addrs) = (0, None);
+    wallet.refresh(Some(0)).await?; // fixme
+    let balance = wallet.get_balance(account, addrs).await?;
+
+    if let Some(path) = wallet_dir_path {
+        if let Some(raw_path) = path.to_str() {
+            if let Err(err) = fs::remove_file(raw_path.to_string() + &wallet_filename).and(
+                fs::remove_file(raw_path.to_string() + &wallet_filename + ".keys"),
+            ) {
+                warn!("Failed to clean balance wallet data after successful balance call. {}. The path used for the wallet directory is probably malformed", err);
+            } else {
+                info!("Successfully removed sweep wallet data after completed sweep.");
+            }
+        }
+    }
+
+    Ok(balance.balance)
+}
+
+fn balance_fetcher(
+    wallet_mutex: Arc<Mutex<monero_rpc::WalletClient>>,
+    wallet_dir_path: Option<PathBuf>,
+    mut balance_get_rx: TokioReceiver<(GetAddressBalance, ServiceId)>,
+    tx_event: TokioSender<BridgeEvent>,
+) -> tokio::task::JoinHandle<()> {
+    tokio::task::spawn(async move {
+        while let Some((get_balance, source)) = balance_get_rx.recv().await {
+            match get_balance.address_secret_key {
+                AddressSecretKey::Monero {
+                    address,
+                    secret_key_info,
+                } => {
+                    match fetch_balance(
+                        Arc::clone(&wallet_mutex),
+                        wallet_dir_path.clone(),
+                        address,
+                        secret_key_info.view,
+                    )
+                    .await
+                    {
+                        Ok(balance) => {
+                            tx_event
+                                .send(BridgeEvent {
+                                    event: Event::AddressBalance(AddressBalance {
+                                        id: get_balance.id,
+                                        address: Address::Monero(address),
+                                        balance: balance.as_pico(),
+                                        err: None,
+                                    }),
+                                    source,
+                                })
+                                .await
+                                .expect("error sending address balance event");
+                        }
+                        Err(e) => {
+                            tx_event
+                                .send(BridgeEvent {
+                                    event: Event::AddressBalance(AddressBalance {
+                                        id: get_balance.id,
+                                        address: Address::Monero(address),
+                                        balance: 0,
+                                        err: Some(e.to_string()),
+                                    }),
+                                    source,
+                                })
+                                .await
+                                .expect("error sending address balance event");
+                            debug!("failed to retrieve address balance: {}", e);
+                        }
+                    }
+                }
+                AddressSecretKey::Bitcoin { address, .. } => {
+                    tx_event
+                        .send(BridgeEvent {
+                            event: Event::AddressBalance(AddressBalance {
+                                address: Address::Bitcoin(address),
+                                id: get_balance.id,
+                                balance: 0,
+                                err: Some(
+                                    "Sent monero address balance to bitcoin syncer".to_string(),
+                                ),
+                            }),
+                            source,
+                        })
+                        .await
+                        .expect("error sending address balance event");
+                    warn!("Received monero address balance task in bitcoin syncer");
+                    continue;
+                }
+            };
+        }
+    })
+}
+
 /// Specific Monero configuration
 #[derive(Default, Debug, Clone, Eq, PartialEq, Hash)]
 pub struct MoneroSyncerServers {
@@ -836,6 +973,11 @@ impl Synclet for MoneroSyncer {
                             monero_rpc::RpcClient::new(syncer_servers.monero_rpc_wallet.clone())
                                 .wallet(),
                         ));
+                        let (balance_get_tx, balance_get_rx): (
+                            TokioSender<(GetAddressBalance, ServiceId)>,
+                            TokioReceiver<(GetAddressBalance, ServiceId)>,
+                        ) = tokio::sync::mpsc::channel(200);
+
                         let (event_tx, event_rx): (
                             TokioSender<BridgeEvent>,
                             TokioReceiver<BridgeEvent>,
@@ -849,6 +991,7 @@ impl Synclet for MoneroSyncer {
                             syncer_servers.clone(),
                             receive_task_channel,
                             Arc::clone(&state),
+                            balance_get_tx,
                             event_tx.clone(),
                         )
                         .await;
@@ -872,14 +1015,22 @@ impl Synclet for MoneroSyncer {
                             Arc::clone(&state),
                             Arc::clone(&wallet_mutex),
                             network,
+                            wallet_dir.clone(),
+                        );
+
+                        let balance_handle = balance_fetcher(
+                            Arc::clone(&wallet_mutex),
                             wallet_dir,
+                            balance_get_rx,
+                            event_tx,
                         );
 
                         let res = tokio::try_join!(
                             address_handle,
                             height_handle,
                             unseen_transaction_handle,
-                            sweep_handle
+                            sweep_handle,
+                            balance_handle,
                         );
                         debug!("exiting monero synclet run routine with: {:?}", res);
                     });
diff --git a/src/syncerd/types.rs b/src/syncerd/types.rs
index db4ff9ea..5ae9453a 100644
--- a/src/syncerd/types.rs
+++ b/src/syncerd/types.rs
@@ -2,6 +2,8 @@
 use serde_with::DisplayFromStr;
 use strict_encoding::{StrictDecode, StrictEncode};
 
+use crate::bus::{info::Address, AddressSecretKey};
+
 // The strict encoding length limit
 pub const STRICT_ENCODE_MAX_ITEMS: u16 = u16::MAX - 1;
 
@@ -252,6 +254,18 @@ pub struct HealthCheck {
     pub id: TaskId,
 }
 
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[cfg_attr(
+    feature = "serde",
+    derive(Serialize, Deserialize),
+    serde(crate = "serde_crate")
+)]
+#[display(Debug)]
+pub struct GetAddressBalance {
+    pub id: TaskId,
+    pub address_secret_key: AddressSecretKey,
+}
+
 /// Tasks created by the daemon and handle by syncers to process a blockchain
 /// and generate [`Event`] back to the syncer.
 #[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
@@ -269,6 +283,7 @@ pub enum Task {
     BroadcastTransaction(BroadcastTransaction),
     SweepAddress(SweepAddress),
     GetTx(GetTx),
+    GetAddressBalance(GetAddressBalance),
     WatchEstimateFee(WatchEstimateFee),
     HealthCheck(HealthCheck),
     Terminate,
@@ -374,6 +389,16 @@ pub enum Health {
     ConfigUnavailable(String),
 }
 
+#[derive(Clone, Debug, Display, StrictEncode, StrictDecode, Eq, PartialEq, Hash)]
+#[display(Debug)]
+// the sats per kvB is because we need u64 for Eq, PartialEq and Hash
+pub struct AddressBalance {
+    pub id: TaskId,
+    pub address: Address,
+    pub balance: u64,
+    pub err: Option<String>,
+}
+
 /// Events returned by syncers to the daemon to update the blockchain states.
 /// Events are identified with a unique 32-bits integer that match the [`Task`]
 /// id.
@@ -394,4 +419,5 @@ pub enum Event {
     /// Empty event to signify that a task with a certain id has not produced an initial result
     Empty(TaskId),
     HealthResult(HealthResult),
+    AddressBalance(AddressBalance),
 }


Test patch:
diff --git a/tests/bitcoin.rs b/tests/bitcoin.rs
index 2dbdba63..91988b63 100644
--- a/tests/bitcoin.rs
+++ b/tests/bitcoin.rs
@@ -3,6 +3,7 @@ use bitcoin::hashes::Hash;
 use bitcoincore_rpc::RpcApi;
 use clap::Parser;
 use farcaster_core::blockchain::{Blockchain, Network};
+use farcaster_node::bus::{AddressSecretKey, BitcoinSecretKeyInfo};
 use farcaster_node::syncerd::bitcoin_syncer::BitcoinSyncer;
 use farcaster_node::syncerd::opts::Opts;
 use farcaster_node::syncerd::runtime::SyncerdTask;
@@ -10,8 +11,8 @@ use farcaster_node::syncerd::types::{
     Abort, AddressAddendum, Boolean, BroadcastTransaction, BtcAddressAddendum, GetTx, SweepAddress,
     SweepAddressAddendum, Task, WatchAddress, WatchEstimateFee, WatchHeight, WatchTransaction,
 };
-use farcaster_node::syncerd::SweepBitcoinAddress;
 use farcaster_node::syncerd::{runtime::Synclet, TaskId, TaskTarget};
+use farcaster_node::syncerd::{GetAddressBalance, SweepBitcoinAddress};
 use farcaster_node::ServiceId;
 use microservices::ZMQ_CONTEXT;
 use ntest::timeout;
@@ -1063,6 +1064,50 @@ fn bitcoin_syncer_sweep_address_test() {
     std::thread::sleep(duration);
 }
 
+#[test]
+#[timeout(600000)]
+#[ignore]
+fn bitcoin_syncer_address_balance_test() {
+    setup_logging();
+    let bitcoin_rpc = bitcoin_setup();
+    let (tx, rx_event) = create_bitcoin_syncer("sweep");
+
+    let target_address = bitcoin_rpc.get_new_address(None, None).unwrap();
+    let target_secret_key = bitcoin_rpc.dump_private_key(&target_address).unwrap().inner;
+
+    // 294 Satoshi is the dust limit for a segwit transaction
+    let amount = bitcoin::Amount::ONE_SAT * 1000;
+    // send some coins to target_address
+    let _txid = bitcoin_rpc
+        .send_to_address(&target_address, amount, None, None, None, None, None, None)
+        .unwrap();
+
+    // allow some time for things to happen, like the electrum server catching up
+    let duration = std::time::Duration::from_secs(10);
+    std::thread::sleep(duration);
+
+    let task = SyncerdTask {
+        task: Task::GetAddressBalance(GetAddressBalance {
+            id: TaskId(0),
+            address_secret_key: AddressSecretKey::Bitcoin {
+                address: target_address,
+                secret_key_info: BitcoinSecretKeyInfo {
+                    swap_id: None,
+                    secret_key: target_secret_key,
+                },
+            },
+        }),
+        source: SOURCE1.clone(),
+    };
+    tx.send(task).unwrap();
+    // allow some time for things to happen, like the electrum server catching up
+    info!("waiting for address balance message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    info!("received address balance message");
+    let request = misc::get_request_from_message(message);
+    assert::address_balance(request, amount.as_sat());
+}
+
 // =========================
 // TODO: move into utils from here
 //
diff --git a/tests/monero.rs b/tests/monero.rs
index b7a7f182..dcf40e3d 100644
--- a/tests/monero.rs
+++ b/tests/monero.rs
@@ -1,5 +1,6 @@
 use clap::Parser;
 use farcaster_core::blockchain::{Blockchain, Network};
+use farcaster_node::bus::{AddressSecretKey, MoneroSecretKeyInfo};
 use farcaster_node::syncerd::monero_syncer::MoneroSyncer;
 use farcaster_node::syncerd::opts::Opts;
 use farcaster_node::syncerd::runtime::SyncerdTask;
@@ -7,6 +8,7 @@ use farcaster_node::syncerd::types::{
     Abort, AddressAddendum, Boolean, BroadcastTransaction, Task, WatchAddress, WatchHeight,
     WatchTransaction,
 };
+use farcaster_node::syncerd::GetAddressBalance;
 use farcaster_node::syncerd::{
     runtime::Synclet, SweepAddress, SweepAddressAddendum, SweepMoneroAddress, TaskId, TaskTarget,
     XmrAddressAddendum,
@@ -195,6 +197,69 @@ async fn monero_syncer_sweep_test() {
     assert!(rx_event.recv_multipart(1).is_err());
 }
 
+#[tokio::test]
+#[timeout(300000)]
+#[ignore]
+async fn monero_syncer_address_balance_test() {
+    setup_logging();
+    let (regtest, wallet) = setup_monero().await;
+    let address = wallet.get_address(0, None).await.unwrap();
+    let height = regtest.generate_blocks(200, address.address).await.unwrap();
+
+    let duration = std::time::Duration::from_secs(20);
+    std::thread::sleep(duration);
+
+    let (tx, rx_event) = create_monero_syncer("sweep", false);
+
+    let target_spend_key = monero::PrivateKey::from_str(
+        "8163466f1883598e6dd14027b8da727057165da91485834314f5500a65846f09",
+    )
+    .unwrap();
+    let target_view_key = monero::PrivateKey::from_str(
+        "8163466f1883598e6dd14027b8da727057165da91485834314f5500a65846f09",
+    )
+    .unwrap();
+    let keypair = monero::KeyPair {
+        view: target_view_key,
+        spend: target_spend_key,
+    };
+    let target_address = monero::Address::from_keypair(monero::Network::Mainnet, &keypair);
+    let amount = 500000000000;
+    send_monero(&wallet, target_address, amount).await;
+
+    let task = SyncerdTask {
+        task: Task::GetAddressBalance(GetAddressBalance {
+            id: TaskId(0),
+            address_secret_key: AddressSecretKey::Monero {
+                address: target_address,
+                secret_key_info: MoneroSecretKeyInfo {
+                    swap_id: None,
+                    spend: target_spend_key,
+                    view: target_view_key,
+                    creation_height: Some(height.height),
+                },
+            },
+        }),
+        source: SOURCE2.clone(),
+    };
+    tx.send(task).unwrap();
+
+    // the minimum amount is not reached, so let it mature and later check that no sweep has been executed
+    regtest.generate_blocks(20, address.address).await.unwrap();
+    let duration = std::time::Duration::from_secs(20);
+    std::thread::sleep(duration);
+
+    info!("waiting for address balance message");
+    let message = rx_event.recv_multipart(0).unwrap();
+    info!("received address balance message");
+    let request = misc::get_request_from_message(message);
+
+    assert::address_balance_min(request, amount);
+
+    // check that only a single sweep message has been received
+    assert!(rx_event.recv_multipart(1).is_err());
+}
+
 /*
 We test for the following scenarios in the address transaction tests:
 
diff --git a/tests/utils/assert.rs b/tests/utils/assert.rs
index 66a187d1..7185c5d4 100644
--- a/tests/utils/assert.rs
+++ b/tests/utils/assert.rs
@@ -136,6 +136,38 @@ pub fn transaction_received(request: BusMsg, expected_txid: bitcoin::Txid) {
     }
 }
 
+pub fn address_balance(request: BusMsg, expected_balance: u64) {
+    match request {
+        BusMsg::Sync(SyncMsg::BridgeEvent(event)) => match event.event {
+            Event::AddressBalance(address_balance) => {
+                assert_eq!(address_balance.balance, expected_balance);
+            }
+            _ => {
+                panic!("expected address balance event");
+            }
+        },
+        _ => {
+            panic!("expected syncerd bridge event");
+        }
+    }
+}
+
+pub fn address_balance_min(request: BusMsg, min_balance: u64) {
+    match request {
+        BusMsg::Sync(SyncMsg::BridgeEvent(event)) => match event.event {
+            Event::AddressBalance(address_balance) => {
+                assert!(address_balance.balance >= min_balance);
+            }
+            _ => {
+                panic!("expected address balance event");
+            }
+        },
+        _ => {
+            panic!("expected syncerd bridge event");
+        }
+    }
+}
+
 pub fn fee_estimation_received(request: BusMsg) {
     match request {
         BusMsg::Sync(SyncMsg::BridgeEvent(BridgeEvent {