Liquidity.Party/tycho-protocol-sdk
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #37 from propeller-heads/zz/update-protobuf

Browse Source 
Update protobuf and Balancer Substreams
This commit is contained in:
Zizou
2024-07-23 15:19:14 +02:00
committed by GitHub
parent ead8e17082 bd0a077917
commit 96f67362bf
14 changed files with 316 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
proto/tycho/evm/v1/common.proto
View File

@@ -106,3 +106,63 @@ message BalanceChange {
// If the protocol component includes multiple contracts, the balance change must be aggregated to reflect how much tokens can be traded.
bytes component_id = 3;
}
// Native entities
// A component is a set of attributes that are associated with a custom entity.
message EntityChanges {
// A unique identifier of the entity within the protocol.
string component_id = 1;
// The set of attributes that are associated with the entity.
repeated Attribute attributes = 2;
}
// VM entities
// A key value entry into contract storage.
message ContractSlot {
// A contract's storage slot.
bytes slot = 2;
// The new value for this storage slot.
bytes value = 3;
}
// Changes made to a single contract's state.
message ContractChange {
// The contract's address
bytes address = 1;
// The new balance of the contract, empty bytes indicates no change.
bytes balance = 2;
// The new code of the contract, empty bytes indicates no change.
bytes code = 3;
// The changes to this contract's slots, empty sequence indicates no change.
repeated ContractSlot slots = 4;
// Whether this is an update, a creation or a deletion.
ChangeType change = 5;
}
// Aggregate entities
// A set of changes aggregated by transaction.
message TransactionChanges {
// The transaction instance that results in the changes.
Transaction tx = 1;
// Contains the changes induced by the above transaction, aggregated on a per-contract basis.
// Contains the contract changes induced by the above transaction, usually for tracking VM components.
repeated ContractChange contract_changes = 2;
// Contains the entity changes induced by the above transaction.
// Usually for tracking native components or used for VM extensions (plugins).
repeated EntityChanges entity_changes = 3;
// An array of newly added components.
repeated ProtocolComponent component_changes = 4;
// An array of balance changes to components.
repeated BalanceChange balance_changes = 5;
}
// A set of transaction changes within a single block.
message BlockChanges {
// The block for which these changes are collectively computed.
Block block = 1;
// The set of transaction changes observed in the specified block.
repeated TransactionChanges changes = 2;
}

9
proto/tycho/evm/v1/entity.proto
View File

