From 7dbf3ffac63064b16bf96f8d342a06f97383a903 Mon Sep 17 00:00:00 2001 From: Florian Pellissier <111426680+flopell@users.noreply.github.com> Date: Fri, 12 Jul 2024 14:46:40 +0200 Subject: [PATCH] feat: Add a generic function to extract balance deltas from Transfer events. --- substreams/crates/tycho-substreams/Cargo.toml | 6 +- .../crates/tycho-substreams/src/abi/erc20.rs | 1066 +++++++++++++++++ .../crates/tycho-substreams/src/abi/mod.rs | 2 + .../crates/tycho-substreams/src/balances.rs | 80 +- substreams/crates/tycho-substreams/src/lib.rs | 1 + 5 files changed, 1152 insertions(+), 3 deletions(-) create mode 100644 substreams/crates/tycho-substreams/src/abi/erc20.rs create mode 100644 substreams/crates/tycho-substreams/src/abi/mod.rs diff --git a/substreams/crates/tycho-substreams/Cargo.toml b/substreams/crates/tycho-substreams/Cargo.toml index 3908f81..a62469d 100644 --- a/substreams/crates/tycho-substreams/Cargo.toml +++ b/substreams/crates/tycho-substreams/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tycho-substreams" -version = "0.1.0" +version = "0.2.0" edition = "2021" [dependencies] @@ -8,4 +8,6 @@ substreams-ethereum.workspace = true substreams.workspace = true prost.workspace = true hex.workspace = true -itertools = "0.12.0" \ No newline at end of file +itertools = "0.12.0" +ethabi.workspace = true +num-bigint = "0.4.4" diff --git a/substreams/crates/tycho-substreams/src/abi/erc20.rs b/substreams/crates/tycho-substreams/src/abi/erc20.rs new file mode 100644 index 0000000..5dd70ca --- /dev/null +++ b/substreams/crates/tycho-substreams/src/abi/erc20.rs @@ -0,0 +1,1066 @@ +const INTERNAL_ERR: &'static str = "`ethabi_derive` internal error"; +/// Contract's functions. +#[allow(dead_code, unused_imports, unused_variables)] +pub mod functions { + use super::INTERNAL_ERR; + #[derive(Debug, Clone, PartialEq)] + pub struct Allowance { + pub owner: Vec, + pub spender: Vec, + } + impl Allowance { + const METHOD_ID: [u8; 4] = [221u8, 98u8, 237u8, 62u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + let maybe_data = call.input.get(4..); + if maybe_data.is_none() { + return Err("no data to decode".to_string()); + } + let mut values = ethabi::decode( + &[ethabi::ParamType::Address, ethabi::ParamType::Address], + maybe_data.unwrap(), + ) + .map_err(|e| format!("unable to decode call.input: {:?}", e))?; + values.reverse(); + Ok(Self { + owner: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + spender: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + }) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[ + ethabi::Token::Address(ethabi::Address::from_slice(&self.owner)), + ethabi::Token::Address(ethabi::Address::from_slice(&self.spender)), + ]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call( + call: &substreams_ethereum::pb::eth::v2::Call, + ) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Uint(256usize)], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok({ + let mut v = [0 as u8; 32]; + values + .pop() + .expect("one output data should have existed") + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for Allowance { + const NAME: &'static str = "allowance"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for Allowance { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct Approve { + pub spender: Vec, + pub value: substreams::scalar::BigInt, + } + impl Approve { + const METHOD_ID: [u8; 4] = [9u8, 94u8, 167u8, 179u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + let maybe_data = call.input.get(4..); + if maybe_data.is_none() { + return Err("no data to decode".to_string()); + } + let mut values = ethabi::decode( + &[ethabi::ParamType::Address, ethabi::ParamType::Uint(256usize)], + maybe_data.unwrap(), + ) + .map_err(|e| format!("unable to decode call.input: {:?}", e))?; + values.reverse(); + Ok(Self { + spender: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + value: { + let mut v = [0 as u8; 32]; + values + .pop() + .expect(INTERNAL_ERR) + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }, + }) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[ + ethabi::Token::Address(ethabi::Address::from_slice(&self.spender)), + ethabi::Token::Uint(ethabi::Uint::from_big_endian( + match self.value.clone().to_bytes_be() { + (num_bigint::Sign::Plus, bytes) => bytes, + (num_bigint::Sign::NoSign, bytes) => bytes, + (num_bigint::Sign::Minus, _) => { + panic!("negative numbers are not supported") + } + } + .as_slice(), + )), + ]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Bool], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok(values + .pop() + .expect("one output data should have existed") + .into_bool() + .expect(INTERNAL_ERR)) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for Approve { + const NAME: &'static str = "approve"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for Approve { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct BalanceOf { + pub owner: Vec, + } + impl BalanceOf { + const METHOD_ID: [u8; 4] = [112u8, 160u8, 130u8, 49u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + let maybe_data = call.input.get(4..); + if maybe_data.is_none() { + return Err("no data to decode".to_string()); + } + let mut values = ethabi::decode(&[ethabi::ParamType::Address], maybe_data.unwrap()) + .map_err(|e| format!("unable to decode call.input: {:?}", e))?; + values.reverse(); + Ok(Self { + owner: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + }) + } + pub fn encode(&self) -> Vec { + let data = + ethabi::encode(&[ethabi::Token::Address(ethabi::Address::from_slice(&self.owner))]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call( + call: &substreams_ethereum::pb::eth::v2::Call, + ) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Uint(256usize)], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok({ + let mut v = [0 as u8; 32]; + values + .pop() + .expect("one output data should have existed") + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for BalanceOf { + const NAME: &'static str = "balanceOf"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for BalanceOf { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct Decimals {} + impl Decimals { + const METHOD_ID: [u8; 4] = [49u8, 60u8, 229u8, 103u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Ok(Self {}) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call( + call: &substreams_ethereum::pb::eth::v2::Call, + ) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Uint(8usize)], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok({ + let mut v = [0 as u8; 32]; + values + .pop() + .expect("one output data should have existed") + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for Decimals { + const NAME: &'static str = "decimals"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for Decimals { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct Name {} + impl Name { + const METHOD_ID: [u8; 4] = [6u8, 253u8, 222u8, 3u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Ok(Self {}) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call( + call: &substreams_ethereum::pb::eth::v2::Call, + ) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::String], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok(values + .pop() + .expect("one output data should have existed") + .into_string() + .expect(INTERNAL_ERR)) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for Name { + const NAME: &'static str = "name"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for Name { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct Symbol {} + impl Symbol { + const METHOD_ID: [u8; 4] = [149u8, 216u8, 155u8, 65u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Ok(Self {}) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call( + call: &substreams_ethereum::pb::eth::v2::Call, + ) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::String], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok(values + .pop() + .expect("one output data should have existed") + .into_string() + .expect(INTERNAL_ERR)) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for Symbol { + const NAME: &'static str = "symbol"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for Symbol { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct TotalSupply {} + impl TotalSupply { + const METHOD_ID: [u8; 4] = [24u8, 22u8, 13u8, 221u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Ok(Self {}) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call( + call: &substreams_ethereum::pb::eth::v2::Call, + ) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Uint(256usize)], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok({ + let mut v = [0 as u8; 32]; + values + .pop() + .expect("one output data should have existed") + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for TotalSupply { + const NAME: &'static str = "totalSupply"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for TotalSupply { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct Transfer { + pub to: Vec, + pub value: substreams::scalar::BigInt, + } + impl Transfer { + const METHOD_ID: [u8; 4] = [169u8, 5u8, 156u8, 187u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + let maybe_data = call.input.get(4..); + if maybe_data.is_none() { + return Err("no data to decode".to_string()); + } + let mut values = ethabi::decode( + &[ethabi::ParamType::Address, ethabi::ParamType::Uint(256usize)], + maybe_data.unwrap(), + ) + .map_err(|e| format!("unable to decode call.input: {:?}", e))?; + values.reverse(); + Ok(Self { + to: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + value: { + let mut v = [0 as u8; 32]; + values + .pop() + .expect(INTERNAL_ERR) + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }, + }) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[ + ethabi::Token::Address(ethabi::Address::from_slice(&self.to)), + ethabi::Token::Uint(ethabi::Uint::from_big_endian( + match self.value.clone().to_bytes_be() { + (num_bigint::Sign::Plus, bytes) => bytes, + (num_bigint::Sign::NoSign, bytes) => bytes, + (num_bigint::Sign::Minus, _) => { + panic!("negative numbers are not supported") + } + } + .as_slice(), + )), + ]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Bool], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok(values + .pop() + .expect("one output data should have existed") + .into_bool() + .expect(INTERNAL_ERR)) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for Transfer { + const NAME: &'static str = "transfer"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for Transfer { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct TransferFrom { + pub from: Vec, + pub to: Vec, + pub value: substreams::scalar::BigInt, + } + impl TransferFrom { + const METHOD_ID: [u8; 4] = [35u8, 184u8, 114u8, 221u8]; + pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + let maybe_data = call.input.get(4..); + if maybe_data.is_none() { + return Err("no data to decode".to_string()); + } + let mut values = ethabi::decode( + &[ + ethabi::ParamType::Address, + ethabi::ParamType::Address, + ethabi::ParamType::Uint(256usize), + ], + maybe_data.unwrap(), + ) + .map_err(|e| format!("unable to decode call.input: {:?}", e))?; + values.reverse(); + Ok(Self { + from: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + to: values + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + value: { + let mut v = [0 as u8; 32]; + values + .pop() + .expect(INTERNAL_ERR) + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }, + }) + } + pub fn encode(&self) -> Vec { + let data = ethabi::encode(&[ + ethabi::Token::Address(ethabi::Address::from_slice(&self.from)), + ethabi::Token::Address(ethabi::Address::from_slice(&self.to)), + ethabi::Token::Uint(ethabi::Uint::from_big_endian( + match self.value.clone().to_bytes_be() { + (num_bigint::Sign::Plus, bytes) => bytes, + (num_bigint::Sign::NoSign, bytes) => bytes, + (num_bigint::Sign::Minus, _) => { + panic!("negative numbers are not supported") + } + } + .as_slice(), + )), + ]); + let mut encoded = Vec::with_capacity(4 + data.len()); + encoded.extend(Self::METHOD_ID); + encoded.extend(data); + encoded + } + pub fn output_call(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::output(call.return_data.as_ref()) + } + pub fn output(data: &[u8]) -> Result { + let mut values = ethabi::decode(&[ethabi::ParamType::Bool], data.as_ref()) + .map_err(|e| format!("unable to decode output data: {:?}", e))?; + Ok(values + .pop() + .expect("one output data should have existed") + .into_bool() + .expect(INTERNAL_ERR)) + } + pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + match call.input.get(0..4) { + Some(signature) => Self::METHOD_ID == signature, + None => false, + } + } + pub fn call(&self, address: Vec) -> Option { + use substreams_ethereum::pb::eth::rpc; + let rpc_calls = rpc::RpcCalls { + calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }], + }; + let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses; + let response = responses + .get(0) + .expect("one response should have existed"); + if response.failed { + return None; + } + match Self::output(response.raw.as_ref()) { + Ok(data) => Some(data), + Err(err) => { + use substreams_ethereum::Function; + substreams::log::info!( + "Call output for function `{}` failed to decode with error: {}", + Self::NAME, + err + ); + None + } + } + } + } + impl substreams_ethereum::Function for TransferFrom { + const NAME: &'static str = "transferFrom"; + fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool { + Self::match_call(call) + } + fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result { + Self::decode(call) + } + fn encode(&self) -> Vec { + self.encode() + } + } + impl substreams_ethereum::rpc::RPCDecodable for TransferFrom { + fn output(data: &[u8]) -> Result { + Self::output(data) + } + } +} +/// Contract's events. +#[allow(dead_code, unused_imports, unused_variables)] +pub mod events { + use super::INTERNAL_ERR; + #[derive(Debug, Clone, PartialEq)] + pub struct Approval { + pub owner: Vec, + pub spender: Vec, + pub value: substreams::scalar::BigInt, + } + impl Approval { + const TOPIC_ID: [u8; 32] = [ + 140u8, 91u8, 225u8, 229u8, 235u8, 236u8, 125u8, 91u8, 209u8, 79u8, 113u8, 66u8, 125u8, + 30u8, 132u8, 243u8, 221u8, 3u8, 20u8, 192u8, 247u8, 178u8, 41u8, 30u8, 91u8, 32u8, + 10u8, 200u8, 199u8, 195u8, 185u8, 37u8, + ]; + pub fn match_log(log: &substreams_ethereum::pb::eth::v2::Log) -> bool { + if log.topics.len() != 3usize { + return false; + } + if log.data.len() != 32usize { + return false; + } + return log + .topics + .get(0) + .expect("bounds already checked") + .as_ref() == + Self::TOPIC_ID; + } + pub fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result { + let mut values = + ethabi::decode(&[ethabi::ParamType::Uint(256usize)], log.data.as_ref()) + .map_err(|e| format!("unable to decode log.data: {:?}", e))?; + values.reverse(); + Ok(Self { + owner: ethabi::decode(&[ethabi::ParamType::Address], log.topics[1usize].as_ref()) + .map_err(|e| { + format!( + "unable to decode param 'owner' from topic of type 'address': {:?}", + e + ) + })? + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + spender: ethabi::decode(&[ethabi::ParamType::Address], log.topics[2usize].as_ref()) + .map_err(|e| { + format!( + "unable to decode param 'spender' from topic of type 'address': {:?}", + e + ) + })? + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + value: { + let mut v = [0 as u8; 32]; + values + .pop() + .expect(INTERNAL_ERR) + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }, + }) + } + } + impl substreams_ethereum::Event for Approval { + const NAME: &'static str = "Approval"; + fn match_log(log: &substreams_ethereum::pb::eth::v2::Log) -> bool { + Self::match_log(log) + } + fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result { + Self::decode(log) + } + } + #[derive(Debug, Clone, PartialEq)] + pub struct Transfer { + pub from: Vec, + pub to: Vec, + pub value: substreams::scalar::BigInt, + } + impl Transfer { + const TOPIC_ID: [u8; 32] = [ + 221u8, 242u8, 82u8, 173u8, 27u8, 226u8, 200u8, 155u8, 105u8, 194u8, 176u8, 104u8, + 252u8, 55u8, 141u8, 170u8, 149u8, 43u8, 167u8, 241u8, 99u8, 196u8, 161u8, 22u8, 40u8, + 245u8, 90u8, 77u8, 245u8, 35u8, 179u8, 239u8, + ]; + pub fn match_log(log: &substreams_ethereum::pb::eth::v2::Log) -> bool { + if log.topics.len() != 3usize { + return false; + } + if log.data.len() != 32usize { + return false; + } + return log + .topics + .get(0) + .expect("bounds already checked") + .as_ref() == + Self::TOPIC_ID; + } + pub fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result { + let mut values = + ethabi::decode(&[ethabi::ParamType::Uint(256usize)], log.data.as_ref()) + .map_err(|e| format!("unable to decode log.data: {:?}", e))?; + values.reverse(); + Ok(Self { + from: ethabi::decode(&[ethabi::ParamType::Address], log.topics[1usize].as_ref()) + .map_err(|e| { + format!( + "unable to decode param 'from' from topic of type 'address': {:?}", + e + ) + })? + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + to: ethabi::decode(&[ethabi::ParamType::Address], log.topics[2usize].as_ref()) + .map_err(|e| { + format!("unable to decode param 'to' from topic of type 'address': {:?}", e) + })? + .pop() + .expect(INTERNAL_ERR) + .into_address() + .expect(INTERNAL_ERR) + .as_bytes() + .to_vec(), + value: { + let mut v = [0 as u8; 32]; + values + .pop() + .expect(INTERNAL_ERR) + .into_uint() + .expect(INTERNAL_ERR) + .to_big_endian(v.as_mut_slice()); + substreams::scalar::BigInt::from_unsigned_bytes_be(&v) + }, + }) + } + } + impl substreams_ethereum::Event for Transfer { + const NAME: &'static str = "Transfer"; + fn match_log(log: &substreams_ethereum::pb::eth::v2::Log) -> bool { + Self::match_log(log) + } + fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result { + Self::decode(log) + } + } +} diff --git a/substreams/crates/tycho-substreams/src/abi/mod.rs b/substreams/crates/tycho-substreams/src/abi/mod.rs new file mode 100644 index 0000000..4542b55 --- /dev/null +++ b/substreams/crates/tycho-substreams/src/abi/mod.rs @@ -0,0 +1,2 @@ +#![allow(clippy::all)] +pub mod erc20; diff --git a/substreams/crates/tycho-substreams/src/balances.rs b/substreams/crates/tycho-substreams/src/balances.rs index 707f60d..ec14fde 100644 --- a/substreams/crates/tycho-substreams/src/balances.rs +++ b/substreams/crates/tycho-substreams/src/balances.rs @@ -21,7 +21,11 @@ //! Through this sequence, the module ensures the transformation from relative to absolute //! balances is conducted with high fidelity, upholding the integrity of transactional data. -use crate::pb::tycho::evm::v1::{BalanceChange, BlockBalanceDeltas, Transaction}; +use crate::{ + abi, + pb::tycho::evm::v1::{BalanceChange, BlockBalanceDeltas, Transaction}, + prelude::BalanceDelta, +}; use itertools::Itertools; use std::{collections::HashMap, str::FromStr}; use substreams::{ @@ -29,6 +33,7 @@ use substreams::{ pb::substreams::StoreDeltas, prelude::{BigInt, StoreAdd}, }; +use substreams_ethereum::{pb::eth::v2::TransactionTrace, Event}; /// Stores relative balance changes in an additive manner. /// @@ -158,6 +163,79 @@ pub fn aggregate_balances_changes( .collect() } +/// Extracts balance deltas from a transaction trace based on a given address predicate. +/// +/// This function processes the logs within a transaction trace to identify ERC-20 token transfer +/// events. It applies the given predicate to determine which addresses are of interest and extracts +/// the balance changes (deltas) for those addresses. The balance deltas are then returned as a +/// vector. +/// +/// # Arguments +/// +/// * `tx` - A reference to a `TransactionTrace` which contains the transaction logs and other +/// details. +/// * `address_predicate` - A predicate function that takes two byte slices representing a token and +/// a component and returns a boolean. This function is used to filter which addresses' balance +/// changes should be extracted. +/// +/// # Returns +/// +/// A vector of `BalanceDelta` structs, each representing a change in balance for a specific address +/// within the transaction. +/// +/// # Example +/// +/// ``` +/// let predicate = |log_address: &[u8], transfer_address: &[u8]| -> bool { +/// // Your predicate logic here, e.g., checking if the address matches a specific pattern. +/// true +/// }; +/// +/// let balance_deltas = extract_balance_deltas_from_tx(&tx, predicate); +/// ``` +/// +/// # Notes +/// +/// - It is assumed that the transactor is the component. If the protocol follows a different +/// design, this function may not be applicable. +/// - The `address_predicate` is applied to both the log address and the `from`/`to` addresses in +/// the transfer event. +pub fn extract_balance_deltas_from_tx bool>( + tx: &TransactionTrace, + address_predicate: F, +) -> Vec { + let mut balance_deltas = vec![]; + + tx.logs_with_calls() + .for_each(|(log, _)| { + if let Some(transfer) = abi::erc20::events::Transfer::match_and_decode(log) { + let mut create_balance_delta = |transactor: &[u8], delta: BigInt| { + balance_deltas.push(BalanceDelta { + ord: log.ordinal, + tx: Some(Transaction { + to: tx.to.clone(), + from: tx.from.clone(), + hash: tx.hash.clone(), + index: tx.index.into(), + }), + token: log.address.clone(), + delta: delta.to_signed_bytes_be(), + component_id: hex::encode(transactor).into(), + }); + }; + + if address_predicate(&log.address, &transfer.from) { + create_balance_delta(&transfer.from, transfer.value.neg()); + } + if address_predicate(&log.address, &transfer.to) { + create_balance_delta(&transfer.to, transfer.value); + } + } + }); + + balance_deltas +} + #[cfg(test)] mod tests { use super::*; diff --git a/substreams/crates/tycho-substreams/src/lib.rs b/substreams/crates/tycho-substreams/src/lib.rs index 03e9e03..60f3047 100644 --- a/substreams/crates/tycho-substreams/src/lib.rs +++ b/substreams/crates/tycho-substreams/src/lib.rs @@ -1,3 +1,4 @@ +mod abi; pub mod balances; pub mod contract; mod mock_store;