fix: small fix and refactor after testing EkuboV2 (#188)
* fix: add `balance_owner` attribute for ekubo_v2 components * refactor: rename `ekubo` into `ekubo-v2` * refactor: freeze `substreams-helper` version in ekubo V2 module * docs: improve integration test comment * docs: add TODO to remove `balance_owner` in favor of `AccountBalances` --------- Co-authored-by: zizou <111426680+flopell@users.noreply.github.com>
This commit is contained in:
Zizou
129
substreams/ethereum-ekubo-v2/src/modules/1_map_events.rs
Normal file
129
substreams/ethereum-ekubo-v2/src/modules/1_map_events.rs
Normal file
@@ -0,0 +1,129 @@
|
||||
use ethabi::Address;
|
||||
use itertools::Itertools;
|
||||
use substreams::scalar::BigInt;
|
||||
use substreams_ethereum::{
|
||||
pb::eth::{self, v2::Log},
|
||||
Event as _,
|
||||
};
|
||||
|
||||
use crate::{
|
||||
abi::core::events as abi_events,
|
||||
deployment_config::DeploymentConfig,
|
||||
pb::ekubo::{
|
||||
block_transaction_events::{
|
||||
transaction_events::{
|
||||
pool_log::{
|
||||
pool_initialized::Extension, Event, FeesAccumulated, PoolInitialized,
|
||||
PositionFeesCollected, PositionUpdated, Swapped,
|
||||
},
|
||||
PoolLog,
|
||||
},
|
||||
TransactionEvents,
|
||||
},
|
||||
BlockTransactionEvents,
|
||||
},
|
||||
pool_config::PoolConfig,
|
||||
sqrt_ratio::float_sqrt_ratio_to_fixed,
|
||||
};
|
||||
|
||||
#[substreams::handlers::map]
|
||||
fn map_events(params: String, block: eth::v2::Block) -> BlockTransactionEvents {
|
||||
let config: DeploymentConfig = serde_qs::from_str(¶ms).unwrap();
|
||||
|
||||
BlockTransactionEvents {
|
||||
block_transaction_events: block
|
||||
.transactions()
|
||||
.flat_map(|trace| {
|
||||
let pool_logs = trace
|
||||
.logs_with_calls()
|
||||
.filter_map(|(log, _)| maybe_pool_log(log, &config))
|
||||
.collect_vec();
|
||||
|
||||
(!pool_logs.is_empty())
|
||||
.then(|| TransactionEvents { transaction: Some(trace.into()), pool_logs })
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
fn maybe_pool_log(log: &Log, config: &DeploymentConfig) -> Option<PoolLog> {
|
||||
if log.address != config.core {
|
||||
return None;
|
||||
}
|
||||
|
||||
let (pool_id, ev) = if log.topics.is_empty() {
|
||||
let data = &log.data;
|
||||
|
||||
assert!(data.len() == 116, "swap event data length mismatch");
|
||||
|
||||
(
|
||||
data[20..52].to_vec(),
|
||||
Event::Swapped(Swapped {
|
||||
delta0: data[52..68].to_vec(),
|
||||
delta1: data[68..84].to_vec(),
|
||||
liquidity_after: data[84..100].to_vec(),
|
||||
sqrt_ratio_after: float_sqrt_ratio_to_fixed(BigInt::from_unsigned_bytes_be(
|
||||
&data[100..112],
|
||||
)),
|
||||
tick_after: i32::from_be_bytes(data[112..116].try_into().unwrap()),
|
||||
}),
|
||||
)
|
||||
} else if let Some(ev) = abi_events::PositionUpdated::match_and_decode(log) {
|
||||
(
|
||||
ev.pool_id.to_vec(),
|
||||
Event::PositionUpdated(PositionUpdated {
|
||||
lower: ev.params.1 .0.to_i32(),
|
||||
upper: ev.params.1 .1.to_i32(),
|
||||
liquidity_delta: ev.params.2.to_signed_bytes_be(),
|
||||
delta0: ev.delta0.to_signed_bytes_be(),
|
||||
delta1: ev.delta1.to_signed_bytes_be(),
|
||||
}),
|
||||
)
|
||||
} else if let Some(ev) = abi_events::PositionFeesCollected::match_and_decode(log) {
|
||||
(
|
||||
ev.pool_id.to_vec(),
|
||||
Event::PositionFeesCollected(PositionFeesCollected {
|
||||
amount0: ev.amount0.to_bytes_be().1,
|
||||
amount1: ev.amount1.to_bytes_be().1,
|
||||
}),
|
||||
)
|
||||
} else if let Some(ev) = abi_events::PoolInitialized::match_and_decode(log) {
|
||||
let pool_config = PoolConfig::from(ev.pool_key.2);
|
||||
|
||||
let extension = {
|
||||
let extension = pool_config.extension;
|
||||
|
||||
if extension == Address::zero().as_bytes() {
|
||||
Extension::Base
|
||||
} else if extension == config.oracle {
|
||||
Extension::Oracle
|
||||
} else {
|
||||
Extension::Unknown
|
||||
}
|
||||
};
|
||||
|
||||
(
|
||||
ev.pool_id.to_vec(),
|
||||
Event::PoolInitialized(PoolInitialized {
|
||||
token0: ev.pool_key.0,
|
||||
token1: ev.pool_key.1,
|
||||
config: ev.pool_key.2.to_vec(),
|
||||
tick: ev.tick.to_i32(),
|
||||
sqrt_ratio: float_sqrt_ratio_to_fixed(ev.sqrt_ratio),
|
||||
extension: extension.into(),
|
||||
}),
|
||||
)
|
||||
} else if let Some(ev) = abi_events::FeesAccumulated::match_and_decode(log) {
|
||||
(
|
||||
ev.pool_id.to_vec(),
|
||||
Event::FeesAccumulated(FeesAccumulated {
|
||||
amount0: ev.amount0.to_bytes_be().1,
|
||||
amount1: ev.amount1.to_bytes_be().1,
|
||||
}),
|
||||
)
|
||||
} else {
|
||||
return None;
|
||||
};
|
||||
|
||||
Some(PoolLog { ordinal: log.ordinal, pool_id, event: Some(ev) })
|
||||
}
|
||||
141
substreams/ethereum-ekubo-v2/src/modules/2_map_components.rs
Normal file
141
substreams/ethereum-ekubo-v2/src/modules/2_map_components.rs
Normal file
@@ -0,0 +1,141 @@
|
||||
use itertools::Itertools;
|
||||
use substreams::{hex, scalar::BigInt};
|
||||
use substreams_helper::hex::Hexable;
|
||||
use tycho_substreams::models::{
|
||||
Attribute, BalanceChange, BlockChanges, ChangeType, EntityChanges, FinancialType,
|
||||
ImplementationType, ProtocolComponent, ProtocolType, TransactionChanges,
|
||||
};
|
||||
|
||||
use crate::{
|
||||
pb::ekubo::{
|
||||
block_transaction_events::transaction_events::{pool_log::Event, PoolLog},
|
||||
BlockTransactionEvents,
|
||||
},
|
||||
pool_config::PoolConfig,
|
||||
};
|
||||
|
||||
#[substreams::handlers::map]
|
||||
fn map_components(block_tx_events: BlockTransactionEvents) -> BlockChanges {
|
||||
BlockChanges {
|
||||
block: None,
|
||||
changes: block_tx_events
|
||||
.block_transaction_events
|
||||
.into_iter()
|
||||
.filter_map(|tx_events| {
|
||||
let (components, entities, balance_changes): (Vec<_>, Vec<_>, Vec<_>) = tx_events
|
||||
.pool_logs
|
||||
.into_iter()
|
||||
.filter_map(maybe_create_component)
|
||||
.multiunzip();
|
||||
|
||||
(!components.is_empty()).then(|| TransactionChanges {
|
||||
tx: Some(tx_events.transaction.unwrap().into()),
|
||||
balance_changes: balance_changes
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.collect(),
|
||||
contract_changes: vec![],
|
||||
entity_changes: entities,
|
||||
component_changes: components,
|
||||
})
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
fn maybe_create_component(
|
||||
log: PoolLog,
|
||||
) -> Option<(ProtocolComponent, EntityChanges, Vec<BalanceChange>)> {
|
||||
if let Event::PoolInitialized(pi) = log.event.unwrap() {
|
||||
let config = PoolConfig::from(<[u8; 32]>::try_from(pi.config).unwrap());
|
||||
let component_id = log.pool_id.to_hex();
|
||||
|
||||
return Some((
|
||||
ProtocolComponent {
|
||||
id: component_id.clone(),
|
||||
tokens: vec![pi.token0.clone(), pi.token1.clone()],
|
||||
contracts: vec![],
|
||||
change: ChangeType::Creation.into(),
|
||||
protocol_type: Some(ProtocolType {
|
||||
name: "ekubo_v2_pool".to_string(),
|
||||
financial_type: FinancialType::Swap.into(),
|
||||
implementation_type: ImplementationType::Custom.into(),
|
||||
attribute_schema: vec![],
|
||||
}),
|
||||
// Order of attributes matters (used in store_pool_details)
|
||||
static_att: vec![
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "token0".to_string(),
|
||||
value: pi.token0.clone(),
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "token1".to_string(),
|
||||
value: pi.token1.clone(),
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "fee".to_string(),
|
||||
value: config.fee,
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "tick_spacing".to_string(),
|
||||
value: config.tick_spacing,
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "extension".to_string(),
|
||||
value: config.extension,
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "extension_id".to_string(),
|
||||
value: pi.extension.to_be_bytes().to_vec(),
|
||||
},
|
||||
],
|
||||
},
|
||||
EntityChanges {
|
||||
component_id: component_id.clone(),
|
||||
attributes: vec![
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "liquidity".to_string(),
|
||||
value: 0_u128.to_be_bytes().to_vec(),
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "tick".to_string(),
|
||||
value: pi.tick.to_be_bytes().to_vec(),
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "sqrt_ratio".to_string(),
|
||||
value: pi.sqrt_ratio,
|
||||
},
|
||||
Attribute {
|
||||
change: ChangeType::Creation.into(),
|
||||
name: "balance_owner".to_string(), /* TODO: We should use AccountBalances
|
||||
* instead */
|
||||
value: hex!("e0e0e08A6A4b9Dc7bD67BCB7aadE5cF48157d444").to_vec(),
|
||||
},
|
||||
],
|
||||
},
|
||||
vec![
|
||||
BalanceChange {
|
||||
component_id: component_id.clone().into_bytes(),
|
||||
token: pi.token0,
|
||||
balance: BigInt::zero().to_signed_bytes_be(),
|
||||
},
|
||||
BalanceChange {
|
||||
component_id: component_id.into_bytes(),
|
||||
token: pi.token1,
|
||||
balance: BigInt::zero().to_signed_bytes_be(),
|
||||
},
|
||||
],
|
||||
));
|
||||
}
|
||||
|
||||
None
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
use substreams::scalar::BigInt;
|
||||
|
||||
use crate::pb::ekubo::{
|
||||
block_transaction_events::transaction_events::pool_log::Event, BlockTransactionEvents,
|
||||
TickDelta, TickDeltas,
|
||||
};
|
||||
|
||||
#[substreams::handlers::map]
|
||||
pub fn map_tick_changes(block_tx_events: BlockTransactionEvents) -> TickDeltas {
|
||||
TickDeltas {
|
||||
deltas: block_tx_events
|
||||
.block_transaction_events
|
||||
.into_iter()
|
||||
.flat_map(|tx_events| {
|
||||
let tx = tx_events.transaction;
|
||||
|
||||
tx_events
|
||||
.pool_logs
|
||||
.into_iter()
|
||||
.flat_map(move |log| {
|
||||
let tx = tx.clone();
|
||||
|
||||
tick_deltas(log.event.unwrap())
|
||||
.into_iter()
|
||||
.map(move |partial| TickDelta {
|
||||
liquidity_net_delta: partial.liquidity_net_delta,
|
||||
pool_id: log.pool_id.clone(),
|
||||
tick_index: partial.tick_index,
|
||||
ordinal: log.ordinal,
|
||||
transaction: tx.clone(),
|
||||
})
|
||||
})
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
struct PartialTickDelta {
|
||||
tick_index: i32,
|
||||
liquidity_net_delta: Vec<u8>,
|
||||
}
|
||||
|
||||
fn tick_deltas(ev: Event) -> Vec<PartialTickDelta> {
|
||||
match ev {
|
||||
Event::PositionUpdated(position_updated) => {
|
||||
vec![
|
||||
PartialTickDelta {
|
||||
tick_index: position_updated.lower,
|
||||
liquidity_net_delta: position_updated.liquidity_delta.clone(),
|
||||
},
|
||||
PartialTickDelta {
|
||||
tick_index: position_updated.upper,
|
||||
liquidity_net_delta: BigInt::from_signed_bytes_be(
|
||||
&position_updated.liquidity_delta,
|
||||
)
|
||||
.neg()
|
||||
.to_signed_bytes_be(),
|
||||
},
|
||||
]
|
||||
}
|
||||
_ => vec![],
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
use substreams::store::{StoreSet, StoreSetInt64};
|
||||
|
||||
use substreams::store::StoreNew;
|
||||
use substreams_helper::hex::Hexable;
|
||||
|
||||
use crate::pb::ekubo::{
|
||||
block_transaction_events::transaction_events::pool_log::Event, BlockTransactionEvents,
|
||||
};
|
||||
|
||||
#[substreams::handlers::store]
|
||||
pub fn store_active_ticks(block_tx_events: BlockTransactionEvents, store: StoreSetInt64) {
|
||||
block_tx_events
|
||||
.block_transaction_events
|
||||
.into_iter()
|
||||
.flat_map(|tx_events| tx_events.pool_logs)
|
||||
.filter_map(|log| {
|
||||
maybe_tick(log.event.unwrap()).map(|tick| (log.pool_id.to_hex(), log.ordinal, tick))
|
||||
})
|
||||
.for_each(|(pool, ordinal, new_tick_index)| {
|
||||
store.set(ordinal, format!("pool:{pool}"), &new_tick_index.into())
|
||||
});
|
||||
}
|
||||
|
||||
fn maybe_tick(ev: Event) -> Option<i32> {
|
||||
match ev {
|
||||
Event::PoolInitialized(pool_initialized) => Some(pool_initialized.tick),
|
||||
Event::Swapped(swapped) => Some(swapped.tick_after),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
use substreams::store::{StoreGet, StoreGetInt64};
|
||||
|
||||
use substreams_helper::hex::Hexable;
|
||||
|
||||
use crate::pb::ekubo::{
|
||||
block_transaction_events::transaction_events::{pool_log::Event, PoolLog},
|
||||
BlockTransactionEvents, LiquidityChange, LiquidityChangeType, LiquidityChanges,
|
||||
};
|
||||
|
||||
#[substreams::handlers::map]
|
||||
pub fn map_liquidity_changes(
|
||||
block_tx_events: BlockTransactionEvents,
|
||||
current_tick_store: StoreGetInt64,
|
||||
) -> LiquidityChanges {
|
||||
LiquidityChanges {
|
||||
changes: block_tx_events
|
||||
.block_transaction_events
|
||||
.into_iter()
|
||||
.flat_map(|tx_events| {
|
||||
let current_tick_store = ¤t_tick_store;
|
||||
|
||||
tx_events
|
||||
.pool_logs
|
||||
.into_iter()
|
||||
.filter_map(move |log| {
|
||||
maybe_liquidity_change(&log, current_tick_store).map(|partial| {
|
||||
LiquidityChange {
|
||||
change_type: partial.change_type.into(),
|
||||
pool_id: log.pool_id,
|
||||
value: partial.value,
|
||||
ordinal: log.ordinal,
|
||||
transaction: tx_events.transaction.clone(),
|
||||
}
|
||||
})
|
||||
})
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
struct PartialLiquidityChange {
|
||||
value: Vec<u8>,
|
||||
change_type: LiquidityChangeType,
|
||||
}
|
||||
|
||||
fn maybe_liquidity_change(
|
||||
log: &PoolLog,
|
||||
current_tick_store: &StoreGetInt64,
|
||||
) -> Option<PartialLiquidityChange> {
|
||||
match log.event.as_ref().unwrap() {
|
||||
Event::Swapped(swapped) => Some(PartialLiquidityChange {
|
||||
value: swapped.liquidity_after.clone(),
|
||||
change_type: LiquidityChangeType::Absolute,
|
||||
}),
|
||||
Event::PositionUpdated(position_updated) => {
|
||||
let current_tick = current_tick_store
|
||||
.get_at(log.ordinal, format!("pool:{0}", log.pool_id.to_hex()))
|
||||
.expect("pool should have active tick when initialized");
|
||||
|
||||
(current_tick >= position_updated.lower.into() &&
|
||||
current_tick < position_updated.upper.into())
|
||||
.then(|| PartialLiquidityChange {
|
||||
value: position_updated.liquidity_delta.clone(),
|
||||
change_type: LiquidityChangeType::Delta,
|
||||
})
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
use substreams::store::{StoreNew, StoreSetIfNotExists, StoreSetIfNotExistsProto};
|
||||
use tycho_substreams::models::BlockChanges;
|
||||
|
||||
use crate::pb::ekubo::PoolDetails;
|
||||
|
||||
// Since only the PoolInitialized event contains the complete pool key we need to store some info
|
||||
// required when processing other events
|
||||
#[substreams::handlers::store]
|
||||
fn store_pool_details(changes: BlockChanges, store: StoreSetIfNotExistsProto<PoolDetails>) {
|
||||
changes
|
||||
.changes
|
||||
.into_iter()
|
||||
.flat_map(|c| c.component_changes.into_iter())
|
||||
.for_each(|component| {
|
||||
let attrs = component.static_att;
|
||||
|
||||
let pool_details = PoolDetails {
|
||||
token0: attrs[0].value.clone(),
|
||||
token1: attrs[1].value.clone(),
|
||||
fee: u64::from_be_bytes(
|
||||
attrs[2]
|
||||
.value
|
||||
.clone()
|
||||
.try_into()
|
||||
.unwrap(),
|
||||
),
|
||||
};
|
||||
|
||||
store.set_if_not_exists(0, component.id, &pool_details);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
use substreams::{
|
||||
scalar::BigInt,
|
||||
store::{StoreAdd, StoreAddBigInt, StoreNew},
|
||||
};
|
||||
use substreams_helper::hex::Hexable;
|
||||
|
||||
use crate::pb::ekubo::TickDeltas;
|
||||
|
||||
#[substreams::handlers::store]
|
||||
pub fn store_tick_liquidities(tick_deltas: TickDeltas, store: StoreAddBigInt) {
|
||||
tick_deltas
|
||||
.deltas
|
||||
.into_iter()
|
||||
.for_each(|delta| {
|
||||
store.add(
|
||||
delta.ordinal,
|
||||
format!("pool:{}:tick:{}", delta.pool_id.to_hex(), delta.tick_index),
|
||||
BigInt::from_signed_bytes_be(&delta.liquidity_net_delta),
|
||||
);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
use substreams::{
|
||||
hex,
|
||||
scalar::BigInt,
|
||||
store::{StoreGet, StoreGetProto},
|
||||
};
|
||||
use substreams_helper::hex::Hexable;
|
||||
use tycho_substreams::models::{BalanceDelta, BlockBalanceDeltas, Transaction};
|
||||
|
||||
use crate::pb::ekubo::{
|
||||
block_transaction_events::transaction_events::pool_log::Event, BlockTransactionEvents,
|
||||
PoolDetails,
|
||||
};
|
||||
|
||||
#[substreams::handlers::map]
|
||||
fn map_balance_changes(
|
||||
block_tx_events: BlockTransactionEvents,
|
||||
store: StoreGetProto<PoolDetails>,
|
||||
) -> BlockBalanceDeltas {
|
||||
BlockBalanceDeltas {
|
||||
balance_deltas: block_tx_events
|
||||
.block_transaction_events
|
||||
.into_iter()
|
||||
.flat_map(|tx_events| {
|
||||
let tx: Transaction = tx_events.transaction.unwrap().into();
|
||||
|
||||
let store = &store;
|
||||
|
||||
tx_events
|
||||
.pool_logs
|
||||
.into_iter()
|
||||
.flat_map(move |log| {
|
||||
let component_id = log.pool_id.to_hex();
|
||||
let pool_details = get_pool_details(store, &component_id);
|
||||
|
||||
let component_id_bytes = component_id.into_bytes();
|
||||
let tx = tx.clone();
|
||||
|
||||
balance_deltas(log.event.unwrap(), pool_details)
|
||||
.into_iter()
|
||||
.map(move |reduced| BalanceDelta {
|
||||
ord: log.ordinal,
|
||||
tx: Some(tx.clone()),
|
||||
token: reduced.token,
|
||||
delta: reduced.delta,
|
||||
component_id: component_id_bytes.clone(),
|
||||
})
|
||||
})
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
struct ReducedBalanceDelta {
|
||||
token: Vec<u8>,
|
||||
delta: Vec<u8>,
|
||||
}
|
||||
|
||||
fn balance_deltas(ev: Event, pool_details: PoolDetails) -> Vec<ReducedBalanceDelta> {
|
||||
match ev {
|
||||
Event::Swapped(swapped) => {
|
||||
vec![
|
||||
ReducedBalanceDelta { token: pool_details.token0, delta: swapped.delta0 },
|
||||
ReducedBalanceDelta { token: pool_details.token1, delta: swapped.delta1 },
|
||||
]
|
||||
}
|
||||
Event::PositionUpdated(position_updated) => {
|
||||
vec![
|
||||
ReducedBalanceDelta {
|
||||
token: pool_details.token0,
|
||||
delta: adjust_delta_by_fee(
|
||||
BigInt::from_signed_bytes_be(&position_updated.delta0),
|
||||
pool_details.fee,
|
||||
)
|
||||
.to_signed_bytes_be(),
|
||||
},
|
||||
ReducedBalanceDelta {
|
||||
token: pool_details.token1,
|
||||
delta: adjust_delta_by_fee(
|
||||
BigInt::from_signed_bytes_be(&position_updated.delta1),
|
||||
pool_details.fee,
|
||||
)
|
||||
.to_signed_bytes_be(),
|
||||
},
|
||||
]
|
||||
}
|
||||
Event::PositionFeesCollected(position_fees_collected) => {
|
||||
vec![
|
||||
ReducedBalanceDelta {
|
||||
token: pool_details.token0,
|
||||
delta: BigInt::from_unsigned_bytes_be(&position_fees_collected.amount0)
|
||||
.neg()
|
||||
.to_signed_bytes_be(),
|
||||
},
|
||||
ReducedBalanceDelta {
|
||||
token: pool_details.token1,
|
||||
delta: BigInt::from_unsigned_bytes_be(&position_fees_collected.amount1)
|
||||
.neg()
|
||||
.to_signed_bytes_be(),
|
||||
},
|
||||
]
|
||||
}
|
||||
Event::FeesAccumulated(fees_accumulated) => {
|
||||
vec![
|
||||
ReducedBalanceDelta { token: pool_details.token0, delta: fees_accumulated.amount0 },
|
||||
ReducedBalanceDelta { token: pool_details.token1, delta: fees_accumulated.amount1 },
|
||||
]
|
||||
}
|
||||
_ => vec![],
|
||||
}
|
||||
}
|
||||
|
||||
// Negative deltas don't include the fees paid by the position owner, thus we need to add it back
|
||||
// here (i.e. subtract from the component's balance)
|
||||
fn adjust_delta_by_fee(delta: BigInt, fee: u64) -> BigInt {
|
||||
if delta < BigInt::zero() {
|
||||
let denom = BigInt::from_signed_bytes_be(&hex!("0100000000000000000000000000000000"));
|
||||
(delta * denom.clone()) / (denom - fee)
|
||||
} else {
|
||||
delta
|
||||
}
|
||||
}
|
||||
|
||||
fn get_pool_details(store: &StoreGetProto<PoolDetails>, component_id: &str) -> PoolDetails {
|
||||
store
|
||||
.get_at(0, component_id)
|
||||
.expect("pool id should exist in store")
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
use substreams::{
|
||||
scalar::BigInt,
|
||||
store::{StoreSetSum, StoreSetSumBigInt},
|
||||
};
|
||||
use substreams_helper::hex::Hexable;
|
||||
|
||||
use crate::pb::ekubo::{LiquidityChangeType, LiquidityChanges};
|
||||
|
||||
#[substreams::handlers::store]
|
||||
pub fn store_liquidities(liquidity_changes: LiquidityChanges, store: StoreSetSumBigInt) {
|
||||
liquidity_changes
|
||||
.changes
|
||||
.into_iter()
|
||||
.for_each(|changes| match changes.change_type() {
|
||||
LiquidityChangeType::Delta => {
|
||||
store.sum(
|
||||
changes.ordinal,
|
||||
format!("pool:{0}", changes.pool_id.to_hex()),
|
||||
BigInt::from_signed_bytes_be(&changes.value),
|
||||
);
|
||||
}
|
||||
LiquidityChangeType::Absolute => {
|
||||
store.set(
|
||||
changes.ordinal,
|
||||
format!("pool:{0}", changes.pool_id.to_hex()),
|
||||
BigInt::from_signed_bytes_be(&changes.value),
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
use substreams::store::{StoreAddBigInt, StoreNew};
|
||||
use tycho_substreams::models::BlockBalanceDeltas;
|
||||
|
||||
#[substreams::handlers::store]
|
||||
fn store_balance_changes(deltas: BlockBalanceDeltas, store: StoreAddBigInt) {
|
||||
tycho_substreams::balances::store_balance_changes(deltas, store);
|
||||
}
|
||||
@@ -0,0 +1,203 @@
|
||||
use std::{collections::HashMap, str::FromStr};
|
||||
|
||||
use itertools::Itertools;
|
||||
use substreams::{key, pb::substreams::StoreDeltas, scalar::BigInt};
|
||||
use substreams_ethereum::pb::eth;
|
||||
use substreams_helper::hex::Hexable;
|
||||
use tycho_substreams::{
|
||||
balances::aggregate_balances_changes,
|
||||
models::{
|
||||
Attribute, BlockBalanceDeltas, BlockChanges, ChangeType, EntityChanges,
|
||||
TransactionChangesBuilder,
|
||||
},
|
||||
};
|
||||
|
||||
use crate::pb::ekubo::{
|
||||
block_transaction_events::transaction_events::pool_log::Event, BlockTransactionEvents,
|
||||
LiquidityChanges, TickDeltas,
|
||||
};
|
||||
|
||||
/// Aggregates protocol components and balance changes by transaction.
|
||||
///
|
||||
/// This is the main method that will aggregate all changes as well as extract all
|
||||
/// relevant contract storage deltas.
|
||||
#[substreams::handlers::map]
|
||||
fn map_protocol_changes(
|
||||
block: eth::v2::Block,
|
||||
new_components: BlockChanges,
|
||||
block_tx_events: BlockTransactionEvents,
|
||||
balances_map_deltas: BlockBalanceDeltas,
|
||||
balances_store_deltas: StoreDeltas,
|
||||
ticks_map_deltas: TickDeltas,
|
||||
ticks_store_deltas: StoreDeltas,
|
||||
liquidity_changes: LiquidityChanges,
|
||||
liquidity_store_deltas: StoreDeltas,
|
||||
) -> Result<BlockChanges, substreams::errors::Error> {
|
||||
let mut transaction_changes: HashMap<_, TransactionChangesBuilder> = HashMap::new();
|
||||
|
||||
// New components
|
||||
new_components
|
||||
.changes
|
||||
.iter()
|
||||
.for_each(|tx_changes| {
|
||||
let tx = tx_changes.tx.as_ref().unwrap();
|
||||
let builder = transaction_changes
|
||||
.entry(tx.index)
|
||||
.or_insert_with(|| TransactionChangesBuilder::new(tx));
|
||||
|
||||
tx_changes
|
||||
.component_changes
|
||||
.iter()
|
||||
.for_each(|component| {
|
||||
builder.add_protocol_component(component);
|
||||
});
|
||||
|
||||
tx_changes
|
||||
.entity_changes
|
||||
.iter()
|
||||
.for_each(|entity_change| {
|
||||
builder.add_entity_change(entity_change);
|
||||
});
|
||||
|
||||
tx_changes
|
||||
.balance_changes
|
||||
.iter()
|
||||
.for_each(|balance_change| {
|
||||
builder.add_balance_change(balance_change);
|
||||
});
|
||||
});
|
||||
|
||||
// Component balances
|
||||
aggregate_balances_changes(balances_store_deltas, balances_map_deltas)
|
||||
.into_iter()
|
||||
.for_each(|(_, (tx, balances))| {
|
||||
let builder = transaction_changes
|
||||
.entry(tx.index)
|
||||
.or_insert_with(|| TransactionChangesBuilder::new(&tx));
|
||||
|
||||
balances
|
||||
.values()
|
||||
.for_each(|token_bc_map| {
|
||||
token_bc_map.values().for_each(|bc| {
|
||||
builder.add_balance_change(bc);
|
||||
})
|
||||
});
|
||||
});
|
||||
|
||||
// Tick liquidities
|
||||
ticks_store_deltas
|
||||
.deltas
|
||||
.into_iter()
|
||||
.zip(ticks_map_deltas.deltas)
|
||||
.for_each(|(store_delta, tick_delta)| {
|
||||
let new_value_bigint = BigInt::from_store_bytes(&store_delta.new_value);
|
||||
|
||||
let is_creation = BigInt::from_store_bytes(&store_delta.old_value).is_zero();
|
||||
|
||||
let attribute = Attribute {
|
||||
name: format!("ticks/{}", tick_delta.tick_index),
|
||||
value: new_value_bigint.to_signed_bytes_be(),
|
||||
change: if is_creation {
|
||||
ChangeType::Creation.into()
|
||||
} else if new_value_bigint.is_zero() {
|
||||
ChangeType::Deletion.into()
|
||||
} else {
|
||||
ChangeType::Update.into()
|
||||
},
|
||||
};
|
||||
let tx = tick_delta.transaction.unwrap();
|
||||
let builder = transaction_changes
|
||||
.entry(tx.index)
|
||||
.or_insert_with(|| TransactionChangesBuilder::new(&tx.into()));
|
||||
|
||||
builder.add_entity_change(&EntityChanges {
|
||||
component_id: tick_delta.pool_id.to_hex(),
|
||||
attributes: vec![attribute],
|
||||
});
|
||||
});
|
||||
|
||||
// Pool liquidities
|
||||
liquidity_store_deltas
|
||||
.deltas
|
||||
.into_iter()
|
||||
.zip(liquidity_changes.changes)
|
||||
.for_each(|(store_delta, change)| {
|
||||
let tx = change.transaction.unwrap();
|
||||
let builder = transaction_changes
|
||||
.entry(tx.index)
|
||||
.or_insert_with(|| TransactionChangesBuilder::new(&tx.into()));
|
||||
|
||||
let new_value_bigint = BigInt::from_str(key::segment_at(
|
||||
&String::from_utf8(store_delta.new_value).unwrap(),
|
||||
1,
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
builder.add_entity_change(&EntityChanges {
|
||||
component_id: change.pool_id.to_hex(),
|
||||
attributes: vec![Attribute {
|
||||
name: "liquidity".to_string(),
|
||||
value: new_value_bigint.to_signed_bytes_be(),
|
||||
change: ChangeType::Update.into(),
|
||||
}],
|
||||
});
|
||||
});
|
||||
|
||||
// Remaining event changes not subject to special treatment
|
||||
block_tx_events
|
||||
.block_transaction_events
|
||||
.into_iter()
|
||||
.flat_map(|tx_events| {
|
||||
let tx = tx_events.transaction.unwrap();
|
||||
|
||||
tx_events
|
||||
.pool_logs
|
||||
.into_iter()
|
||||
.flat_map(move |log| {
|
||||
let tx = tx.clone();
|
||||
|
||||
maybe_attribute_updates(log.event.unwrap()).map(|attrs| {
|
||||
(
|
||||
tx,
|
||||
EntityChanges { component_id: log.pool_id.to_hex(), attributes: attrs },
|
||||
)
|
||||
})
|
||||
})
|
||||
})
|
||||
.for_each(|(tx, entity_changes)| {
|
||||
let builder = transaction_changes
|
||||
.entry(tx.index)
|
||||
.or_insert_with(|| TransactionChangesBuilder::new(&tx.into()));
|
||||
builder.add_entity_change(&entity_changes);
|
||||
});
|
||||
|
||||
Ok(BlockChanges {
|
||||
block: Some((&block).into()),
|
||||
changes: transaction_changes
|
||||
.drain()
|
||||
.sorted_unstable_by_key(|(index, _)| *index)
|
||||
.filter_map(|(_, builder)| builder.build())
|
||||
.collect(),
|
||||
})
|
||||
}
|
||||
|
||||
fn maybe_attribute_updates(ev: Event) -> Option<Vec<Attribute>> {
|
||||
match ev {
|
||||
Event::Swapped(swapped) => Some(vec![
|
||||
Attribute {
|
||||
name: "tick".into(),
|
||||
value: swapped
|
||||
.tick_after
|
||||
.to_be_bytes()
|
||||
.to_vec(),
|
||||
change: ChangeType::Update.into(),
|
||||
},
|
||||
Attribute {
|
||||
name: "sqrt_ratio".into(),
|
||||
value: swapped.sqrt_ratio_after,
|
||||
change: ChangeType::Update.into(),
|
||||
},
|
||||
]),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
48
substreams/ethereum-ekubo-v2/src/modules/mod.rs
Normal file
48
substreams/ethereum-ekubo-v2/src/modules/mod.rs
Normal file
@@ -0,0 +1,48 @@
|
||||
use substreams_ethereum::pb::eth::v2::TransactionTrace;
|
||||
|
||||
use crate::pb::ekubo::Transaction;
|
||||
|
||||
#[path = "1_map_events.rs"]
|
||||
mod map_events;
|
||||
|
||||
#[path = "2_map_components.rs"]
|
||||
mod map_components;
|
||||
#[path = "2_map_tick_changes.rs"]
|
||||
mod map_tick_changes;
|
||||
#[path = "2_store_active_ticks.rs"]
|
||||
mod store_active_ticks;
|
||||
|
||||
#[path = "3_map_liquidity_changes.rs"]
|
||||
mod map_liquidity_changes;
|
||||
#[path = "3_store_pool_details.rs"]
|
||||
mod store_pool_details;
|
||||
#[path = "3_store_tick_liquidities.rs"]
|
||||
mod store_tick_liquidities;
|
||||
|
||||
#[path = "4_map_balance_changes.rs"]
|
||||
mod map_balance_changes;
|
||||
#[path = "4_store_liquidities.rs"]
|
||||
mod store_liquidities;
|
||||
|
||||
#[path = "5_store_balance_changes.rs"]
|
||||
mod store_balance_changes;
|
||||
|
||||
#[path = "6_map_protocol_changes.rs"]
|
||||
mod map_protocol_changes;
|
||||
|
||||
impl From<&TransactionTrace> for Transaction {
|
||||
fn from(value: &TransactionTrace) -> Self {
|
||||
Self {
|
||||
hash: value.hash.clone(),
|
||||
from: value.from.clone(),
|
||||
to: value.to.clone(),
|
||||
index: value.index.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Transaction> for tycho_substreams::prelude::Transaction {
|
||||
fn from(value: Transaction) -> Self {
|
||||
Self { hash: value.hash, from: value.from, to: value.to, index: value.index }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user