feat: Update into_tx_graph_update and add syncing

Update into_tx_graph_update to take a Vec of tuples of
blocks and transactions as input to create a TxGraph.

Add syncing with Nakamoto client. We start by creating a list of scripts
to sync and we continue to scan the chain until we have satisfied
stop_gap condition. On each scan we get a fresh list of script to watch.
We then update the IndexedGraph and return the IndexedAdditions.

crates/bdk_cbf/src/lib.rs

@@ -1,18 +1,25 @@
 use std::{net, thread};
 
+use bdk_chain::keychain::DerivationAdditions;
 use nakamoto::client::network::Services;
 use nakamoto::client::traits::Handle as HandleTrait;
 use nakamoto::client::Handle;
 use nakamoto::client::{chan, Client, Config, Error, Event};
 use nakamoto::common::block::Height;
 use nakamoto::net::poll;
 
+pub use nakamoto::client::network::Network;
+
 use bdk_chain::{
     bitcoin::{Script, Transaction},
-    BlockId, ChainOracle, TxGraph,
+    collections::BTreeMap,
+    indexed_tx_graph::{IndexedAdditions, IndexedTxGraph, Indexer},
+    keychain::KeychainTxOutIndex,
+    BlockId, ChainOracle, ConfirmationHeightAnchor, TxGraph,
 };
 
-/// The network reactor we're going to use.
+use core::fmt::Debug;
+
 type Reactor = poll::Reactor<net::TcpStream>;
 
 impl ChainOracle for CBFClient {
@@ -147,21 +154,18 @@ impl CBFClient {
     }
 
     // Turns a CBFUpdate into a TxGraph update
-    pub fn into_tx_graph_update<F>(
+    pub fn into_tx_graph_update(
         &self,
-        txs: Vec<Transaction>,
-        block: BlockId,
-        is_relevant: F,
-    ) -> TxGraph<BlockId>
-    where
-        F: Fn(&Transaction) -> bool,
-    {
+        block_txs: Vec<(BlockId, Vec<Transaction>)>,
+    ) -> TxGraph<ConfirmationHeightAnchor> {
         let mut tx_graph = TxGraph::default();
-        let filtered_txs = txs.into_iter().filter(|tx| is_relevant(tx));
-        for tx in filtered_txs {
-            let txid = tx.txid();
-            let _ = tx_graph.insert_anchor(txid, block);
-            let _ = tx_graph.insert_tx(tx);
+
+        for (blockid, txs) in block_txs.into_iter() {
+            for tx in txs {
+                let txid = tx.txid();
+                let _ = tx_graph.insert_anchor(txid, to_confirmation_height_anchor(blockid));
+                let _ = tx_graph.insert_tx(tx);
+            }
         }
         tx_graph
     }
@@ -171,4 +175,121 @@ impl CBFClient {
             client: self.clone(),
         }
     }
+
+    pub fn sync<K>(
+        &self,
+        lookahead: u32,
+        start_height: Height,
+        indexed_tx_graph: &mut IndexedTxGraph<ConfirmationHeightAnchor, KeychainTxOutIndex<K>>,
+        stop_gap: u32,
+    ) -> Result<IndexedAdditions<ConfirmationHeightAnchor, DerivationAdditions<K>>, Error>
+    where
+        K: Ord + Clone + Debug,
+    {
+        indexed_tx_graph.index.set_lookahead_for_all(lookahead);
+        let mut keychain_spks = indexed_tx_graph.index.spks_of_all_keychains();
+        let mut empty_scripts_counter = BTreeMap::<K, u32>::new();
+        keychain_spks.keys().for_each(|k| {
+            empty_scripts_counter.insert(k.clone(), 0);
+        });
+
+        let mut updates = Vec::new();
+
+        while let Some(keychains) = Self::check_stop_gap(stop_gap, &empty_scripts_counter) {
+            keychains.iter().for_each(|k| {
+                indexed_tx_graph.index.set_lookahead(k, lookahead);
+            });
+
+            let mut spk_watchlist = BTreeMap::<K, Vec<Script>>::new();
+            for (k, script_iter) in keychain_spks.iter_mut() {
+                (0..lookahead).for_each(|_| {
+                    if let Some((_, script)) = script_iter.next() {
+                        let spks = spk_watchlist.entry(k.clone()).or_insert(vec![]);
+                        spks.push(script);
+                    }
+                });
+            }
+
+            self.start_scanning(start_height, spk_watchlist.values().flatten().cloned())?;
+
+            for update in self.iter() {
+                match update {
+                    Ok(CBFUpdate::BlockMatched {
+                        transactions,
+                        block,
+                    }) => {
+                        let relevant_txs = transactions
+                            .into_iter()
+                            .filter(|tx| indexed_tx_graph.index.is_tx_relevant(tx))
+                            .collect::<Vec<_>>();
+                        updates.push((block, relevant_txs));
+                    }
+                    Ok(CBFUpdate::BlockDisconnected { .. }) => {
+                        //TODO: Don't know how to handle re-orgs yet
+                        //I will love to get your comments on this.
+                    }
+                    Ok(CBFUpdate::Synced { .. }) => {
+                        break;
+                    }
+                    Err(e) => {
+                        return Err(e);
+                    }
+                }
+            }
+
+            // Determine which scripts are part of the update.
+            for (k, scripts) in spk_watchlist.iter() {
+                for script in scripts {
+                    let counter = empty_scripts_counter.get_mut(k).unwrap();
+                    if Self::is_script_in_udpate(script.clone(), &updates) {
+                        *counter = 0;
+                    } else {
+                        *counter += 1;
+                    }
+                }
+            }
+        }
+
+        //apply the updates to IndexedGraph
+        let graph_update = self.into_tx_graph_update(updates);
+        let additions = indexed_tx_graph.apply_update(graph_update);
+
+        Ok(additions)
+    }
+
+    fn is_script_in_udpate(script: Script, updates: &Vec<(BlockId, Vec<Transaction>)>) -> bool {
+        for update in updates {
+            for tx in update.1.iter() {
+                for output in tx.output.iter() {
+                    if output.script_pubkey == script {
+                        return true;
+                    }
+                }
+            }
+        }
+        false
+    }
+
+    fn check_stop_gap<K>(stop_gap: u32, empty_scripts_counter: &BTreeMap<K, u32>) -> Option<Vec<K>>
+    where
+        K: Ord + Clone + Debug,
+    {
+        let keychains = empty_scripts_counter
+            .iter()
+            .filter(|(_, counter)| **counter < stop_gap)
+            .map(|(k, _)| k.clone())
+            .collect::<Vec<_>>();
+        if keychains.is_empty() {
+            None
+        } else {
+            Some(keychains)
+        }
+    }
+}
+
+fn to_confirmation_height_anchor(blockid: BlockId) -> ConfirmationHeightAnchor {
+    ConfirmationHeightAnchor {
+        anchor_block: blockid,
+        confirmation_height: blockid.height,
+    }
 }