Add a electrum syncing example

The electrum sync is done in two steps.
 - `wallet_txid_scan` returns a candidate `KeychainScan` with TxGraph
 data.
 - The candidate is compared with tracker's `SparseChain`, and list of
 new tx data to fetch is derived from the `ChangeSet`.
 - New transactions are fetched in batch and added into the `KeychainScan`
 update.
 - Apply the final `KeychainScan` on tracker and DB.

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "bdk_core",
     "bdk_cli_lib",
     "bdk_esplora_example",
+    "bdk_electrum_example",
     "bdk_keychain",
     "bdk_file_store"
 ]

bdk_electrum_example/.gitignore

@@ -0,0 +1 @@
+/target

bdk_electrum_example/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "bdk_electrum_example"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[dependencies]
+# BDK Core
+bdk_core = { path = "../bdk_core", features = ["serde"] }
+bdk_cli = { path = "../bdk_cli_lib"}
+bdk_keychain = { path = "../bdk_keychain" }
+
+# Electrum
+electrum-client = { version = "0.12" }
+
+# Auxiliary
+log = "0.4"
+env_logger = "0.7"

bdk_electrum_example/src/electrum.rs

@@ -0,0 +1,150 @@
+use std::{collections::BTreeMap, ops::Deref};
+
+use bdk_cli::{anyhow::Result, Broadcast};
+use bdk_core::{
+    bitcoin::{Script, Txid},
+    sparse_chain::SparseChain,
+    BlockId, TxHeight,
+};
+use electrum_client::{Client, Config, ElectrumApi};
+
+pub struct ElectrumClient {
+    client: Client,
+}
+
+impl ElectrumClient {
+    pub fn new(url: &str) -> Result<Self> {
+        let client = Client::from_config(url, Config::default())?;
+        Ok(Self { client })
+    }
+}
+
+impl Deref for ElectrumClient {
+    type Target = Client;
+    fn deref(&self) -> &Self::Target {
+        &self.client
+    }
+}
+
+impl Broadcast for ElectrumClient {
+    type Error = electrum_client::Error;
+    fn broadcast(&self, tx: &bdk_core::bitcoin::Transaction) -> Result<(), Self::Error> {
+        let _ = self.client.transaction_broadcast(tx)?;
+        Ok(())
+    }
+}
+
+/// We will detect reorg if it has depth less than this
+const REORG_DETECTION_DEPTH: u32 = 100;
+const DEFAULT_STOP_GAP: usize = 10;
+
+impl ElectrumClient {
+    /// Fetch latest block height.
+    pub fn get_height(&self) -> Result<u32> {
+        // TODO: unsubscribe when added to the client, or is there a better call to use here?
+        Ok(self
+            .client
+            .block_headers_subscribe()
+            .map(|data| data.height as u32)?)
+    }
+
+    /// Scan for a list of scripts, and create an initial [`ChainGraph`] candidate update.
+    /// This update will only contain [`Txid`]s in SparseChain, and no actual transaction data.
+    ///
+    /// User needs to fetch the required transaction data and update the [`ChainGraph`] before applying it.
+    pub fn spk_txid_scan(
+        &self,
+        spks: impl Iterator<Item = Script>,
+        last_known_height: Option<u32>,
+    ) -> Result<SparseChain> {
+        let mut dummy_keychains = BTreeMap::new();
+        dummy_keychains.insert((), spks.enumerate().map(|(i, spk)| (i as u32, spk)));
+
+        Ok(self
+            .wallet_txid_scan(dummy_keychains, None, last_known_height)?
+            .0)
+    }
+
+    /// Scan for a keychain tracker, and create an initial [`KeychainScan`] candidate update.
+    /// This update will only contain [`Txid`]s in SparseChain, and no actual transaction data.
+    ///
+    /// User needs to fetch the required transaction data and update the [`KeychainScan`] before applying it.
+    pub fn wallet_txid_scan<K: Ord + Clone>(
+        &self,
+        scripts: BTreeMap<K, impl Iterator<Item = (u32, Script)>>,
+        stop_gap: Option<usize>,
+        last_known_height: Option<u32>,
+    ) -> Result<(SparseChain, BTreeMap<K, u32>)> {
+        let mut sparse_chain = SparseChain::default();
+
+        // 1. Create checkpoint data from last_known_height - REORG_DEPTH
+        // Reorg situation will he handled at the time of applying this KeychainScan.
+        // If there's reorg deeper than the assumed depth, update process will throw error.
+        let current_height = self.get_height()?;
+        let check_from = last_known_height
+            .map(|ht| ht.saturating_sub(REORG_DETECTION_DEPTH))
+            .unwrap_or(0);
+        let required_block_count = (current_height - check_from) + 1;
+        let headers = self.block_headers(check_from as usize, required_block_count as usize)?;
+
+        let block_ids =
+            (check_from..=current_height)
+                .zip(headers.headers)
+                .map(|(height, header)| BlockId {
+                    height,
+                    hash: header.block_hash(),
+                });
+
+        for block in block_ids {
+            sparse_chain.insert_checkpoint(block)?;
+        }
+
+        let mut keychain_index_update = BTreeMap::new();
+
+        // 2. Fetch Keychain's last_active_index and all related txids.
+        // Add them into the KeyChainScan
+        for (keychain, mut scripts) in scripts.into_iter() {
+            let mut last_active_index = 0;
+            let mut unused_script_count = 0usize;
+            let mut script_history_txid = Vec::<(Txid, TxHeight)>::new();
+
+            loop {
+                let (index, script) = scripts.next().expect("its an infinite iterator");
+
+                let history = self
+                    .script_get_history(&script)?
+                    .iter()
+                    .map(|history_result| {
+                        if history_result.height > 0 {
+                            return (
+                                history_result.tx_hash,
+                                TxHeight::Confirmed(history_result.height as u32),
+                            );
+                        } else {
+                            return (history_result.tx_hash, TxHeight::Unconfirmed);
+                        };
+                    })
+                    .collect::<Vec<(Txid, TxHeight)>>();
+
+                if history.is_empty() {
+                    unused_script_count += 1;
+                } else {
+                    last_active_index = index;
+                    script_history_txid.extend(history.iter());
+                }
+
+                if unused_script_count >= stop_gap.unwrap_or(DEFAULT_STOP_GAP) {
+                    break;
+                }
+            }
+
+            for (txid, index) in script_history_txid {
+                sparse_chain.insert_tx(txid, index)?;
+            }
+
+            keychain_index_update.insert(keychain, last_active_index);
+        }
+
+        Ok((sparse_chain, keychain_index_update))
+    }
+}

