add balance changes

Cargo.toml

@@ -1,9 +1,5 @@
 [workspace]
-members = [
-    "blocks/evm",
-    "common"
-]
-
+members = ["blocks/evm", "common"]
 resolver = "2"
 
 [workspace.package]

README.md

@@ -14,7 +14,7 @@
 - [ ] **Traces**
   - [ ] **Creation Traces**
 - [ ] **Beacon**
-- [ ] **BalanceChanges**
+- [x] **BalanceChanges**
 
 ## Supported Networks
 
@@ -107,11 +107,11 @@ CREATE TABLE IF NOT EXISTS blocks
 - Snowflake
 - BigQuery
 - Databricks
-  https://docs.databricks.com/en/connect/storage/amazon-s3.html
+  <https://docs.databricks.com/en/connect/storage/amazon-s3.html>
 - Clickhouse
 - Postgres
 - Amazon Redshift
-  https://aws.amazon.com/redshift/
+  <https://aws.amazon.com/redshift/>
 
 ## Graph
 
@@ -138,4 +138,4 @@ FROM blocks
 WHERE date='2015-07-31'
 INTO OUTFILE 'eth_2015-07-31_blocks.csv'
 FORMAT CSVWithNames
-```
+```

blocks/evm/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "raw_blocks_evm"
-edition.workspace = true
-version.workspace = true
+edition = { workspace = true }
+version = { workspace = true }
 
 [lib]
 crate-type = ["cdylib"]

blocks/evm/Makefile

@@ -27,7 +27,7 @@ run:
 
 .PHONY: gui
 gui:
-	substreams gui -e eth.substreams.pinax.network:443 ch_out -s 1500000
+	substreams gui -e eth.substreams.pinax.network:443 ch_out -s 15000000
 
 .PHONY: sql-setup
 sql-setup:

blocks/evm/schema.sql

@@ -68,4 +68,21 @@ CREATE TABLE IF NOT EXISTS logs
     ENGINE = ReplacingMergeTree()
         PRIMARY KEY (block_date, block_time, block_number, log_index, tx_hash)
         ORDER BY (block_date, block_time, block_number, log_index, tx_hash)
-        COMMENT 'Ethereum event logs';
+        COMMENT 'Ethereum event logs';
+
+CREATE TABLE IF NOT EXISTS balance_changes
+(
+    block_time          DateTime('UTC'),
+    block_number        UInt64,
+    block_hash          String,
+    block_date          LowCardinality(String),
+    address             String,
+    new_value           String,
+    old_value           String,
+    ordinal             UInt64,
+    reason              Int32
+)
+    ENGINE = ReplacingMergeTree()
+        PRIMARY KEY (block_date, block_time, block_number, log_index, tx_hash)
+        ORDER BY (block_date, block_time, block_number, log_index, tx_hash)
+        COMMENT 'Ethereum balance changes';

blocks/evm/src/balance_changes.rs

@@ -0,0 +1,27 @@
+use common::keys::balance_changes_keys;
+use common::sinks::insert_timestamp;
+use common::utils::bytes_to_hex;
+use substreams::pb::substreams::Clock;
+use substreams_database_change::pb::database::{table_change, DatabaseChanges};
+use substreams_ethereum::pb::eth::v2::Block;
+
+// https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L658
+pub fn insert_balance_changes(tables: &mut DatabaseChanges, clock: &Clock, block: &Block) {
+    for balance_change in block.balance_changes.clone() {
+        let address = bytes_to_hex(balance_change.address);
+        let new_value = bytes_to_hex(balance_change.new_value.unwrap_or_default().bytes);
+        let old_value = bytes_to_hex(balance_change.old_value.unwrap_or_default().bytes);
+        let ordinal = balance_change.ordinal.to_string();
+        let reason = balance_change.reason.to_string();
+        let keys = balance_changes_keys(&clock, &ordinal);
+        let row = tables
+            .push_change_composite("balance_changes", keys, 0, table_change::Operation::Create)
+            .change("address", ("", address.as_str()))
+            .change("new_value", ("", new_value.as_str()))
+            .change("old_value", ("", old_value.as_str()))
+            .change("ordinal", ("", ordinal.as_str()))
+            .change("reason", ("", reason.as_str()));
+
+        insert_timestamp(row, clock, false);
+    }
+}

blocks/evm/src/blocks.rs

@@ -1,6 +1,6 @@
-use common::{keys::blocks_keys, utils::optional_bigint_to_string};
-use common::utils::{bytes_to_hex, optional_uint64_to_string};
 use common::sinks::insert_timestamp;
+use common::utils::{bytes_to_hex, optional_uint64_to_string};
+use common::{keys::blocks_keys, utils::optional_bigint_to_string};
 use substreams::pb::substreams::Clock;
 use substreams_database_change::pb::database::{table_change, DatabaseChanges};
 use substreams_ethereum::pb::eth::v2::Block;
@@ -24,6 +24,12 @@ pub fn insert_blocks(tables: &mut DatabaseChanges, clock: &Clock, block: &Block)
     let withdrawals_root = bytes_to_hex(header.withdrawals_root);
     let parent_beacon_root = bytes_to_hex(header.parent_beacon_root);
 
+    let difficulty = optional_bigint_to_string(header.difficulty);
+    let total_difficulty = optional_bigint_to_string(header.total_difficulty);
+    let base_fee_per_gas = optional_bigint_to_string(header.base_fee_per_gas);
+    let excess_blob_gas = header.excess_blob_gas.unwrap_or_default().to_string();
+    let blob_gas_used = header.blob_gas_used.unwrap_or_default().to_string();
+
     // blocks
     let keys = blocks_keys(&clock);
     let row = tables
