From bfb762128d47e2640444077a17c6ee7508cc62ae Mon Sep 17 00:00:00 2001 From: Mark Sinclair Date: Wed, 23 Jul 2025 17:01:54 +0100 Subject: [PATCH] move message parsing and change webhook --- Cargo.lock | 2 +- .../src/storage/block_storage.rs | 20 ++----- .../src/storage/transaction.rs | 27 +++------- .../src/block_processor/types.rs | 3 ++ common/nyxd-scraper-shared/src/rpc_client.rs | 53 ++++++++++++++----- nym-data-observatory/Cargo.toml | 3 +- .../src/chain_scraper/webhook.rs | 28 ++++------ 7 files changed, 65 insertions(+), 71 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3985bd4ea6..078a1a2f9f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5739,7 +5739,7 @@ dependencies = [ [[package]] name = "nym-data-observatory" -version = "0.1.0" +version = "1.0.0" dependencies = [ "anyhow", "async-trait", diff --git a/common/nyxd-scraper-psql/src/storage/block_storage.rs b/common/nyxd-scraper-psql/src/storage/block_storage.rs index b16f6c5891..e87e72b1ef 100644 --- a/common/nyxd-scraper-psql/src/storage/block_storage.rs +++ b/common/nyxd-scraper-psql/src/storage/block_storage.rs @@ -11,17 +11,13 @@ use crate::storage::transaction::PostgresStorageTransaction; use async_trait::async_trait; use nyxd_scraper_shared::storage::helpers::log_db_operation_time; use nyxd_scraper_shared::storage::{NyxdScraperStorage, NyxdScraperStorageError}; -use nyxd_scraper_shared::{default_message_registry, MessageRegistry}; use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime}; use tokio::time::Instant; -use tracing::{debug, error, info, instrument}; +use tracing::{debug, error, info, instrument, warn}; #[derive(Clone)] pub struct PostgresScraperStorage { pub(crate) manager: StorageManager, - - // kinda like very limited cosmos sdk codec - pub(crate) message_registry: MessageRegistry, } impl PostgresScraperStorage { @@ -41,8 +37,8 @@ impl PostgresScraperStorage { .run(&connection_pool) .await { - error!("Failed to initialize SQLx database: {err}"); - return Err(err.into()); + warn!("Failed to initialize SQLx database: {err}"); + // return Err(err.into()); } info!("Database migration finished!"); @@ -50,10 +46,7 @@ impl PostgresScraperStorage { let manager = StorageManager { connection_pool }; manager.set_initial_metadata().await?; - let storage = PostgresScraperStorage { - manager, - message_registry: default_message_registry(), - }; + let storage = PostgresScraperStorage { manager }; Ok(storage) } @@ -94,10 +87,7 @@ impl PostgresScraperStorage { .connection_pool .begin() .await - .map(|inner| PostgresStorageTransaction { - inner, - registry: self.message_registry.clone(), - }) + .map(|inner| PostgresStorageTransaction { inner }) .map_err(|source| PostgresScraperError::StorageTxBeginFailure { source }) } diff --git a/common/nyxd-scraper-psql/src/storage/transaction.rs b/common/nyxd-scraper-psql/src/storage/transaction.rs index af118f300a..a499a19885 100644 --- a/common/nyxd-scraper-psql/src/storage/transaction.rs +++ b/common/nyxd-scraper-psql/src/storage/transaction.rs @@ -21,7 +21,7 @@ use nyxd_scraper_shared::storage::validators::Response; use nyxd_scraper_shared::storage::{ validators, Block, Commit, CommitSig, NyxdScraperStorageError, NyxdScraperTransaction, }; -use nyxd_scraper_shared::{Any, MessageRegistry, ParsedTransactionResponse}; +use nyxd_scraper_shared::ParsedTransactionResponse; use serde_json::{json, Value}; use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime}; use sqlx::{Postgres, Transaction}; @@ -30,8 +30,6 @@ use tracing::{debug, trace, warn}; pub struct PostgresStorageTransaction { pub(super) inner: Transaction<'static, Postgres>, - - pub(super) registry: MessageRegistry, } impl Deref for PostgresStorageTransaction { @@ -48,16 +46,6 @@ impl DerefMut for PostgresStorageTransaction { } impl PostgresStorageTransaction { - fn decode_or_skip(&self, msg: &Any) -> Option { - match self.registry.try_decode(msg) { - Ok(decoded) => Some(decoded), - Err(err) => { - warn!("{err}"); - None - } - } - } - async fn persist_validators( &mut self, validators: &validators::Response, @@ -169,11 +157,9 @@ impl PostgresStorageTransaction { .collect(); let messages = chain_tx - .tx - .body - .messages - .iter() - .filter_map(|msg| self.decode_or_skip(msg)) + .parsed_messages + .values() + .map(|v| v.clone()) .collect::>(); let signer_infos = chain_tx @@ -220,7 +206,8 @@ impl PostgresStorageTransaction { let mut wasm_message_type: Option = None; let mut funds: Option> = None; - let value = serde_json::to_value(self.decode_or_skip(msg))?; + let parsed_message = chain_tx.parsed_messages.get(&index); + let value = serde_json::to_value(parsed_message)?; if msg.type_url == "/cosmwasm.wasm.v1.MsgExecuteContract" { if let Ok(wasm_execute) = MsgExecuteContract::decode(msg.value.as_ref()) { @@ -270,7 +257,7 @@ impl PostgresStorageTransaction { } fn get_first_field_name(value: &Value) -> Option { - debug!("value:\n{value}"); + trace!("value:\n{value}"); match value.as_object() { Some(map) => map.keys().next().cloned(), None => None, diff --git a/common/nyxd-scraper-shared/src/block_processor/types.rs b/common/nyxd-scraper-shared/src/block_processor/types.rs index 1c456b9318..cb17bb698b 100644 --- a/common/nyxd-scraper-shared/src/block_processor/types.rs +++ b/common/nyxd-scraper-shared/src/block_processor/types.rs @@ -3,6 +3,7 @@ use crate::error::ScraperError; use crate::helpers; +use std::collections::HashMap; use tendermint::{Block, Hash, abci, block, tx}; use tendermint_rpc::endpoint::{block as block_endpoint, block_results, validators}; use tendermint_rpc::event::{Event, EventData}; @@ -26,6 +27,8 @@ pub struct ParsedTransactionResponse { pub tx: cosmrs::tx::Tx, pub proof: Option, + + pub parsed_messages: HashMap, } #[derive(Debug)] diff --git a/common/nyxd-scraper-shared/src/rpc_client.rs b/common/nyxd-scraper-shared/src/rpc_client.rs index 5a3621abd4..879b15bb6c 100644 --- a/common/nyxd-scraper-shared/src/rpc_client.rs +++ b/common/nyxd-scraper-shared/src/rpc_client.rs @@ -6,15 +6,16 @@ use crate::block_processor::types::{ }; use crate::error::ScraperError; use crate::helpers::tx_hash; +use crate::{default_message_registry, Any, MessageRegistry}; use futures::StreamExt; use futures::future::join3; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; use tendermint::Hash; use tendermint_rpc::endpoint::{block, block_results, tx, validators}; use tendermint_rpc::{Client, HttpClient, Paging}; use tokio::sync::Mutex; -use tracing::{debug, instrument}; +use tracing::{debug, instrument, warn}; use url::Url; #[derive(Clone)] @@ -22,6 +23,9 @@ pub struct RpcClient { // right now I don't care about anything nym specific, so a simple http client is sufficient, // once this is inadequate, we can switch to a NyxdClient inner: Arc, + + // kinda like very limited cosmos sdk codec + pub(crate) message_registry: MessageRegistry, } impl RpcClient { @@ -35,9 +39,20 @@ impl RpcClient { Ok(RpcClient { inner: Arc::new(http_client), + message_registry: default_message_registry(), }) } + fn decode_or_skip(&self, msg: &Any) -> Option { + match self.message_registry.try_decode(msg) { + Ok(decoded) => Some(decoded), + Err(err) => { + warn!("Failed to decode raw message: {err}"); + None + } + } + } + #[instrument(skip(self, block), fields(height = block.height))] pub async fn try_get_full_details( &self, @@ -56,19 +71,29 @@ impl RpcClient { let raw_transactions = raw_transactions?; let mut transactions = Vec::with_capacity(raw_transactions.len()); - for tx in raw_transactions { + for raw_tx in raw_transactions { + let mut parsed_messages = HashMap::new(); + let tx = cosmrs::Tx::from_bytes(&raw_tx.tx).map_err(|source| { + ScraperError::TxParseFailure { + hash: raw_tx.hash, + source, + } + })?; + + for (index, msg) in tx.body.messages.iter().enumerate() { + if let Some(value) = self.decode_or_skip(msg) { + parsed_messages.insert(index, value); + } + } + transactions.push(ParsedTransactionResponse { - hash: tx.hash, - height: tx.height, - index: tx.index, - tx_result: tx.tx_result, - tx: cosmrs::Tx::from_bytes(&tx.tx).map_err(|source| { - ScraperError::TxParseFailure { - hash: tx.hash, - source, - } - })?, - proof: tx.proof, + hash: raw_tx.hash, + height: raw_tx.height, + index: raw_tx.index, + tx_result: raw_tx.tx_result, + tx, + proof: raw_tx.proof, + parsed_messages, }) } diff --git a/nym-data-observatory/Cargo.toml b/nym-data-observatory/Cargo.toml index e215a48daa..75d07705fe 100644 --- a/nym-data-observatory/Cargo.toml +++ b/nym-data-observatory/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "nym-data-observatory" -version = "0.1.0" +version = "1.0.0" authors.workspace = true repository.workspace = true homepage.workspace = true @@ -11,7 +11,6 @@ documentation.workspace = true edition.workspace = true license.workspace = true rust-version.workspace = true -readme.workspace = true [dependencies] anyhow = { workspace = true } diff --git a/nym-data-observatory/src/chain_scraper/webhook.rs b/nym-data-observatory/src/chain_scraper/webhook.rs index 221d99e008..50a06596e8 100644 --- a/nym-data-observatory/src/chain_scraper/webhook.rs +++ b/nym-data-observatory/src/chain_scraper/webhook.rs @@ -9,14 +9,12 @@ use nym_validator_client::nyxd::{Any, Msg, MsgSend, Name}; use nyxd_scraper_psql::{ MsgModule, NyxdScraperTransaction, ParsedTransactionResponse, ScraperError, }; -use nyxd_scraper_shared::{default_message_registry, MessageRegistry}; use reqwest::{Client, Url}; -use tracing::{error, info, warn}; +use tracing::{error, info}; use utoipa::gen::serde_json; pub struct WebhookModule { webhooks: Vec, - registry: MessageRegistry, } impl WebhookModule { @@ -27,20 +25,7 @@ impl WebhookModule { .iter() .map(|watcher_cfg| Webhook::new(watcher_cfg.clone())) .collect::>>()?; - Ok(Self { - webhooks, - registry: default_message_registry(), - }) - } - - fn decode_or_skip(&self, msg: &Any) -> Option { - match self.registry.try_decode(msg) { - Ok(decoded) => Some(decoded), - Err(err) => { - warn!("webhook processing failed {err}"); - None - } - } + Ok(Self { webhooks }) } } @@ -53,11 +38,11 @@ impl MsgModule for WebhookModule { async fn handle_msg( &mut self, index: usize, - msg: &Any, + _msg: &Any, tx: &ParsedTransactionResponse, _storage_tx: &mut dyn NyxdScraperTransaction, ) -> Result<(), ScraperError> { - let message = serde_json::to_value(self.decode_or_skip(msg)).ok(); + let message = serde_json::to_value(tx.parsed_messages.get(&index)).ok(); let payload = WebhookPayload { height: tx.height.value(), @@ -66,6 +51,11 @@ impl MsgModule for WebhookModule { message, }; + println!( + "->>>>>>>>>>>>>>>>>>>>>>>>> {}", + serde_json::to_string(&payload).unwrap() + ); + for webhook in self.webhooks.clone() { let payload = payload.clone(); tokio::spawn(async move {