GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix scraper

Browse Source
This commit is contained in:
Sachin Kamath
2024-11-27 20:07:24 +05:30
parent c5cb45fdfa
commit dc9a4b14fe
3 changed files with 181 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nyxd-scraper/Cargo.toml
+1
View File
@@ -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}
common/nyxd-scraper/src/storage/manager.rs
+93 -33
View File
@@ -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<cosmrs::Any>,
memo: String,
signatures: Vec<String>,
signer_infos: Vec<SignerInfo>,
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::<Vec<_>>()),
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::<Vec<_>>()),
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<String>,
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<Coin>,
pub gas_limit: i64,
pub payer: String,
pub granter: String,
}
common/nyxd-scraper/src/storage/mod.rs
+87 -43
View File
@@ -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(())
}
fn extract_involved_accounts(_msg: &cosmrs::Any) -> Vec<String> {
// This is a placeholder implementation
// TODO: Implement proper account extraction based on message type
vec![]
}