Merge pull request #14 from propeller-heads/ah/factor-tycho-substreams

Add shared tycho-substreams lib.

crates/tycho-substreams/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "tycho-substreams"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+substreams-ethereum = "0.9.9"
+substreams = "0.5"
+prost = "0.11"
+hex = "0.4.3"
+itertools = "0.12.0"

crates/tycho-substreams/Readme.md

@@ -0,0 +1,18 @@
+# Tycho Substreams SDK
+
+Some shared functionality that is used to create tycho substream packages.
+
+## Protobuf Models
+
+Protobuf models are manually synced from the `tycho-indexer` repository whenever they 
+changed.
+
+To generate the rust structs run the following command from within the `./proto` 
+directory:
+
+```bash
+buf generate \
+  --path tycho \
+  --template ../crates/tycho-substreams/buf.gen.yaml \
+  --output ../crates/tycho-substreams/ 
+```

crates/tycho-substreams/buf.gen.yaml

@@ -0,0 +1,12 @@
+version: v1
+plugins:
+  - plugin: buf.build/community/neoeinstein-prost:v0.2.2
+    out: src/pb
+    opt:
+      - file_descriptor_set=false
+      - type_attribute=.tycho.evm.v1.Transaction=#[derive(Eq\, Hash)]
+
+  - plugin: buf.build/community/neoeinstein-prost-crate:v0.3.1
+    out: src/pb
+    opt:
+      - no_features

crates/tycho-substreams/rustfmt.toml

@@ -0,0 +1,13 @@
+reorder_imports = true
+imports_granularity = "Crate"
+use_small_heuristics = "Max"
+comment_width = 100
+wrap_comments = true
+binop_separator = "Back"
+trailing_comma = "Vertical"
+trailing_semicolon = false
+use_field_init_shorthand = true
+chain_width = 40
+ignore = [
+    "src/pb",
+]

crates/tycho-substreams/src/balances.rs

