diff --git a/common/nyxd-scraper-psql/src/lib.rs b/common/nyxd-scraper-psql/src/lib.rs index b6f3bcec3a..85fcbb46ef 100644 --- a/common/nyxd-scraper-psql/src/lib.rs +++ b/common/nyxd-scraper-psql/src/lib.rs @@ -13,7 +13,6 @@ pub use nyxd_scraper_shared::{ pub use storage::models; pub mod error; -mod proto_registry; pub mod storage; pub type PostgresNyxdScraper = NyxdScraper; diff --git a/common/nyxd-scraper-psql/src/proto_registry.rs b/common/nyxd-scraper-psql/src/proto_registry.rs deleted file mode 100644 index 58b4768c4c..0000000000 --- a/common/nyxd-scraper-psql/src/proto_registry.rs +++ /dev/null @@ -1,6 +0,0 @@ -// Copyright 2025 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use std::collections::HashMap; - -pub struct ProtoRegistry(HashMap<(), ()>); diff --git a/common/nyxd-scraper-psql/src/storage/block_storage.rs b/common/nyxd-scraper-psql/src/storage/block_storage.rs index 43f23199c4..b16f6c5891 100644 --- a/common/nyxd-scraper-psql/src/storage/block_storage.rs +++ b/common/nyxd-scraper-psql/src/storage/block_storage.rs @@ -11,6 +11,7 @@ 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}; @@ -18,6 +19,9 @@ use tracing::{debug, error, info, instrument}; #[derive(Clone)] pub struct PostgresScraperStorage { pub(crate) manager: StorageManager, + + // kinda like very limited cosmos sdk codec + pub(crate) message_registry: MessageRegistry, } impl PostgresScraperStorage { @@ -46,7 +50,10 @@ impl PostgresScraperStorage { let manager = StorageManager { connection_pool }; manager.set_initial_metadata().await?; - let storage = PostgresScraperStorage { manager }; + let storage = PostgresScraperStorage { + manager, + message_registry: default_message_registry(), + }; Ok(storage) } @@ -68,7 +75,8 @@ impl PostgresScraperStorage { update_last_pruned(current_height.into(), &mut **tx).await?; let commit_start = Instant::now(); - tx.0.commit() + tx.inner + .commit() .await .map_err(|source| PostgresScraperError::StorageTxCommitFailure { source })?; log_db_operation_time("committing pruning tx", commit_start); @@ -86,7 +94,10 @@ impl PostgresScraperStorage { .connection_pool .begin() .await - .map(PostgresStorageTransaction) + .map(|inner| PostgresStorageTransaction { + inner, + registry: self.message_registry.clone(), + }) .map_err(|source| PostgresScraperError::StorageTxBeginFailure { source }) } diff --git a/common/nyxd-scraper-psql/src/storage/helpers.rs b/common/nyxd-scraper-psql/src/storage/helpers.rs index 79adc68a8a..5561e57cb9 100644 --- a/common/nyxd-scraper-psql/src/storage/helpers.rs +++ b/common/nyxd-scraper-psql/src/storage/helpers.rs @@ -1,35 +1,8 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -// use crate::error::PostgresScraperError; -// use nyxd_scraper_shared::Any; -// use sqlx::types::JsonValue; -// -// pub(crate) fn proto_to_json(proto: &Any) -> Result { -// todo!() -// } - -use nyxd_scraper_shared::Any; use serde::Serialize; -#[derive(Serialize)] -pub(crate) struct PlaceholderMessage { - #[serde(rename = "@type")] - pub(crate) typ: String, - - pub(crate) placeholder: String, -} - -impl<'a> From<&'a Any> for PlaceholderMessage { - fn from(value: &'a Any) -> Self { - PlaceholderMessage { - typ: value.type_url.to_ascii_lowercase(), - placeholder: "PLACEHOLDER CONTENT - TODO: IMPLEMENT PROPER PROTO -> JSON PARSING" - .to_string(), - } - } -} - #[derive(Serialize)] pub(crate) struct PlaceholderStruct { pub(crate) typ: String, diff --git a/common/nyxd-scraper-psql/src/storage/transaction.rs b/common/nyxd-scraper-psql/src/storage/transaction.rs index 135e8321fb..f7774703d4 100644 --- a/common/nyxd-scraper-psql/src/storage/transaction.rs +++ b/common/nyxd-scraper-psql/src/storage/transaction.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::error::PostgresScraperError; -use crate::storage::helpers::{PlaceholderMessage, PlaceholderStruct}; +use crate::storage::helpers::PlaceholderStruct; use crate::storage::manager::{ insert_block, insert_message, insert_precommit, insert_transaction, insert_validator, }; @@ -16,29 +16,43 @@ use nyxd_scraper_shared::storage::validators::Response; use nyxd_scraper_shared::storage::{ validators, Block, Commit, CommitSig, NyxdScraperStorageError, NyxdScraperTransaction, }; -use nyxd_scraper_shared::ParsedTransactionResponse; +use nyxd_scraper_shared::{Any, MessageRegistry, ParsedTransactionResponse}; use serde_json::json; use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime}; use sqlx::{Postgres, Transaction}; use std::ops::{Deref, DerefMut}; use tracing::{debug, trace, warn}; -pub struct PostgresStorageTransaction(pub(crate) Transaction<'static, Postgres>); +pub struct PostgresStorageTransaction { + pub(super) inner: Transaction<'static, Postgres>, + + pub(super) registry: MessageRegistry, +} impl Deref for PostgresStorageTransaction { type Target = Transaction<'static, Postgres>; fn deref(&self) -> &Self::Target { - &self.0 + &self.inner } } impl DerefMut for PostgresStorageTransaction { fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.0 + &mut self.inner } } 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, @@ -51,7 +65,7 @@ impl PostgresStorageTransaction { insert_validator( consensus_address.to_string(), consensus_pubkey.to_string(), - self.0.as_mut(), + self.inner.as_mut(), ) .await?; } @@ -77,7 +91,7 @@ impl PostgresStorageTransaction { total_gas, proposer_address, time, - self.0.as_mut(), + self.inner.as_mut(), ) .await?; Ok(()) @@ -126,7 +140,7 @@ impl PostgresStorageTransaction { time, validator.power.into(), validator.proposer_priority.value(), - self.0.as_mut(), + self.inner.as_mut(), ) .await?; } @@ -149,14 +163,12 @@ impl PostgresStorageTransaction { .map(|sig| general_purpose::STANDARD.encode(sig)) .collect(); - // TODO: uncover the secrets of juno's usage of `jsonpb` and how they're recovering - // field names from proto data let messages = chain_tx .tx .body .messages .iter() - .map(|msg| PlaceholderMessage::from(msg)) + .filter_map(|msg| self.decode_or_skip(msg)) .collect::>(); // TODO: missing cosmrs' derives @@ -173,7 +185,7 @@ impl PostgresStorageTransaction { chain_tx.height.into(), chain_tx.index as i32, chain_tx.tx_result.code.is_ok(), - serde_json::to_value(messages)?, + serde_json::Value::Array(messages), chain_tx.tx.body.memo.clone(), signatures, serde_json::to_value(signer_infos)?, @@ -182,7 +194,7 @@ impl PostgresStorageTransaction { chain_tx.tx_result.gas_used, chain_tx.tx_result.log.clone(), json!({ "value": "yep, another todo. on first glance corresponding field doesn't exist in rust" }), - self.0.as_mut(), + self.inner.as_mut(), ) .await?; } @@ -202,10 +214,10 @@ impl PostgresStorageTransaction { chain_tx.hash.to_string(), index as i64, msg.type_url.clone(), - serde_json::to_value(PlaceholderMessage::from(msg))?, + serde_json::to_value(self.decode_or_skip(msg))?, vec!["PLACEHOLDER".to_owned()], chain_tx.height.into(), - self.0.as_mut(), + self.inner.as_mut(), ) .await? } @@ -218,7 +230,7 @@ impl PostgresStorageTransaction { #[async_trait] impl NyxdScraperTransaction for PostgresStorageTransaction { async fn commit(self) -> Result<(), NyxdScraperStorageError> { - self.0 + self.inner .commit() .await .map_err(PostgresScraperError::from) diff --git a/common/nyxd-scraper-shared/src/block_processor/mod.rs b/common/nyxd-scraper-shared/src/block_processor/mod.rs index d99b83677a..a6c37ac984 100644 --- a/common/nyxd-scraper-shared/src/block_processor/mod.rs +++ b/common/nyxd-scraper-shared/src/block_processor/mod.rs @@ -4,7 +4,6 @@ use crate::block_processor::helpers::split_request_range; use crate::block_processor::types::BlockToProcess; use crate::block_requester::BlockRequest; -use crate::cosmos_module::message_registry::{default_message_registry, MessageRegistry}; use crate::error::ScraperError; use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; @@ -92,7 +91,6 @@ pub struct BlockProcessor { incoming: UnboundedReceiverStream, block_requester: Sender, storage: S, - message_registry: MessageRegistry, // future work: rather than sending each msg to every msg module, // let them subscribe based on `type_url` inside the message itself @@ -137,7 +135,6 @@ where incoming: incoming.into(), block_requester, storage, - message_registry: default_message_registry(), block_modules: vec![], tx_modules: vec![], msg_modules: vec![], diff --git a/common/nyxd-scraper-shared/src/cosmos_module/message_registry.rs b/common/nyxd-scraper-shared/src/cosmos_module/message_registry.rs index c972c6b1b8..ef5d0e7b8f 100644 --- a/common/nyxd-scraper-shared/src/cosmos_module/message_registry.rs +++ b/common/nyxd-scraper-shared/src/cosmos_module/message_registry.rs @@ -60,7 +60,7 @@ pub(crate) fn default_proto_to_json( type ConvertFn = fn(&Any) -> Result; -#[derive(Default)] +#[derive(Default, Clone)] pub struct MessageRegistry { // type url to function converting bytes to proto and finally to json registered_types: HashMap, diff --git a/common/nyxd-scraper-shared/src/cosmos_module/mod.rs b/common/nyxd-scraper-shared/src/cosmos_module/mod.rs index 5b89ff7ec3..d60bb7caf1 100644 --- a/common/nyxd-scraper-shared/src/cosmos_module/mod.rs +++ b/common/nyxd-scraper-shared/src/cosmos_module/mod.rs @@ -3,7 +3,7 @@ use crate::cosmos_module::message_registry::MessageRegistry; -pub(crate) mod message_registry; +pub mod message_registry; mod modules; pub trait CosmosModule { diff --git a/common/nyxd-scraper-shared/src/lib.rs b/common/nyxd-scraper-shared/src/lib.rs index 594d0c129f..7d3f198b7c 100644 --- a/common/nyxd-scraper-shared/src/lib.rs +++ b/common/nyxd-scraper-shared/src/lib.rs @@ -14,6 +14,10 @@ pub mod storage; pub use block_processor::pruning::{PruningOptions, PruningStrategy}; pub use block_processor::types::ParsedTransactionResponse; +pub use cosmos_module::{ + message_registry::{default_message_registry, MessageRegistry}, + CosmosModule, +}; pub use cosmrs::Any; pub use modules::{BlockModule, MsgModule, TxModule}; pub use scraper::{Config, NyxdScraper, StartingBlockOpts};