Add traces

blocks/evm/schema-temp.sql

@@ -57,25 +57,3 @@ CREATE TABLE IF NOT EXISTS transactions
 
 -- trace_type
 -- status
-
-CREATE TABLE IF NOT EXISTS logs
-(
-    block_time          DateTime('UTC'),
-    block_number        UInt64,
-    block_hash          String,
-    block_date          LowCardinality(String), -- Date
-    contract_address    String,
-    topic0              String,
-    topic1              String,
-    topic2              String,
-    topic3              String,
-    data                String,
-    log_index           UInt32,
-    tx_hash             String,
-    tx_index            UInt32,
-    tx_from             String,
-    tx_to               String
-)
-    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);

blocks/evm/schema.sql

@@ -86,4 +86,48 @@ CREATE TABLE IF NOT EXISTS balance_changes
     ENGINE = ReplacingMergeTree()
         PRIMARY KEY (block_date, block_time, block_number, ordinal)
         ORDER BY (block_date, block_time, block_number, ordinal)
-        COMMENT 'Ethereum balance changes';
+        COMMENT 'Ethereum balance changes';
+
+CREATE TABLE IF NOT EXISTS traces
+(
+    -- block --
+    block_time          DateTime('UTC'),
+    block_number        UInt64,
+    block_hash          String,
+    block_date          Date,
+
+    -- transaction --
+    tx_hash             String,
+    tx_index            UInt32,
+    tx_status           LowCardinality(String),
+    tx_status_code      Int32,
+    tx_is_successful    Bool,
+    from                String,
+    to                  String,
+
+    -- trace --
+    index               UInt32,
+    parent_index        UInt32,
+    depth               UInt32,
+    caller              String,
+    call_type           LowCardinality(String),
+    call_type_code      Int32,
+    address             String,
+    value               UInt256,
+    gas_limit           UInt64,
+    gas_consumed        UInt64,
+    return_data         String,
+    input               String,
+    suicide             Bool,
+    failure_reason      LowCardinality(String),
+    state_reverted      Bool,
+    status_reverted     Bool,
+    status_failed       Bool,
+    executed_code       Bool,
+    begin_ordinal       UInt64,
+    end_ordinal         UInt64
+)
+    ENGINE = ReplacingMergeTree()
+        PRIMARY KEY (block_date, block_time, block_number, tx_hash, tx_index, index)
+        ORDER BY (block_date, block_time, block_number, tx_hash, tx_index, index)
+        COMMENT 'Ethereum traces';

blocks/evm/src/lib.rs

@@ -2,5 +2,5 @@ mod balance_changes;
 mod blocks;
 mod logs;
 mod sinks;
-// mod traces;
+mod traces;
 // mod transactions;

blocks/evm/src/traces.rs