@@ -0,0 +1,343 @@
+//! Utilities to handle relative balances.
+//!
+//!
+//! To aggregate relative balances changes to absolute balances the general approach is:
+//!
+//! 1. Use a map function that will extract a `BlockBalanceDeltas` message. BalanceDeltas
+//!     within this message are required to have increasing ordinals so that
+//!     the order of relative balance changes is unambiguous.
+//! 2. Store the balances changes with a store handler. You can use the
+//!     `store_balance_changes` library method directly for this.
+//! 3. In the output module, use aggregate_balance_changes to receive an
+//!     aggregated map of absolute balances.
+//!
+use crate::pb::tycho::evm::v1::{BalanceChange, BlockBalanceDeltas, Transaction};
+use itertools::Itertools;
+use std::collections::HashMap;
+use std::str::FromStr;
+use substreams::key;
+use substreams::pb::substreams::StoreDeltas;
+use substreams::prelude::{BigInt, StoreAdd};
+
+/// Store relative balances changes in a additive manner.
+///
+/// Effectively aggregates the relative balances changes into an absolute balances.
+///
+/// ## Arguments
+///
+/// * `deltas` - A `BlockBalanceDeltas` message containing the relative balances changes.
+///     Note: relative balance deltas must have strictly increasing ordinals per token
+///     address, will panic otherwise.
+/// * `store` - An AddStore that will add relative balance changes.
+///
+/// This method is meant to be used in combination with `aggregate_balances_changes`
+/// which consumes the store filled with this methods in
+/// [deltas mode](https://substreams.streamingfast.io/documentation/develop/manifest-modules/types#deltas-mode).
+pub fn store_balance_changes(deltas: BlockBalanceDeltas, store: impl StoreAdd<BigInt>) {
+    let mut previous_ordinal = HashMap::<String, u64>::new();
+    deltas
+        .balance_deltas
+        .iter()
+        .for_each(|delta| {
+            let balance_key = format!(
+                "{0}:{1}",
+                String::from_utf8(delta.component_id.clone())
+                    .expect("delta.component_id is not valid utf-8!"),
+                hex::encode(&delta.token)
+            );
+            let current_ord = delta.ord;
+            previous_ordinal
+                .entry(balance_key.clone())
+                .and_modify(|ord| {
+                    // ordinals must arrive in increasing order
+                    if *ord >= current_ord {
+                        panic!(
+                            "Invalid ordinal sequence for {}: {} >= {}",
+                            balance_key, *ord, current_ord
+                        );
+                    }
+                    *ord = current_ord;
+                })
+                .or_insert(delta.ord);
+            store.add(delta.ord, balance_key, BigInt::from_signed_bytes_be(&delta.delta));
+        });
+}
+
+/// Aggregates absolute balances per transaction and token.
+///
+/// ## Arguments
+/// * `balance_store` - A `StoreDeltas` with all changes that occured in the source
+///     store module.
+/// * `deltas` - A `BlockBalanceDeltas` message containing the relative balances changes.
+///
+/// Reads absolute balance values from the additive store (see `store_balance_changes`
+/// on how to create such a store), proceeds to zip them with the relative balance
+/// deltas to associate balance values to token and component.
+///
+/// Will keep the last balance change per token per transaction if there are multiple
+/// changes.
+///
+/// Returns a map of transactions hashes to the full transaction and aggregated
+/// absolute balance changes.
+pub fn aggregate_balances_changes(
+    balance_store: StoreDeltas,
+    deltas: BlockBalanceDeltas,
+) -> HashMap<Vec<u8>, (Transaction, HashMap<Vec<u8>, BalanceChange>)> {
+    balance_store
+        .deltas
+        .into_iter()
+        .zip(deltas.balance_deltas)
+        .map(|(store_delta, balance_delta)| {
+            let component_id = key::segment_at(&store_delta.key, 0);
+            let token_id = key::segment_at(&store_delta.key, 1);
+            // store_delta.new_value is an ASCII string representing an integer
+            let ascii_string =
+                String::from_utf8(store_delta.new_value.clone()).expect("Invalid UTF-8 sequence");
+            let balance = BigInt::from_str(&ascii_string).expect("Failed to parse integer");
+            let big_endian_bytes_balance = balance.to_bytes_be().1;
+
+            (
+                balance_delta
+                    .tx
+                    .expect("Missing transaction on delta"),
+                BalanceChange {
+                    token: hex::decode(token_id).expect("Token ID not valid hex"),
+                    balance: big_endian_bytes_balance,
+                    component_id: component_id.as_bytes().to_vec(),
+                },
+            )
+        })
+        // We need to group the balance changes by tx hash for the `TransactionContractChanges` agg
+        .group_by(|(tx, _)| tx.hash.clone())
+        .into_iter()
+        .map(|(txh, group)| {
+            let (mut transactions, balance_changes): (Vec<_>, Vec<_>) = group.into_iter().unzip();
+
+            let balances = balance_changes
+                .into_iter()
+                .map(|balance_change| (balance_change.token.clone(), balance_change))
+                .collect();
+            (txh, (transactions.pop().unwrap(), balances))
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mock_store::MockStore;
+    use crate::pb::tycho::evm::v1::{BalanceDelta, Transaction};
+    use substreams::pb::substreams::StoreDelta;
+    use substreams::prelude::{StoreGet, StoreNew};
+
+    fn block_balance_deltas() -> BlockBalanceDeltas {
+        let comp_id = "0x42c0ffee"
+            .to_string()
+            .as_bytes()
+            .to_vec();
+        let token_0 = hex::decode("bad999").unwrap();
+        let token_1 = hex::decode("babe00").unwrap();
+        BlockBalanceDeltas {
+            balance_deltas: vec![
+                BalanceDelta {
+                    ord: 0,
+                    tx: Some(Transaction {
+                        hash: vec![0, 1],
+                        from: vec![9, 9],
+                        to: vec![8, 8],
+                        index: 0,
+                    }),
+                    token: token_0.clone(),
+                    delta: BigInt::from_str("+1000")
+                        .unwrap()
+                        .to_signed_bytes_be(),
+                    component_id: comp_id.clone(),
+                },
+                BalanceDelta {
+                    ord: 2,
+                    tx: Some(Transaction {
+                        hash: vec![0, 1],
+                        from: vec![9, 9],
+                        to: vec![8, 8],
+                        index: 0,
+                    }),
+                    token: token_1.clone(),
+                    delta: BigInt::from_str("+100")
+                        .unwrap()
+                        .to_signed_bytes_be(),
+                    component_id: comp_id.clone(),
+                },
+                BalanceDelta {
+                    ord: 3,
+                    tx: Some(Transaction {
+                        hash: vec![0, 1],
+                        from: vec![9, 9],
+                        to: vec![8, 8],
+                        index: 0,
+                    }),
+                    token: token_1.clone(),
+                    delta: BigInt::from_str("50")
+                        .unwrap()
+                        .to_signed_bytes_be(),
+                    component_id: comp_id.clone(),
+                },
+                BalanceDelta {
+                    ord: 10,
+                    tx: Some(Transaction {
+                        hash: vec![0, 1],
+                        from: vec![9, 9],
+                        to: vec![8, 8],
+                        index: 0,
+                    }),
+                    token: token_0.clone(),
+                    delta: BigInt::from_str("-1")
+                        .unwrap()
+                        .to_signed_bytes_be(),
+                    component_id: comp_id.clone(),
+                },
+            ],
+        }
+    }
+    fn store_deltas() -> StoreDeltas {
+        let comp_id = "0x42c0ffee"
+            .to_string()
+            .as_bytes()
+            .to_vec();
+        let token_0 = hex::decode("bad999").unwrap();
+        let token_1 = hex::decode("babe00").unwrap();
+
+        let t0_key =
+            format!("{}:{}", String::from_utf8(comp_id.clone()).unwrap(), hex::encode(&token_0));
+        let t1_key =
+            format!("{}:{}", String::from_utf8(comp_id.clone()).unwrap(), hex::encode(&token_1));
+        StoreDeltas {
+            deltas: vec![
+                StoreDelta {
+                    operation: 0,
+                    ordinal: 0,
+                    key: t0_key.clone(),
+                    old_value: BigInt::from(0)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                    new_value: BigInt::from(1000)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                },
+                StoreDelta {
+                    operation: 0,
+                    ordinal: 2,
+                    key: t1_key.clone(),
+                    old_value: BigInt::from(0)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                    new_value: BigInt::from(100)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                },
+                StoreDelta {
+                    operation: 0,
+                    ordinal: 3,
+                    key: t1_key.clone(),
+                    old_value: BigInt::from(100)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                    new_value: BigInt::from(150)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                },
+                StoreDelta {
+                    operation: 0,
+                    ordinal: 10,
+                    key: t0_key.clone(),
+                    old_value: BigInt::from(1000)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                    new_value: BigInt::from(999)
+                        .to_string()
+                        .as_bytes()
+                        .to_vec(),
+                },
+            ],
+        }
+    }
+
+    #[test]
+    fn test_store_balances() {
+        let comp_id = "0x42c0ffee"
+            .to_string()
+            .as_bytes()
+            .to_vec();
+        let token_0 = hex::decode("bad999").unwrap();
+        let token_1 = hex::decode("babe00").unwrap();
+        let deltas = block_balance_deltas();
+        let store = <MockStore as StoreNew>::new();
+
+        store_balance_changes(deltas, store.clone());
+        let res_0 = store.get_last(format!(
+            "{}:{}",
+            String::from_utf8(comp_id.clone()).unwrap(),
+            hex::encode(&token_0)
+        ));
+        let res_1 = store.get_last(format!(
+            "{}:{}",
+            String::from_utf8(comp_id.clone()).unwrap(),
+            hex::encode(&token_1)
+        ));
+
+        assert_eq!(res_0, Some(BigInt::from_str("+999").unwrap()));
+        assert_eq!(res_1, Some(BigInt::from_str("+150").unwrap()));
+    }
+
+    #[test]
+    fn test_aggregate_balances_changes() {
+        let store_deltas = store_deltas();
+        let balance_deltas = block_balance_deltas();
+        let comp_id = "0x42c0ffee"
+            .to_string()
+            .as_bytes()
+            .to_vec();
+        let token_0 = hex::decode("bad999").unwrap();
+        let token_1 = hex::decode("babe00").unwrap();
+
+        let exp = [(
+            vec![0, 1],
+            (
+                Transaction { hash: vec![0, 1], from: vec![9, 9], to: vec![8, 8], index: 0 },
+                [
+                    (
+                        token_0.clone(),
+                        BalanceChange {
+                            token: token_0,
+                            balance: BigInt::from(999)
+                                .to_signed_bytes_be()
+                                .to_vec(),
+                            component_id: comp_id.clone(),
+                        },
+                    ),
+                    (
+                        token_1.clone(),
+                        BalanceChange {
+                            token: token_1,
+                            balance: vec![150],
+                            component_id: comp_id.clone(),
+                        },
+                    ),
+                ]
+                .into_iter()
+                .collect::<HashMap<_, _>>(),
+            ),
+        )]
+        .into_iter()
+        .collect::<HashMap<_, _>>();
+
+        let res = aggregate_balances_changes(store_deltas, balance_deltas);
+        assert_eq!(res, exp);
+    }
+}

crates/tycho-substreams/src/contract.rs

@@ -0,0 +1,211 @@
+/// Helpers to extract relevant contract storage.
+///
+/// This file contains helpers to capture contract changes from the expanded block
+/// model. These leverage the `code_changes`, `balance_changes`, and `storage_changes`
+/// fields available on the `Call` type provided by block model in a substream
+/// (i.e. `logs_and_calls`, etc).
+///
+/// ## Warning
+/// ⚠️ These helpers *only* work if the **extended block model** is available,
+/// more [here](https://streamingfastio.medium.com/new-block-model-to-accelerate-chain-integration-9f65126e5425)
+use std::collections::HashMap;
+
+use substreams_ethereum::pb::eth;
+use substreams_ethereum::pb::eth::v2::block::DetailLevel;
+use substreams_ethereum::pb::eth::v2::StorageChange;
+
+use crate::pb::tycho::evm::v1::{self as tycho};
+
+struct SlotValue {
+    new_value: Vec<u8>,
+    start_value: Vec<u8>,
+}
+
+impl From<&StorageChange> for SlotValue {
+    fn from(change: &StorageChange) -> Self {
+        Self { new_value: change.new_value.clone(), start_value: change.old_value.clone() }
+    }
+}
+
+impl SlotValue {
+    fn has_changed(&self) -> bool {
+        self.start_value != self.new_value
+    }
+}
+
+// Uses a map for slots, protobuf does not allow bytes in hashmap keys
+struct InterimContractChange {
+    address: Vec<u8>,
+    balance: Vec<u8>,
+    code: Vec<u8>,
+    slots: HashMap<Vec<u8>, SlotValue>,
+    change: tycho::ChangeType,
+}
+
+impl InterimContractChange {
+    fn new(address: &[u8], creation: bool) -> Self {
+        Self {
+            address: address.to_vec(),
+            balance: vec![],
+            code: vec![],
+            slots: Default::default(),
+            change: if creation {
+                tycho::ChangeType::Creation.into()
+            } else {
+                tycho::ChangeType::Update.into()
+            },
+        }
+    }
+}
+
+impl From<InterimContractChange> for tycho::ContractChange {
+    fn from(value: InterimContractChange) -> Self {
+        tycho::ContractChange {
+            address: value.address,
+            balance: value.balance,
+            code: value.code,
+            slots: value
+                .slots
+                .into_iter()
+                .filter(|(_, value)| value.has_changed())
+                .map(|(slot, value)| tycho::ContractSlot { slot, value: value.new_value })
+                .collect(),
+            change: value.change.into(),
+        }
+    }
+}
+
+/// Extracts relevant contract changes from the block.
+///
+/// Contract changes include changes in storage, code and native balance.
+///
+/// ## Arguments
+///
+/// * `block` - The block to extract changes from. Must be the extended block model.
+/// * `inclusion_predicate` - A predicate function that determines if a contract address is relevant.
+/// * `transaction_contract_changes` - A mutable map to store the contract changes in.
+///
+/// ## Panics
+/// Will panic in case the detail level of the block is not extended.
+pub fn extract_contract_changes<F: Fn(&[u8]) -> bool>(
+    block: &eth::v2::Block,
+    inclusion_predicate: F,
+    transaction_contract_changes: &mut HashMap<u64, tycho::TransactionContractChanges>,
+) {
+    if block.detail_level != Into::<i32>::into(DetailLevel::DetaillevelExtended) {
+        panic!("Only extended blocks are supported");
+    }
+    let mut changed_contracts: HashMap<Vec<u8>, InterimContractChange> = HashMap::new();
+
+    // Collect all accounts created in this block
+    let created_accounts: HashMap<_, _> = block
+        .transactions()
+        .flat_map(|tx| {
+            tx.calls.iter().flat_map(|call| {
+                call.account_creations
+                    .iter()
+                    .map(|ac| (&ac.account, ac.ordinal))
+            })
+        })
+        .collect();
+
+    block
+        .transactions()
+        .for_each(|block_tx| {
+            let mut storage_changes = Vec::new();
+            let mut balance_changes = Vec::new();
+            let mut code_changes = Vec::new();
+
+            block_tx
+                .calls
+                .iter()
+                .filter(|call| !call.state_reverted && inclusion_predicate(&call.address))
+                .for_each(|call| {
+                    storage_changes.extend(call.storage_changes.iter());
+                    balance_changes.extend(call.balance_changes.iter());
+                    code_changes.extend(call.code_changes.iter());
+                });
+
+            storage_changes.sort_unstable_by_key(|change| change.ordinal);
+            balance_changes.sort_unstable_by_key(|change| change.ordinal);
+            code_changes.sort_unstable_by_key(|change| change.ordinal);
+
+            storage_changes
+                .iter()
+                .filter(|changes| inclusion_predicate(&changes.address))
+                .for_each(|&storage_change| {
+                    let contract_change = changed_contracts
+                        .entry(storage_change.address.clone())
+                        .or_insert_with(|| {
+                            InterimContractChange::new(
+                                &storage_change.address,
+                                created_accounts.contains_key(&storage_change.address),
+                            )
+                        });
+
+                    let slot_value = contract_change
+                        .slots
+                        .entry(storage_change.key.clone())
+                        .or_insert_with(|| storage_change.into());
+
+                    slot_value
+                        .new_value
+                        .copy_from_slice(&storage_change.new_value);
+                });
+
+            balance_changes
+                .iter()
+                .filter(|changes| inclusion_predicate(&changes.address))
+                .for_each(|balance_change| {
+                    let contract_change = changed_contracts
+                        .entry(balance_change.address.clone())
+                        .or_insert_with(|| {
+                            InterimContractChange::new(
+                                &balance_change.address,
+                                created_accounts.contains_key(&balance_change.address),
+                            )
+                        });
+
+                    if let Some(new_balance) = &balance_change.new_value {
+                        contract_change.balance.clear();
+                        contract_change
+                            .balance
+                            .extend_from_slice(&new_balance.bytes);
+                    }
+                });
+
+            code_changes
+                .iter()
+                .filter(|changes| inclusion_predicate(&changes.address))
+                .for_each(|code_change| {
+                    let contract_change = changed_contracts
+                        .entry(code_change.address.clone())
+                        .or_insert_with(|| {
+                            InterimContractChange::new(
+                                &code_change.address,
+                                created_accounts.contains_key(&code_change.address),
+                            )
+                        });
+
+                    contract_change.code.clear();
+                    contract_change
+                        .code
+                        .extend_from_slice(&code_change.new_code);
+                });
+
+            if !storage_changes.is_empty()
+                || !balance_changes.is_empty()
+                || !code_changes.is_empty()
+            {
+                transaction_contract_changes
+                    .entry(block_tx.index.into())
+                    .or_insert_with(|| tycho::TransactionContractChanges::new(&(block_tx.into())))
+                    .contract_changes
+                    .extend(
+                        changed_contracts
+                            .drain()
+                            .map(|(_, change)| change.into()),
+                    );
+            }
+        });
+}

crates/tycho-substreams/src/lib.rs

@@ -0,0 +1,9 @@
+pub mod balances;
+pub mod contract;
+mod mock_store;
+pub mod models;
+mod pb;
+
+pub mod prelude {
+    pub use super::models::*;
+}

crates/tycho-substreams/src/mock_store.rs

@@ -0,0 +1,93 @@
+//! Contains a mock store for internal testing.
+//!
+//! Might make this public alter to users can test their store handlers.
+use std::cell::RefCell;
+use std::collections::HashMap;
+use std::rc::Rc;
+use substreams::prelude::{BigInt, StoreDelete, StoreGet, StoreNew};
+use substreams::store::StoreAdd;
+
+#[derive(Debug, Clone)]
+pub struct MockStore {
+    data: Rc<RefCell<HashMap<String, Vec<(u64, BigInt)>>>>,
+}
+
+impl StoreDelete for MockStore {
+    fn delete_prefix(&self, _ord: i64, prefix: &String) {
+        self.data
+            .borrow_mut()
+            .retain(|k, _| !k.starts_with(prefix));
+    }
+}
+
+impl StoreNew for MockStore {
+    fn new() -> Self {
+        Self { data: Rc::new(RefCell::new(HashMap::new())) }
+    }
+}
+
+impl StoreAdd<BigInt> for MockStore {
+    fn add<K: AsRef<str>>(&self, ord: u64, key: K, value: BigInt) {
+        let mut guard = self.data.borrow_mut();
+        guard
+            .entry(key.as_ref().to_string())
+            .and_modify(|v| {
+                let prev_value = v.last().unwrap().1.clone();
+                v.push((ord, prev_value + value.clone()));
+            })
+            .or_insert(vec![(ord, value)]);
+    }
+
+    fn add_many<K: AsRef<str>>(&self, _ord: u64, _keys: &Vec<K>, _value: BigInt) {
+        todo!()
+    }
+}
+
+impl StoreGet<BigInt> for MockStore {
+    fn new(_idx: u32) -> Self {
+        Self { data: Rc::new(RefCell::new(HashMap::new())) }
+    }
+
+    fn get_at<K: AsRef<str>>(&self, ord: u64, key: K) -> Option<BigInt> {
+        self.data
+            .borrow()
+            .get(&key.as_ref().to_string())
+            .map(|v| {
+                v.iter()
+                    .find(|(current_ord, _)| *current_ord == ord)
+                    .unwrap()
+                    .1
+                    .clone()
+            })
+    }
+
+    fn get_last<K: AsRef<str>>(&self, key: K) -> Option<BigInt> {
+        self.data
+            .borrow()
+            .get(&key.as_ref().to_string())
+            .map(|v| v.last().unwrap().1.clone())
+    }
+
+    fn get_first<K: AsRef<str>>(&self, key: K) -> Option<BigInt> {
+        self.data
+            .borrow()
+            .get(&key.as_ref().to_string())
+            .map(|v| v.first().unwrap().1.clone())
+    }
+
+    fn has_at<K: AsRef<str>>(&self, ord: u64, key: K) -> bool {
+        self.data
+            .borrow()
+            .get(&key.as_ref().to_string())
+            .map(|v| v.iter().any(|(v, _)| *v == ord))
+            .unwrap_or(false)
+    }
+
+    fn has_last<K: AsRef<str>>(&self, _key: K) -> bool {
+        todo!()
+    }
+
+    fn has_first<K: AsRef<str>>(&self, _key: K) -> bool {
+        todo!()
+    }
+}

crates/tycho-substreams/src/models.rs

@@ -0,0 +1,123 @@
+use substreams_ethereum::pb::eth::v2::{self as sf};
+
+// re-export the protobuf types here.
+pub use crate::pb::tycho::evm::v1::*;
+
+impl TransactionContractChanges {
+    /// Creates a new empty `TransactionContractChanges` instance.
+    pub fn new(tx: &Transaction) -> Self {
+        Self {
+            tx: Some(tx.clone()),
+            contract_changes: vec![],
+            component_changes: vec![],
+            balance_changes: vec![],
+        }
+    }
+}
+
+impl From<&sf::TransactionTrace> for Transaction {
+    fn from(tx: &sf::TransactionTrace) -> Self {
+        Self {
+            hash: tx.hash.clone(),
+            from: tx.from.clone(),
+            to: tx.to.clone(),
+            index: tx.index.into(),
+        }
+    }
+}
+
+impl From<&sf::Block> for Block {
+    fn from(block: &sf::Block) -> Self {
+        Self {
+            number: block.number,
+            hash: block.hash.clone(),
+            parent_hash: block
+                .header
+                .as_ref()
+                .expect("Block header not present")
+                .parent_hash
+                .clone(),
+            ts: block.timestamp_seconds(),
+        }
+    }
+}
+
+impl ProtocolComponent {
+    /// Creates a new empty `ProtocolComponent` instance.
+    ///
+    /// You can use the `with_*` methods to set the fields in a convience way.
+    pub fn new(id: &str, tx: &Transaction) -> Self {
+        Self {
+            id: id.to_string(),
+            tokens: vec![],
+            contracts: vec![],
+            static_att: vec![],
+            change: ChangeType::Creation.into(),
+            protocol_type: None,
+            tx: Some(tx.clone()),
+        }
+    }
+
+    /// Shorthand to create a component with a 1-1 relationship to a contract.
+    ///
+    /// Will set the component id to a hex encoded address with a 0x prefix
+    /// and add the contract to contracts attributes.
+    pub fn at_contract(id: &[u8], tx: &Transaction) -> Self {
+        Self {
+            id: format!("0x{}", hex::encode(id)),
+            tokens: vec![],
+            contracts: vec![id.to_vec()],
+            static_att: vec![],
+            change: ChangeType::Creation.into(),
+            protocol_type: None,
+            tx: Some(tx.clone()),
+        }
+    }
+
+    /// Replaces the tokens on this component.
+    pub fn with_tokens<B: AsRef<[u8]>>(mut self, tokens: &[B]) -> Self {
+        self.tokens = tokens
+            .iter()
+            .map(|e| e.as_ref().to_vec())
+            .collect::<Vec<Vec<u8>>>();
+        self
+    }
+
+    /// Replaces the contracts associated with this component.
+    pub fn with_contracts<B: AsRef<[u8]>>(mut self, contracts: &[B]) -> Self {
+        self.contracts = contracts
+            .iter()
+            .map(|e| e.as_ref().to_vec())
+            .collect::<Vec<Vec<u8>>>();
+        self
+    }
+
+    /// Replaces the static attributes on this component.
+    ///
+    /// The change type will be set to Creation.
+    pub fn with_attributes<K: AsRef<str>, V: AsRef<[u8]>>(mut self, attributes: &[(K, V)]) -> Self {
+        self.static_att = attributes
+            .iter()
+            .map(|(k, v)| Attribute {
+                name: k.as_ref().to_string(),
+                value: v.as_ref().to_vec(),
+                change: ChangeType::Creation.into(),
+            })
+            .collect::<Vec<Attribute>>();
+        self
+    }
+
+    /// Sets the protocol_type on this component.
+    ///
+    /// Will set the `financial_type` to FinancialType::Swap and the
+    /// `attribute_schema` to an empty list.
+    pub fn as_swap_type(mut self, name: &str, implementation_type: ImplementationType) -> Self {
+        self.protocol_type = Some(ProtocolType {
+            name: name.to_string(),
+            financial_type: FinancialType::Swap.into(),
+            attribute_schema: vec![],
+            implementation_type: implementation_type.into(),
+        });
+        self
+    }
+}

crates/tycho-substreams/src/pb/tycho.evm.v1.rs

@@ -19,6 +19,7 @@ pub struct Block {
     pub ts: u64,
 }
 /// A struct describing a transaction.
+#[derive(Eq, Hash)]
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Transaction {
@@ -205,6 +206,93 @@ impl ImplementationType {
         }
     }
 }
+// This file contains the definition for the native integration of Substreams.
+
+/// A component is a set of attributes that are associated with a custom entity.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EntityChanges {
+    /// A unique identifier of the entity within the protocol.
+    #[prost(string, tag="1")]
+    pub component_id: ::prost::alloc::string::String,
+    /// The set of attributes that are associated with the entity.
+    #[prost(message, repeated, tag="2")]
+    pub attributes: ::prost::alloc::vec::Vec<Attribute>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TransactionEntityChanges {
+    #[prost(message, optional, tag="1")]
+    pub tx: ::core::option::Option<Transaction>,
+    #[prost(message, repeated, tag="2")]
+    pub entity_changes: ::prost::alloc::vec::Vec<EntityChanges>,
+    /// An array of newly added components.
+    #[prost(message, repeated, tag="3")]
+    pub component_changes: ::prost::alloc::vec::Vec<ProtocolComponent>,
+    /// An array of balance changes to components.
+    #[prost(message, repeated, tag="4")]
+    pub balance_changes: ::prost::alloc::vec::Vec<BalanceChange>,
+}
+/// A set of transaction changes within a single block.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BlockEntityChanges {
+    /// The block for which these changes are collectively computed.
+    #[prost(message, optional, tag="1")]
+    pub block: ::core::option::Option<Block>,
+    /// The set of transaction changes observed in the specified block.
+    #[prost(message, repeated, tag="2")]
+    pub changes: ::prost::alloc::vec::Vec<TransactionEntityChanges>,
+}
+/// A message containing relative balance changes.
+///
+/// Used to track token balances of protocol components in case they are only
+/// available as relative values within a block.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BalanceDelta {
+    /// The ordinal of the balance change. Must be unique & deterministic over all balances
+    /// changes within a block.
+    #[prost(uint64, tag="1")]
+    pub ord: u64,
+    /// The tx hash of the transaction that caused the balance change.
+    #[prost(message, optional, tag="2")]
+    pub tx: ::core::option::Option<Transaction>,
+    /// The address of the ERC20 token whose balance changed.
+    #[prost(bytes="vec", tag="3")]
+    pub token: ::prost::alloc::vec::Vec<u8>,
+    /// The delta balance of the token.
+    #[prost(bytes="vec", tag="4")]
+    pub delta: ::prost::alloc::vec::Vec<u8>,
+    /// The id of the component whose TVL is tracked.
+    /// If the protocol component includes multiple contracts, the balance change must be
+    /// aggregated to reflect how much tokens can be traded.
+    #[prost(bytes="vec", tag="5")]
+    pub component_id: ::prost::alloc::vec::Vec<u8>,
+}
+/// A set of balances deltas, usually a group of changes within a single block.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BlockBalanceDeltas {
+    #[prost(message, repeated, tag="1")]
+    pub balance_deltas: ::prost::alloc::vec::Vec<BalanceDelta>,
+}
+/// A message containing protocol components that were created by a single tx.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TransactionProtocolComponents {
+    #[prost(message, optional, tag="1")]
+    pub tx: ::core::option::Option<Transaction>,
+    #[prost(message, repeated, tag="2")]
+    pub components: ::prost::alloc::vec::Vec<ProtocolComponent>,
+}
+/// All protocol components that were created within a block with their corresponding tx.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BlockTransactionProtocolComponents {
+    #[prost(message, repeated, tag="1")]
+    pub tx_components: ::prost::alloc::vec::Vec<TransactionProtocolComponents>,
+}
 // This file contains proto definitions specific to the VM integration.
 
 /// A key value entry into contract storage.