@@ -4,16 +4,9 @@ package tycho.evm.v1;
import "tycho/evm/v1/common.proto";
// WARNING: DEPRECATED. Please use common.proto's TransactionChanges and BlockChanges instead.
// 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.
message EntityChanges {
// A unique identifier of the entity within the protocol.
string component_id = 1;
// The set of attributes that are associated with the entity.
repeated Attribute attributes = 2;
}
message TransactionEntityChanges {
Transaction tx = 1;
repeated EntityChanges entity_changes = 2;

26
proto/tycho/evm/v1/vm.proto
View File

@@ -4,38 +4,16 @@ package tycho.evm.v1;
import "tycho/evm/v1/common.proto";
// WARNING: DEPRECATED. Please use common.proto's TransactionChanges and BlockChanges instead.
// This file contains proto definitions specific to the VM integration.
// A key value entry into contract storage.
message ContractSlot {
// A contract's storage slot.
bytes slot = 2;
// The new value for this storage slot.
bytes value = 3;
}
// Changes made to a single contract's state.
message ContractChange {
// The contract's address
bytes address = 1;
// The new native balance of the contract, empty bytes indicates no change.
bytes balance = 2;
// The new code of the contract, empty bytes indicates no change.
bytes code = 3;
// The changes to this contract's slots, empty sequence indicates no change.
repeated ContractSlot slots = 4;
// Whether this is an update, a creation or a deletion.
ChangeType change = 5;
}
// A set of changes aggregated by transaction.
message TransactionContractChanges {
// The transaction instance that results in the changes.
Transaction tx = 1;
// 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.
repeated ContractChange contract_changes = 2;
// An array of any component changes.
// An array of newly added components.
repeated ProtocolComponent component_changes = 3;
// An array of balance changes to components.
repeated BalanceChange balance_changes = 4;

2
substreams/Cargo.lock generated
View File

@@ -210,7 +210,7 @@ dependencies = [
[[package]]
name = "ethereum-balancer"
version = "0.1.0"
version = "0.2.0"
dependencies = [
"anyhow",
"bytes",

12
substreams/crates/tycho-substreams/Changelog.md Normal file
View File

@@ -0,0 +1,12 @@
# Changelog
## 0.2.0
### Updated
- Protobuf struct updated to align with recent changes in the indexer.
### Changed
- Removed the distinction between VM and native implementations. Now, there is a single implementation type that can extract both contracts and protocol state.
- Enabled the attachment of dynamic attributes to protocol components.

13
substreams/crates/tycho-substreams/src/contract.rs
View File

@@ -83,9 +83,8 @@ impl From<InterimContractChange> for tycho::ContractChange {
/// model.
/// * `inclusion_predicate` - A closure that determines if a contract's address is of interest for
/// the collection of changes. Only contracts satisfying this predicate are included.
/// * `transaction_contract_changes` - A mutable reference to a map where extracted contract changes
/// are stored. Keyed by transaction index, it aggregates changes into
/// `tycho::TransactionContractChanges`.
/// * `transaction_changes` - A mutable reference to a map where extracted contract changes are
/// stored. Keyed by transaction index, it aggregates changes into `tycho::TransactionChanges`.
///
/// ## Panics
/// Panics if the provided block is not an extended block model, as indicated by its detail level.
@@ -94,7 +93,7 @@ impl From<InterimContractChange> for tycho::ContractChange {
/// The function iterates over transactions and their calls within the block, collecting contract
/// changes (storage, balance, code) that pass the inclusion predicate. Changes are then sorted by
/// their ordinals to maintain the correct sequence of events. Aggregated changes for each contract
/// are stored in `transaction_contract_changes`, categorized by transaction index.
/// are stored in `transaction_changes`, categorized by transaction index.
///
/// Contracts created within the block are tracked to differentiate between new and existing
/// contracts. The aggregation process respects transaction boundaries, ensuring that changes are
@@ -102,7 +101,7 @@ impl From<InterimContractChange> for tycho::ContractChange {
pub fn extract_contract_changes<F: Fn(&[u8]) -> bool>(
block: &eth::v2::Block,
inclusion_predicate: F,
transaction_contract_changes: &mut HashMap<u64, tycho::TransactionContractChanges>,
transaction_changes: &mut HashMap<u64, tycho::TransactionChanges>,
) {
if block.detail_level != Into::<i32>::into(DetailLevel::DetaillevelExtended) {
panic!("Only extended blocks are supported");
@@ -215,9 +214,9 @@ pub fn extract_contract_changes<F: Fn(&[u8]) -> bool>(
!balance_changes.is_empty() ||
!code_changes.is_empty()
{
transaction_contract_changes
transaction_changes
.entry(block_tx.index.into())
.or_insert_with(|| tycho::TransactionContractChanges::new(&(block_tx.into())))
.or_insert_with(|| tycho::TransactionChanges::new(&(block_tx.into())))
.contract_changes
.extend(
changed_contracts

12
substreams/crates/tycho-substreams/src/models.rs
View File

@@ -6,12 +6,14 @@ 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![],
Self { tx: Some(tx.clone()), ..Default::default() }
}
}
impl TransactionChanges {
/// Creates a new empty `TransactionChanges` instance.
pub fn new(tx: &Transaction) -> Self {
Self { tx: Some(tx.clone()), ..Default::default() }
}
}

128
substreams/crates/tycho-substreams/src/pb/tycho.evm.v1.rs
View File

@@ -115,6 +115,87 @@ pub struct BalanceChange {
#[prost(bytes="vec", tag="3")]
pub component_id: ::prost::alloc::vec::Vec<u8>,
}
// Native entities
/// 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>,
}
// VM entities
/// 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 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,
}
// Aggregate entities
/// A set of changes aggregated by transaction.
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct TransactionChanges {
/// 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.
/// Contains the contract changes induced by the above transaction, usually for tracking VM components.
#[prost(message, repeated, tag="2")]
pub contract_changes: ::prost::alloc::vec::Vec<ContractChange>,
/// Contains the entity changes induced by the above transaction.
/// Usually for tracking native components or used for VM extensions (plugins).
#[prost(message, repeated, tag="3")]
pub entity_changes: ::prost::alloc::vec::Vec<EntityChanges>,
/// An array of newly added components.
#[prost(message, repeated, tag="4")]
pub component_changes: ::prost::alloc::vec::Vec<ProtocolComponent>,
/// An array of balance changes to components.
#[prost(message, repeated, tag="5")]
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 BlockChanges {
/// 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<TransactionChanges>,
}
/// Enum to specify the type of a change.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
@@ -206,19 +287,9 @@ impl ImplementationType {
}
}
}
// WARNING: DEPRECATED. Please use common.proto's TransactionChanges and BlockChanges instead.
// 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 {
@@ -293,39 +364,9 @@ pub struct BlockTransactionProtocolComponents {
#[prost(message, repeated, tag="1")]
pub tx_components: ::prost::alloc::vec::Vec<TransactionProtocolComponents>,
}
// WARNING: DEPRECATED. Please use common.proto's TransactionChanges and BlockChanges instead.
// 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)]
@@ -334,10 +375,9 @@ pub struct TransactionContractChanges {
#[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.
/// 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.

2
substreams/ethereum-balancer/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "ethereum-balancer"
version = "0.1.0"
version = "0.2.0"
edition = "2021"
[lib]

59
substreams/ethereum-balancer/src/modules.rs
View File

@@ -12,7 +12,7 @@ use tycho_substreams::{
balances::aggregate_balances_changes, contract::extract_contract_changes, prelude::*,
};
const VAULT_ADDRESS: &[u8] = &hex!("BA12222222228d8Ba445958a75a0704d566BF2C8");
pub const VAULT_ADDRESS: &[u8] = &hex!("BA12222222228d8Ba445958a75a0704d566BF2C8");
#[substreams::handlers::map]
pub fn map_components(block: eth::v2::Block) -> Result<BlockTransactionProtocolComponents> {
@@ -24,13 +24,12 @@ pub fn map_components(block: eth::v2::Block) -> Result<BlockTransactionProtocolC
.filter_map(|tx| {
let components = tx
.logs_with_calls()
.filter(|(_, call)| !call.call.state_reverted)
.filter_map(|(log, call)| {
pool_factories::address_map(
call.call.address.as_slice(),
log,
call.call,
&(tx.into()),
tx,
)
})
.collect::<Vec<_>>();
@@ -132,11 +131,11 @@ pub fn store_balances(deltas: BlockBalanceDeltas, store: StoreAddBigInt) {
}
/// This is the main map that handles most of the indexing of this substream.
/// Every contract change is grouped by transaction index via the `transaction_contract_changes`
/// map. Each block of code will extend the `TransactionContractChanges` struct with the
/// Every contract change is grouped by transaction index via the `transaction_changes`
/// map. Each block of code will extend the `TransactionChanges` struct with the
/// cooresponding changes (balance, component, contract), inserting a new one if it doesn't exist.
/// At the very end, the map can easily be sorted by index to ensure the final
/// `BlockContractChanges` is ordered by transactions properly.
/// `BlockChanges` is ordered by transactions properly.
#[substreams::handlers::map]
pub fn map_protocol_changes(
block: eth::v2::Block,
@@ -144,23 +143,43 @@ pub fn map_protocol_changes(
deltas: BlockBalanceDeltas,
components_store: StoreGetInt64,
balance_store: StoreDeltas, // Note, this map module is using the `deltas` mode for the store.
) -> Result<BlockContractChanges> {
) -> Result<BlockChanges> {
// 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<_, TransactionContractChanges> = HashMap::new();
let mut transaction_changes: HashMap<_, TransactionChanges> = HashMap::new();
// `ProtocolComponents` are gathered from `map_pools_created` which just need a bit of work to
// convert into `TransactionContractChanges`
// convert into `TransactionChanges`
grouped_components
.tx_components
.iter()
.for_each(|tx_component| {
let tx = tx_component.tx.as_ref().unwrap();
transaction_contract_changes
transaction_changes
.entry(tx.index)
.or_insert_with(|| TransactionContractChanges::new(tx))
.or_insert_with(|| TransactionChanges::new(tx))
.component_changes
.extend_from_slice(&tx_component.components);
tx_component
.components
.iter()
.for_each(|component| {
transaction_changes
.entry(tx.index)
.or_insert_with(|| TransactionChanges::new(tx))
.entity_changes
.push(EntityChanges {
component_id: component.id.clone(),
attributes: vec![Attribute {
name: "balance_owner".to_string(),
value: "0xBA12222222228d8Ba445958a75a0704d566BF2C8"
.to_string()
.as_bytes()
.to_vec(),
change: ChangeType::Creation.into(),
}],
});
});
});
// Balance changes are gathered by the `StoreDelta` based on `PoolBalanceChanged` creating
@@ -170,9 +189,9 @@ pub fn map_protocol_changes(
aggregate_balances_changes(balance_store, deltas)
.into_iter()
.for_each(|(_, (tx, balances))| {
transaction_contract_changes
transaction_changes
.entry(tx.index)
.or_insert_with(|| TransactionContractChanges::new(&tx))
.or_insert_with(|| TransactionChanges::new(&tx))
.balance_changes
.extend(balances.into_values());
});
@@ -183,22 +202,24 @@ pub fn map_protocol_changes(
|addr| {
components_store
.get_last(format!("pool:0x{0}", hex::encode(addr)))
.is_some()
.is_some() ||
addr.eq(VAULT_ADDRESS)
},
&mut transaction_contract_changes,
&mut transaction_changes,
);
// Process all `transaction_contract_changes` for final output in the `BlockContractChanges`,
// Process all `transaction_changes` for final output in the `BlockChanges`,
// sorted by transaction index (the key).
Ok(BlockContractChanges {
Ok(BlockChanges {
block: Some((&block).into()),
changes: transaction_contract_changes
changes: transaction_changes
.drain()
.sorted_unstable_by_key(|(index, _)| *index)
.filter_map(|(_, change)| {
if change.contract_changes.is_empty() &&
change.component_changes.is_empty() &&
change.balance_changes.is_empty()
change.balance_changes.is_empty() &&
change.entity_changes.is_empty()
{
None
} else {

79
substreams/ethereum-balancer/src/pool_factories.rs
View File

@@ -1,7 +1,7 @@
use crate::abi;
use crate::{abi, modules::VAULT_ADDRESS};
use substreams::{hex, scalar::BigInt};
use substreams_ethereum::{
pb::eth::v2::{Call, Log},
pb::eth::v2::{Call, Log, TransactionTrace},
Event, Function,
};
use tycho_substreams::prelude::*;
@@ -29,6 +29,19 @@ impl SerializableVecBigInt for Vec<BigInt> {
}
}
/// Helper function to get pool_registered event
fn get_pool_registered(
tx: &TransactionTrace,
pool_address: &Vec<u8>,
) -> abi::vault::events::PoolRegistered {
tx.logs_with_calls()
.filter(|(log, _)| log.address == VAULT_ADDRESS)
.filter_map(|(log, _)| abi::vault::events::PoolRegistered::match_and_decode(log))
.find(|pool| pool.pool_address == *pool_address)
.unwrap()
.clone()
}
/// This is the main function that handles the creation of `ProtocolComponent`s with `Attribute`s
/// based on the specific factory address. There's 3 factory groups that are represented here:
/// - Weighted Pool Factories
@@ -45,7 +58,7 @@ pub fn address_map(
pool_factory_address: &[u8],
log: &Log,
call: &Call,
tx: &Transaction,
tx: &TransactionTrace,
) -> Option<ProtocolComponent> {
match *pool_factory_address {
hex!("897888115Ada5773E02aA29F775430BFB5F34c51") => {
@@ -53,9 +66,11 @@ pub fn address_map(
abi::weighted_pool_factory::functions::Create::match_and_decode(call)?;
let pool_created =
abi::weighted_pool_factory::events::PoolCreated::match_and_decode(log)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&create_call.tokens)
.with_attributes(&[
("pool_type", "WeightedPoolFactory".as_bytes()),
@@ -65,6 +80,10 @@ pub fn address_map(
.normalized_weights
.serialize_bytes(),
),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)
@@ -74,11 +93,19 @@ pub fn address_map(
abi::composable_stable_pool_factory::functions::Create::match_and_decode(call)?;
let pool_created =
abi::composable_stable_pool_factory::events::PoolCreated::match_and_decode(log)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&create_call.tokens)
.with_attributes(&[("pool_type", "ComposableStablePoolFactory".as_bytes())])
.with_attributes(&[
("pool_type", "ComposableStablePoolFactory".as_bytes()),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)
}
@@ -87,9 +114,11 @@ pub fn address_map(
abi::erc_linear_pool_factory::functions::Create::match_and_decode(call)?;
let pool_created =
abi::erc_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&[create_call.main_token, create_call.wrapped_token])
.with_attributes(&[
("pool_type", "ERC4626LinearPoolFactory".as_bytes()),
@@ -99,6 +128,10 @@ pub fn address_map(
.upper_target
.to_signed_bytes_be(),
),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)
@@ -108,9 +141,11 @@ pub fn address_map(
abi::euler_linear_pool_factory::functions::Create::match_and_decode(call)?;
let pool_created =
abi::euler_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&[create_call.main_token, create_call.wrapped_token])
.with_attributes(&[
("pool_type", "EulerLinearPoolFactory".as_bytes()),
@@ -120,6 +155,10 @@ pub fn address_map(
.upper_target
.to_signed_bytes_be(),
),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)
@@ -177,9 +216,11 @@ pub fn address_map(
abi::silo_linear_pool_factory::functions::Create::match_and_decode(call)?;
let pool_created =
abi::silo_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&[create_call.main_token, create_call.wrapped_token])
.with_attributes(&[
("pool_type", "SiloLinearPoolFactory".as_bytes()),
@@ -189,6 +230,10 @@ pub fn address_map(
.upper_target
.to_signed_bytes_be(),
),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)
@@ -198,9 +243,11 @@ pub fn address_map(
abi::yearn_linear_pool_factory::functions::Create::match_and_decode(call)?;
let pool_created =
abi::yearn_linear_pool_factory::events::PoolCreated::match_and_decode(log)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&[create_call.main_token, create_call.wrapped_token])
.with_attributes(&[
("pool_type", "YearnLinearPoolFactory".as_bytes()),
@@ -210,6 +257,10 @@ pub fn address_map(
.upper_target
.to_signed_bytes_be(),
),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)
@@ -221,13 +272,19 @@ pub fn address_map(
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)?;
let pool_registered = get_pool_registered(tx, &pool_created.pool);
Some(
ProtocolComponent::at_contract(&pool_created.pool, tx)
ProtocolComponent::at_contract(&pool_created.pool, &(tx.into()))
.with_contracts(&[pool_created.pool, VAULT_ADDRESS.to_vec()])
.with_tokens(&create_call.tokens)
.with_attributes(&[
("pool_type", "WeightedPool2TokensFactory".as_bytes()),
("weights", &create_call.weights.serialize_bytes()),
(
"pool_id",
format!("0x{}", hex::encode(pool_registered.pool_id)).as_bytes(),
),
])
.as_swap_type("balancer_pool", ImplementationType::Vm),
)

18
substreams/ethereum-balancer/substreams.yaml
View File

@@ -1,7 +1,7 @@
specVersion: v0.1.0
package:
name: "ethereum_balancer"
version: v0.1.0
version: v0.2.0
protobuf:
files:
@@ -19,15 +19,15 @@ binaries:
modules:
- name: map_components
kind: map
initialBlock: 12369300
initialBlock: 12272146
inputs:
- source: sf.ethereum.type.v2.Block
output:
type: proto:tycho.evm.v1.GroupedTransactionProtocolComponents
type: proto:tycho.evm.v1.BlockTransactionProtocolComponents
- name: store_components
kind: store
initialBlock: 12369300
initialBlock: 12272146
updatePolicy: add
valueType: int64
inputs:
@@ -35,16 +35,16 @@ modules:
- name: map_relative_balances
kind: map
initialBlock: 12369300 # An arbitrary block that should change based on your requirements
initialBlock: 12272146
inputs:
- source: sf.ethereum.type.v2.Block
- store: store_components
output:
type: proto:tycho.evm.v1.BalanceDeltas
type: proto:tycho.evm.v1.BlockBalanceDeltas
- name: store_balances
kind: store
initialBlock: 12369300
initialBlock: 12272146
updatePolicy: add
valueType: bigint
inputs:
@@ -52,7 +52,7 @@ modules:
- name: map_protocol_changes
kind: map
initialBlock: 12369300
initialBlock: 12272146
inputs:
- source: sf.ethereum.type.v2.Block
- map: map_components
@@ -61,4 +61,4 @@ modules:
- store: store_balances
mode: deltas # This is the key property that simplifies `BalanceChange` handling
output:
type: proto:tycho.evm.v1.BlockContractChanges
type: proto:tycho.evm.v1.BlockChanges

32
substreams/ethereum-curve/src/modules.rs
View File

@@ -142,8 +142,8 @@ pub fn store_balances(deltas: BlockBalanceDeltas, store: StoreAddBigInt) {
}
/// This is the main map that handles most of the indexing of this substream.
/// Every contract change is grouped by transaction index via the `transaction_contract_changes`
/// map. Each block of code will extend the `TransactionContractChanges` struct with the
/// Every change is grouped by transaction index via the `transaction_changes`
/// map. Each block of code will extend the `TransactionChanges` struct with the
/// cooresponding changes (balance, component, contract), inserting a new one if it doesn't exist.
/// At the very end, the map can easily be sorted by index to ensure the final
/// `BlockContractChanges` is ordered by transactions properly.
@@ -154,25 +154,26 @@ pub fn map_protocol_changes(
deltas: BlockBalanceDeltas,
components_store: StoreGetString,
balance_store: StoreDeltas, // Note, this map module is using the `deltas` mode for the store.
) -> Result<BlockContractChanges> {
) -> Result<BlockChanges> {
// 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<_, TransactionContractChanges> = HashMap::new();
let mut transaction_changes: HashMap<_, TransactionChanges> = HashMap::new();
// `ProtocolComponents` are gathered from `map_pools_created` which just need a bit of work to
// convert into `TransactionContractChanges`
// convert into `TransactionChanges`
grouped_components
.tx_components
.into_iter()
.for_each(|tx_component| {
let tx = tx_component.tx.as_ref().unwrap();
transaction_contract_changes
transaction_changes
.entry(tx.index)
.or_insert_with(|| TransactionContractChanges {
.or_insert_with(|| TransactionChanges {
tx: Some(tx.clone()),
contract_changes: vec![],
component_changes: vec![],
balance_changes: vec![],
entity_changes: vec![],
})
.component_changes
.extend_from_slice(
@@ -214,19 +215,20 @@ pub fn map_protocol_changes(
},
)
})
// We need to group the balance changes by tx hash for the `TransactionContractChanges` agg
// We need to group the balance changes by tx hash for the `TransactionChanges` agg
.chunk_by(|(tx, _)| TransactionWrapper(tx.clone()))
.into_iter()
.for_each(|(tx_wrapped, group)| {
let tx = tx_wrapped.0;
transaction_contract_changes
transaction_changes
.entry(tx.index)
.or_insert_with(|| TransactionContractChanges {
.or_insert_with(|| TransactionChanges {
tx: Some(tx.clone()),
contract_changes: vec![],
component_changes: vec![],
balance_changes: vec![],
entity_changes: vec![],
})
.balance_changes
.extend(group.map(|(_, change)| change));
@@ -242,10 +244,10 @@ pub fn map_protocol_changes(
.get_last(format!("pool:{0}", hex::encode(addr)))
.is_some()
},
&mut transaction_contract_changes,
&mut transaction_changes,
);
for change in transaction_contract_changes.values_mut() {
for change in transaction_changes.values_mut() {
for balance_change in change.balance_changes.iter_mut() {
replace_eth_address(&mut balance_change.token);
}
@@ -257,9 +259,9 @@ pub fn map_protocol_changes(
}
}
// Process all `transaction_contract_changes` for final output in the `BlockContractChanges`,
// Process all `transaction_changes` for final output in the `BlockContractChanges`,
// sorted by transaction index (the key).
Ok(BlockContractChanges {
Ok(BlockChanges {
block: Some(Block {
number: block.number,
hash: block.hash.clone(),
@@ -271,7 +273,7 @@ pub fn map_protocol_changes(
.clone(),
ts: block.timestamp_seconds(),
}),
changes: transaction_contract_changes
changes: transaction_changes
.drain()
.sorted_unstable_by_key(|(index, _)| *index)
.filter_map(|(_, change)| {

2
substreams/ethereum-curve/substreams.yaml
View File

@@ -63,7 +63,7 @@ modules:
- store: store_balances
mode: deltas # This is the key property that simplifies `BalanceChange` handling
output:
type: proto:tycho.evm.v1.BlockContractChanges
type: proto:tycho.evm.v1.BlockChanges
params:
map_components: "address=bebc44782c7db0a1a60cb6fe97d0b483032ff1c7&tx_hash=20793bbf260912aae189d5d261ff003c9b9166da8191d8f9d63ff1c7722f3ac6&tokens[]=6b175474e89094c44da98b954eedeac495271d0f&tokens[]=a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&tokens[]=dac17f958d2ee523a2206206994597c13d831ec7&attribute_keys[]=name&attribute_vals[]=3pool&attribute_keys[]=factory_name&attribute_vals[]=NA&attribute_keys[]=factory&attribute_vals[]=0x0000000000000000000000000000000000000000,address=dc24316b9ae028f1497c275eb9192a3ea0f67022&tx_hash=fac67ecbd423a5b915deff06045ec9343568edaec34ae95c43d35f2c018afdaa&tokens[]=eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee&tokens[]=ae7ab96520de3a18e5e111b5eaab095312d7fe84&attribute_keys[]=name&attribute_vals[]=steth&attribute_keys[]=factory_name&attribute_vals[]=NA&attribute_keys[]=factory&attribute_vals[]=0x0000000000000000000000000000000000000000,address=d51a44d3fae010294c616388b506acda1bfaae46&tx_hash=dafb6385ed988ce8aacecfe1d97b38ea5e60b1ebce74d2423f71ddd621680138&tokens[]=dac17f958d2ee523a2206206994597c13d831ec7&tokens[]=2260fac5e5542a773aa44fbcfedf7c193bc2c599&tokens[]=c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2&attribute_keys[]=name&attribute_vals[]=tricrypto2&attribute_keys[]=factory_name&attribute_vals[]=NA&attribute_keys[]=factory&attribute_vals[]=0x0000000000000000000000000000000000000000,address=a5407eae9ba41422680e2e00537571bcc53efbfd&tx_hash=51aca4a03a395de8855fa2ca59b7febe520c2a223e69c502066162f7c1a95ec2&tokens[]=6b175474e89094c44da98b954eedeac495271d0f&tokens[]=a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&tokens[]=dac17f958d2ee523a2206206994597c13d831ec7&tokens[]=57ab1ec28d129707052df4df418d58a2d46d5f51&attribute_keys[]=name&attribute_vals[]=susd&attribute_keys[]=factory_name&attribute_vals[]=NA&attribute_keys[]=factory&attribute_vals[]=0x0000000000000000000000000000000000000000,address=dcef968d416a41cdac0ed8702fac8128a64241a2&tx_hash=1f4254004ce9e19d4eb742ee5a69d30f29085902d976f73e97c44150225ef775&tokens[]=853d955acef822db058eb8505911ed77f175b99e&tokens[]=a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&attribute_keys[]=name&attribute_vals[]=fraxusdc&attribute_keys[]=factory_name&attribute_vals[]=NA&attribute_keys[]=factory&attribute_vals[]=0x0000000000000000000000000000000000000000"