Ekubo Integration (#172)

* fix: Implement ethereum-ekubo

* fix: Remove unnecessary store

* fix: Correct balance accounting

* Adjust deltas by fee at PositionUpdated event

* Add partial Ekubo integration

* Generalize Hexable

* Native Ekubo integration

* cargo fmt & clippy

---------

Co-authored-by: kayibal <alan@datarevenue.com>
Co-authored-by: Zizou <111426680+zizou0x@users.noreply.github.com>

substreams/ethereum-ekubo/src/abi/mod.rs

@@ -0,0 +1,2 @@
+#![allow(clippy::all)]
+pub mod core;

substreams/ethereum-ekubo/src/deployment_config.rs

@@ -0,0 +1,9 @@
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+pub struct DeploymentConfig {
+    #[serde(with = "hex::serde")]
+    pub core: Vec<u8>,
+    #[serde(with = "hex::serde")]
+    pub oracle: Vec<u8>,
+}

substreams/ethereum-ekubo/src/lib.rs

@@ -0,0 +1,6 @@
+mod abi;
+mod deployment_config;
+mod modules;
+mod pb;
+mod pool_config;
+mod sqrt_ratio;

substreams/ethereum-ekubo/src/modules/1_map_events.rs

@@ -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(&params).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) })
+}

substreams/ethereum-ekubo/src/modules/2_map_components.rs

@@ -0,0 +1,135 @@
+use itertools::Itertools;
+use substreams::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".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,
+                    },
+                ],
+            },
+            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
+}

substreams/ethereum-ekubo/src/modules/2_map_tick_changes.rs

@@ -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![],
+    }
+}

substreams/ethereum-ekubo/src/modules/2_store_active_ticks.rs

@@ -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,
+    }
+}

substreams/ethereum-ekubo/src/modules/3_map_liquidity_changes.rs

@@ -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 = &current_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,
+    }
+}

substreams/ethereum-ekubo/src/modules/3_store_pool_details.rs

@@ -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);
+        });
+}

substreams/ethereum-ekubo/src/modules/3_store_tick_liquidities.rs

@@ -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),
+            );
+        });
+}

substreams/ethereum-ekubo/src/modules/4_map_balance_changes.rs

@@ -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")
+}

substreams/ethereum-ekubo/src/modules/4_store_liquidities.rs

@@ -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),
+                );
+            }
+        });
+}

substreams/ethereum-ekubo/src/modules/5_store_balance_changes.rs

@@ -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);
+}

substreams/ethereum-ekubo/src/modules/6_map_protocol_changes.rs

@@ -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,
+    }
+}

substreams/ethereum-ekubo/src/modules/mod.rs

@@ -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 }
+    }
+}

substreams/ethereum-ekubo/src/pb/ekubo.rs

