feat: Add Maverick V2 adapter and substreams (#167)

Co-authored-by: Thales <thales@datarevenue.com>
Co-authored-by: zizou <111426680+flopell@users.noreply.github.com>

substreams/ethereum-maverick-v2/src/modules/1_map_components.rs

@@ -0,0 +1,55 @@
+use anyhow::{Ok, Result};
+use ethabi::ethereum_types::Address;
+use substreams_ethereum::pb::eth::v2::{Block, Log, TransactionTrace};
+use tycho_substreams::prelude::*;
+
+use crate::{abi::factory::events::PoolCreated, modules::utils::Params};
+use substreams_helper::{event_handler::EventHandler, hex::Hexable};
+
+#[substreams::handlers::map]
+pub fn map_components(params: String, block: Block) -> Result<BlockTransactionProtocolComponents> {
+    let mut new_pools: Vec<TransactionProtocolComponents> = vec![];
+    let params = Params::parse_from_query(&params)?;
+    get_new_pools(params, &block, &mut new_pools);
+
+    Ok(BlockTransactionProtocolComponents { tx_components: new_pools })
+}
+
+fn get_new_pools(
+    params: Params,
+    block: &Block,
+    new_pools: &mut Vec<TransactionProtocolComponents>,
+) {
+    let (factory_address, quoter_address) = params.decode_addresses().unwrap();
+
+    let mut on_pool_created = |event: PoolCreated, _tx: &TransactionTrace, _log: &Log| {
+        let tycho_tx: Transaction = _tx.into();
+        let contracts = vec![
+            event.pool_address.as_slice(),
+            factory_address.as_slice(),
+            quoter_address.as_slice(),
+        ];
+        let new_pool_component = ProtocolComponent::new(&event.pool_address.to_hex())
+            .with_tokens(&[event.token_a.as_slice(), event.token_b.as_slice()])
+            .with_contracts(&contracts)
+            .with_attributes(&[
+                ("fee_a_in", &event.fee_a_in.to_signed_bytes_be()),
+                ("fee_b_in", &event.fee_b_in.to_signed_bytes_be()),
+                ("tick_spacing", &event.tick_spacing.to_signed_bytes_be()),
+                ("kinds", &event.kinds.to_signed_bytes_be()),
+            ])
+            .as_swap_type("maverick_v2_pool", ImplementationType::Vm);
+
+        new_pools.push(TransactionProtocolComponents {
+            tx: Some(tycho_tx.clone()),
+            components: vec![new_pool_component],
+        });
+    };
+
+    let mut eh = EventHandler::new(block);
+
+    eh.filter_by_address(vec![Address::from_slice(&factory_address)]);
+
+    eh.on::<PoolCreated, _>(&mut on_pool_created);
+    eh.handle_events();
+}

substreams/ethereum-maverick-v2/src/modules/2_store_components.rs

@@ -0,0 +1,25 @@
+use crate::pb::maverick::v2::Pool;
+use substreams::{
+    prelude::{StoreSetIfNotExists, StoreSetIfNotExistsProto},
+    store::StoreNew,
+};
+use tycho_substreams::prelude::*;
+
+#[substreams::handlers::store]
+pub fn store_components(
+    map: BlockTransactionProtocolComponents,
+    store: StoreSetIfNotExistsProto<Pool>,
+) {
+    for tx_pc in map.tx_components {
+        for pc in tx_pc.components {
+            let pool_address = &pc.id;
+            let pool = Pool {
+                address: hex::decode(pool_address.trim_start_matches("0x")).unwrap(),
+                token_a: pc.tokens[0].clone(),
+                token_b: pc.tokens[1].clone(),
+                created_tx_hash: tx_pc.tx.as_ref().unwrap().hash.clone(),
+            };
+            store.set_if_not_exists(0, format!("Pool:{pool_address}"), &pool);
+        }
+    }
+}

substreams/ethereum-maverick-v2/src/modules/3_map_relative_balances.rs

@@ -0,0 +1,39 @@
+use crate::{events::get_log_changed_balances, pb::maverick::v2::Pool};
+use anyhow::{Ok, Result};
+use substreams::{prelude::StoreGetProto, store::StoreGet};
+use substreams_ethereum::pb::eth::v2::Block;
+use substreams_helper::hex::Hexable;
+use tycho_substreams::prelude::*;
+
+#[substreams::handlers::map]
+pub fn map_relative_balances(
+    block: Block,
+    pools_store: StoreGetProto<Pool>,
+) -> Result<BlockBalanceDeltas, anyhow::Error> {
+    let mut balance_deltas = Vec::new();
+    for trx in block.transactions() {
+        let mut tx_deltas = Vec::new();
+        let tx = Transaction {
+            to: trx.to.clone(),
+            from: trx.from.clone(),
+            hash: trx.hash.clone(),
+            index: trx.index.into(),
+        };
+        for log in trx
+            .calls
+            .iter()
+            .filter(|call| !call.state_reverted)
+            .flat_map(|call| &call.logs)
+        {
+            if let Some(pool) = pools_store.get_last(format!("Pool:{}", &log.address.to_hex())) {
+                tx_deltas.extend(get_log_changed_balances(&tx, log, &pool));
+            } else {
+                continue;
+            }
+        }
+        if !tx_deltas.is_empty() {
+            balance_deltas.extend(tx_deltas);
+        }
+    }
+    Ok(BlockBalanceDeltas { balance_deltas })
+}

substreams/ethereum-maverick-v2/src/modules/4_store_balances.rs

@@ -0,0 +1,7 @@
+use substreams::store::{StoreAddBigInt, StoreNew};
+use tycho_substreams::prelude::*;
+
+#[substreams::handlers::store]
+pub fn store_balances(balances_deltas: BlockBalanceDeltas, store: StoreAddBigInt) {
+    tycho_substreams::balances::store_balance_changes(balances_deltas, store);
+}

substreams/ethereum-maverick-v2/src/modules/5_map_protocol_changes.rs

@@ -0,0 +1,118 @@
+use crate::{modules::utils::Params, pb::maverick::v2::Pool};
+use anyhow::Result;
+use itertools::Itertools;
+use std::collections::HashMap;
+use substreams::{pb::substreams::StoreDeltas, prelude::StoreGetProto, store::StoreGet};
+use substreams_ethereum::pb::eth::v2::Block;
+use substreams_helper::hex::Hexable;
+use tycho_substreams::{
+    balances::aggregate_balances_changes, contract::extract_contract_changes_builder, prelude::*,
+};
+
+#[substreams::handlers::map]
+pub fn map_protocol_changes(
+    params: String,
+    block: Block,
+    protocol_components: BlockTransactionProtocolComponents,
+    balance_deltas: BlockBalanceDeltas,
+    pool_store: StoreGetProto<Pool>,
+    balance_store: StoreDeltas,
+) -> Result<BlockChanges> {
+    let params = Params::parse_from_query(&params)?;
+    let (factory_address, quoter_address) = params.decode_addresses()?;
+    let mut transaction_changes: HashMap<_, TransactionChangesBuilder> = HashMap::new();
+
+    protocol_components
+        .tx_components
+        .iter()
+        .for_each(|tx_component| {
+            let tx = tx_component.tx.as_ref().unwrap();
+            let builder = transaction_changes
+                .entry(tx.index)
+                .or_insert_with(|| TransactionChangesBuilder::new(tx));
+
+            tx_component
+                .components
+                .iter()
+                .for_each(|c| {
+                    builder.add_protocol_component(c);
+                });
+        });
+
+    aggregate_balances_changes(balance_store, balance_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))
+                });
+        });
+
+    extract_contract_changes_builder(
+        &block,
+        |addr| {
+            pool_store
+                .get_last(format!("Pool:0x{}", hex::encode(addr)))
+                .is_some() ||
+                addr.eq(factory_address.as_slice())
+        },
+        &mut transaction_changes,
+    );
+
+    block
+        .transactions()
+        .for_each(|block_tx| {
+            block_tx.calls.iter().for_each(|call| {
+                if call.address == quoter_address {
+                    let mut contract_change =
+                        InterimContractChange::new(call.address.as_slice(), true);
+
+                    if let Some(code_change) = &call.code_changes.first() {
+                        contract_change.set_code(&code_change.new_code);
+                    }
+
+                    let builder = transaction_changes
+                        .entry(block_tx.index.into())
+                        .or_insert_with(|| TransactionChangesBuilder::new(&(block_tx.into())));
+                    builder.add_contract_changes(&contract_change);
+                }
+            });
+        });
+
+    transaction_changes
+        .iter_mut()
+        .for_each(|(_, change)| {
+            // this indirection is necessary due to borrowing rules.
+            let addresses = change
+                .changed_contracts()
+                .map(|e| e.to_vec())
+                .collect::<Vec<_>>();
+            addresses
+                .into_iter()
+                .for_each(|address| {
+                    // check if the address is not a pool
+                    if address != factory_address.as_slice() && address != quoter_address.as_slice()
+                    {
+                        let pool = pool_store
+                            .get_last(format!("Pool:0x{}", hex::encode(address)))
+                            .unwrap();
+                        change.mark_component_as_updated(&pool.address.to_hex());
+                    }
+                })
+        });
+
+    Ok(BlockChanges {
+        block: Some((&block).into()),
+        changes: transaction_changes
+            .drain()
+            .sorted_unstable_by_key(|(index, _)| *index)
+            .filter_map(|(_, builder)| builder.build())
+            .collect::<Vec<_>>(),
+    })
+}

