Refactor to use same transaction for loop

blocks/antelope/src/account_ram_deltas.rs

@@ -1,6 +1,7 @@
 use crate::pb::antelope::AccountRamDelta as RawAccountRamDelta;
 use crate::transactions::is_transaction_success;
 use common::structs::BlockTimestamp;
+use substreams_antelope::pb::TransactionTrace;
 use substreams_antelope::Block;
 
 pub fn collect_account_ram_deltas(block: &Block, timestamp: &BlockTimestamp) -> Vec<RawAccountRamDelta> {
@@ -32,3 +33,28 @@ pub fn collect_account_ram_deltas(block: &Block, timestamp: &BlockTimestamp) ->
 
     account_ram_deltas
 }
+
+pub fn collect_tx_account_ram_deltas(transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> Vec<RawAccountRamDelta> {
+    let mut account_ram_deltas = Vec::new();
+
+    for action_trace in transaction.action_traces.iter() {
+        let action_index = action_trace.execution_index;
+
+        for (index, delta) in action_trace.account_ram_deltas.iter().enumerate() {
+            account_ram_deltas.push(RawAccountRamDelta {
+                block_time: Some(timestamp.time.clone()),
+                block_number: timestamp.number,
+                block_hash: timestamp.hash.clone(),
+                block_date: timestamp.date.clone(),
+                tx_hash: transaction.id.clone(),
+                tx_success,
+                action_index,
+                index: index as u32,
+                account: delta.account.clone(),
+                delta: delta.delta,
+            });
+        }
+    }
+
+    account_ram_deltas
+}

blocks/antelope/src/actions.rs

@@ -1,6 +1,7 @@
 use common::structs::BlockTimestamp;
 
 use substreams::Hex;
+use substreams_antelope::pb::TransactionTrace;
 use substreams_antelope::Block;
 
 use crate::pb::antelope::Action as RawAction;
@@ -53,3 +54,45 @@ pub fn collect_actions(block: &Block, timestamp: &BlockTimestamp) -> Vec<RawActi
 
     actions
 }
+
+pub fn collect_tx_actions(block: &Block, transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> Vec<RawAction> {
+    let header = block.header.clone().unwrap_or_default();
+    let mut actions: Vec<RawAction> = Vec::new();
+
+    for trace in transaction.action_traces.iter() {
+        let action = trace.action.clone().unwrap_or_default();
+        let receipt = trace.receipt.clone().unwrap_or_default();
+
+        actions.push(RawAction {
+            block_time: Some(timestamp.time.clone()),
+            block_number: timestamp.number,
+            block_hash: timestamp.hash.clone(),
+            block_date: timestamp.date.clone(),
+            tx_hash: transaction.id.clone(),
+            tx_success,
+            abi_sequence: receipt.abi_sequence,
+            code_sequence: receipt.code_sequence,
+            digest: receipt.digest,
+            global_sequence: receipt.global_sequence,
+            receipt_receiver: receipt.receiver,
+            recv_sequence: receipt.recv_sequence,
+            account: action.account,
+            name: action.name,
+            json_data: action.json_data,
+            raw_data: Hex::encode(&action.raw_data),
+            index: trace.execution_index,
+            action_ordinal: trace.action_ordinal,
+            receiver: trace.receiver.clone(),
+            context_free: trace.context_free,
+            elapsed: trace.elapsed,
+            console: trace.console.clone(),
+            raw_return_value: Hex::encode(&trace.raw_return_value),
+            json_return_value: trace.json_return_value.clone(),
+            creator_action_ordinal: trace.creator_action_ordinal,
+            closest_unnotified_ancestor_action_ordinal: trace.closest_unnotified_ancestor_action_ordinal,
+            action_mroot: Hex::encode(&header.action_mroot),
+        });
+    }
+
+    actions
+}

blocks/antelope/src/auth_sequences.rs

@@ -1,4 +1,5 @@
 use common::structs::BlockTimestamp;
+use substreams_antelope::pb::TransactionTrace;
 use substreams_antelope::Block;
 
 use crate::pb::antelope::AuthSequence as RawAuthSequence;
@@ -35,3 +36,29 @@ pub fn collect_auth_sequences(block: &Block, timestamp: &BlockTimestamp) -> Vec<
 
     auth_sequences
 }
