Merge pull request #115 from bitfinity-network/EPROD-704_genesys_account_data

[Eprod-704] genesys account data

Author: ufoscout

src/ethereum-json-rpc-client/src/lib.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 use anyhow::Context;
 use did::transaction::StorableExecutionResult;
 use ethers_core::types::{
-    Block, BlockNumber, Log, Transaction, TransactionReceipt, H160, H256, U64,
+    Block, BlockNumber, Log, Transaction, TransactionReceipt, H160, H256, U256, U64,
 };
 use itertools::Itertools;
 use jsonrpc_core::{Call, Id, MethodCall, Output, Params, Request, Response, Version};
@@ -25,7 +25,8 @@ const ETH_BLOCK_NUMBER_METHOD: &str = "eth_blockNumber";
 const ETH_GET_TRANSACTION_RECEIPT_METHOD: &str = "eth_getTransactionReceipt";
 const ETH_SEND_RAW_TRANSACTION_METHOD: &str = "eth_sendRawTransaction";
 const ETH_GET_LOGS_METHOD: &str = "eth_getLogs";
-const ETH_GET_TX_EXECUTION_RESULT_BY_HASH_METHOD: &str = "ic_getExeResultByHash";
+const IC_GET_TX_EXECUTION_RESULT_BY_HASH_METHOD: &str = "ic_getExeResultByHash";
+const IC_GET_GENESIS_BALANCES: &str = "ic_getGenesisBalances";
 
 /// A client for interacting with an Ethereum node over JSON-RPC.
 #[derive(Clone)]