@@ -267,53 +355,4 @@ pub struct BlockContractChanges {
     #[prost(message, repeated, tag="2")]
     pub changes: ::prost::alloc::vec::Vec<TransactionContractChanges>,
 }
-/// A message containing relative balance changes.
-///
-/// Used to track token balances of protocol components in case they are only
-/// available as relative values within a block.
-#[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct BalanceDelta {
-    /// The ordinal of the balance change. Must be unique & deterministic over all balances
-    /// changes within a block.
-    #[prost(uint64, tag="1")]
-    pub ord: u64,
-    /// The tx hash of the transaction that caused the balance change.
-    #[prost(message, optional, tag="2")]
-    pub tx: ::core::option::Option<Transaction>,
-    /// The address of the ERC20 token whose balance changed.
-    #[prost(bytes="vec", tag="3")]
-    pub token: ::prost::alloc::vec::Vec<u8>,
-    /// The delta balance of the token.
-    #[prost(bytes="vec", tag="4")]
-    pub delta: ::prost::alloc::vec::Vec<u8>,
-    /// The id of the component whose TVL is tracked.
-    /// If the protocol component includes multiple contracts, the balance change must be
-    /// aggregated to reflect how much tokens can be traded.
-    #[prost(bytes="vec", tag="5")]
-    pub component_id: ::prost::alloc::vec::Vec<u8>,
-}
-/// A set of balances deltas, usually a group of changes within a single block.
-#[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct BlockBalanceDeltas {
-    #[prost(message, repeated, tag="1")]
-    pub balance_deltas: ::prost::alloc::vec::Vec<BalanceDelta>,
-}
-/// A message containing protocol components that were created by a single tx.
-#[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct TransactionProtocolComponents {
-    #[prost(message, optional, tag="1")]
-    pub tx: ::core::option::Option<Transaction>,
-    #[prost(message, repeated, tag="2")]
-    pub components: ::prost::alloc::vec::Vec<ProtocolComponent>,
-}
-/// All protocol components that were created within a block with their corresponding tx.
-#[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct BlockTransactionProtocolComponents {
-    #[prost(message, repeated, tag="1")]
-    pub tx_components: ::prost::alloc::vec::Vec<TransactionProtocolComponents>,
-}
 // @@protoc_insertion_point(module)

