Restructure indexers, collect deposit requests and da entries

crates/btcio/src/reader/mod.rs

@@ -1,3 +1,3 @@
-mod ops_visitor;
 pub mod query;
 mod state;
+mod tx_indexer;

crates/btcio/src/reader/query.rs

@@ -9,7 +9,7 @@ use bitcoin::{Block, BlockHash};
 use strata_config::btcio::ReaderConfig;
 use strata_l1tx::{
     filter::{
-        visitor::{BlockIndexer, DepositRequestIndexer, OpIndexer},
+        indexer::{BlockIndexer, OpIndexer},
         TxFilterConfig,
     },
     messages::{BlockData, L1Event},
@@ -24,7 +24,7 @@ use tokio::sync::mpsc;
 use tracing::*;
 
 use crate::{
-    reader::{ops_visitor::ClientOpsVisitor, state::ReaderState},
+    reader::{state::ReaderState, tx_indexer::ClientTxIndexer},
     rpc::traits::ReaderRpc,
     status::{apply_status_updates, L1StatusUpdate},
 };
@@ -327,15 +327,12 @@ async fn process_block<R: ReaderRpc>(
     let params = ctx.params.clone();
 
     // Index ops
-    let ops_indexer = OpIndexer::new(ClientOpsVisitor::new());
-    let tx_entries = ops_indexer
+    let ops_indexer = OpIndexer::new(ClientTxIndexer::new());
+    let (tx_entries, _dep_reqs, _da_entries) = ops_indexer
         .index_block(&block, state.filter_config())
         .collect();
 
-    // Index deposit requests
-    let _dep_reqs = DepositRequestIndexer::new()
-        .index_block(&block, state.filter_config())
-        .collect();
+    // TODO: do stuffs with dep_reqs and da_entries
 
     let block_data = BlockData::new(height, block, tx_entries);
 

crates/btcio/src/reader/ops_visitor.rs → crates/btcio/src/reader/tx_indexer.rs

@@ -1,51 +1,73 @@
 use digest::Digest;
 use sha2::Sha256;
-use strata_l1tx::filter::visitor::OpsVisitor;
+use strata_l1tx::{filter::indexer::TxIndexer, messages::DaEntry};
+use strata_primitives::buf::Buf32;
 use strata_state::{
     batch::SignedBatchCheckpoint,
     tx::{DepositInfo, DepositRequestInfo, ProtocolOperation},
 };
 
-/// Ops visitor for rollup client. This is intended to get Da commitment as well as process/store da
-/// blob.
+/// Ops indexer for rollup client. Collects extra info like da blobs and deposit requets
 #[derive(Clone, Debug)]
-pub struct ClientOpsVisitor {
+pub struct ClientTxIndexer {
     ops: Vec<ProtocolOperation>,
-    // TODO: Add l1 manager to store da to db
+    deposit_requests: Vec<DepositRequestInfo>,
+    da_entries: Vec<DaEntry>,
 }
 
-impl ClientOpsVisitor {
+impl ClientTxIndexer {
     pub fn new() -> Self {
-        Self { ops: Vec::new() }
+        Self {
+            ops: Vec::new(),
+            deposit_requests: Vec::new(),
+            da_entries: Vec::new(),
+        }
     }
-}
 
-impl OpsVisitor for ClientOpsVisitor {
-    fn collect(self) -> Vec<ProtocolOperation> {
-        self.ops
+    fn ops(&self) -> &[ProtocolOperation] {
+        &self.ops
     }
+}
 
+impl TxIndexer for ClientTxIndexer {
     fn visit_da<'a>(&mut self, chunks: impl Iterator<Item = &'a [u8]>) {
         let mut hasher = Sha256::new();
+        let mut blob = Vec::new();
+
         for chunk in chunks {
             hasher.update(chunk);
+            blob.extend_from_slice(chunk);
         }
         let hash: [u8; 32] = hasher.finalize().into();
-        // TODO: store da in db
-        self.ops.push(ProtocolOperation::DaCommitment(hash.into()));
+        let commitment: Buf32 = hash.into();
+
+        self.ops.push(ProtocolOperation::DaCommitment(commitment));
+        // Collect da
+        self.da_entries.push(DaEntry::new(commitment, blob));
     }
 
     fn visit_deposit(&mut self, d: DepositInfo) {
         self.ops.push(ProtocolOperation::Deposit(d));
     }
 
     fn visit_deposit_request(&mut self, d: DepositRequestInfo) {
-        self.ops.push(ProtocolOperation::DepositRequest(d));
+        self.ops.push(ProtocolOperation::DepositRequest(d.clone()));
+        self.deposit_requests.push(d);
     }
 
     fn visit_checkpoint(&mut self, chkpt: SignedBatchCheckpoint) {
         self.ops.push(ProtocolOperation::Checkpoint(chkpt));
     }
+
+    fn collect(
+        self,
+    ) -> (
+        Vec<ProtocolOperation>,
+        Vec<DepositRequestInfo>,
+        Vec<DaEntry>,
+    ) {
+        (self.ops, self.deposit_requests, self.da_entries)
+    }
 }
 
 #[cfg(test)]
@@ -56,7 +78,7 @@ mod test {
         Amount, Block, BlockHash, CompactTarget, ScriptBuf, Transaction, TxMerkleNode,
     };
     use strata_l1tx::filter::{
-        visitor::{BlockIndexer, DepositRequestIndexer, OpIndexer},
+        indexer::{BlockIndexer, OpIndexer},
         TxFilterConfig,
     };
     use strata_primitives::{
@@ -73,7 +95,7 @@ mod test {
         ArbitraryGenerator,
     };
 
-    use crate::{reader::ops_visitor::ClientOpsVisitor, test_utils::create_checkpoint_envelope_tx};
+    use crate::{reader::tx_indexer::ClientTxIndexer, test_utils::create_checkpoint_envelope_tx};
 
     const TEST_ADDR: &str = "bcrt1q6u6qyya3sryhh42lahtnz2m7zuufe7dlt8j0j5";
 
@@ -114,8 +136,8 @@ mod test {
 
         let block = create_test_block(vec![tx]);
 
-        let ops_indexer = OpIndexer::new(ClientOpsVisitor::new());
-        let tx_entries = ops_indexer.index_block(&block, &filter_config).collect();
+        let ops_indexer = OpIndexer::new(ClientTxIndexer::new());
+        let (tx_entries, _, _) = ops_indexer.index_block(&block, &filter_config).collect();
 
         assert_eq!(tx_entries.len(), 1, "Should find one relevant transaction");
 
@@ -158,10 +180,8 @@ mod test {
 
         let block = create_test_block(vec![tx]);
 
-        let deposit_req_indexer = DepositRequestIndexer::new();
-        let dep_reqs = deposit_req_indexer
-            .index_block(&block, &filter_config)
-            .collect();
+        let ops_indexer = OpIndexer::new(ClientTxIndexer::new());
+        let (_, dep_reqs, _) = ops_indexer.index_block(&block, &filter_config).collect();
 
         assert_eq!(dep_reqs.len(), 1, "Should find one deposit request");
 
@@ -187,8 +207,8 @@ mod test {
 
         let block = create_test_block(vec![irrelevant_tx]);
 
-        let ops_indexer = OpIndexer::new(ClientOpsVisitor::new());
-        let tx_entries = ops_indexer.index_block(&block, &filter_config).collect();
+        let ops_indexer = OpIndexer::new(ClientTxIndexer::new());
+        let (tx_entries, _, _) = ops_indexer.index_block(&block, &filter_config).collect();
 
         assert!(
             tx_entries.is_empty(),
@@ -222,8 +242,8 @@ mod test {
 
         let block = create_test_block(vec![tx1, tx2]);
 
-        let ops_indexer = OpIndexer::new(ClientOpsVisitor::new());
-        let tx_entries = ops_indexer.index_block(&block, &filter_config).collect();
+        let ops_indexer = OpIndexer::new(ClientTxIndexer::new());
+        let (tx_entries, _, _) = ops_indexer.index_block(&block, &filter_config).collect();
 
         assert_eq!(tx_entries.len(), 2, "Should find two relevant transactions");
 
@@ -286,9 +306,8 @@ mod test {
         // Create a block with single tx that has multiple ops
         let block = create_test_block(vec![tx]);
 
-        let ops_indexer = OpIndexer::new(ClientOpsVisitor::new());
-        let tx_entries = ops_indexer.index_block(&block, &filter_config).collect();
-        println!("TX ENTRIES: {:?}", tx_entries);
+        let ops_indexer = OpIndexer::new(ClientTxIndexer::new());
+        let (tx_entries, _, _) = ops_indexer.index_block(&block, &filter_config).collect();
 
         assert_eq!(
             tx_entries.len(),

crates/l1tx/src/filter/visitor.rs → crates/l1tx/src/filter/indexer.rs

@@ -8,9 +8,9 @@ use super::{
     parse_checkpoint_envelopes, parse_da_blobs, parse_deposit_requests, parse_deposits,
     TxFilterConfig,
 };
-use crate::messages::ProtocolTxEntry;
+use crate::messages::{DaEntry, ProtocolTxEntry};
 
-pub trait OpsVisitor {
+pub trait TxIndexer {
     // Do stuffs with `SignedBatchCheckpoint`.
     fn visit_checkpoint(&mut self, _chkpt: SignedBatchCheckpoint) {}
 
@@ -23,14 +23,17 @@ pub trait OpsVisitor {
     // Do stuffs with DA.
     fn visit_da<'a>(&mut self, _d: impl Iterator<Item = &'a [u8]>) {}
 
-    fn collect(self) -> Vec<ProtocolOperation>;
+    // Collect data
+    fn collect(
+        self,
+    ) -> (
+        Vec<ProtocolOperation>,
+        Vec<DepositRequestInfo>,
+        Vec<DaEntry>,
+    );
 }
 
 pub trait BlockIndexer {
-    type Output;
-
-    fn collect(self) -> Self::Output;
-
     fn index_tx(&mut self, txidx: u32, tx: &Transaction, config: &TxFilterConfig);
 
     fn index_block(mut self, block: &Block, config: &TxFilterConfig) -> Self
@@ -45,36 +48,29 @@ pub trait BlockIndexer {
 }
 
 #[derive(Clone, Debug)]
-pub struct OpIndexer<V: OpsVisitor> {
+pub struct OpIndexer<V: TxIndexer> {
     visitor: V,
     tx_entries: Vec<ProtocolTxEntry>,
+    dep_reqs: Vec<DepositRequestInfo>,
+    da_entries: Vec<DaEntry>,
 }
 
-impl<V: OpsVisitor> OpIndexer<V> {
+impl<V: TxIndexer> OpIndexer<V> {
     pub fn new(visitor: V) -> Self {
         Self {
             visitor,
             tx_entries: Vec::new(),
+            dep_reqs: Vec::new(),
+            da_entries: Vec::new(),
         }
     }
-}
-
-#[derive(Clone, Default)]
-pub struct DepositRequestIndexer {
-    requests: Vec<DepositRequestInfo>,
-}
 
-impl DepositRequestIndexer {
-    pub fn new() -> Self {
-        Self {
-            ..Default::default()
-        }
+    pub fn collect(self) -> (Vec<ProtocolTxEntry>, Vec<DepositRequestInfo>, Vec<DaEntry>) {
+        (self.tx_entries, self.dep_reqs, self.da_entries)
     }
 }
 
-impl<V: OpsVisitor + Clone> BlockIndexer for OpIndexer<V> {
-    type Output = Vec<ProtocolTxEntry>;
-
+impl<V: TxIndexer + Clone> BlockIndexer for OpIndexer<V> {
     fn index_tx(&mut self, txidx: u32, tx: &Transaction, config: &TxFilterConfig) {
         let mut visitor = self.visitor.clone();
         for chp in parse_checkpoint_envelopes(tx, config) {
@@ -94,26 +90,13 @@ impl<V: OpsVisitor + Clone> BlockIndexer for OpIndexer<V> {
             visitor.visit_da(da);
         }
 
-        let ops = visitor.collect();
+        let (ops, dep_reqs, da_entries) = visitor.collect();
         if !ops.is_empty() {
             let entry = ProtocolTxEntry::new(txidx, ops);
             self.tx_entries.push(entry);
         }
-    }
-
-    fn collect(self) -> Self::Output {
-        self.tx_entries
-    }
-}
-
-impl BlockIndexer for DepositRequestIndexer {
-    type Output = Vec<DepositRequestInfo>;
-
-    fn collect(self) -> Self::Output {
-        self.requests
-    }
 
-    fn index_tx(&mut self, _txidx: u32, tx: &Transaction, config: &TxFilterConfig) {
-        self.requests.extend(parse_deposit_requests(tx, config));
+        self.dep_reqs.extend_from_slice(&dep_reqs);
+        self.da_entries.extend_from_slice(&da_entries);
     }
 }

crates/l1tx/src/filter/mod.rs

@@ -6,8 +6,8 @@ use strata_state::{
 };
 use tracing::warn;
 
+pub mod indexer;
 pub mod types;
-pub mod visitor;
 
 pub use types::TxFilterConfig;
 

crates/l1tx/src/messages.rs

@@ -1,5 +1,6 @@
 use bitcoin::Block;
 use borsh::{BorshDeserialize, BorshSerialize};
+use strata_primitives::buf::Buf32;
 use strata_state::{l1::HeaderVerificationState, tx::ProtocolOperation};
 
 /// L1 events that we observe and want the persistence task to work on.
@@ -45,6 +46,21 @@ impl ProtocolTxEntry {
     }
 }
 
+/// Da data retrieved from L1 transaction.
+#[derive(Clone, Debug)]
+pub struct DaEntry {
+    #[allow(unused)]
+    commitment: Buf32,
+    #[allow(unused)]
+    blob: Vec<u8>,
+}
+
+impl DaEntry {
+    pub fn new(commitment: Buf32, blob: Vec<u8>) -> Self {
+        Self { commitment, blob }
+    }
+}
+
 /// Store the bitcoin block and references to the relevant transactions within the block
 #[derive(Clone, Debug)]
 pub struct BlockData {

crates/proof-impl/btc-blockspace/src/filter.rs

@@ -3,7 +3,7 @@
 
 use bitcoin::Block;
 use strata_l1tx::filter::{
-    visitor::{BlockIndexer, OpIndexer},
+    indexer::{BlockIndexer, OpIndexer},
     TxFilterConfig,
 };
 use strata_primitives::{block_credential::CredRule, params::RollupParams};
@@ -12,7 +12,7 @@ use strata_state::{
     tx::{DepositInfo, ProtocolOperation},
 };
 
-use crate::ops_visitor::ProverOpsVisitor;
+use crate::tx_indexer::ProverTxIndexer;
 
 pub fn extract_relevant_info(
     block: &Block,
@@ -24,8 +24,8 @@ pub fn extract_relevant_info(
 
     // Just pass a no-op to the filter function as prover does not have to do anything with the raw
     // data like storing in db.
-    let indexer = OpIndexer::new(ProverOpsVisitor::new());
-    let tx_refs = indexer.index_block(block, filter_config).collect();
+    let indexer = OpIndexer::new(ProverTxIndexer::new());
+    let (tx_refs, _, _) = indexer.index_block(block, filter_config).collect();
 
     for op in tx_refs.into_iter().flat_map(|t| t.proto_ops().to_vec()) {
         match op {

crates/proof-impl/btc-blockspace/src/lib.rs

@@ -4,7 +4,8 @@ pub mod block;
 pub mod filter;
 pub mod logic;
 pub mod merkle;
-mod ops_visitor;
 pub mod prover;
 pub mod scan;
 pub mod tx;
+mod tx_indexer;
+