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

move message parsing and change webhook

Browse Source
This commit is contained in:
Mark Sinclair
2025-07-23 17:01:54 +01:00
parent b7c99f802d
commit bfb762128d
7 changed files with 65 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -5739,7 +5739,7 @@ dependencies = [
[[package]]
name = "nym-data-observatory"
version = "0.1.0"
version = "1.0.0"
dependencies = [
"anyhow",
"async-trait",
common/nyxd-scraper-psql/src/storage/block_storage.rs
+5 -15
View File
@@ -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 })
}
common/nyxd-scraper-psql/src/storage/transaction.rs
+7 -20
View File
@@ -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<serde_json::Value> {
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::<Vec<_>>();
let signer_infos = chain_tx
@@ -220,7 +206,8 @@ impl PostgresStorageTransaction {
let mut wasm_message_type: Option<String> = None;
let mut funds: Option<Vec<Coin>> = 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<String> {
debug!("value:\n{value}");
trace!("value:\n{value}");
match value.as_object() {
Some(map) => map.keys().next().cloned(),
None => None,
common/nyxd-scraper-shared/src/block_processor/types.rs
+3
View File
@@ -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<tx::Proof>,
pub parsed_messages: HashMap<usize, serde_json::Value>,
}
#[derive(Debug)]
common/nyxd-scraper-shared/src/rpc_client.rs
+39 -14
View File
@@ -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<HttpClient>,
// 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<serde_json::Value> {
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,
})
}
nym-data-observatory/Cargo.toml
+1 -2
View File
@@ -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 }
nym-data-observatory/src/chain_scraper/webhook.rs
+9 -19
View File
@@ -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<Webhook>,
registry: MessageRegistry,
}
impl WebhookModule {
@@ -27,20 +25,7 @@ impl WebhookModule {
.iter()
.map(|watcher_cfg| Webhook::new(watcher_cfg.clone()))
.collect::<anyhow::Result<Vec<_>>>()?;
Ok(Self {
webhooks,
registry: default_message_registry(),
})
}
fn decode_or_skip(&self, msg: &Any) -> Option<serde_json::Value> {
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 {