bdk_electrum_example/src/main.rs

@@ -0,0 +1,169 @@
+mod electrum;
+
+use bdk_core::BlockId;
+use bdk_keychain::KeychainScan;
+use electrum::ElectrumClient;
+
+use bdk_cli::{
+    anyhow::{self, Context},
+    handle_commands, Commands,
+};
+use log::debug;
+
+use electrum_client::ElectrumApi;
+
+fn main() -> anyhow::Result<()> {
+    let (args, keymap, mut tracker, mut db) = bdk_cli::init()?;
+
+    let client = ElectrumClient::new("ssl://electrum.blockstream.info:60002")?;
+
+    match args.command {
+        Commands::Scan { stop_gap } => {
+            let last_known_height = tracker
+                .chain()
+                .checkpoints()
+                .iter()
+                .last()
+                .map(|(&ht, _)| ht);
+
+            let scripts = tracker.txout_index.iter_all_script_pubkeys_by_keychain();
+
+            let mut scan_update = KeychainScan::default();
+
+            // Wallet scan returns a sparse chain that contains new All the BlockIds and Txids
+            // relevant to the wallet, along with keychain index update if required.
+            let (new_sparsechain, keychain_index_update) =
+                client.wallet_txid_scan(scripts, Some(stop_gap), last_known_height)?;
+
+            // Set the last active indexes in update struct
+            scan_update.last_active_indexes = keychain_index_update;
+
+            // Inserting everything from the new_sparsechain should be okay as duplicate
+            // data would be rejected at the time of update application.
+            for (height, hash) in new_sparsechain.checkpoints() {
+                let _ = scan_update.update.insert_checkpoint(BlockId {
+                    height: *height,
+                    hash: *hash,
+                })?;
+            }
+
+            // New Txids that we don't know the full data of.
+            let txid_changeset = tracker
+                .chain()
+                .determine_changeset(&new_sparsechain)?
+                .0
+                .txids;
+
+            let new_transactions =
+                client.batch_transaction_get(txid_changeset.iter().map(|(txid, _)| txid))?;
+
+            for (tx, index) in new_transactions
+                .into_iter()
+                .zip(txid_changeset.into_iter().map(|(_, index)| index))
+            {
+                scan_update.update.insert_tx(tx, index)?;
+            }
+
+            // Apply the full scan update
+            let changeset = tracker.determine_changeset(&scan_update)?;
+            db.append_changeset(&changeset)?;
+            tracker.apply_changeset(changeset);
+            debug!("sync completed!!")
+        }
+        Commands::Sync {
+            mut unused,
+            mut unspent,
+            all,
+        } => {
+            let txout_index = &tracker.txout_index;
+            if !(all || unused || unspent) {
+                unused = true;
+                unspent = true;
+            } else if all {
+                unused = false;
+                unspent = false
+            }
+            let mut spks: Box<dyn Iterator<Item = bdk_core::bitcoin::Script>> =
+                Box::new(core::iter::empty());
+            if unused {
+                spks = Box::new(spks.chain(txout_index.iter_unused().map(|(index, script)| {
+                    eprintln!("Checking if address at {:?} has been used", index);
+                    script.clone()
+                })));
+            }
+
+            if all {
+                spks = Box::new(spks.chain(txout_index.script_pubkeys().iter().map(
+                    |(index, script)| {
+                        eprintln!("scanning {:?}", index);
+                        script.clone()
+                    },
+                )));
+            }
+
+            if unspent {
+                spks = Box::new(spks.chain(tracker.utxos().map(|(_index, ftxout)| {
+                    eprintln!("checking if {} has been spent", ftxout.outpoint);
+                    ftxout.txout.script_pubkey
+                })));
+            }
+
+            let last_known_height = tracker
+                .chain()
+                .checkpoints()
+                .iter()
+                .last()
+                .map(|(&ht, _)| ht);
+
+            let mut scan_update = KeychainScan::default();
+
+            // Wallet scan returns a sparse chain that contains new All the BlockIds and Txids
+            // relevant to the wallet, along with keychain index update if required.
+            let new_sparsechain = client
+                .spk_txid_scan(spks, last_known_height)
+                .context("scanning the blockchain")?;
+
+            // Inserting everything from the new_sparsechain should be okay as duplicate
+            // data would be rejected at the time of update application.
+            for (height, hash) in new_sparsechain.checkpoints() {
+                let _ = scan_update.update.insert_checkpoint(BlockId {
+                    height: *height,
+                    hash: *hash,
+                })?;
+            }
+
+            // New Txids that we don't know the full data of.
+            let txid_changeset = tracker
+                .chain()
+                .determine_changeset(&new_sparsechain)?
+                .0
+                .txids;
+
+            let new_transactions =
+                client.batch_transaction_get(txid_changeset.iter().map(|(txid, _)| txid))?;
+
+            for (tx, index) in new_transactions
+                .into_iter()
+                .zip(txid_changeset.into_iter().map(|(_, index)| index))
+            {
+                scan_update.update.insert_tx(tx, index)?;
+            }
+
+            // Apply the full scan update
+            let changeset = tracker.determine_changeset(&scan_update)?;
+            db.append_changeset(&changeset)?;
+            tracker.apply_changeset(changeset);
+            debug!("sync completed!!")
+        }
+        // For everything else run handler
+        _ => handle_commands(
+            args.command,
+            client,
+            &mut tracker,
+            &mut db,
+            args.network,
+            &keymap,
+        )?,
+    }
+    Ok(())
+}