@@ -0,0 +1,102 @@
+use common::keys::traces_keys;
+use common::sinks::insert_timestamp;
+use common::utils::bytes_to_hex;
+use common::{keys::balance_changes_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, TransactionTrace};
+
+pub fn trace_status_to_string(status: i32) -> String {
+    match status {
+        0 => "Unknown".to_string(),
+        1 => "Succeeded".to_string(),
+        2 => "Failed".to_string(),
+        3 => "Reverted".to_string(),
+        _ => "Unknown".to_string(),
+    }
+}
+
+pub fn call_types_to_string(call_type: i32) -> String {
+    match call_type {
+        0 => "Unspecified".to_string(),
+        1 => "Call".to_string(),
+        2 => "Callcode".to_string(),
+        3 => "Delegate".to_string(),
+        4 => "Static".to_string(),
+        5 => "Create".to_string(),
+        _ => "Unknown".to_string(),
+    }
+}
+
+// https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L546
+pub fn insert_traces(tables: &mut DatabaseChanges, clock: &Clock, transaction: &TransactionTrace) {
+    let tx_index = transaction.index.to_string();
+    let tx_hash = bytes_to_hex(transaction.hash);
+    let from = bytes_to_hex(transaction.from); // does trace contain `from`?
+    let to = bytes_to_hex(transaction.to); // does trace contain `to`?
+    let tx_status = trace_status_to_string(transaction.status);
+    let tx_status_code = transaction.status.to_string();
+    let tx_is_successful = (transaction.status == 1).to_string();
+
+    // traces
+    for trace in transaction.calls.iter() {
+        let address = bytes_to_hex(trace.address); // additional `trace_address`?
+        let begin_ordinal = trace.begin_ordinal.to_string();
+        let call_type = call_types_to_string(trace.call_type);
+        let call_type_code = trace.call_type.to_string();
+        let caller = bytes_to_hex(trace.caller);
+        let depth = trace.depth.to_string();
+        let end_ordinal = trace.end_ordinal.to_string();
+        let executed_code = trace.executed_code.to_string();
+        let failure_reason = trace.failure_reason.to_string();
+        let gas_consumed = trace.gas_consumed.to_string();
+        let gas_limit = trace.gas_limit.to_string();
+        let index = trace.index.to_string(); // or `subtraces`?
+        let input = bytes_to_hex(trace.input);
+        let parent_index = trace.parent_index.to_string();
+        let return_data = bytes_to_hex(trace.return_data);
+        let state_reverted = trace.state_reverted.to_string();
+        let status_failed = trace.status_failed.to_string();
+        let status_reverted = trace.status_reverted.to_string();
+        let suicide = trace.suicide.to_string(); // or `selfdestruct`?
+        let value = optional_bigint_to_string(trace.value); // UInt256
+
+        // TODO: trace.code_changes
+        // TODO: trace.storage_changes
+
+        let keys = traces_keys(&clock, &tx_hash, &tx_index, &index);
+        let row = tables
+            .push_change_composite("traces", keys, 0, table_change::Operation::Create)
+            // transaction
+            .change("tx_index", ("", tx_index.as_str()))
+            .change("tx_hash", ("", tx_hash.as_str()))
+            .change("from", ("", from.as_str()))
+            .change("to", ("", to.as_str()))
+            .change("tx_status", ("", tx_status.as_str()))
+            .change("tx_status_code", ("", tx_status_code.as_str()))
+            .change("tx_is_successful", ("", tx_is_successful.as_str()))
+            // trace
+            .change("address", ("", address.as_str()))
+            .change("begin_ordinal", ("", begin_ordinal.as_str()))
+            .change("call_type", ("", call_type.as_str()))
+            .change("call_type_code", ("", call_type_code.as_str()))
+            .change("caller", ("", caller.as_str()))
+            .change("depth", ("", depth.as_str()))
+            .change("end_ordinal", ("", end_ordinal.as_str()))
+            .change("executed_code", ("", executed_code.as_str()))
+            .change("failure_reason", ("", failure_reason.as_str()))
+            .change("gas_consumed", ("", gas_consumed.as_str()))
+            .change("gas_limit", ("", gas_limit.as_str()))
+            .change("index", ("", index.as_str()))
+            .change("input", ("", input.as_str()))
+            .change("parent_index", ("", parent_index.as_str()))
+            .change("return_data", ("", return_data.as_str()))
+            .change("state_reverted", ("", state_reverted.as_str()))
+            .change("status_failed", ("", status_failed.as_str()))
+            .change("status_reverted", ("", status_reverted.as_str()))
+            .change("suicide", ("", suicide.as_str()))
+            .change("value", ("", value.as_str()));
+
+        insert_timestamp(row, clock, false);
+    }
+}

blocks/evm/src/transactions.rs

@@ -0,0 +1,33 @@
+use common::sinks::insert_timestamp;
+use common::utils::bytes_to_hex;
+use common::{keys::balance_changes_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;
+
+use crate::traces::insert_traces;
+
+// https://github.com/streamingfast/firehose-ethereum/blob/1bcb32a8eb3e43347972b6b5c9b1fcc4a08c751e/proto/sf/ethereum/type/v2/type.proto#L658
+pub fn insert_transactions(tables: &mut DatabaseChanges, clock: &Clock, block: &Block) {
+    // transactions
+    for transaction in block.transaction_traces.iter() {
+        insert_traces(tables, clock, transaction);
+        // traces
+        for trace in transaction.calls.iter() {}
+        // 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);
+    }
+}

common/src/keys.rs

@@ -47,3 +47,19 @@ pub fn balance_changes_keys(clock: &Clock, ordinal: &String) -> HashMap<String,
         ("ordinal".to_string(), ordinal.to_string()),
     ])
 }
+
+pub fn traces_keys(clock: &Clock, tx_hash: &String, tx_index: &String, index: &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),
+        ("tx_hash".to_string(), tx_hash.to_string()),
+        ("tx_index".to_string(), tx_index.to_string()),
+        ("index".to_string(), index.to_string()),
+    ])
+}