@@ -41,20 +47,16 @@ pub fn insert_blocks(tables: &mut DatabaseChanges, clock: &Clock, block: &Block)
         .change("extra_data", ("", extra_data.as_str()))
         .change("gas_limit", ("", gas_limit.as_str()))
         .change("gas_used", ("", gas_used.as_str()))
-        .change("difficulty", ("", optional_bigint_to_string(header.difficulty).as_str()))
-        .change("total_difficulty", ("", optional_bigint_to_string(header.total_difficulty).as_str()))
-
+        .change("difficulty", ("", difficulty.as_str()))
+        .change("total_difficulty", ("", total_difficulty.as_str()))
         // EIP-1559 (London Fork)
-        .change("base_fee_per_gas", ("", optional_bigint_to_string(header.base_fee_per_gas).as_str()))
-
+        .change("base_fee_per_gas", ("", base_fee_per_gas.as_str()))
         // EIP-4895 (Shangai Fork)
         .change("withdrawals_root", ("", withdrawals_root.as_str()))
-
         // EIP-4844 & EIP-4788 (Dencun Fork)
         .change("parent_beacon_root", ("", parent_beacon_root.as_str()))
-        .change("excess_blob_gas", ("", optional_uint64_to_string(header.excess_blob_gas).as_str()))
-        .change("blob_gas_used", ("", optional_uint64_to_string(header.blob_gas_used).as_str()))
-    ;
+        .change("excess_blob_gas", ("", excess_blob_gas.as_str()))
+        .change("blob_gas_used", ("", blob_gas_used.as_str()));
 
     insert_timestamp(row, clock, true);
 
@@ -73,5 +75,4 @@ pub fn insert_blocks(tables: &mut DatabaseChanges, clock: &Clock, block: &Block)
     row.change("total_transactions", ("", total_transactions.to_string().as_str()))
         .change("successful_transactions", ("", successful_transactions.to_string().as_str()))
         .change("failed_transactions", ("", failed_transactions.to_string().as_str()));
-
-}
+}

blocks/evm/src/lib.rs

@@ -1,3 +1,4 @@
-mod sinks;
+mod balance_changes;
 mod blocks;
-mod logs;
+mod logs;
+mod sinks;

blocks/evm/src/logs.rs

@@ -1,13 +1,12 @@
 use common::keys::logs_keys;
-use common::utils::{bytes_to_hex, extract_topic};
 use common::sinks::insert_timestamp;
+use common::utils::{bytes_to_hex, extract_topic};
 use substreams::pb::substreams::Clock;
 use substreams_database_change::pb::database::{table_change, DatabaseChanges};
 use substreams_ethereum::pb::eth::v2::Block;
 
 // https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L512
 pub fn insert_logs(tables: &mut DatabaseChanges, clock: &Clock, block: &Block) {
-    // logs
     for log in block.logs() {
         let log_index = log.index().to_string();
         let transaction = log.receipt.transaction;
@@ -40,4 +39,4 @@ pub fn insert_logs(tables: &mut DatabaseChanges, clock: &Clock, block: &Block) {
 
         insert_timestamp(row, clock, false);
     }
-}
+}

blocks/evm/src/sinks.rs

@@ -3,6 +3,7 @@ use substreams::pb::substreams::Clock;
 use substreams_database_change::pb::database::DatabaseChanges;
 use substreams_ethereum::pb::eth::v2::Block;
 
+use crate::balance_changes::insert_balance_changes;
 use crate::blocks::insert_blocks;
 use crate::logs::insert_logs;
 
@@ -11,5 +12,6 @@ pub fn ch_out(clock: Clock, block: Block) -> Result<DatabaseChanges, Error> {
     let mut tables: DatabaseChanges = DatabaseChanges::default();
     insert_blocks(&mut tables, &clock, &block);
     insert_logs(&mut tables, &clock, &block);
+    insert_balance_changes(&mut tables, &clock, &block);
     Ok(tables)
 }

common/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "common"
-edition.workspace = true
-version.workspace = true
+edition = { workspace = true }
+version = { workspace = true }
 
 [dependencies]
 substreams = { workspace = true }

common/src/keys.rs

@@ -15,7 +15,7 @@ pub fn blocks_keys(clock: &Clock) -> HashMap<String, String> {
         ("block_date".to_string(), block_date),
         ("block_time".to_string(), block_time),
         ("block_number".to_string(), block_number),
-        ("block_hash".to_string(), block_hash)
+        ("block_hash".to_string(), block_hash),
     ])
 }
 
@@ -32,4 +32,18 @@ pub fn logs_keys(clock: &Clock, log_index: &String, tx_hash: &String) -> HashMap
         ("log_index".to_string(), log_index.to_string()),
         ("tx_hash".to_string(), tx_hash.to_string()),
     ])
-}
+}
+
+pub fn balance_changes_keys(clock: &Clock, ordinal: &String) -> HashMap<String, String> {
+    let timestamp = clock.clone().timestamp.unwrap();
+    let block_date = block_time_to_date(&timestamp.to_string()).to_string();
+    let block_time = timestamp.seconds.to_string();
+    let block_number = clock.number.to_string();
+
+    HashMap::from([
+        ("block_date".to_string(), block_date),
+        ("block_time".to_string(), block_time),
+        ("block_number".to_string(), block_number),
+        ("ordinal".to_string(), ordinal.to_string()),
+    ])
+}

rustfmt.toml

@@ -0,0 +1 @@
+max_width = 200