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

message content parsing in psql

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2025-07-07 14:54:16 +01:00
parent 2320a2f249
commit 42c0b69f4d
9 changed files with 48 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nyxd-scraper-psql/src/lib.rs
-1
View File
@@ -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<PostgresScraperStorage>;
common/nyxd-scraper-psql/src/proto_registry.rs
-6
View File
@@ -1,6 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use std::collections::HashMap;
pub struct ProtoRegistry(HashMap<(), ()>);
common/nyxd-scraper-psql/src/storage/block_storage.rs
+14 -3
View File
@@ -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 })
}
common/nyxd-scraper-psql/src/storage/helpers.rs
-27
View File
@@ -1,35 +1,8 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// 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<JsonValue, PostgresScraperError> {
// 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,
common/nyxd-scraper-psql/src/storage/transaction.rs
+28 -16
View File
@@ -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<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,
@@ -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::<Vec<_>>();
// 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)
common/nyxd-scraper-shared/src/block_processor/mod.rs
-3
View File
@@ -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<S> {
incoming: UnboundedReceiverStream<BlockToProcess>,
block_requester: Sender<BlockRequest>,
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![],
common/nyxd-scraper-shared/src/cosmos_module/message_registry.rs
+1 -1
View File
@@ -60,7 +60,7 @@ pub(crate) fn default_proto_to_json<T: Message + Default + Serialize>(
type ConvertFn = fn(&Any) -> Result<serde_json::Value, ScraperError>;
#[derive(Default)]
#[derive(Default, Clone)]
pub struct MessageRegistry {
// type url to function converting bytes to proto and finally to json
registered_types: HashMap<String, ConvertFn>,
common/nyxd-scraper-shared/src/cosmos_module/mod.rs
+1 -1
View File
@@ -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 {
common/nyxd-scraper-shared/src/lib.rs
+4
View File
@@ -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};