Add shared tycho-substreams lib.

crates/tycho-substreams/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "tycho-substreams"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+substreams-ethereum = "0.9.9"

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/src/contract.rs

@@ -0,0 +1,199 @@
+/// 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 [here](https://streamingfastio.medium.com/new-block-model-to-accelerate-chain-integration-9f65126e5425)
+use std::collections::HashMap;
+
+use substreams_ethereum::pb::eth;
+
+use crate::pb::tycho::evm::v1::{self as tycho};
+
+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<F: Fn(&[u8]) -> bool >(
+    block: &eth::v2::Block,
+    inclusion_predicate: F,
+    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
+                    && 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 {
+                        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| {
+                inclusion_predicate(&changes.address)
+            })
+            .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| {
+                inclusion_predicate(&changes.address)
+            })
+            .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()));
+        }
+    });
+}

crates/tycho-substreams/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod contract;
+pub mod pb;

crates/tycho-substreams/src/pb/mod.rs

@@ -0,0 +1,10 @@
+// @generated
+pub mod tycho {
+    pub mod evm {
+        // @@protoc_insertion_point(attribute:tycho.evm.v1)
+        pub mod v1 {
+            include!("tycho.evm.v1.rs");
+            // @@protoc_insertion_point(tycho.evm.v1)
+        }
+    }
+}

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

