Add a electrum syncing example

The electrum sync is done in two steps.
 - `wallet_txid_scan` returns a candidate `SparseChain` containing all
 the related txids and a `(keychain, last_known_index)` map.
 - The candidate `SparseChain` 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,151 @@
+use std::{collections::BTreeMap, ops::Deref};
+
+use bdk_cli::{anyhow::Result, Broadcast};
+use bdk_core::{
+    bitcoin::{BlockHash, 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(())
+    }
+}
+
+impl ElectrumClient {
+    /// Fetch latest block height.
+    pub fn get_tip(&self) -> Result<(u32, BlockHash)> {
+        // 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, data.header.block_hash()))?)
+    }
+
+    /// 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>,
+        local_chain: &BTreeMap<u32, BlockHash>,
+    ) -> 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, local_chain)?.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>,
+        local_chain: &BTreeMap<u32, BlockHash>,
+    ) -> Result<(SparseChain, BTreeMap<K, u32>)> {
+        let mut sparse_chain = SparseChain::default();
+
+        // Check for reorgs.
+        // In case of reorg, new checkpoints until the last common checkpoint is added to the structure
+        for (&existing_height, &existing_hash) in local_chain.iter().rev() {
+            let current_hash = self
+                .client
+                .block_header(existing_height as usize)?
+                .block_hash();
+            sparse_chain
+                .insert_checkpoint(BlockId {
+                    height: existing_height,
+                    hash: current_hash,
+                })
+                .expect("should not collide");
+
+            if current_hash == existing_hash {
+                break;
+            }
+        }
+
+        // Insert the new tip
+        let (tip_height, tip_hash) = self.get_tip()?;
+        sparse_chain
+            .insert_checkpoint(BlockId {
+                height: tip_height,
+                hash: tip_hash,
+            })
+            .expect("Should not collide");
+
+        let mut keychain_index_update = BTreeMap::new();
+
+        // 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());
+                    unused_script_count = 0;
+                }
+
+                if unused_script_count >= stop_gap.unwrap_or(usize::MAX) {
+                    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,201 @@
+mod electrum;
+
+use std::fmt::Debug;
+
+use bdk_core::{bitcoin::Transaction, sparse_chain::SparseChain, BlockId, TxHeight};
+use bdk_keychain::{KeychainScan, KeychainTracker};
+use electrum::ElectrumClient;
+
+use bdk_cli::{
+    anyhow::{self, Context, Result},
+    clap::{self, Subcommand},
+};
+use log::debug;
+
+use electrum_client::ElectrumApi;
+
+#[derive(Subcommand, Debug, Clone)]
+enum ElectrumCommands {
+    /// Scans the addresses in the wallet using esplora API.
+    Scan {
+        /// When a gap this large has been found for a keychain it will stop.
+        #[clap(long, default_value = "5")]
+        stop_gap: usize,
+    },
+    /// Scans particular addresses using esplora API
+    Sync {
+        /// Scan all the unused addresses
+        #[clap(long)]
+        unused: bool,
+        /// Scan the script addresses that have unspent outputs
+        #[clap(long)]
+        unspent: bool,
+        /// Scan every address that you have derived
+        #[clap(long)]
+        all: bool,
+    },
+}
+
+fn fetch_transactions<K: Debug + Ord + Clone>(
+    new_sparsechain: &SparseChain<TxHeight>,
+    client: &ElectrumClient,
+    tracker: &KeychainTracker<K, TxHeight>,
+) -> Result<Vec<(Transaction, Option<TxHeight>)>> {
+    // Changeset of txids, both new and old.
+    let txid_changeset = tracker.chain().determine_changeset(new_sparsechain)?.txids;
+
+    // Only filter for txids that are new to us.
+    let new_txids = txid_changeset
+        .iter()
+        .filter_map(|(txid, index)| {
+            if !tracker.graph().contains_txid(*txid) {
+                Some((txid, index))
+            } else {
+                None
+            }
+        })
+        .collect::<Vec<_>>();
+
+    // Remaining of the transactions that only changed in Index
+    let existing_txs = txid_changeset
+        .iter()
+        .filter_map(|(txid, index)| match tracker.graph().tx(*txid) {
+            Some(tx) => Some((tx.clone(), *index)),
+            // We don't keep the index for `TxNode::Partial`s
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+
+    let new_transactions = client.batch_transaction_get(new_txids.iter().map(|(txid, _)| *txid))?;
+
+    // Add all the transaction, new and existing into scan_update
+    let transaction_update = new_transactions
+        .into_iter()
+        .zip(new_txids.into_iter().map(|(_, index)| *index))
+        .chain(existing_txs)
+        .collect::<Vec<_>>();
+
+    Ok(transaction_update)
+}
+
+fn main() -> anyhow::Result<()> {
+    let (args, keymap, mut tracker, mut db) = bdk_cli::init::<ElectrumCommands, _>()?;
+
+    let client = ElectrumClient::new("ssl://electrum.blockstream.info:60002")?;
+
+    let electrum_cmd = match args.command {
+        bdk_cli::Commands::ChainSpecific(electrum_cmd) => electrum_cmd,
+        general_command => {
+            return bdk_cli::handle_commands(
+                general_command,
+                client,
+                &mut tracker,
+                &mut db,
+                args.network,
+                &keymap,
+            )
+        }
+    };
+
+    match electrum_cmd {
+        ElectrumCommands::Scan { stop_gap } => {
+            let scripts = tracker.txout_index.iter_all_script_pubkeys_by_keychain();
+
+            let mut keychain_scan = 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), tracker.chain().checkpoints())?;
+
+            keychain_scan.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 _ = keychain_scan.update.insert_checkpoint(BlockId {
+                    height: *height,
+                    hash: *hash,
+                })?;
+            }
+
+            // Fetch the new and old transactions to be added in update structure
+            for (tx, index) in fetch_transactions(&new_sparsechain, &client, &tracker)? {
+                keychain_scan.update.insert_tx(tx, index)?;
+            }
+
+            // Apply the full scan update
+            let changeset = tracker.determine_changeset(&keychain_scan)?;
+            db.append_changeset(&changeset)?;
+            tracker.apply_changeset(changeset);
+            debug!("sync completed!!")
+        }
+        ElectrumCommands::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 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, tracker.chain().checkpoints())
+                .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,
+                })?;
+            }
+
+            // Fetch the new and old transactions to be added in update structure
+            for (tx, index) in fetch_transactions(&new_sparsechain, &client, &tracker)? {
+                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!!")
+        }
+    }
+    Ok(())
+}