substreams/ethereum-balancer/Cargo.lock

@@ -931,6 +931,7 @@ dependencies = [
  "prost-types 0.12.3",
  "substreams",
  "substreams-ethereum",
+ "tycho-substreams",
 ]
 
 [[package]]
@@ -1064,6 +1065,17 @@ dependencies = [
  "winnow",
 ]
 
+[[package]]
+name = "tycho-substreams"
+version = "0.1.0"
+dependencies = [
+ "hex",
+ "itertools 0.12.0",
+ "prost 0.11.9",
+ "substreams",
+ "substreams-ethereum",
+]
+
 [[package]]
 name = "typenum"
 version = "1.17.0"

substreams/ethereum-balancer/Cargo.toml

@@ -19,6 +19,7 @@ anyhow = "1.0.75"
 prost-types = "0.12.3"
 num-bigint = "0.4.4"
 itertools = "0.12.0"
+tycho-substreams = {path  ="../../crates/tycho-substreams"}
 
 [build-dependencies]
 anyhow = "1"

substreams/ethereum-balancer/src/contract_changes.rs

@@ -1,211 +0,0 @@
-/// This file contains helpers to capture contract changes from the expanded block model. These
-///  leverage the `code_changes`, `balance_changes`, and `storage_changes` fields available on the
-///  `Call` type provided by block model in a substream (i.e. `logs_and_calls`, etc).
-///
-/// ⚠️ These helpers *only* work if the **expanded block model** is available, more info blow.
-/// https://streamingfastio.medium.com/new-block-model-to-accelerate-chain-integration-9f65126e5425
-use std::collections::HashMap;
-
-use substreams_ethereum::pb::eth;
-
-use pb::tycho::evm::v1::{self as tycho};
-
-use substreams::store::{StoreGet, StoreGetInt64};
-
-use crate::pb;
-
-struct SlotValue {
-    new_value: Vec<u8>,
-    start_value: Vec<u8>,
-}
-
-impl SlotValue {
-    fn has_changed(&self) -> bool {
-        self.start_value != self.new_value
-    }
-}
-
-// Uses a map for slots, protobuf does not allow bytes in hashmap keys
-pub struct InterimContractChange {
-    address: Vec<u8>,
-    balance: Vec<u8>,
-    code: Vec<u8>,
-    slots: HashMap<Vec<u8>, SlotValue>,
-    change: tycho::ChangeType,
-}
-
-impl From<InterimContractChange> for tycho::ContractChange {
-    fn from(value: InterimContractChange) -> Self {
-        tycho::ContractChange {
-            address: value.address,
-            balance: value.balance,
-            code: value.code,
-            slots: value
-                .slots
-                .into_iter()
-                .filter(|(_, value)| value.has_changed())
-                .map(|(slot, value)| tycho::ContractSlot {
-                    slot,
-                    value: value.new_value,
-                })
-                .collect(),
-            change: value.change.into(),
-        }
-    }
-}
-
-pub fn extract_contract_changes(
-    block: &eth::v2::Block,
-    contracts: StoreGetInt64,
-    transaction_contract_changes: &mut HashMap<u64, tycho::TransactionContractChanges>,
-) {
-    let mut changed_contracts: HashMap<Vec<u8>, InterimContractChange> = HashMap::new();
-
-    // Collect all accounts created in this block
-    let created_accounts: HashMap<_, _> = block
-        .transactions()
-        .flat_map(|tx| {
-            tx.calls.iter().flat_map(|call| {
-                call.account_creations
-                    .iter()
-                    .map(|ac| (&ac.account, ac.ordinal))
-            })
-        })
-        .collect();
-
-    block.transactions().for_each(|block_tx| {
-        let mut storage_changes = Vec::new();
-        let mut balance_changes = Vec::new();
-        let mut code_changes = Vec::new();
-
-        block_tx
-            .calls
-            .iter()
-            .filter(|call| {
-                !call.state_reverted
-                    && contracts
-                        .get_last(format!("pool:{0}", hex::encode(&call.address)))
-                        .is_some()
-            })
-            .for_each(|call| {
-                storage_changes.extend(call.storage_changes.iter());
-                balance_changes.extend(call.balance_changes.iter());
-                code_changes.extend(call.code_changes.iter());
-            });
-
-        storage_changes.sort_unstable_by_key(|change| change.ordinal);
-        balance_changes.sort_unstable_by_key(|change| change.ordinal);
-        code_changes.sort_unstable_by_key(|change| change.ordinal);
-
-        storage_changes
-            .iter()
-            .filter(|changes| {
-                contracts
-                    .get_last(format!("pool:{0}", hex::encode(&changes.address)))
-                    .is_some()
-            })
-            .for_each(|storage_change| {
-                let contract_change = changed_contracts
-                    .entry(storage_change.address.clone())
-                    .or_insert_with(|| InterimContractChange {
-                        address: storage_change.address.clone(),
-                        balance: Vec::new(),
-                        code: Vec::new(),
-                        slots: HashMap::new(),
-                        change: if created_accounts.contains_key(&storage_change.address) {
-                            tycho::ChangeType::Creation
-                        } else {
-                            tycho::ChangeType::Update
-                        },
-                    });
-
-                let slot_value = contract_change
-                    .slots
-                    .entry(storage_change.key.clone())
-                    .or_insert_with(|| SlotValue {
-                        new_value: storage_change.new_value.clone(),
-                        start_value: storage_change.old_value.clone(),
-                    });
-
-                slot_value
-                    .new_value
-                    .copy_from_slice(&storage_change.new_value);
-            });
-
-        balance_changes
-            .iter()
-            .filter(|changes| {
-                contracts
-                    .get_last(format!("pool:{0}", hex::encode(&changes.address)))
-                    .is_some()
-            })
-            .for_each(|balance_change| {
-                let contract_change = changed_contracts
-                    .entry(balance_change.address.clone())
-                    .or_insert_with(|| InterimContractChange {
-                        address: balance_change.address.clone(),
-                        balance: Vec::new(),
-                        code: Vec::new(),
-                        slots: HashMap::new(),
-                        change: if created_accounts.contains_key(&balance_change.address) {
-                            tycho::ChangeType::Creation
-                        } else {
-                            tycho::ChangeType::Update
-                        },
-                    });
-
-                if let Some(new_balance) = &balance_change.new_value {
-                    contract_change.balance.clear();
-                    contract_change
-                        .balance
-                        .extend_from_slice(&new_balance.bytes);
-                }
-            });
-
-        code_changes
-            .iter()
-            .filter(|changes| {
-                contracts
-                    .get_last(format!("pool:{0}", hex::encode(&changes.address)))
-                    .is_some()
-            })
-            .for_each(|code_change| {
-                let contract_change = changed_contracts
-                    .entry(code_change.address.clone())
-                    .or_insert_with(|| InterimContractChange {
-                        address: code_change.address.clone(),
-                        balance: Vec::new(),
-                        code: Vec::new(),
-                        slots: HashMap::new(),
-                        change: if created_accounts.contains_key(&code_change.address) {
-                            tycho::ChangeType::Creation
-                        } else {
-                            tycho::ChangeType::Update
-                        },
-                    });
-
-                contract_change.code.clear();
-                contract_change
-                    .code
-                    .extend_from_slice(&code_change.new_code);
-            });
-
-        if !storage_changes.is_empty() || !balance_changes.is_empty() || !code_changes.is_empty() {
-            transaction_contract_changes
-                .entry(block_tx.index.into())
-                .or_insert_with(|| tycho::TransactionContractChanges {
-                    tx: Some(tycho::Transaction {
-                        hash: block_tx.hash.clone(),
-                        from: block_tx.from.clone(),
-                        to: block_tx.to.clone(),
-                        index: block_tx.index as u64,
-                    }),
-                    contract_changes: vec![],
-                    component_changes: vec![],
-                    balance_changes: vec![],
-                })
-                .contract_changes
-                .extend(changed_contracts.drain().map(|(_, change)| change.into()));
-        }
-    });
-}