@@ -0,0 +1,256 @@
+// @generated
+/// Copy of tycho.evm.v1.Transaction to be able to implement conversions to/from TransactionTrace
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Transaction {
+    #[prost(bytes="vec", tag="1")]
+    pub hash: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes="vec", tag="2")]
+    pub from: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes="vec", tag="3")]
+    pub to: ::prost::alloc::vec::Vec<u8>,
+    #[prost(uint64, tag="4")]
+    pub index: u64,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TickDelta {
+    /// bytes32
+    #[prost(bytes="vec", tag="1")]
+    pub pool_id: ::prost::alloc::vec::Vec<u8>,
+    #[prost(int32, tag="2")]
+    pub tick_index: i32,
+    /// int128
+    #[prost(bytes="vec", tag="3")]
+    pub liquidity_net_delta: ::prost::alloc::vec::Vec<u8>,
+    #[prost(uint64, tag="4")]
+    pub ordinal: u64,
+    #[prost(message, optional, tag="5")]
+    pub transaction: ::core::option::Option<Transaction>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TickDeltas {
+    #[prost(message, repeated, tag="1")]
+    pub deltas: ::prost::alloc::vec::Vec<TickDelta>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct LiquidityChange {
+    /// bytes32
+    #[prost(bytes="vec", tag="1")]
+    pub pool_id: ::prost::alloc::vec::Vec<u8>,
+    /// uint128 or int128, depending on change_type
+    #[prost(bytes="vec", tag="2")]
+    pub value: ::prost::alloc::vec::Vec<u8>,
+    #[prost(enumeration="LiquidityChangeType", tag="3")]
+    pub change_type: i32,
+    #[prost(uint64, tag="4")]
+    pub ordinal: u64,
+    #[prost(message, optional, tag="5")]
+    pub transaction: ::core::option::Option<Transaction>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct LiquidityChanges {
+    #[prost(message, repeated, tag="1")]
+    pub changes: ::prost::alloc::vec::Vec<LiquidityChange>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PoolDetails {
+    /// address
+    #[prost(bytes="vec", tag="1")]
+    pub token0: ::prost::alloc::vec::Vec<u8>,
+    /// address
+    #[prost(bytes="vec", tag="2")]
+    pub token1: ::prost::alloc::vec::Vec<u8>,
+    #[prost(fixed64, tag="3")]
+    pub fee: u64,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BlockTransactionEvents {
+    #[prost(message, repeated, tag="1")]
+    pub block_transaction_events: ::prost::alloc::vec::Vec<block_transaction_events::TransactionEvents>,
+}
+/// Nested message and enum types in `BlockTransactionEvents`.
+pub mod block_transaction_events {
+    #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+    pub struct TransactionEvents {
+        #[prost(message, optional, tag="1")]
+        pub transaction: ::core::option::Option<super::Transaction>,
+        #[prost(message, repeated, tag="2")]
+        pub pool_logs: ::prost::alloc::vec::Vec<transaction_events::PoolLog>,
+    }
+    /// Nested message and enum types in `TransactionEvents`.
+    pub mod transaction_events {
+        #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+        pub struct PoolLog {
+            #[prost(uint64, tag="1")]
+            pub ordinal: u64,
+            /// bytes32
+            #[prost(bytes="vec", tag="2")]
+            pub pool_id: ::prost::alloc::vec::Vec<u8>,
+            #[prost(oneof="pool_log::Event", tags="3, 4, 5, 6, 7")]
+            pub event: ::core::option::Option<pool_log::Event>,
+        }
+        /// Nested message and enum types in `PoolLog`.
+        pub mod pool_log {
+            #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+            pub struct Swapped {
+                /// int128
+                #[prost(bytes="vec", tag="1")]
+                pub delta0: ::prost::alloc::vec::Vec<u8>,
+                /// int128
+                #[prost(bytes="vec", tag="2")]
+                pub delta1: ::prost::alloc::vec::Vec<u8>,
+                /// uint192
+                #[prost(bytes="vec", tag="3")]
+                pub sqrt_ratio_after: ::prost::alloc::vec::Vec<u8>,
+                /// uint128
+                #[prost(bytes="vec", tag="4")]
+                pub liquidity_after: ::prost::alloc::vec::Vec<u8>,
+                /// int32
+                #[prost(sint32, tag="5")]
+                pub tick_after: i32,
+            }
+            #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+            pub struct PositionUpdated {
+                /// int32
+                #[prost(sint32, tag="1")]
+                pub lower: i32,
+                /// int32
+                #[prost(sint32, tag="2")]
+                pub upper: i32,
+                /// int128
+                #[prost(bytes="vec", tag="3")]
+                pub liquidity_delta: ::prost::alloc::vec::Vec<u8>,
+                /// int128
+                #[prost(bytes="vec", tag="4")]
+                pub delta0: ::prost::alloc::vec::Vec<u8>,
+                /// int128
+                #[prost(bytes="vec", tag="5")]
+                pub delta1: ::prost::alloc::vec::Vec<u8>,
+            }
+            #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+            pub struct PositionFeesCollected {
+                /// uint128
+                #[prost(bytes="vec", tag="1")]
+                pub amount0: ::prost::alloc::vec::Vec<u8>,
+                /// uint128
+                #[prost(bytes="vec", tag="2")]
+                pub amount1: ::prost::alloc::vec::Vec<u8>,
+            }
+            #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+            pub struct PoolInitialized {
+                /// address
+                #[prost(bytes="vec", tag="1")]
+                pub token0: ::prost::alloc::vec::Vec<u8>,
+                /// address
+                #[prost(bytes="vec", tag="2")]
+                pub token1: ::prost::alloc::vec::Vec<u8>,
+                /// bytes32
+                #[prost(bytes="vec", tag="3")]
+                pub config: ::prost::alloc::vec::Vec<u8>,
+                /// int32
+                #[prost(sint32, tag="4")]
+                pub tick: i32,
+                /// uint192
+                #[prost(bytes="vec", tag="5")]
+                pub sqrt_ratio: ::prost::alloc::vec::Vec<u8>,
+                #[prost(enumeration="pool_initialized::Extension", tag="6")]
+                pub extension: i32,
+            }
+            /// Nested message and enum types in `PoolInitialized`.
+            pub mod pool_initialized {
+                #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+                #[repr(i32)]
+                pub enum Extension {
+                    Unknown = 0,
+                    Base = 1,
+                    Oracle = 2,
+                }
+                impl Extension {
+                    /// 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 {
+                            Extension::Unknown => "EXTENSION_UNKNOWN",
+                            Extension::Base => "EXTENSION_BASE",
+                            Extension::Oracle => "EXTENSION_ORACLE",
+                        }
+                    }
+                    /// Creates an enum from field names used in the ProtoBuf definition.
+                    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+                        match value {
+                            "EXTENSION_UNKNOWN" => Some(Self::Unknown),
+                            "EXTENSION_BASE" => Some(Self::Base),
+                            "EXTENSION_ORACLE" => Some(Self::Oracle),
+                            _ => None,
+                        }
+                    }
+                }
+            }
+            #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+            pub struct FeesAccumulated {
+                /// uint128
+                #[prost(bytes="vec", tag="1")]
+                pub amount0: ::prost::alloc::vec::Vec<u8>,
+                /// uint128
+                #[prost(bytes="vec", tag="2")]
+                pub amount1: ::prost::alloc::vec::Vec<u8>,
+            }
+            #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Oneof)]
+            pub enum Event {
+                #[prost(message, tag="3")]
+                Swapped(Swapped),
+                #[prost(message, tag="4")]
+                PositionUpdated(PositionUpdated),
+                #[prost(message, tag="5")]
+                PositionFeesCollected(PositionFeesCollected),
+                #[prost(message, tag="6")]
+                PoolInitialized(PoolInitialized),
+                #[prost(message, tag="7")]
+                FeesAccumulated(FeesAccumulated),
+            }
+        }
+    }
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum LiquidityChangeType {
+    Delta = 0,
+    Absolute = 1,
+}
+impl LiquidityChangeType {
+    /// 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 {
+            LiquidityChangeType::Delta => "DELTA",
+            LiquidityChangeType::Absolute => "ABSOLUTE",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "DELTA" => Some(Self::Delta),
+            "ABSOLUTE" => Some(Self::Absolute),
+            _ => None,
+        }
+    }
+}
+// @@protoc_insertion_point(module)

