From dc9a4b14fe14ef606526de9f46eff9e2425f5da7 Mon Sep 17 00:00:00 2001 From: Sachin Kamath Date: Wed, 27 Nov 2024 20:07:24 +0530 Subject: [PATCH] fix scraper --- common/nyxd-scraper/Cargo.toml | 1 + common/nyxd-scraper/src/storage/manager.rs | 126 ++++++++++++++------ common/nyxd-scraper/src/storage/mod.rs | 130 ++++++++++++++------- 3 files changed, 181 insertions(+), 76 deletions(-) diff --git a/common/nyxd-scraper/Cargo.toml b/common/nyxd-scraper/Cargo.toml index 32125f841f..8266c23262 100644 --- a/common/nyxd-scraper/Cargo.toml +++ b/common/nyxd-scraper/Cargo.toml @@ -12,6 +12,7 @@ license.workspace = true [dependencies] async-trait.workspace = true +base64.workspace = true const_format = { workspace = true } cosmrs.workspace = true chrono = {workspace = true} diff --git a/common/nyxd-scraper/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs index 2e60b1868f..1fb9ae4d06 100644 --- a/common/nyxd-scraper/src/storage/manager.rs +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -5,6 +5,8 @@ use std::vec; use crate::storage::log_db_operation_time; use crate::storage::models::{CommitSignature, Validator}; +use base64; +use serde_json::Value as JsonValue; use sqlx::types::time::OffsetDateTime; use sqlx::{Executor, Postgres}; use tokio::time::Instant; @@ -348,9 +350,12 @@ pub(crate) async fn insert_transaction<'a, E>( height: i64, index: i64, success: bool, - message_len: i64, + num_messages: i64, messages: Vec, memo: String, + signatures: Vec, + signer_infos: Vec, + fee: Fee, gas_wanted: i64, gas_used: i64, raw_log: String, @@ -364,32 +369,57 @@ where sqlx::query!( r#" - INSERT INTO "transaction" (hash, height, "index", success, num_messages, messages, memo, gas_wanted, gas_used, raw_log) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) - ON CONFLICT (hash) DO UPDATE - SET height = excluded.height, - "index" = excluded."index", - success = excluded.success, - num_messages = excluded.num_messages, - messages = excluded.messages - memo = excluded.memo, - gas_wanted = excluded.gas_wanted, - gas_used = excluded.gas_used, - raw_log = excluded.raw_log + INSERT INTO transaction (hash, height, "index", success, num_messages, messages, memo, signatures, signer_infos, fee, gas_wanted, gas_used, raw_log) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) + ON CONFLICT (hash) DO UPDATE + SET height = excluded.height, + "index" = excluded."index", + success = excluded.success, + num_messages = excluded.num_messages, + messages = excluded.messages, + memo = excluded.memo, + signatures = excluded.signatures, + signer_infos = excluded.signer_infos, + fee = excluded.fee, + gas_wanted = excluded.gas_wanted, + gas_used = excluded.gas_used, + raw_log = excluded.raw_log "#, - hash, - height, - index as i32, - success, - messages as i32, - messages, - memo, - gas_wanted, - gas_used, - raw_log, + hash, + height, + index as i32, + success, + num_messages as i32, + serde_json::json!(messages.iter().map(|msg| { + serde_json::json!({ + "type_url": msg.type_url, + "value": base64::encode(&msg.value) + }) + }).collect::>()), + memo, + &signatures, + serde_json::json!(signer_infos.iter().map(|info| { + serde_json::json!({ + "public_key": { + "type_url": info.public_key.type_url, + "value": base64::encode(&info.public_key.value) + }, + "mode_info": info.mode_info, + "sequence": info.sequence + }) + }).collect::>()), + serde_json::json!({ + "amount": fee.amount, + "gas_limit": fee.gas_limit, + "payer": fee.payer, + "granter": fee.granter + }), + gas_wanted, + gas_used, + raw_log ) - .execute(executor) - .await?; + .execute(executor) + .await?; log_db_operation_time("insert_transaction", start); Ok(()) @@ -400,6 +430,8 @@ pub(crate) async fn insert_message<'a, E>( transaction_hash: String, index: i64, typ: String, + value: cosmrs::Any, + involved_accounts: Vec, height: i64, executor: E, ) -> Result<(), sqlx::Error> @@ -411,16 +443,23 @@ where sqlx::query!( r#" - INSERT INTO message (transaction_hash, "index", type, height) - VALUES ($1, $2, $3, $4) + INSERT INTO message (transaction_hash, "index", type, value, involved_accounts_addresses, height) + VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (transaction_hash, "index") DO UPDATE SET height = excluded.height, - type = excluded.type + type = excluded.type, + value = excluded.value, + involved_accounts_addresses = excluded.involved_accounts_addresses "#, - transaction_hash, - index, - typ, - height + transaction_hash, + index, + typ, + serde_json::json!({ + "type_url": value.type_url, + "value": base64::encode(&value.value) + }), + &involved_accounts, + height ) .execute(executor) .await?; @@ -511,7 +550,7 @@ where let start = Instant::now(); sqlx::query!( - "DELETE FROM \"transaction\" WHERE height < $1", + "DELETE FROM transaction WHERE height < $1", oldest_to_keep ) .execute(executor) @@ -538,3 +577,24 @@ where Ok(()) } + +#[derive(Debug, Clone, serde::Serialize)] +pub struct Coin { + pub denom: String, + pub amount: String, +} + +#[derive(Debug, Clone)] +pub struct SignerInfo { + pub public_key: cosmrs::Any, + pub mode_info: String, + pub sequence: i64, +} + +#[derive(Debug, Clone)] +pub struct Fee { + pub amount: Vec, + pub gas_limit: i64, + pub payer: String, + pub granter: String, +} diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index 6fc3cf7b24..8358109a8b 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -4,9 +4,9 @@ use crate::block_processor::types::{FullBlockInformation, ParsedTransactionResponse}; use crate::error::ScraperError; use crate::storage::manager::{ - insert_block, insert_message, insert_precommit, insert_transaction, insert_validator, - prune_blocks, prune_messages, prune_pre_commits, prune_transactions, update_last_processed, - update_last_pruned, StorageManager, + insert_block, insert_message, insert_precommit, insert_transaction, + insert_validator, prune_blocks, prune_messages, prune_pre_commits, prune_transactions, + update_last_processed, update_last_pruned, StorageManager, Coin, Fee, SignerInfo, }; use crate::storage::models::{CommitSignature, Validator}; use sqlx::types::time::OffsetDateTime; @@ -227,10 +227,9 @@ pub async fn persist_block( } // persist txs - persist_txs(&block.transactions, tx).await?; - - // persist messages (inside the transactions) - persist_messages(&block.transactions, tx).await?; + for chain_tx in &block.transactions { + persist_transaction(chain_tx, tx).await?; + } update_last_processed(block.block.header.height.into(), tx).await?; @@ -326,24 +325,86 @@ async fn persist_commits( Ok(()) } -async fn persist_txs( - txs: &[ParsedTransactionResponse], - tx: &mut StorageTransaction, +async fn persist_transaction( + chain_tx: &ParsedTransactionResponse, + tx: &mut Transaction<'_, Postgres>, ) -> Result<(), ScraperError> { - debug!("persisting {} txs", txs.len()); + let signer_infos = chain_tx + .tx + .auth_info + .signer_infos + .iter() + .map(|info| SignerInfo { + public_key: info.public_key.clone() + .map(|pk| pk.into()) + .unwrap_or_default(), + mode_info: serde_json::to_string(&serde_json::json!({ + "single": { + "mode": match &info.mode_info { + cosmrs::tx::ModeInfo::Single(s) => s.mode as u32, + _ => 0, + } + } + })).unwrap_or_default(), + sequence: info.sequence as i64, + }) + .collect(); - for chain_tx in txs { - insert_transaction( + let fee = Fee { + amount: chain_tx + .tx + .auth_info + .fee + .amount + .iter() + .map(|coin| Coin { + denom: coin.denom.to_string(), + amount: coin.amount.to_string(), + }) + .collect(), + gas_limit: chain_tx.tx.auth_info.fee.gas_limit as i64, + payer: chain_tx.tx.auth_info.fee.payer + .clone() + .map(|id| id.to_string()) + .unwrap_or_default(), + granter: chain_tx.tx.auth_info.fee.granter + .clone() + .map(|id| id.to_string()) + .unwrap_or_default(), + }; + + let signatures = chain_tx.tx.signatures + .iter() + .map(|sig| base64::encode(sig)) + .collect(); + + insert_transaction( + chain_tx.hash.to_string(), + chain_tx.height.into(), + chain_tx.index as i64, + chain_tx.tx_result.code.is_ok(), + chain_tx.tx.body.messages.len() as i64, + chain_tx.tx.body.messages.clone(), + chain_tx.tx.body.memo.clone(), + signatures, + signer_infos, + fee, + chain_tx.tx_result.gas_wanted, + chain_tx.tx_result.gas_used, + chain_tx.tx_result.log.clone(), + &mut *tx, + ) + .await?; + + for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() { + let involved_accounts = extract_involved_accounts(msg); + insert_message( chain_tx.hash.to_string(), + index as i64, + msg.type_url.clone(), + msg.clone(), + involved_accounts, chain_tx.height.into(), - chain_tx.index as i64, - chain_tx.tx_result.code.is_ok(), - chain_tx.tx.body.messages.len() as i64, - chain_tx.tx.body.messages.clone(), - chain_tx.tx.body.memo.clone(), - chain_tx.tx_result.gas_wanted, - chain_tx.tx_result.gas_used, - chain_tx.tx_result.log.clone(), &mut *tx, ) .await?; @@ -352,25 +413,8 @@ async fn persist_txs( Ok(()) } -async fn persist_messages( - txs: &[ParsedTransactionResponse], - tx: &mut StorageTransaction, -) -> Result<(), ScraperError> { - debug!("persisting messages"); - - for chain_tx in txs { - for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() { - insert_message( - chain_tx.hash.to_string(), - index as i64, - msg.type_url.clone(), - chain_tx.height.into(), - &mut *tx, - ) - .await? - } - } - - Ok(()) -} - \ No newline at end of file +fn extract_involved_accounts(_msg: &cosmrs::Any) -> Vec { + // This is a placeholder implementation + // TODO: Implement proper account extraction based on message type + vec![] +} \ No newline at end of file