substreams/ethereum-balancer/src/lib.rs

@@ -1,5 +1,3 @@
 mod abi;
-mod contract_changes;
 mod modules;
-mod pb;
 mod pool_factories;

substreams/ethereum-balancer/src/modules.rs

@@ -1,41 +1,20 @@
-use std::collections::HashMap;
+use crate::{abi, pool_factories};
-use std::str::FromStr;
-
 use anyhow::Result;
+use itertools::Itertools;
+use std::collections::HashMap;
 use substreams::hex;
 use substreams::pb::substreams::StoreDeltas;
 use substreams::store::{
     StoreAdd, StoreAddBigInt, StoreAddInt64, StoreGet, StoreGetInt64, StoreNew,
 };
-
-use substreams::key;
-use substreams::scalar::BigInt;
-
 use substreams_ethereum::pb::eth;
-
-use itertools::Itertools;
-
-
-use contract_changes::extract_contract_changes;
 use substreams_ethereum::Event;
-
+use tycho_substreams::balances::aggregate_balances_changes;
-use crate::pb::tycho::evm::v1::{self as tycho};
+use tycho_substreams::contract::extract_contract_changes;
-use crate::pb::tycho::evm::v1::{BalanceDelta, BlockBalanceDeltas, BlockTransactionProtocolComponents, TransactionProtocolComponents};
+use tycho_substreams::prelude::*;
-use crate::{abi, contract_changes, pool_factories};
 
 const VAULT_ADDRESS: &[u8] = &hex!("BA12222222228d8Ba445958a75a0704d566BF2C8");
 
