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

Merge pull request #13 from propeller-heads/zz/fix-balancer-substream-naming

Browse Source 
WIP: Balancer Substream integration fixes
This commit is contained in:
Alan Höng
2024-03-14 10:50:33 +00:00
committed by GitHub
parent b3909a4140 fdd43ad38e
commit d10c4e3ea3
20 changed files with 472 additions and 509 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,7 @@ message Transaction {
// The receiver of the transaction.
bytes to = 3;
// The transactions index within the block.
// TODO: should this be uint32? to match the type from the native substream type?
uint64 index = 4;
}
@@ -47,6 +48,26 @@ message Attribute {
ChangeType change = 3;
}
enum FinancialType{
SWAP = 0;
LEND = 1;
LEVERAGE = 2;
PSM = 3;
}
enum ImplementationType {
VM = 0;
CUSTOM = 1;
}
message ProtocolType{
string name = 1;
FinancialType financial_type = 2;
repeated Attribute attribute_schema = 3;
ImplementationType implementation_type = 4;
}
// 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
@@ -67,6 +88,10 @@ message ProtocolComponent {
repeated Attribute static_att = 4;
// Type of change the component underwent.
ChangeType change = 5;
/// Represents the functionality of the component.
ProtocolType protocol_type = 6;
// Transaction where this component was created
Transaction tx = 7;
}
// A struct for following the changes of Total Value Locked (TVL) of a protocol component.
@@ -75,9 +100,9 @@ message ProtocolComponent {
message BalanceChange {
// The address of the ERC20 token whose balance changed.
bytes token = 1;
// The new balance of the token.
// The new balance of the token. Note: it must be a big endian encoded int.
bytes balance = 2;
// The id of the component whose TVL is tracked.
// 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.
bytes component_id = 3;
}

41
proto/tycho/evm/v1/utils.proto Normal file
View File

@@ -0,0 +1,41 @@
syntax = "proto3";
package tycho.evm.v1;
import "tycho/evm/v1/common.proto";
// 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.
message BalanceDelta {
// The ordinal of the balance change. Must be unique & deterministic over all balances
// changes within a block.
uint64 ord = 1;
// The tx hash of the transaction that caused the balance change.
Transaction tx = 2;
// The address of the ERC20 token whose balance changed.
bytes token = 3;
// The delta balance of the token.
bytes delta = 4;
// 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.
bytes component_id = 5;
}
// A set of balances deltas, usually a group of changes within a single block.
message BlockBalanceDeltas {
repeated BalanceDelta balance_deltas = 1;
}
// A message containing protocol components that were created by a single tx.
message TransactionProtocolComponents {
Transaction tx = 1;
repeated ProtocolComponent components = 2;
}
// All protocol components that were created within a block with their corresponding tx.
message BlockTransactionProtocolComponents {
repeated TransactionProtocolComponents tx_components = 1;
}

2
substreams/Readme.md
View File

@@ -47,7 +47,7 @@ The example from the official docs will serve us just well:
```toml
[package]
name = "substreams_[CHAIN]_[PROTOCOL_SYSTEM]"
name = "substreams-[CHAIN]-[PROTOCOL_SYSTEM]"
version = "0.1.0"
edition = "2021"

36
substreams/ethereum-balancer/Cargo.lock generated
View File

@@ -884,24 +884,6 @@ dependencies = [
"thiserror",
]
[[package]]
name = "substreams-balancer"
version = "0.1.0"
dependencies = [
"anyhow",
"bytes",
"ethabi 18.0.0",
"getrandom",
"hex",
"hex-literal 0.4.1",
"itertools 0.12.0",
"num-bigint",
"prost 0.11.9",
"prost-types 0.12.3",
"substreams",
"substreams-ethereum",
]
[[package]]
name = "substreams-ethereum"
version = "0.9.9"
@@ -933,6 +915,24 @@ dependencies = [
"syn 1.0.109",
]
[[package]]
name = "substreams-ethereum-balancer"
version = "0.1.0"
dependencies = [
"anyhow",
"bytes",
"ethabi 18.0.0",
"getrandom",
"hex",
"hex-literal 0.4.1",
"itertools 0.12.0",
"num-bigint",
"prost 0.11.9",
"prost-types 0.12.3",
"substreams",
"substreams-ethereum",
]
[[package]]
name = "substreams-ethereum-core"
version = "0.9.9"

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

@@ -1,10 +1,10 @@
[package]
name = "substreams-balancer"
name = "substreams-ethereum-balancer"
version = "0.1.0"
edition = "2021"
[lib]
name = "substreams_balancer"
name = "substreams_ethereum_balancer"
crate-type = ["cdylib"]
[dependencies]

1
substreams/ethereum-balancer/build.rs
View File

@@ -8,6 +8,7 @@ fn main() -> Result<()> {
let files = fs::read_dir(abi_folder)?;
let mut mod_rs_content = String::new();
mod_rs_content.push_str("#![allow(clippy::all)]\n");
for file in files {
let file = file?;

19
substreams/ethereum-balancer/proto/factory.proto
View File

@@ -1,19 +0,0 @@
syntax = "proto3";
package eth.factory.v1;
message Pools {
repeated Pool pools = 1;
}
message Pool {
bytes pool_id = 1;
fixed64 log_ordinal = 2;
}
message Transfer {
bytes from = 1;
bytes to = 2;
string token = 3;
string amount = 4;
}

15
substreams/ethereum-balancer/proto/pool.proto
View File

@@ -1,15 +0,0 @@
syntax = "proto3";
package eth.pool.v1;
message Transfers {
repeated Transfer transfers = 1;
}
message Transfer {
string from = 1;
string to = 2;
uint64 token_id = 3;
string trx_hash = 4;
uint64 ordinal = 5;
}

113
substreams/ethereum-balancer/proto/tycho/evm/v1/common.proto
View File

@@ -1,113 +0,0 @@
syntax = "proto3";
package tycho.evm.v1;
// This file contains the proto definitions for Substreams common to all integrations.
// A struct describing a block.
message Block {
// The blocks hash.
bytes hash = 1;
// The parent blocks hash.
bytes parent_hash = 2;
// The block number.
uint64 number = 3;
// The block timestamp.
uint64 ts = 4;
}
// A struct describing a transaction.
message Transaction {
// The transaction hash.
bytes hash = 1;
// The sender of the transaction.
bytes from = 2;
// The receiver of the transaction.
bytes to = 3;
// The transactions index within the block.
// TODO: should this be uint32? to match the type from the native substream type?
uint64 index = 4;
}
// Enum to specify the type of a change.
enum ChangeType {
CHANGE_TYPE_UNSPECIFIED = 0;
CHANGE_TYPE_UPDATE = 1;
CHANGE_TYPE_CREATION = 2;
CHANGE_TYPE_DELETION = 3;
}
// 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.
message Attribute {
// The name of the attribute.
string name = 1;
// The value of the attribute.
bytes value = 2;
// The type of change the attribute underwent.
ChangeType change = 3;
}
// 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".
message ProtocolComponent {
// A unique identifier for the component within the protocol.
// Can be e.g. a stringified address or a string describing the trading pair.
string id = 1;
// Addresses of the ERC20 tokens used by the component.
repeated bytes tokens = 2;
// Addresses of the contracts used by the component.
// Usually it is a single contract, but some protocols use multiple contracts.
repeated bytes contracts = 3;
// Attributes of the component. Used mainly be the native integration.
// The inner ChangeType of the attribute has to match the ChangeType of the ProtocolComponent.
repeated Attribute static_att = 4;
// Type of change the component underwent.
ChangeType change = 5;
}
message TransactionProtocolComponents {
Transaction tx = 1;
repeated ProtocolComponent components = 2;
}
message GroupedTransactionProtocolComponents {
repeated TransactionProtocolComponents tx_components = 1;
}
// 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.
message BalanceChange {
// The address of the ERC20 token whose balance changed.
bytes token = 1;
// The new balance of the token.
bytes balance = 2;
// 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.
bytes component_id = 3;
}
// 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.
message BalanceDelta {
uint64 ord = 1;
// The tx hash of the transaction that caused the balance change.
Transaction tx = 2;
// The address of the ERC20 token whose balance changed.
bytes token = 3;
// The delta balance of the token.
bytes delta = 4;
// 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.
bytes component_id = 5;
}
message BalanceDeltas {
repeated BalanceDelta balance_deltas = 1;
}

32
substreams/ethereum-balancer/proto/tycho/evm/v1/entity.proto
View File

@@ -1,32 +0,0 @@
syntax = "proto3";
package tycho.evm.v1;
import "tycho/evm/v1/common.proto";
// 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;
// An array of newly added components.
repeated ProtocolComponent component_changes = 3;
// An array of balance changes to components.
repeated BalanceChange balance_changes = 4;
}
// A set of transaction changes within a single block.
message BlockEntityChanges {
// The block for which these changes are collectively computed.
Block block = 1;
// The set of transaction changes observed in the specified block.
repeated TransactionEntityChanges changes = 2;
}

50
substreams/ethereum-balancer/proto/tycho/evm/v1/vm.proto
View File

@@ -1,50 +0,0 @@
syntax = "proto3";
package tycho.evm.v1;
import "tycho/evm/v1/common.proto";
// 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.
repeated ProtocolComponent component_changes = 3;
// An array of balance changes to components.
repeated BalanceChange balance_changes = 4;
}
// A set of transaction changes within a single block.
message BlockContractChanges {
// The block for which these changes are collectively computed.
Block block = 1;
// The set of transaction changes observed in the specified block.
repeated TransactionContractChanges changes = 2;
}

1
substreams/ethereum-balancer/src/abi/mod.rs
View File

@@ -1,3 +1,4 @@
#![allow(clippy::all)]
pub mod yearn_linear_pool_factory;
pub mod composable_stable_pool_factory;
pub mod vault;

153
substreams/ethereum-balancer/src/contract_changes.rs
View File

@@ -97,77 +97,98 @@ pub fn extract_contract_changes(
balance_changes.sort_unstable_by_key(|change| change.ordinal);
code_changes.sort_unstable_by_key(|change| change.ordinal);
storage_changes.iter().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
},
});
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(),
});
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);
});
slot_value
.new_value
.copy_from_slice(&storage_change.new_value);
});
balance_changes.iter().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
},
});
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();
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
.balance
.extend_from_slice(&new_balance.bytes);
}
});
code_changes.iter().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);
});
.code
.extend_from_slice(&code_change.new_code);
});
if !storage_changes.is_empty() || !balance_changes.is_empty() || !code_changes.is_empty() {
transaction_contract_changes

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

@@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::str::FromStr;
use anyhow::Result;
use substreams::hex;
@@ -13,11 +14,14 @@ use substreams::scalar::BigInt;
use substreams_ethereum::pb::eth;
use itertools::Itertools;
use pb::tycho::evm::v1::{self as tycho};
use contract_changes::extract_contract_changes;
use substreams_ethereum::Event;
use crate::{abi, contract_changes, pb, pool_factories};
use crate::pb::tycho::evm::v1::{self as tycho};
use crate::pb::tycho::evm::v1::{BalanceDelta, BlockBalanceDeltas, BlockTransactionProtocolComponents, TransactionProtocolComponents};
use crate::{abi, contract_changes, pool_factories};
const VAULT_ADDRESS: &[u8] = &hex!("BA12222222228d8Ba445958a75a0704d566BF2C8");
@@ -33,12 +37,10 @@ impl PartialEq for TransactionWrapper {
}
#[substreams::handlers::map]
pub fn map_pools_created(
block: eth::v2::Block,
) -> Result<tycho::GroupedTransactionProtocolComponents> {
pub fn map_pools_created(block: eth::v2::Block) -> Result<BlockTransactionProtocolComponents> {
// Gather contract changes by indexing `PoolCreated` events and analysing the `Create` call
// We store these as a hashmap by tx hash since we need to agg by tx hash later
Ok(tycho::GroupedTransactionProtocolComponents {
Ok(BlockTransactionProtocolComponents {
tx_components: block
.transactions()
.filter_map(|tx| {
@@ -46,16 +48,22 @@ pub fn map_pools_created(
.logs_with_calls()
.filter(|(_, call)| !call.call.state_reverted)
.filter_map(|(log, call)| {
Some(pool_factories::address_map(
pool_factories::address_map(
call.call.address.as_slice(),
log,
call.call,
)?)
&tycho::Transaction {
hash: tx.hash.clone(),
from: tx.from.clone(),
to: tx.to.clone(),
index: tx.index.into(),
},
)
})
.collect::<Vec<_>>();
if !components.is_empty() {
Some(tycho::TransactionProtocolComponents {
Some(TransactionProtocolComponents {
tx: Some(tycho::Transaction {
hash: tx.hash.clone(),
from: tx.from.clone(),
@@ -74,7 +82,7 @@ pub fn map_pools_created(
/// Simply stores the `ProtocolComponent`s with the pool id as the key
#[substreams::handlers::store]
pub fn store_pools_created(map: tycho::GroupedTransactionProtocolComponents, store: StoreAddInt64) {
pub fn store_pools_created(map: BlockTransactionProtocolComponents, store: StoreAddInt64) {
store.add_many(
0,
&map.tx_components
@@ -86,54 +94,88 @@ pub fn store_pools_created(map: tycho::GroupedTransactionProtocolComponents, sto
);
}
/// Since the `PoolBalanceChanged` events administer only deltas, we need to leverage a map and a
/// Since the `PoolBalanceChanged` and `Swap` events administer only deltas, we need to leverage a map and a
/// store to be able to tally up final balances for tokens in a pool.
#[substreams::handlers::map]
pub fn map_balance_deltas(
block: eth::v2::Block,
store: StoreGetInt64,
) -> Result<tycho::BalanceDeltas, anyhow::Error> {
Ok(tycho::BalanceDeltas {
balance_deltas: block
.events::<abi::vault::events::PoolBalanceChanged>(&[&VAULT_ADDRESS])
.flat_map(|(event, log)| {
event
.tokens
.iter()
.zip(event.deltas.iter())
.filter_map(|(token, delta)| {
let component_id: Vec<_> = event.pool_id.into();
) -> Result<BlockBalanceDeltas, anyhow::Error> {
let balance_deltas = block
.logs()
.filter(|log| log.address() == VAULT_ADDRESS)
.flat_map(|vault_log| {
let mut deltas = Vec::new();
if store
.get_last(format!("pool:{0}", hex::encode(&component_id)))
.is_none()
{
return None;
}
Some(tycho::BalanceDelta {
ord: log.log.ordinal,
if let Some(ev) =
abi::vault::events::PoolBalanceChanged::match_and_decode(vault_log.log)
{
let component_id = ev.pool_id[..20].to_vec();
if store
.get_last(format!("pool:{}", hex::encode(&component_id)))
.is_some()
{
for (token, delta) in ev.tokens.iter().zip(ev.deltas.iter()) {
deltas.push(BalanceDelta {
ord: vault_log.ordinal(),
tx: Some(tycho::Transaction {
hash: log.receipt.transaction.hash.clone(),
from: log.receipt.transaction.from.clone(),
to: log.receipt.transaction.to.clone(),
index: Into::<u64>::into(log.receipt.transaction.index),
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.clone(),
token: token.to_vec(),
delta: delta.to_signed_bytes_be(),
component_id: component_id.clone(),
})
})
.collect::<Vec<_>>()
})
.collect::<Vec<_>>(),
})
});
}
}
} else if let Some(ev) = abi::vault::events::Swap::match_and_decode(vault_log.log) {
let component_id = ev.pool_id[..20].to_vec();
if store
.get_last(format!("pool:{}", hex::encode(&component_id)))
.is_some()
{
deltas.extend_from_slice(&[
BalanceDelta {
ord: vault_log.ordinal(),
tx: Some(tycho::Transaction {
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 {
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,
},
]);
}
}
deltas
})
.collect::<Vec<_>>();
Ok(BlockBalanceDeltas { balance_deltas })
}
/// It's significant to include both the `pool_id` and the `token_id` for each balance delta as the
/// store key to ensure that there's a unique balance being tallied for each.
#[substreams::handlers::store]
pub fn store_balance_changes(deltas: tycho::BalanceDeltas, store: StoreAddBigInt) {
pub fn store_balance_changes(deltas: BlockBalanceDeltas, store: StoreAddBigInt) {
deltas.balance_deltas.iter().for_each(|delta| {
store.add(
delta.ord,
@@ -156,8 +198,8 @@ pub fn store_balance_changes(deltas: tycho::BalanceDeltas, store: StoreAddBigInt
#[substreams::handlers::map]
pub fn map_changes(
block: eth::v2::Block,
grouped_components: tycho::GroupedTransactionProtocolComponents,
deltas: tycho::BalanceDeltas,
grouped_components: BlockTransactionProtocolComponents,
deltas: BlockBalanceDeltas,
components_store: StoreGetInt64,
balance_store: StoreDeltas, // Note, this map module is using the `deltas` mode for the store.
) -> Result<tycho::BlockContractChanges> {
@@ -187,7 +229,7 @@ pub fn map_changes(
});
// Balance changes are gathered by the `StoreDelta` based on `PoolBalanceChanged` creating
// `BalanceDeltas`. We essentially just process the changes that occured to the `store` this
// `BlockBalanceDeltas`. We essentially just process the changes that occured 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
@@ -197,12 +239,18 @@ pub fn map_changes(
.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: store_delta.new_value,
component_id: hex::decode(pool_id).expect("Token ID not valid hex"),
balance: big_endian_bytes_balance,
component_id: pool_id.as_bytes().to_vec(),
},
)
})
@@ -245,7 +293,7 @@ pub fn map_changes(
}),
changes: transaction_contract_changes
.drain()
.sorted_unstable_by_key(|(index, _)| index.clone())
.sorted_unstable_by_key(|(index, _)| *index)
.filter_map(|(_, change)| {
if change.contract_changes.is_empty()
&& change.component_changes.is_empty()

28
substreams/ethereum-balancer/src/pb/eth.balancer.v1.rs
View File

@@ -1,28 +0,0 @@
// @generated
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Pools {
#[prost(message, repeated, tag="1")]
pub pools: ::prost::alloc::vec::Vec<Pool>,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Pool {
#[prost(bytes="vec", tag="1")]
pub pool_id: ::prost::alloc::vec::Vec<u8>,
#[prost(fixed64, tag="2")]
pub log_ordinal: u64,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Transfer {
#[prost(bytes="vec", tag="1")]
pub from: ::prost::alloc::vec::Vec<u8>,
#[prost(bytes="vec", tag="2")]
pub to: ::prost::alloc::vec::Vec<u8>,
#[prost(string, tag="3")]
pub token: ::prost::alloc::string::String,
#[prost(string, tag="4")]
pub amount: ::prost::alloc::string::String,
}
// @@protoc_insertion_point(module)

28
substreams/ethereum-balancer/src/pb/eth.factory.v1.rs
View File

@@ -1,28 +0,0 @@
// @generated
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Pools {
#[prost(message, repeated, tag="1")]
pub pools: ::prost::alloc::vec::Vec<Pool>,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Pool {
#[prost(bytes="vec", tag="1")]
pub pool_id: ::prost::alloc::vec::Vec<u8>,
#[prost(fixed64, tag="2")]
pub log_ordinal: u64,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Transfer {
#[prost(bytes="vec", tag="1")]
pub from: ::prost::alloc::vec::Vec<u8>,
#[prost(bytes="vec", tag="2")]
pub to: ::prost::alloc::vec::Vec<u8>,
#[prost(string, tag="3")]
pub token: ::prost::alloc::string::String,
#[prost(string, tag="4")]
pub amount: ::prost::alloc::string::String,
}
// @@protoc_insertion_point(module)

22
substreams/ethereum-balancer/src/pb/eth.pool.v1.rs
View File

@@ -1,22 +0,0 @@
// @generated
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Transfers {
#[prost(message, repeated, tag="1")]
pub transfers: ::prost::alloc::vec::Vec<Transfer>,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Transfer {
#[prost(string, tag="1")]
pub from: ::prost::alloc::string::String,
#[prost(string, tag="2")]
pub to: ::prost::alloc::string::String,
#[prost(uint64, tag="3")]
pub token_id: u64,
#[prost(string, tag="4")]
pub trx_hash: ::prost::alloc::string::String,
#[prost(uint64, tag="5")]
pub ordinal: u64,
}
// @@protoc_insertion_point(module)

169
substreams/ethereum-balancer/src/pb/tycho.evm.v1.rs
View File

@@ -51,6 +51,18 @@ pub struct Attribute {
#[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
@@ -78,20 +90,12 @@ pub struct ProtocolComponent {
/// Type of change the component underwent.
#[prost(enumeration="ChangeType", tag="5")]
pub change: i32,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct TransactionProtocolComponents {
#[prost(message, optional, tag="1")]
/// / 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>,
#[prost(message, repeated, tag="2")]
pub components: ::prost::alloc::vec::Vec<ProtocolComponent>,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct GroupedTransactionProtocolComponents {
#[prost(message, repeated, tag="1")]
pub tx_components: ::prost::alloc::vec::Vec<TransactionProtocolComponents>,
}
/// 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.
@@ -102,42 +106,14 @@ 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.
/// The new balance of the token. Note: it must be a big endian encoded int.
#[prost(bytes="vec", tag="2")]
pub balance: ::prost::alloc::vec::Vec<u8>,
/// The id of the component whose TVL is tracked.
/// 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>,
}
/// 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 BalanceDelta {
#[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>,
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BalanceDeltas {
#[prost(message, repeated, tag="1")]
pub balance_deltas: ::prost::alloc::vec::Vec<BalanceDelta>,
}
/// Enum to specify the type of a change.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
@@ -171,6 +147,64 @@ impl ChangeType {
}
}
}
#[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 proto definitions specific to the VM integration.
/// A key value entry into contract storage.
@@ -233,4 +267,53 @@ 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)

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

@@ -3,6 +3,7 @@ use substreams_ethereum::{Event, Function};
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;
@@ -13,6 +14,7 @@ use substreams::scalar::BigInt;
/// handled by any downstream application.
trait SerializableVecBigInt {
fn serialize_bytes(&self) -> Vec<u8>;
#[allow(dead_code)]
fn deserialize_bytes(bytes: &[u8]) -> Vec<BigInt>;
}
@@ -25,7 +27,7 @@ impl SerializableVecBigInt for Vec<BigInt> {
fn deserialize_bytes(bytes: &[u8]) -> Vec<BigInt> {
bytes
.chunks_exact(32)
.map(|chunk| BigInt::from_signed_bytes_be(chunk))
.map(BigInt::from_signed_bytes_be)
.collect::<Vec<BigInt>>()
}
}
@@ -45,6 +47,7 @@ pub fn address_map(
pool_factory_address: &[u8],
log: &Log,
call: &Call,
tx: &Transaction,
) -> Option<tycho::ProtocolComponent> {
match *pool_factory_address {
hex!("897888115Ada5773E02aA29F775430BFB5F34c51") => {
@@ -56,7 +59,7 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: create_call.tokens,
contracts: vec![pool_factory_address.into(), pool_created.pool],
contracts: vec![pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
@@ -70,6 +73,13 @@ pub fn address_map(
},
],
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") => {
@@ -81,15 +91,20 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: create_call.tokens,
contracts: vec![pool_factory_address.into(), pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
value: "ComposableStablePoolFactory".into(),
change: tycho::ChangeType::Creation.into(),
},
],
contracts: vec![pool_created.pool],
static_att: vec![tycho::Attribute {
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") => {
@@ -101,7 +116,7 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: vec![create_call.main_token, create_call.wrapped_token],
contracts: vec![pool_factory_address.into(), pool_created.pool],
contracts: vec![pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
@@ -117,6 +132,13 @@ pub fn address_map(
// 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") => {
@@ -128,7 +150,7 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: vec![create_call.main_token, create_call.wrapped_token],
contracts: vec![pool_factory_address.into(), pool_created.pool],
contracts: vec![pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
@@ -142,6 +164,13 @@ pub fn address_map(
},
],
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
@@ -200,7 +229,7 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: vec![create_call.main_token, create_call.wrapped_token],
contracts: vec![pool_factory_address.into(), pool_created.pool],
contracts: vec![pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
@@ -214,6 +243,13 @@ pub fn address_map(
},
],
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") => {
@@ -225,7 +261,7 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: vec![create_call.main_token, create_call.wrapped_token],
contracts: vec![pool_factory_address.into(), pool_created.pool],
contracts: vec![pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
@@ -239,6 +275,13 @@ pub fn address_map(
},
],
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
@@ -252,7 +295,7 @@ pub fn address_map(
Some(tycho::ProtocolComponent {
id: hex::encode(&pool_created.pool),
tokens: create_call.tokens,
contracts: vec![pool_factory_address.into(), pool_created.pool],
contracts: vec![pool_created.pool],
static_att: vec![
tycho::Attribute {
name: "pool_type".into(),
@@ -266,6 +309,13 @@ pub fn address_map(
},
],
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,

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

@@ -1,20 +1,20 @@
specVersion: v0.1.0
package:
name: "substreams_balancer"
name: "substreams_ethereum_balancer"
version: v0.1.0
protobuf:
files:
- tycho/evm/v1/vm.proto
- tycho/evm/v1/common.proto
- tycho/evm/v1/utils.proto
importPaths:
- ../../proto/tycho/evm/v1/
- ./proto
- ../../proto
binaries:
default:
type: wasm/rust-v1
file: target/wasm32-unknown-unknown/release/substreams_balancer.wasm
file: target/wasm32-unknown-unknown/release/substreams_ethereum_balancer.wasm
modules:
- name: map_pools_created
@@ -35,7 +35,7 @@ modules:
- name: map_balance_deltas
kind: map
initialBlock: 12369300 # An arbitrary block that should change based on your requirements
initialBlock: 12369300 # An arbitrary block that should change based on your requirements
inputs:
- source: sf.ethereum.type.v2.Block
- store: store_pools_created
@@ -59,6 +59,6 @@ modules:
- map: map_balance_deltas
- store: store_pools_created
- store: store_balance_changes
mode: deltas # This is the key property that simplifies `BalanceChange` handling
mode: deltas # This is the key property that simplifies `BalanceChange` handling
output:
type: proto:tycho.evm.v1.BlockContractChanges