@@ -0,0 +1,309 @@
+// @generated
+// This file contains the proto definitions for Substreams common to all integrations.
+
+/// A struct describing a block.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Block {
+    /// The blocks hash.
+    #[prost(bytes="vec", tag="1")]
+    pub hash: ::prost::alloc::vec::Vec<u8>,
+    /// The parent blocks hash.
+    #[prost(bytes="vec", tag="2")]
+    pub parent_hash: ::prost::alloc::vec::Vec<u8>,
+    /// The block number.
+    #[prost(uint64, tag="3")]
+    pub number: u64,
+    /// The block timestamp.
+    #[prost(uint64, tag="4")]
+    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 {
+    /// The transaction hash.
+    #[prost(bytes="vec", tag="1")]
+    pub hash: ::prost::alloc::vec::Vec<u8>,
+    /// The sender of the transaction.
+    #[prost(bytes="vec", tag="2")]
+    pub from: ::prost::alloc::vec::Vec<u8>,
+    /// The receiver of the transaction.
+    #[prost(bytes="vec", tag="3")]
+    pub to: ::prost::alloc::vec::Vec<u8>,
+    /// The transactions index within the block.
+    /// TODO: should this be uint32? to match the type from the native substream type?
+    #[prost(uint64, tag="4")]
+    pub index: u64,
+}
+/// A custom struct representing an arbitrary attribute of a protocol component.
+/// This is mainly used by the native integration to track the necessary information about the protocol.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Attribute {
+    /// The name of the attribute.
+    #[prost(string, tag="1")]
+    pub name: ::prost::alloc::string::String,
+    /// The value of the attribute.
+    #[prost(bytes="vec", tag="2")]
+    pub value: ::prost::alloc::vec::Vec<u8>,
+    /// The type of change the attribute underwent.
+    #[prost(enumeration="ChangeType", tag="3")]
+    pub change: i32,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ProtocolType {
+    #[prost(string, tag="1")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(enumeration="FinancialType", tag="2")]
+    pub financial_type: i32,
+    #[prost(message, repeated, tag="3")]
+    pub attribute_schema: ::prost::alloc::vec::Vec<Attribute>,
+    #[prost(enumeration="ImplementationType", tag="4")]
+    pub implementation_type: i32,
+}
+/// A struct describing a part of the protocol.
+/// Note: For example this can be a UniswapV2 pair, that tracks the two ERC20 tokens used by the pair, 
+/// the component would represent a single contract. In case of VM integration, such component would 
+/// not need any attributes, because all the relevant info would be tracked via storage slots and balance changes.
+/// It can also be a wrapping contract, like WETH, that has a constant price, but it allows swapping tokens. 
+/// This is why the name ProtocolComponent is used instead of "Pool" or "Pair".
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ProtocolComponent {
+    /// A unique identifier for the component within the protocol.
+    /// Can be e.g. a stringified address or a string describing the trading pair.
+    #[prost(string, tag="1")]
+    pub id: ::prost::alloc::string::String,
+    /// Addresses of the ERC20 tokens used by the component.
+    #[prost(bytes="vec", repeated, tag="2")]
+    pub tokens: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    /// Addresses of the contracts used by the component.
+    /// Usually it is a single contract, but some protocols use multiple contracts.
+    #[prost(bytes="vec", repeated, tag="3")]
+    pub contracts: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    /// Attributes of the component. Used mainly be the native integration.
+    /// The inner ChangeType of the attribute has to match the ChangeType of the ProtocolComponent.
+    #[prost(message, repeated, tag="4")]
+    pub static_att: ::prost::alloc::vec::Vec<Attribute>,
+    /// Type of change the component underwent.
+    #[prost(enumeration="ChangeType", tag="5")]
+    pub change: i32,
+    /// / Represents the functionality of the component.
+    #[prost(message, optional, tag="6")]
+    pub protocol_type: ::core::option::Option<ProtocolType>,
+    /// Transaction where this component was created
+    #[prost(message, optional, tag="7")]
+    pub tx: ::core::option::Option<Transaction>,
+}
+/// A struct for following the changes of Total Value Locked (TVL) of a protocol component.
+/// Note that if a ProtocolComponent contains multiple contracts, the TVL is tracked for the component as a whole.
+/// E.g. for UniswapV2 pair WETH/USDC, this tracks the USDC and WETH balance of the pair contract.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BalanceChange {
+    /// The address of the ERC20 token whose balance changed.
+    #[prost(bytes="vec", tag="1")]
+    pub token: ::prost::alloc::vec::Vec<u8>,
+    /// The new balance of the token.
+    #[prost(bytes="vec", tag="2")]
+    pub balance: ::prost::alloc::vec::Vec<u8>,
+    /// The id of the component whose TVL is tracked.  Note: This MUST be utf8 encoded.
+    /// 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="3")]
+    pub component_id: ::prost::alloc::vec::Vec<u8>,
+}
+/// Enum to specify the type of a change.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum ChangeType {
+    Unspecified = 0,
+    Update = 1,
+    Creation = 2,
+    Deletion = 3,
+}
+impl ChangeType {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            ChangeType::Unspecified => "CHANGE_TYPE_UNSPECIFIED",
+            ChangeType::Update => "CHANGE_TYPE_UPDATE",
+            ChangeType::Creation => "CHANGE_TYPE_CREATION",
+            ChangeType::Deletion => "CHANGE_TYPE_DELETION",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "CHANGE_TYPE_UNSPECIFIED" => Some(Self::Unspecified),
+            "CHANGE_TYPE_UPDATE" => Some(Self::Update),
+            "CHANGE_TYPE_CREATION" => Some(Self::Creation),
+            "CHANGE_TYPE_DELETION" => Some(Self::Deletion),
+            _ => None,
+        }
+    }
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum FinancialType {
+    Swap = 0,
+    Lend = 1,
+    Leverage = 2,
+    Psm = 3,
+}
+impl FinancialType {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            FinancialType::Swap => "SWAP",
+            FinancialType::Lend => "LEND",
+            FinancialType::Leverage => "LEVERAGE",
+            FinancialType::Psm => "PSM",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "SWAP" => Some(Self::Swap),
+            "LEND" => Some(Self::Lend),
+            "LEVERAGE" => Some(Self::Leverage),
+            "PSM" => Some(Self::Psm),
+            _ => None,
+        }
+    }
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum ImplementationType {
+    Vm = 0,
+    Custom = 1,
+}
+impl ImplementationType {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            ImplementationType::Vm => "VM",
+            ImplementationType::Custom => "CUSTOM",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "VM" => Some(Self::Vm),
+            "CUSTOM" => Some(Self::Custom),
+            _ => None,
+        }
+    }
+}
+// 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>,
+}
+// This file contains proto definitions specific to the VM integration.
+
+/// A key value entry into contract storage.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ContractSlot {
+    /// A contract's storage slot.
+    #[prost(bytes="vec", tag="2")]
+    pub slot: ::prost::alloc::vec::Vec<u8>,
+    /// The new value for this storage slot.
+    #[prost(bytes="vec", tag="3")]
+    pub value: ::prost::alloc::vec::Vec<u8>,
+}
+/// Changes made to a single contract's state.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ContractChange {
+    /// The contract's address
+    #[prost(bytes="vec", tag="1")]
+    pub address: ::prost::alloc::vec::Vec<u8>,
+    /// The new native balance of the contract, empty bytes indicates no change.
+    #[prost(bytes="vec", tag="2")]
+    pub balance: ::prost::alloc::vec::Vec<u8>,
+    /// The new code of the contract, empty bytes indicates no change.
+    #[prost(bytes="vec", tag="3")]
+    pub code: ::prost::alloc::vec::Vec<u8>,
+    /// The changes to this contract's slots, empty sequence indicates no change.
+    #[prost(message, repeated, tag="4")]
+    pub slots: ::prost::alloc::vec::Vec<ContractSlot>,
+    /// Whether this is an update, a creation or a deletion.
+    #[prost(enumeration="ChangeType", tag="5")]
+    pub change: i32,
+}
+/// A set of changes aggregated by transaction.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TransactionContractChanges {
+    /// The transaction instance that results in the changes.
+    #[prost(message, optional, tag="1")]
+    pub tx: ::core::option::Option<Transaction>,
+    /// Contains the changes induced by the above transaction, aggregated on a per-contract basis.
+    /// Must include changes to every contract that is tracked by all ProtocolComponents.
+    #[prost(message, repeated, tag="2")]
+    pub contract_changes: ::prost::alloc::vec::Vec<ContractChange>,
+    /// An array of any component changes.
+    #[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 BlockContractChanges {
+    /// 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<TransactionContractChanges>,
+}
+// @@protoc_insertion_point(module)