-/// This struct purely exists to spoof the `PartialEq` trait for `Transaction` so we can use it in
-///  a later groupby operation.
-#[derive(Debug)]
-struct TransactionWrapper(tycho::Transaction);
-
-impl PartialEq for TransactionWrapper {
-    fn eq(&self, other: &Self) -> bool {
-        self.0.hash == other.0.hash
-    }
-}
-
 #[substreams::handlers::map]
 pub fn map_pools_created(block: eth::v2::Block) -> Result<BlockTransactionProtocolComponents> {
     // Gather contract changes by indexing `PoolCreated` events and analysing the `Create` call
@@ -52,24 +31,14 @@ pub fn map_pools_created(block: eth::v2::Block) -> Result<BlockTransactionProtoc
                             call.call.address.as_slice(),
                             log,
                             call.call,
-                            &tycho::Transaction {
+                            &(tx.into()),
-                                hash: tx.hash.clone(),
-                                from: tx.from.clone(),
-                                to: tx.to.clone(),
-                                index: tx.index.into(),
-                            },
                         )
                     })
                     .collect::<Vec<_>>();
 
                 if !components.is_empty() {
                     Some(TransactionProtocolComponents {
-                        tx: Some(tycho::Transaction {
+                        tx: Some(tx.into()),
-                            hash: tx.hash.clone(),
-                            from: tx.from.clone(),
-                            to: tx.to.clone(),
-                            index: Into::<u64>::into(tx.index),
-                        }),
                         components,
                     })
                 } else {
@@ -110,7 +79,13 @@ pub fn map_balance_deltas(
             if let Some(ev) =
                 abi::vault::events::PoolBalanceChanged::match_and_decode(vault_log.log)
             {
-                let component_id = ev.pool_id[..20].to_vec();
+                let component_id = format!(
+                    "0x{}",
+                    String::from_utf8(ev.pool_id[..20].to_vec()).unwrap()
+                )
+                .as_bytes()
+                .to_vec();
+
                 if store
                     .get_last(format!("pool:{}", hex::encode(&component_id)))
                     .is_some()
@@ -118,12 +93,7 @@ pub fn map_balance_deltas(
                     for (token, delta) in ev.tokens.iter().zip(ev.deltas.iter()) {
                         deltas.push(BalanceDelta {
                             ord: vault_log.ordinal(),
-                            tx: Some(tycho::Transaction {
+                            tx: Some(vault_log.receipt.transaction.into()),
-                                hash: vault_log.receipt.transaction.hash.clone(),
-                                from: vault_log.receipt.transaction.from.clone(),
-                                to: vault_log.receipt.transaction.to.clone(),
-                                index: vault_log.receipt.transaction.index.into(),
-                            }),
                             token: token.to_vec(),
                             delta: delta.to_signed_bytes_be(),
                             component_id: component_id.clone(),
@@ -131,7 +101,13 @@ pub fn map_balance_deltas(
                     }
                 }
             } else if let Some(ev) = abi::vault::events::Swap::match_and_decode(vault_log.log) {
-                let component_id = ev.pool_id[..20].to_vec();
+                let component_id = format!(
+                    "0x{}",
+                    String::from_utf8(ev.pool_id[..20].to_vec()).unwrap()
+                )
+                .as_bytes()
+                .to_vec();
+
                 if store
                     .get_last(format!("pool:{}", hex::encode(&component_id)))
                     .is_some()
@@ -139,24 +115,14 @@ pub fn map_balance_deltas(
                     deltas.extend_from_slice(&[
                         BalanceDelta {
                             ord: vault_log.ordinal(),
-                            tx: Some(tycho::Transaction {
+                            tx: Some(vault_log.receipt.transaction.into()),
-                                hash: vault_log.receipt.transaction.hash.clone(),
-                                from: vault_log.receipt.transaction.from.clone(),
-                                to: vault_log.receipt.transaction.to.clone(),
-                                index: vault_log.receipt.transaction.index.into(),
-                            }),
                             token: ev.token_in.to_vec(),
                             delta: ev.amount_in.to_signed_bytes_be(),
                             component_id: component_id.clone(),
                         },
                         BalanceDelta {
                             ord: vault_log.ordinal(),
-                            tx: Some(tycho::Transaction {
+                            tx: Some(vault_log.receipt.transaction.into()),
-                                hash: vault_log.receipt.transaction.hash.clone(),
-                                from: vault_log.receipt.transaction.from.clone(),
-                                to: vault_log.receipt.transaction.to.clone(),
-                                index: vault_log.receipt.transaction.index.into(),
-                            }),
                             token: ev.token_out.to_vec(),
                             delta: ev.amount_out.neg().to_signed_bytes_be(),
                             component_id,
@@ -176,17 +142,7 @@ pub fn map_balance_deltas(
 ///  store key to ensure that there's a unique balance being tallied for each.
 #[substreams::handlers::store]
 pub fn store_balance_changes(deltas: BlockBalanceDeltas, store: StoreAddBigInt) {
-    deltas.balance_deltas.iter().for_each(|delta| {
+    tycho_substreams::balances::store_balance_changes(deltas, store);
-        store.add(
-            delta.ord,
-            format!(
-                "pool:{0}:token:{1}",
-                hex::encode(&delta.component_id),
-                hex::encode(&delta.token)
-            ),
-            BigInt::from_signed_bytes_be(&delta.delta),
-        );
-    });
 }
 
 /// This is the main map that handles most of the indexing of this substream.
@@ -202,11 +158,10 @@ pub fn map_changes(
     deltas: BlockBalanceDeltas,
     components_store: StoreGetInt64,
     balance_store: StoreDeltas, // Note, this map module is using the `deltas` mode for the store.
-) -> Result<tycho::BlockContractChanges> {
+) -> Result<BlockContractChanges> {
     // We merge contract changes by transaction (identified by transaction index) making it easy to
     //  sort them at the very end.
-    let mut transaction_contract_changes: HashMap<_, tycho::TransactionContractChanges> =
+    let mut transaction_contract_changes: HashMap<_, TransactionContractChanges> = HashMap::new();
-        HashMap::new();
 
     // `ProtocolComponents` are gathered from `map_pools_created` which just need a bit of work to
     //   convert into `TransactionContractChanges`
@@ -215,82 +170,42 @@ pub fn map_changes(
         .iter()
         .for_each(|tx_component| {
             let tx = tx_component.tx.as_ref().unwrap();
-
             transaction_contract_changes
                 .entry(tx.index)
-                .or_insert_with(|| tycho::TransactionContractChanges {
+                .or_insert_with(|| TransactionContractChanges::new(&tx))
-                    tx: Some(tx.clone()),
-                    contract_changes: vec![],
-                    component_changes: vec![],
-                    balance_changes: vec![],
-                })
                 .component_changes
                 .extend_from_slice(&tx_component.components);
         });
 
     // Balance changes are gathered by the `StoreDelta` based on `PoolBalanceChanged` creating
-    //  `BlockBalanceDeltas`. We essentially just process the changes that occured to the `store` this
+    //  `BlockBalanceDeltas`. We essentially just process the changes that occurred to the `store` this
     //  block. Then, these balance changes are merged onto the existing map of tx contract changes,
     //  inserting a new one if it doesn't exist.
-    balance_store
+    aggregate_balances_changes(balance_store, deltas)
-        .deltas
         .into_iter()
-        .zip(deltas.balance_deltas)
+        .for_each(|(_, (tx, balances))| {
-        .map(|(store_delta, balance_delta)| {
-            let pool_id = key::segment_at(&store_delta.key, 1);
-            let token_id = key::segment_at(&store_delta.key, 3);
-            // store_delta.new_value is an ASCII string representing an integer
-            let ascii_string =
-                String::from_utf8(store_delta.new_value.clone()).expect("Invalid UTF-8 sequence");
-            let balance = BigInt::from_str(&ascii_string).expect("Failed to parse integer");
-            let big_endian_bytes_balance = balance.to_bytes_be().1;
-
-            (
-                balance_delta.tx.unwrap(),
-                tycho::BalanceChange {
-                    token: hex::decode(token_id).expect("Token ID not valid hex"),
-                    balance: big_endian_bytes_balance,
-                    component_id: pool_id.as_bytes().to_vec(),
-                },
-            )
-        })
-        // We need to group the balance changes by tx hash for the `TransactionContractChanges` agg
-        .group_by(|(tx, _)| TransactionWrapper(tx.clone()))
-        .into_iter()
-        .for_each(|(tx_wrapped, group)| {
-            let tx = tx_wrapped.0;
-
             transaction_contract_changes
                 .entry(tx.index)
-                .or_insert_with(|| tycho::TransactionContractChanges {
+                .or_insert_with(|| TransactionContractChanges::new(&tx))
-                    tx: Some(tx.clone()),
-                    contract_changes: vec![],
-                    component_changes: vec![],
-                    balance_changes: vec![],
-                })
                 .balance_changes
-                .extend(group.map(|(_, change)| change));
+                .extend(balances.into_iter().map(|(_, change)| change));
         });
 
-    // General helper for extracting contract changes. Uses block, our component store which holds
+    // Extract and insert any storage changes that happened for any of the components.
-    //  all of our tracked deployed pool addresses, and the map of tx contract changes which we
+    extract_contract_changes(
-    //  output into for final processing later.
+        &block,
-    extract_contract_changes(&block, components_store, &mut transaction_contract_changes);
+        |addr| {
+            components_store
+                .get_last(format!("pool:{0}", hex::encode(&addr)))
+                .is_some()
+        },
+        &mut transaction_contract_changes,
+    );
 
     // Process all `transaction_contract_changes` for final output in the `BlockContractChanges`,
     //  sorted by transaction index (the key).
-    Ok(tycho::BlockContractChanges {
+    Ok(BlockContractChanges {
-        block: Some(tycho::Block {
+        block: Some((&block).into()),
-            number: block.number,
-            hash: block.hash.clone(),
-            parent_hash: block
-                .header
-                .as_ref()
-                .expect("Block header not present")
-                .parent_hash
-                .clone(),
-            ts: block.timestamp_seconds(),
-        }),
         changes: transaction_contract_changes
             .drain()
             .sorted_unstable_by_key(|(index, _)| *index)

substreams/ethereum-balancer/src/pool_factories.rs

@@ -1,13 +1,9 @@
+use crate::abi;
+use substreams::hex;
+use substreams::scalar::BigInt;
 use substreams_ethereum::pb::eth::v2::{Call, Log};
 use substreams_ethereum::{Event, Function};
-
+use tycho_substreams::prelude::*;
-use crate::abi;
-use crate::pb;
-use crate::pb::tycho::evm::v1::{FinancialType, ImplementationType, ProtocolType, Transaction};
-use pb::tycho::evm::v1::{self as tycho};
-use substreams::hex;
-
-use substreams::scalar::BigInt;
 
 /// This trait defines some helpers for serializing and deserializing `Vec<BigInt` which is needed
 ///  to be able to encode the `normalized_weights` and `weights` `Attribute`s. This should also be
@@ -39,16 +35,16 @@ impl SerializableVecBigInt for Vec<BigInt> {
 ///  - Stable Pool Factories
 /// (Balancer does have a bit more (esp. in the deprecated section) that could be implemented as
 ///  desired.)
-/// We use the specific ABIs to decode both the log event and cooresponding call to gather
+/// We use the specific ABIs to decode both the log event and corresponding call to gather
 ///  `PoolCreated` event information alongside the `Create` call data that provide us details to
-///  fufill both the required details + any extra `Attributes`
+///  fulfill both the required details + any extra `Attributes`
 /// Ref: https://docs.balancer.fi/reference/contracts/deployment-addresses/mainnet.html
 pub fn address_map(
     pool_factory_address: &[u8],
     log: &Log,
     call: &Call,
     tx: &Transaction,
-) -> Option<tycho::ProtocolComponent> {
+) -> Option<ProtocolComponent> {
     match *pool_factory_address {
         hex!("897888115Ada5773E02aA29F775430BFB5F34c51") => {
             let create_call =
@@ -56,31 +52,18 @@ pub fn address_map(
             let pool_created =
                 abi::weighted_pool_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: create_call.tokens,
+                    .with_tokens(&create_call.tokens)
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[
-                static_att: vec![
+                        ("pool_type", "WeightedPoolFactory".as_bytes()),
-                    tycho::Attribute {
+                        (
-                        name: "pool_type".into(),
+                            "normalized_weights",
-                        value: "WeightedPoolFactory".into(),
+                            &create_call.normalized_weights.serialize_bytes(),
-                        change: tycho::ChangeType::Creation.into(),
+                        ),
-                    },
+                    ])
-                    tycho::Attribute {
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                        name: "normalized_weights".into(),
+            )
-                        value: create_call.normalized_weights.serialize_bytes(),
-                        change: tycho::ChangeType::Creation.into(),
-                    },
-                ],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
         hex!("DB8d758BCb971e482B2C45f7F8a7740283A1bd3A") => {
             let create_call =
@@ -88,24 +71,12 @@ pub fn address_map(
             let pool_created =
                 abi::composable_stable_pool_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: create_call.tokens,
+                    .with_tokens(&create_call.tokens)
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[("pool_type", "ComposableStablePoolFactory".as_bytes())])
-                static_att: vec![tycho::Attribute {
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                    name: "pool_type".into(),
+            )
-                    value: "ComposableStablePoolFactory".into(),
-                    change: tycho::ChangeType::Creation.into(),
-                }],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
         hex!("813EE7a840CE909E7Fea2117A44a90b8063bd4fd") => {
             let create_call =
@@ -113,33 +84,18 @@ pub fn address_map(
             let pool_created =
                 abi::erc_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: vec![create_call.main_token, create_call.wrapped_token],
+                    .with_tokens(&[create_call.main_token, create_call.wrapped_token])
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[
-                static_att: vec![
+                        ("pool_type", "ERC4626LinearPoolFactory".as_bytes()),
-                    tycho::Attribute {
+                        (
-                        name: "pool_type".into(),
+                            "upper_target",
-                        value: "ERC4626LinearPoolFactory".into(),
+                            &create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
+                        ),
-                    },
+                    ])
-                    tycho::Attribute {
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                        name: "upper_target".into(),
+            )
-                        value: create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
-                    },
-                    // Note, `lower_target` is generally hardcoded for all pools, not located in call data
-                    // Note, rate provider might be provided as `create.protocol_id`, but as a BigInt. needs investigation
-                ],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
         hex!("5F43FBa61f63Fa6bFF101a0A0458cEA917f6B347") => {
             let create_call =
@@ -147,31 +103,18 @@ pub fn address_map(
             let pool_created =
                 abi::euler_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: vec![create_call.main_token, create_call.wrapped_token],
+                    .with_tokens(&[create_call.main_token, create_call.wrapped_token])
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[
-                static_att: vec![
+                        ("pool_type", "EulerLinearPoolFactory".as_bytes()),
-                    tycho::Attribute {
+                        (
-                        name: "pool_type".into(),
+                            "upper_target",
-                        value: "EulerLinearPoolFactory".into(),
+                            &create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
+                        ),
-                    },
+                    ])
-                    tycho::Attribute {
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                        name: "upper_target".into(),
+            )
-                        value: create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
-                    },
-                ],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
         // ❌ Reading the deployed factory for Gearbox showcases that it's currently disabled
         // hex!("39A79EB449Fc05C92c39aA6f0e9BfaC03BE8dE5B") => {
@@ -226,31 +169,18 @@ pub fn address_map(
             let pool_created =
                 abi::silo_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: vec![create_call.main_token, create_call.wrapped_token],
+                    .with_tokens(&[create_call.main_token, create_call.wrapped_token])
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[
-                static_att: vec![
+                        ("pool_type", "SiloLinearPoolFactory".as_bytes()),
-                    tycho::Attribute {
+                        (
-                        name: "pool_type".into(),
+                            "upper_target",
-                        value: "SiloLinearPoolFactory".into(),
+                            &create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
+                        ),
-                    },
+                    ])
-                    tycho::Attribute {
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                        name: "upper_target".into(),
+            )
-                        value: create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
-                    },
-                ],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
         hex!("5F5222Ffa40F2AEd6380D022184D6ea67C776eE0") => {
             let create_call =
@@ -258,65 +188,36 @@ pub fn address_map(
             let pool_created =
                 abi::yearn_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: vec![create_call.main_token, create_call.wrapped_token],
+                    .with_tokens(&[create_call.main_token, create_call.wrapped_token])
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[
-                static_att: vec![
+                        ("pool_type", "YearnLinearPoolFactory".as_bytes()),
-                    tycho::Attribute {
+                        (
-                        name: "pool_type".into(),
+                            "upper_target",
-                        value: "YearnLinearPoolFactory".into(),
+                            &create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
+                        ),
-                    },
+                    ])
-                    tycho::Attribute {
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                        name: "upper_target".into(),
+            )
-                        value: create_call.upper_target.to_signed_bytes_be(),
-                        change: tycho::ChangeType::Creation.into(),
-                    },
-                ],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
-        // The `WeightedPool2TokenFactory` is a deprecated contract but we've included it since one
+        // The `WeightedPool2TokenFactory` is a deprecated contract, but we've included
-        //  of the highest TVL pools, 80BAL-20WETH, is able to be tracked.
+        // it to be able to track one of the highest TVL pools: 80BAL-20WETH.
         hex!("A5bf2ddF098bb0Ef6d120C98217dD6B141c74EE0") => {
             let create_call =
                 abi::weighted_pool_tokens_factory::functions::Create::match_and_decode(call)?;
             let pool_created =
                 abi::weighted_pool_tokens_factory::events::PoolCreated::match_and_decode(log)?;
 
-            Some(tycho::ProtocolComponent {
+            Some(
-                id: hex::encode(&pool_created.pool),
+                ProtocolComponent::at_contract(&pool_created.pool, &tx)
-                tokens: create_call.tokens,
+                    .with_tokens(&create_call.tokens)
-                contracts: vec![pool_created.pool],
+                    .with_attributes(&[
-                static_att: vec![
+                        ("pool_type", "WeightedPool2TokensFactory".as_bytes()),
-                    tycho::Attribute {
+                        ("weights", &create_call.weights.serialize_bytes()),
-                        name: "pool_type".into(),
+                    ])
-                        value: "WeightedPool2TokensFactory".into(),
+                    .as_swap_type("balancer_pool", ImplementationType::Vm),
-                        change: tycho::ChangeType::Creation.into(),
+            )
-                    },
-                    tycho::Attribute {
-                        name: "weights".into(),
-                        value: create_call.weights.serialize_bytes(),
-                        change: tycho::ChangeType::Creation.into(),
-                    },
-                ],
-                change: tycho::ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
-                    name: "balancer_pool".to_string(),
-                    financial_type: FinancialType::Swap.into(),
-                    attribute_schema: vec![],
-                    implementation_type: ImplementationType::Vm.into(),
-                }),
-                tx: Some(tx.clone()),
-            })
         }
         _ => None,
     }