substreams/ethereum-maverick-v2/src/modules/mod.rs

@@ -0,0 +1,21 @@
+pub use map_components::map_components;
+pub use map_protocol_changes::map_protocol_changes;
+pub use map_relative_balances::map_relative_balances;
+pub use store_balances::store_balances;
+pub use store_components::store_components;
+
+#[path = "1_map_components.rs"]
+mod map_components;
+
+#[path = "2_store_components.rs"]
+mod store_components;
+
+#[path = "3_map_relative_balances.rs"]
+mod map_relative_balances;
+
+#[path = "4_store_balances.rs"]
+mod store_balances;
+
+#[path = "5_map_protocol_changes.rs"]
+mod map_protocol_changes;
+mod utils;

substreams/ethereum-maverick-v2/src/modules/utils.rs

@@ -0,0 +1,26 @@
+use anyhow::{anyhow, Result};
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+pub struct Params {
+    pub factory: String,
+    pub quoter: String,
+}
+
+impl Params {
+    pub fn parse_from_query(input: &str) -> Result<Self> {
+        serde_qs::from_str(input).map_err(|e| anyhow!("Failed to parse query params: {}", e))
+    }
+
+    pub fn decode_addresses(&self) -> Result<([u8; 20], [u8; 20])> {
+        let factory =
+            hex::decode(&self.factory).map_err(|e| anyhow!("Invalid factory hex: {}", e))?;
+        let quoter = hex::decode(&self.quoter).map_err(|e| anyhow!("Invalid quoter hex: {}", e))?;
+
+        if factory.len() != 20 || quoter.len() != 20 {
+            return Err(anyhow!("Addresses must be 20 bytes"));
+        }
+
+        Ok((factory.try_into().unwrap(), quoter.try_into().unwrap()))
+    }
+}