@@ -176,7 +177,7 @@ impl<C: Client> EthJsonRcpClient<C> {
     ) -> anyhow::Result<StorableExecutionResult> {
         let transaction = self
             .single_request::<Option<StorableExecutionResult>>(
-                ETH_GET_TX_EXECUTION_RESULT_BY_HASH_METHOD.to_string(),
+                IC_GET_TX_EXECUTION_RESULT_BY_HASH_METHOD.to_string(),
                 make_params_array!(hash),
                 Id::Str(hash.to_string()),
             )
@@ -202,7 +203,7 @@ impl<C: Client> EthJsonRcpClient<C> {
 
         Ok(self
             .batch_request::<Option<StorableExecutionResult>>(
-                ETH_GET_TX_EXECUTION_RESULT_BY_HASH_METHOD.to_string(),
+                IC_GET_TX_EXECUTION_RESULT_BY_HASH_METHOD.to_string(),
                 params,
                 max_batch_size,
             )
@@ -212,6 +213,16 @@ impl<C: Client> EthJsonRcpClient<C> {
             .collect())
     }
 
+    /// Returns the genesis accounts
+    pub async fn get_genesis_balances(&self) -> anyhow::Result<Vec<(H160, U256)>> {
+        self.single_request(
+            IC_GET_GENESIS_BALANCES.to_string(),
+            make_params_array!(),
+            Id::Str(IC_GET_GENESIS_BALANCES.to_string()),
+        )
+        .await
+    }
+
     /// Performs a request.
     pub async fn request(&self, request: Request) -> anyhow::Result<Response> {
         self.client.send_rpc_request(request).await

src/evm-block-extractor/src/block_extractor.rs

@@ -5,7 +5,7 @@ use ethereum_json_rpc_client::EthJsonRcpClient;
 use itertools::Itertools;
 use tokio::time::Duration;
 
-use crate::database::DatabaseClient;
+use crate::database::{AccountBalance, DatabaseClient};
 
 /// Extracts blocks from an EVMC and stores them in a database
 pub struct BlockExtractor {
@@ -32,6 +32,7 @@ impl BlockExtractor {
 
     /// Collects blocks from the EVMC and stores them in the database.
     /// Returns the inclusive range of blocks that were collected.
+    /// This collects also the genesis accounts if needed.
     pub async fn collect_blocks(
         &mut self,
         from_block_inclusive: u64,
@@ -43,6 +44,8 @@ impl BlockExtractor {
             to_block_inclusive
         );
 
+        self.collect_genesis_balances().await?;
+
         let client = self.client.clone();
 
         let request_time_out_secs = self.request_time_out_secs;
@@ -116,4 +119,36 @@ impl BlockExtractor {
 
         Ok((from_block_inclusive, to_block_inclusive))
     }
+
+    /// Collects blocks from the EVMC and stores them in the database.
+    /// Returns the inclusive range of blocks that were collected.
+    /// This collects also the genesis accounts if needed.
+    async fn collect_genesis_balances(&self) -> anyhow::Result<()> {
+        if self.blockchain.get_genesis_balances().await?.is_some() {
+            log::debug!("Genesis balances already present in the DB. Skipping");
+            return Ok(());
+        }
+
+        log::info!("Genesis balances not present in the DB. Collecting them");
+
+        match self.client.get_genesis_balances().await {
+            Ok(genesis_balances) => {
+                let genesis_balances = genesis_balances
+                    .into_iter()
+                    .map(|(address, balance)| AccountBalance {
+                        address: address.into(),
+                        balance: balance.into(),
+                    })
+                    .collect::<Vec<_>>();
+                self.blockchain
+                    .insert_genesis_balances(&genesis_balances)
+                    .await?;
+            }
+            Err(e) => {
+                log::error!("Error getting genesis balances: {:?}. The process will not be stopped but there will be missing genesis balances in the DB", e);
+            }
+        }
+
+        Ok(())
+    }
 }

src/evm-block-extractor/src/database/big_query_db_client.rs

@@ -18,11 +18,14 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use serde_json::Value;
 
-use super::DatabaseClient;
+use super::{AccountBalance, DatabaseClient};
 
 const BQ_EXE_RESULTS_TABLE_ID: &str = "exe_results";
 const BQ_BLOCKS_TABLE_ID: &str = "blocks";
 const BQ_TRANSACTIONS_TABLE_ID: &str = "transactions";
+const BQ_KEY_VALUE_TABLE_ID: &str = "key_value_data";
+
+const GENESIS_BALANCES_KEY: &str = "genesis_balances";
 
 #[derive(Clone)]
 /// A client for BigQuery that can be used to query and insert data
@@ -67,7 +70,16 @@ impl BigQueryDbClient {
         })
     }
 
-    async fn execute_query<T: DeserializeOwned>(&self, query: QueryRequest) -> anyhow::Result<T> {
+    async fn query_one<T: DeserializeOwned>(&self, query: QueryRequest) -> anyhow::Result<T> {
+        self.query_one_optional(query)
+            .await?
+            .ok_or(anyhow::anyhow!("No data found for the query"))
+    }
+
+    async fn query_one_optional<T: DeserializeOwned>(
+        &self,
+        query: QueryRequest,
+    ) -> anyhow::Result<Option<T>> {
         let mut response = self.client.job().query(&self.project_id, query).await?;
 
         if response.next_row() {
@@ -79,12 +91,11 @@ impl BigQueryDbClient {
 
             let result: T = serde_json::from_str(&result_str)?;
 
-            Ok(result)
+            Ok(Some(result))
         } else {
-            Err(anyhow::anyhow!("No data found for the query"))
+            Ok(None)
         }
     }
-
     async fn insert_batch_data(
         &self,
         table_id: &str,
@@ -110,6 +121,8 @@ impl BigQueryDbClient {
     async fn create_tables_if_not_present(&self) -> anyhow::Result<()> {
         let dataset = Dataset::new(&self.project_id, &self.dataset_id);
 
+        log::info!("Creating tables if not present");
+
         // Make sure the dataset exists
         if self
             .client
@@ -145,6 +158,13 @@ impl BigQueryDbClient {
                     TableFieldSchema::integer("block_number"),
                 ],
             ),
+            (
+                BQ_KEY_VALUE_TABLE_ID,
+                vec![
+                    TableFieldSchema::string("key"),
+                    TableFieldSchema::json("data"),
+                ],
+            ),
         ];
 
         // Check each table and create if it does not exist
@@ -173,6 +193,51 @@ impl BigQueryDbClient {
 
         Ok(())
     }
+
+    async fn fetch_key_value_data<D: DeserializeOwned>(
+        &self,
+        key: &str,
+    ) -> anyhow::Result<Option<D>> {
+        let query_request = QueryRequest {
+            query_parameters: Some(vec![QueryParameter {
+                name: Some("key".to_string()),
+                parameter_type: Some(QueryParameterType {
+                    r#type: "STRING".to_string(),
+                    ..Default::default()
+                }),
+                parameter_value: Some(QueryParameterValue {
+                    value: Some(key.to_string()),
+                    ..Default::default()
+                }),
+            }]),
+            query: format!(
+                "SELECT data FROM `{project_id}.{dataset_id}.{table_id}` WHERE key = @key",
+                project_id = self.project_id,
+                dataset_id = self.dataset_id,
+                table_id = BQ_KEY_VALUE_TABLE_ID,
+            ),
+            ..Default::default()
+        };
+
+        self.query_one_optional(query_request).await
+    }
+
+    async fn insert_key_value_data<D: Serialize>(&self, key: &str, data: D) -> anyhow::Result<()> {
+        let json = KeyValueDataRow {
+            key: key.to_string(),
+            data: serde_json::to_value(data).expect("Failed to serialize data"),
+        };
+
+        let key_value_row = TableDataInsertAllRequestRows {
+            insert_id: Some(key.to_string()),
+            json: serde_json::to_value(json)?,
+        };
+
+        log::debug!("Inserting key value data with key [{}]", key);
+
+        self.insert_batch_data(BQ_KEY_VALUE_TABLE_ID, vec![key_value_row])
+            .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -185,7 +250,6 @@ impl DatabaseClient for BigQueryDbClient {
                 if !self.check_if_same_block_hash(&block).await? {
                     if reset_database {
                         self.clear().await?;
-                        self.create_tables_if_not_present().await?;
                     } else {
                         return Err(anyhow::anyhow!(
                             "The block hash in the database is different from the one in the block"
@@ -210,8 +274,9 @@ impl DatabaseClient for BigQueryDbClient {
         delete_table(BQ_BLOCKS_TABLE_ID.to_owned()).await?;
         delete_table(BQ_EXE_RESULTS_TABLE_ID.to_owned()).await?;
         delete_table(BQ_TRANSACTIONS_TABLE_ID.to_owned()).await?;
+        delete_table(BQ_KEY_VALUE_TABLE_ID.to_owned()).await?;
 
-        Ok(())
+        self.create_tables_if_not_present().await
     }
 
     async fn get_block_by_number(&self, block_number: u64) -> anyhow::Result<Block<H256>> {
@@ -236,7 +301,7 @@ impl DatabaseClient for BigQueryDbClient {
             ..Default::default()
         };
 
-        self.execute_query(query_request).await
+        self.query_one(query_request).await
     }
 
     async fn get_full_block_by_number(
@@ -402,7 +467,7 @@ impl DatabaseClient for BigQueryDbClient {
             ..Default::default()
         };
 
-        let exe_result: StorableExecutionResult = self.execute_query(query_request).await?;
+        let exe_result: StorableExecutionResult = self.query_one(query_request).await?;
 
         Ok(TransactionReceipt::from(exe_result))
     }
@@ -458,6 +523,18 @@ impl DatabaseClient for BigQueryDbClient {
             ))
         }
     }
+
+    async fn get_genesis_balances(&self) -> anyhow::Result<Option<Vec<AccountBalance>>> {
+        self.fetch_key_value_data(GENESIS_BALANCES_KEY).await
+    }
+
+    async fn insert_genesis_balances(
+        &self,
+        genesis_balances: &[AccountBalance],
+    ) -> anyhow::Result<()> {
+        self.insert_key_value_data(GENESIS_BALANCES_KEY, genesis_balances)
+            .await
+    }
 }
 
 /// A row in the BigQuery table
@@ -482,3 +559,11 @@ pub struct TransactionRow {
     transaction: Value,
     block_number: u64,
 }
+
+/// A row in the BigQuery table
+#[derive(Debug, Serialize)]
+pub struct KeyValueDataRow {
+    key: String,
+    #[serde(serialize_with = "serialize_json_as_string")]
+    data: Value,
+}

src/evm-block-extractor/src/database/mod.rs

@@ -2,7 +2,15 @@ pub mod big_query_db_client;
 pub mod postgres_db_client;
 
 use did::transaction::StorableExecutionResult;
-use did::{Block, Transaction, TransactionReceipt, H256};
+use did::{Block, Transaction, TransactionReceipt, H160, H256, U256};
+use serde::{Deserialize, Serialize};
+
+/// Account balance
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct AccountBalance {
+    pub address: H160,
+    pub balance: U256,
+}
 
 /// A trait for interacting with a blockchain database
 #[async_trait::async_trait]
@@ -37,6 +45,15 @@ pub trait DatabaseClient: Send + Sync {
         transactions: &[Transaction],
     ) -> anyhow::Result<()>;
 
+    /// Get genesis balances
+    async fn get_genesis_balances(&self) -> anyhow::Result<Option<Vec<AccountBalance>>>;
+
+    /// Insert genesis balances
+    async fn insert_genesis_balances(
+        &self,
+        genesis_balances: &[AccountBalance],
+    ) -> anyhow::Result<()>;
+
     /// Get a transaction receipt from the database
     async fn get_transaction_receipt(&self, tx_hash: H256) -> anyhow::Result<TransactionReceipt>;