+
+pub fn collect_tx_auth_sequences(transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> Vec<RawAuthSequence> {
+    let mut auth_sequences = Vec::new();
+
+    for action_trace in transaction.action_traces.iter() {
+        let receipt = action_trace.receipt.as_ref().expect("Action trace receipt is required");
+        let action_index = action_trace.execution_index;
+
+        for (index, auth) in receipt.auth_sequence.iter().enumerate() {
+            auth_sequences.push(RawAuthSequence {
+                block_time: Some(timestamp.time.clone()),
+                block_number: timestamp.number,
+                block_hash: timestamp.hash.clone(),
+                block_date: timestamp.date.clone(),
+                tx_hash: transaction.id.clone(),
+                tx_success,
+                action_index,
+                index: index as u32,
+                account_name: auth.account_name.clone(),
+                sequence: auth.sequence,
+            });
+        }
+    }
+
+    auth_sequences
+}

blocks/antelope/src/authority.rs

@@ -3,7 +3,7 @@ use crate::{
     transactions::is_transaction_success,
 };
 use common::structs::BlockTimestamp;
-use substreams_antelope::Block;
+use substreams_antelope::{pb::TransactionTrace, Block};
 
 pub struct AuthorityVectors {
     pub accounts: Vec<RawAccount>,
@@ -81,3 +81,69 @@ pub fn collect_authority_vectors(block: &Block, timestamp: &BlockTimestamp) -> A
 
     AuthorityVectors { accounts, keys, waits }
 }
+
+pub fn collect_tx_authority_vectors(transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> AuthorityVectors {
+    let mut accounts: Vec<RawAccount> = vec![];
+    let mut keys: Vec<RawKey> = vec![];
+    let mut waits: Vec<RawWait> = vec![];
+
+    for perm_op in transaction.perm_ops.iter() {
+        if let Some(new_perm) = &perm_op.new_perm {
+            let authority = new_perm.authority.as_ref().unwrap();
+            let action_index = perm_op.action_index;
+
+            // Process authority accounts
+            for (index, account) in authority.accounts.iter().enumerate() {
+                if let Some(permission) = &account.permission {
+                    accounts.push(RawAccount {
+                        block_time: Some(timestamp.time.clone()),
+                        block_number: timestamp.number,
+                        block_hash: timestamp.hash.clone(),
+                        block_date: timestamp.date.clone(),
+                        tx_hash: transaction.id.clone(),
+                        tx_success,
+                        index: index as u32,
+                        action_index: action_index as u32,
+                        actor: permission.actor.clone(),
+                        permission: permission.permission.clone(),
+                        weight: account.weight,
+                    });
+                }
+            }
+
+            // Process authority keys
+            for (index, key) in authority.keys.iter().enumerate() {
+                keys.push(RawKey {
+                    block_time: Some(timestamp.time.clone()),
+                    block_number: timestamp.number,
+                    block_hash: timestamp.hash.clone(),
+                    block_date: timestamp.date.clone(),
+                    tx_hash: transaction.id.clone(),
+                    tx_success,
+                    index: index as u32,
+                    action_index: action_index as u32,
+                    public_key: key.public_key.clone(),
+                    weight: key.weight,
+                });
+            }
+
+            // Process authority waits
+            for (index, wait) in authority.waits.iter().enumerate() {
+                waits.push(RawWait {
+                    block_time: Some(timestamp.time.clone()),
+                    block_number: timestamp.number,
+                    block_hash: timestamp.hash.clone(),
+                    block_date: timestamp.date.clone(),
+                    tx_hash: transaction.id.clone(),
+                    tx_success,
+                    index: index as u32,
+                    action_index: action_index as u32,
+                    wait_sec: wait.wait_sec,
+                    weight: wait.weight,
+                });
+            }
+        }
+    }
+
+    AuthorityVectors { accounts, keys, waits }
+}

blocks/antelope/src/authorizations.rs

@@ -1,5 +1,5 @@
 use common::structs::BlockTimestamp;
-use substreams_antelope::Block;
+use substreams_antelope::{pb::TransactionTrace, Block};
 
 use crate::{pb::antelope::Authorization as RawAuthorization, transactions::is_transaction_success};
 
@@ -31,3 +31,26 @@ pub fn collect_authorizations(block: &Block, timestamp: &BlockTimestamp) -> Vec<
 
     authorizations
 }
+
+pub fn collect_tx_authorizations(transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> Vec<RawAuthorization> {
+    let mut authorizations = Vec::new();
+
+    for action_trace in transaction.action_traces.iter() {
+        for (index, authorization) in action_trace.action.as_ref().unwrap().authorization.iter().enumerate() {
+            authorizations.push(RawAuthorization {
+                block_time: Some(timestamp.time.clone()),
+                block_number: timestamp.number,
+                block_hash: timestamp.hash.clone(),
+                block_date: timestamp.date.clone(),
+                tx_hash: transaction.id.clone(),
+                tx_success,
+                action_index: action_trace.execution_index,
+                index: index as u32,
+                actor: authorization.actor.clone(),
+                permission: authorization.permission.clone(),
+            });
+        }
+    }
+
+    authorizations
+}

blocks/antelope/src/blocks.rs

@@ -1,5 +1,5 @@
 use crate::{pb::antelope::Block as RawBlock, size::collect_size};
-use common::{structs::BlockTimestamp, utils::bytes_to_hex};
+use common::structs::BlockTimestamp;
 use substreams::Hex;
 use substreams_antelope::Block;
 
@@ -24,7 +24,7 @@ pub fn collect_block(block: &Block, timestamp: &BlockTimestamp) -> RawBlock {
         dpos_irreversible_blocknum: block.dpos_irreversible_blocknum,
         transaction_mroot: Hex::encode(&header.transaction_mroot.to_vec()),
         action_mroot: Hex::encode(&header.action_mroot.to_vec()),
-        blockroot_merkle_active_nodes: blockroot_merkle.active_nodes.iter().map(|row| bytes_to_hex(row)).collect::<Vec<String>>().join(","),
+        blockroot_merkle_active_nodes: blockroot_merkle.active_nodes.iter().map(|row| Hex::encode(row)).collect::<Vec<String>>(),
         blockroot_merkle_node_count: blockroot_merkle.node_count,
         size: size.size,
         total_transactions: size.total_transactions,

blocks/antelope/src/collect_events.rs

@@ -0,0 +1,60 @@
+use common::structs::BlockTimestamp;
+use substreams_antelope::Block;
+
+use crate::{
+    account_ram_deltas::collect_tx_account_ram_deltas,
+    actions::collect_tx_actions,
+    auth_sequences::collect_tx_auth_sequences,
+    authority::collect_tx_authority_vectors,
+    authorizations::collect_tx_authorizations,
+    blocks::collect_block,
+    creation_tree::collect_tx_creation_trees,
+    db_ops::collect_tx_db_ops,
+    feature_ops::collect_tx_feature_ops,
+    pb::antelope::Events as RawEvents,
+    perm_ops::collect_tx_perm_ops,
+    ram_ops::collect_tx_ram_ops,
+    table_ops::collect_tx_table_ops,
+    transactions::{collect_transaction, is_transaction_success},
+};
+
+pub fn collect_events(block: &Block, timestamp: &BlockTimestamp) -> RawEvents {
+    let mut events = RawEvents {
+        blocks: vec![collect_block(block, timestamp)],
+        transactions: vec![],
+        actions: vec![],
+        db_ops: vec![],
+        feature_ops: vec![],
+        perm_ops: vec![],
+        table_ops: vec![],
+        accounts: vec![],
+        keys: vec![],
+        waits: vec![],
+        ram_ops: vec![],
+        authorizations: vec![],
+        auth_sequences: vec![],
+        account_ram_deltas: vec![],
+        creation_trees: vec![],
+    };
+
+    for transaction in block.transaction_traces() {
+        let tx_success = is_transaction_success(transaction.receipt.clone().unwrap_or_default().status);
+        events.transactions.push(collect_transaction(block, transaction, timestamp, tx_success));
+        events.actions.extend(collect_tx_actions(block, transaction, timestamp, tx_success));
+        events.db_ops.extend(collect_tx_db_ops(transaction, timestamp, tx_success));
+        events.feature_ops.extend(collect_tx_feature_ops(transaction, timestamp, tx_success));
+        events.perm_ops.extend(collect_tx_perm_ops(transaction, timestamp, tx_success));
+        events.table_ops.extend(collect_tx_table_ops(transaction, timestamp, tx_success));
+        let authority_vectors = collect_tx_authority_vectors(transaction, timestamp, tx_success);
+        events.accounts.extend(authority_vectors.accounts);
+        events.keys.extend(authority_vectors.keys);
+        events.waits.extend(authority_vectors.waits);
+        events.ram_ops.extend(collect_tx_ram_ops(transaction, timestamp, tx_success));
+        events.authorizations.extend(collect_tx_authorizations(transaction, timestamp, tx_success));
+        events.auth_sequences.extend(collect_tx_auth_sequences(transaction, timestamp, tx_success));
+        events.account_ram_deltas.extend(collect_tx_account_ram_deltas(transaction, timestamp, tx_success));
+        events.creation_trees.extend(collect_tx_creation_trees(transaction, timestamp, tx_success));
+    }
+
+    events
+}

blocks/antelope/src/creation_tree.rs

@@ -1,6 +1,7 @@
 use crate::pb::antelope::CreationTree as RawCreationTree;
 use crate::transactions::is_transaction_success;
 use common::structs::BlockTimestamp;
+use substreams_antelope::pb::TransactionTrace;
 use substreams_antelope::Block;
 
 pub fn collect_creation_trees(block: &Block, timestamp: &BlockTimestamp) -> Vec<RawCreationTree> {
@@ -26,3 +27,22 @@ pub fn collect_creation_trees(block: &Block, timestamp: &BlockTimestamp) -> Vec<
 
     creation_trees
 }
+
+pub fn collect_tx_creation_trees(transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> Vec<RawCreationTree> {
+    let mut creation_trees = Vec::new();
+
+    for creation_flat_node in transaction.creation_tree.iter() {
+        creation_trees.push(RawCreationTree {
+            block_time: Some(timestamp.time.clone()),
+            block_number: timestamp.number,
+            block_hash: timestamp.hash.clone(),
+            block_date: timestamp.date.clone(),
+            tx_hash: transaction.id.clone(),
+            tx_success,
+            creator_action_index: creation_flat_node.creator_action_index,
+            execution_action_index: creation_flat_node.execution_action_index,
+        });
+    }
+
+    creation_trees
+}

blocks/antelope/src/db_ops.rs

@@ -1,6 +1,6 @@
 use common::structs::BlockTimestamp;
 use substreams::Hex;
-use substreams_antelope::Block;
+use substreams_antelope::{pb::TransactionTrace, Block};
 
 use crate::{pb::antelope::DbOp as RawDbOp, transactions::is_transaction_success};
 
@@ -50,3 +50,34 @@ pub fn collect_db_ops(block: &Block, timestamp: &BlockTimestamp) -> Vec<RawDbOp>
 
     db_ops
 }
+
+pub fn collect_tx_db_ops(transaction: &TransactionTrace, timestamp: &BlockTimestamp, tx_success: bool) -> Vec<RawDbOp> {
+    let mut db_ops: Vec<RawDbOp> = Vec::new();
+
+    for (index, db_op) in transaction.db_ops.iter().enumerate() {
+        db_ops.push(RawDbOp {
+            block_time: Some(timestamp.time.clone()),
+            block_number: timestamp.number,
+            block_hash: timestamp.hash.clone(),
+            block_date: timestamp.date.clone(),
+            tx_hash: transaction.id.clone(),
+            tx_success,
+            index: index as u32,
+            operation: operation_to_string(db_op.operation),
+            operation_code: db_op.operation as u32,
+            action_index: db_op.action_index,
+            code: db_op.code.clone(),
+            scope: db_op.scope.clone(),
+            table_name: db_op.table_name.clone(),
+            primary_key: db_op.primary_key.clone(),
+            old_payer: db_op.old_payer.clone(),
+            new_payer: db_op.new_payer.clone(),
+            old_data: Hex::encode(&db_op.old_data),
+            new_data: Hex::encode(&db_op.new_data),
+            old_data_json: db_op.old_data_json.clone(),
+            new_data_json: db_op.new_data_json.clone(),
+        });
+    }
+
+    db_ops
+}