Refactored parse event

blocks/evm/src/account_creations.rs

@@ -1,56 +1,47 @@
 use common::structs::BlockTimestamp;
 use common::utils::bytes_to_hex;
 
-use substreams_ethereum::pb::eth::v2::Block;
+use substreams_ethereum::pb::eth::v2::{AccountCreation, Block, TransactionTrace};
 
-use crate::pb::evm::AccountCreation;
+use crate::pb::evm::AccountCreation as AccountCreationEvent;
 
 // https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L736
 // DetailLevel: EXTENDED
-pub fn collect_account_creations(block: &Block, timestamp: &BlockTimestamp) -> Vec<AccountCreation> {
-    let mut account_creations: Vec<AccountCreation> = vec![];
+pub fn collect_account_creations(block: &Block, timestamp: &BlockTimestamp) -> Vec<AccountCreationEvent> {
+    let mut account_creations: Vec<AccountCreationEvent> = vec![];
 
     // Collect account creations from system calls
     for call in &block.system_calls {
         for account_creation in &call.account_creations {
-            account_creations.push(AccountCreation {
-                // block
-                block_time: Some(timestamp.time),
-                block_number: timestamp.number,
-                block_hash: timestamp.hash.clone(),
-                block_date: timestamp.date.clone(),
-
-                // transaction
-                tx_hash: Some("".to_string()),
-
-                // account creation
-                account: bytes_to_hex(&account_creation.account),
-                ordinal: account_creation.ordinal,
-            });
+            account_creations.push(parse_account_creation(account_creation, &TransactionTrace::default(), timestamp));
         }
     }
 
     // Collect account creations from transaction traces
     for transaction in &block.transaction_traces {
         for call in &transaction.calls {
             for account_creation in &call.account_creations {
-                account_creations.push(AccountCreation {
-                    // block
-                    block_time: Some(timestamp.time),
-                    block_number: timestamp.number,
-                    block_hash: timestamp.hash.clone(),
-                    block_date: timestamp.date.clone(),
-
-                    // transaction
-                    tx_hash: Some(bytes_to_hex(&transaction.hash)),
-
-                    // account creation
-                    account: bytes_to_hex(&account_creation.account),
-                    ordinal: account_creation.ordinal,
-                });
+                account_creations.push(parse_account_creation(account_creation, transaction, timestamp));
             }
         }
     }
 
     account_creations
 }
+
+pub fn parse_account_creation(account_creation: &AccountCreation, transaction: &TransactionTrace, timestamp: &BlockTimestamp) -> AccountCreationEvent {
+    AccountCreationEvent {
+        // block
+        block_time: Some(timestamp.time),
+        block_number: timestamp.number,
+        block_hash: timestamp.hash.clone(),
+        block_date: timestamp.date.clone(),
+
+        // transaction
+        tx_hash: Some(bytes_to_hex(&transaction.hash)),
+
+        // account creation
+        account: bytes_to_hex(&account_creation.account),
+        ordinal: account_creation.ordinal,
+    }
+}

blocks/evm/src/balance_changes.rs

@@ -1,9 +1,9 @@
 use common::structs::BlockTimestamp;
 use common::utils::optional_bigint_to_string;
 use common::utils::{bytes_to_hex, optional_bigint_to_decimal};
-use substreams_ethereum::pb::eth::v2::Block;
+use substreams_ethereum::pb::eth::v2::{BalanceChange, Block, TransactionTrace};
 
-use crate::pb::evm::BalanceChange;
+use crate::pb::evm::BalanceChange as BalanceChangeEvent;
 
 pub fn balance_change_reason_to_string(reason: i32) -> String {
     match reason {
@@ -35,62 +35,47 @@ pub fn balance_change_reason_to_string(reason: i32) -> String {
 
 // https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L658
 // DetailLevel: EXTENDED
-pub fn collect_balance_changes(block: &Block, timestamp: &BlockTimestamp) -> Vec<BalanceChange> {
-    let mut balance_changes: Vec<BalanceChange> = vec![];
+pub fn collect_balance_changes(block: &Block, timestamp: &BlockTimestamp) -> Vec<BalanceChangeEvent> {
+    let mut balance_changes: Vec<BalanceChangeEvent> = vec![];
 
     // Collect balance changes from system calls
     for call in &block.system_calls {
         for balance_change in &call.balance_changes {
-            let amount = optional_bigint_to_decimal(balance_change.new_value.clone()) - optional_bigint_to_decimal(balance_change.old_value.clone());
-            balance_changes.push(BalanceChange {
-                // block
-                block_time: Some(timestamp.time),
-                block_number: timestamp.number,
-                block_hash: timestamp.hash.clone(),
-                block_date: timestamp.date.clone(),
-
-                // transaction
-                tx_hash: Some("".to_string()),
-
-                // balance changes
-                address: bytes_to_hex(&balance_change.address),
-                new_balance: optional_bigint_to_string(&balance_change.new_value, "0"),
-                old_balance: optional_bigint_to_string(&balance_change.old_value, "0"),
-                amount: amount.to_string(),
-                ordinal: balance_change.ordinal,
-                reason: balance_change_reason_to_string(balance_change.reason),
-                reason_code: balance_change.reason as u32,
-            });
+            balance_changes.push(parse_balance_change(balance_change, &TransactionTrace::default(), timestamp));
         }
     }
 
     // Collect balance changes from transaction traces
     for transaction in &block.transaction_traces {
         for call in &transaction.calls {
             for balance_change in &call.balance_changes {
-                let amount = optional_bigint_to_decimal(balance_change.new_value.clone()) - optional_bigint_to_decimal(balance_change.old_value.clone());
-                balance_changes.push(BalanceChange {
-                    // block
-                    block_time: Some(timestamp.time),
-                    block_number: timestamp.number,
-                    block_hash: timestamp.hash.clone(),
-                    block_date: timestamp.date.clone(),
-
-                    // transaction
-                    tx_hash: Some(bytes_to_hex(&transaction.hash)),
-
-                    // balance changes
-                    address: bytes_to_hex(&balance_change.address),
-                    new_balance: optional_bigint_to_string(&balance_change.new_value, "0"),
-                    old_balance: optional_bigint_to_string(&balance_change.old_value, "0"),
-                    amount: amount.to_string(),
-                    ordinal: balance_change.ordinal,
-                    reason: balance_change_reason_to_string(balance_change.reason),
-                    reason_code: balance_change.reason as u32,
-                });
+                balance_changes.push(parse_balance_change(balance_change, transaction, timestamp));
             }
         }
     }
 
     balance_changes
 }
+
+pub fn parse_balance_change(balance_change: &BalanceChange, transaction: &TransactionTrace, timestamp: &BlockTimestamp) -> BalanceChangeEvent {
+    let amount = optional_bigint_to_decimal(balance_change.new_value.clone()) - optional_bigint_to_decimal(balance_change.old_value.clone());
+    BalanceChangeEvent {
+        // block
+        block_time: Some(timestamp.time),
+        block_number: timestamp.number,
+        block_hash: timestamp.hash.clone(),
+        block_date: timestamp.date.clone(),
+
+        // transaction
+        tx_hash: Some(bytes_to_hex(&transaction.hash)),
+
+        // balance changes
+        address: bytes_to_hex(&balance_change.address),
+        new_balance: optional_bigint_to_string(&balance_change.new_value, "0"),
+        old_balance: optional_bigint_to_string(&balance_change.old_value, "0"),
+        amount: amount.to_string(),
+        ordinal: balance_change.ordinal,
+        reason: balance_change_reason_to_string(balance_change.reason),
+        reason_code: balance_change.reason as u32,
+    }
+}

blocks/evm/src/code_changes.rs

@@ -1,63 +1,50 @@
 use common::structs::BlockTimestamp;
 use common::utils::bytes_to_hex;
-use substreams_ethereum::pb::eth::v2::Block;
+use substreams_ethereum::pb::eth::v2::{Block, CodeChange, TransactionTrace};
 
-use crate::pb::evm::CodeChange;
+use crate::pb::evm::CodeChange as CodeChangeEvent;
 
 // https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L744
 // DetailLevel: EXTENDED
-pub fn collect_code_changes(block: &Block, timestamp: &BlockTimestamp) -> Vec<CodeChange> {
-    let mut code_changes: Vec<CodeChange> = vec![];
+pub fn collect_code_changes(block: &Block, timestamp: &BlockTimestamp) -> Vec<CodeChangeEvent> {
+    let mut code_changes: Vec<CodeChangeEvent> = vec![];
 
     // Collect code changes from system calls
     for call in &block.system_calls {
         for code_change in &call.code_changes {
-            code_changes.push(CodeChange {
-                // block
-                block_time: Some(timestamp.time),
-                block_number: timestamp.number,
-                block_hash: timestamp.hash.clone(),
-                block_date: timestamp.date.clone(),
-
-                // transaction
-                tx_hash: Some(String::new()),
-
-                // code changes
-                address: bytes_to_hex(&code_change.address),
-                old_hash: bytes_to_hex(&code_change.old_hash),
-                old_code: bytes_to_hex(&code_change.old_code),
-                new_hash: bytes_to_hex(&code_change.new_hash),
-                new_code: bytes_to_hex(&code_change.new_code),
-                ordinal: code_change.ordinal,
-            });
+            code_changes.push(parse_code_change(code_change, &TransactionTrace::default(), timestamp));
         }
     }
 
     // Collect code changes from transaction traces
     for transaction in &block.transaction_traces {
         for call in &transaction.calls {
             for code_change in &call.code_changes {
-                code_changes.push(CodeChange {
-                    // block
-                    block_time: Some(timestamp.time),
-                    block_number: timestamp.number,
-                    block_hash: timestamp.hash.clone(),
-                    block_date: timestamp.date.clone(),
-
-                    // transaction
-                    tx_hash: Some(bytes_to_hex(&transaction.hash)),
-
-                    // code changes
-                    address: bytes_to_hex(&code_change.address),
-                    old_hash: bytes_to_hex(&code_change.old_hash),
-                    old_code: bytes_to_hex(&code_change.old_code),
-                    new_hash: bytes_to_hex(&code_change.new_hash),
-                    new_code: bytes_to_hex(&code_change.new_code),
-                    ordinal: code_change.ordinal,
-                });
+                code_changes.push(parse_code_change(code_change, transaction, timestamp));
             }
         }
     }
 
     code_changes
 }
+
+pub fn parse_code_change(code_change: &CodeChange, transaction: &TransactionTrace, timestamp: &BlockTimestamp) -> CodeChangeEvent {
+    CodeChangeEvent {
+        // block
+        block_time: Some(timestamp.time),
+        block_number: timestamp.number,
+        block_hash: timestamp.hash.clone(),
+        block_date: timestamp.date.clone(),
+
+        // transaction
+        tx_hash: Some(bytes_to_hex(&transaction.hash)),
+
+        // code changes
+        address: bytes_to_hex(&code_change.address),
+        old_hash: bytes_to_hex(&code_change.old_hash),
+        old_code: bytes_to_hex(&code_change.old_code),
+        new_hash: bytes_to_hex(&code_change.new_hash),
+        new_code: bytes_to_hex(&code_change.new_code),
+        ordinal: code_change.ordinal,
+    }
+}

blocks/evm/src/events.rs

@@ -20,6 +20,7 @@ use crate::transactions::collect_transactions;
 #[substreams::handlers::map]
 pub fn map_events(clock: Clock, block: Block) -> Result<Events, Error> {
     let timestamp = build_timestamp_with_prefix(&clock);
+    let mut events = Events::default();
 
     Ok(Events {
         blocks: vec![collect_block(&block, &timestamp)],

blocks/evm/src/gas_changes.rs

@@ -1,7 +1,7 @@
 use common::{structs::BlockTimestamp, utils::bytes_to_hex};
-use substreams_ethereum::pb::eth::v2::Block;
+use substreams_ethereum::pb::eth::v2::{Block, GasChange, TransactionTrace};
 
-use crate::pb::evm::GasChange;
+use crate::pb::evm::GasChange as GasChangeEvent;
 
 pub fn gas_change_reason_to_string(reason: i32) -> String {
     match reason {
@@ -37,56 +37,44 @@ pub fn gas_change_reason_to_string(reason: i32) -> String {
 
 // https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L726C9-L726C20
 // DetailLevel: EXTENDED
-pub fn collect_gas_changes(block: &Block, timestamp: &BlockTimestamp) -> Vec<GasChange> {
-    let mut gas_changes: Vec<GasChange> = vec![];
+pub fn collect_gas_changes(block: &Block, timestamp: &BlockTimestamp) -> Vec<GasChangeEvent> {
+    let mut gas_changes: Vec<GasChangeEvent> = vec![];
 
     // Collect gas changes from system calls
     for call in &block.system_calls {
         for gas_change in &call.gas_changes {
-            gas_changes.push(GasChange {
-                // block
-                block_time: Some(timestamp.time),
-                block_number: timestamp.number,
-                block_hash: timestamp.hash.clone(),
-                block_date: timestamp.date.clone(),
-
-                // transaction
-                tx_hash: Some(String::new()),
-
-                // gas changes
-                old_value: gas_change.old_value,
-                new_value: gas_change.new_value,
-                reason: gas_change_reason_to_string(gas_change.reason),
-                reason_code: gas_change.reason as u32,
-                ordinal: gas_change.ordinal,
-            });
+            gas_changes.push(parse_gas_changes(gas_change, &TransactionTrace::default(), timestamp));
         }
     }
 
     // Collect gas changes from transaction traces
     for transaction in &block.transaction_traces {
         for call in &transaction.calls {
             for gas_change in &call.gas_changes {
-                gas_changes.push(GasChange {
-                    // block
-                    block_time: Some(timestamp.time),
-                    block_number: timestamp.number,
-                    block_hash: timestamp.hash.clone(),
-                    block_date: timestamp.date.clone(),
-
-                    // transaction
-                    tx_hash: Some(bytes_to_hex(&transaction.hash)),
-
-                    // gas changes
-                    old_value: gas_change.old_value,
-                    new_value: gas_change.new_value,
-                    reason: gas_change_reason_to_string(gas_change.reason),
-                    reason_code: gas_change.reason as u32,
-                    ordinal: gas_change.ordinal,
-                });
+                gas_changes.push(parse_gas_changes(gas_change, transaction, timestamp));
             }
         }
     }
 
     gas_changes
 }
+
+pub fn parse_gas_changes(gas_change: &GasChange, transaction: &TransactionTrace, timestamp: &BlockTimestamp) -> GasChangeEvent {
+    GasChangeEvent {
+        // block
+        block_time: Some(timestamp.time),
+        block_number: timestamp.number,
+        block_hash: timestamp.hash.clone(),
+        block_date: timestamp.date.clone(),
+
+        // transaction
+        tx_hash: Some(bytes_to_hex(&transaction.hash)),
+
+        // gas changes
+        old_value: gas_change.old_value,
+        new_value: gas_change.new_value,
+        reason: gas_change_reason_to_string(gas_change.reason),
+        reason_code: gas_change.reason as u32,
+        ordinal: gas_change.ordinal,
+    }
+}