substreams/ethereum-ekubo/src/pb/mod.rs

@@ -0,0 +1,6 @@
+// @generated
+// @@protoc_insertion_point(attribute:ekubo)
+pub mod ekubo {
+    include!("ekubo.rs");
+    // @@protoc_insertion_point(ekubo)
+}

substreams/ethereum-ekubo/src/pool_config.rs

@@ -0,0 +1,15 @@
+pub struct PoolConfig {
+    pub fee: Vec<u8>,
+    pub tick_spacing: Vec<u8>,
+    pub extension: Vec<u8>,
+}
+
+impl From<[u8; 32]> for PoolConfig {
+    fn from(value: [u8; 32]) -> Self {
+        Self {
+            tick_spacing: value[28..32].into(),
+            fee: value[20..28].into(),
+            extension: value[..20].into(),
+        }
+    }
+}

substreams/ethereum-ekubo/src/sqrt_ratio.rs

@@ -0,0 +1,11 @@
+use substreams::scalar::BigInt;
+
+const BIT_MASK: u128 = 0xc00000000000000000000000;
+const NOT_BIT_MASK: u128 = 0x3fffffffffffffffffffffff;
+
+pub fn float_sqrt_ratio_to_fixed(sqrt_ratio_float: BigInt) -> Vec<u8> {
+    let sqrt_ratio_fixed = (sqrt_ratio_float.clone() & NOT_BIT_MASK) <<
+        <BigInt as Into<u32>>::into(2_u64 + ((sqrt_ratio_float & BIT_MASK) >> 89_u8));
+
+    sqrt_ratio_fixed.to_bytes_be().1
+}