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

chain-scraper : use tx module for parsing transactions

Browse Source
This commit is contained in:
Sachin Kamath
2024-12-12 01:09:18 +05:30
committed by Mark Sinclair
parent a029be3456
commit 91f0321479
9 changed files with 233 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -6843,6 +6843,7 @@ name = "nyx-chain-watcher"
version = "0.1.5"
dependencies = [
"anyhow",
"async-trait",
"axum 0.7.7",
"chrono",
"clap 4.5.20",
common/nyxd-scraper/src/lib.rs
+1
View File
@@ -15,6 +15,7 @@ pub(crate) mod scraper;
pub mod storage;
pub use block_processor::pruning::{PruningOptions, PruningStrategy};
pub use block_processor::types::ParsedTransactionResponse;
pub use modules::{BlockModule, MsgModule, TxModule};
pub use scraper::{Config, NyxdScraper, StartingBlockOpts};
pub use storage::models;
common/nyxd-scraper/src/storage/manager.rs
-52
View File
@@ -237,58 +237,6 @@ impl StorageManager {
Ok(-1)
}
}
#[allow(dead_code)]
pub async fn get_transactions_after_height(
&self,
min_height: i64,
message_type: Option<&str>,
) -> Result<Vec<TransactionWithBlock>, sqlx::Error> {
match message_type {
Some(msg_type) => {
sqlx::query_as!(
TransactionWithBlock,
r#"
SELECT t.hash, t.height, t.memo, t.raw_log
FROM message m
JOIN "transaction" t ON m.transaction_hash = t.hash
JOIN block b ON t.height = b.height
WHERE t.height > ?
AND m.type = ?
ORDER BY t.height ASC
"#,
min_height,
msg_type
)
.fetch_all(&self.connection_pool)
.await
}
None => {
sqlx::query_as!(
TransactionWithBlock,
r#"
SELECT t.hash, t.height, t.memo, t.raw_log
FROM message m
JOIN "transaction" t ON m.transaction_hash = t.hash
JOIN block b ON t.height = b.height
WHERE t.height > ?
ORDER BY t.height ASC
"#,
min_height
)
.fetch_all(&self.connection_pool)
.await
}
}
}
}
#[derive(Debug, sqlx::FromRow)]
pub struct TransactionWithBlock {
pub hash: String,
pub height: i64,
pub memo: Option<String>,
pub raw_log: Option<String>,
}
// make those generic over executor so that they could be performed over connection pool and a tx
common/nyxd-scraper/src/storage/mod.rs
-12
View File
@@ -13,7 +13,6 @@ use crate::{
models::{CommitSignature, Validator},
},
};
use manager::TransactionWithBlock;
use sqlx::{
sqlite::{SqliteAutoVacuum, SqliteSynchronous},
types::time::OffsetDateTime,
@@ -221,17 +220,6 @@ impl ScraperStorage {
pub async fn get_pruned_height(&self) -> Result<i64, ScraperError> {
Ok(self.manager.get_pruned_height().await?)
}
pub async fn get_transactions_after_height(
&self,
min_height: i64,
message_type: Option<&str>,
) -> Result<Vec<TransactionWithBlock>, ScraperError> {
Ok(self
.manager
.get_transactions_after_height(min_height, message_type)
.await?)
}
}
pub async fn persist_block(
nyx-chain-watcher/Cargo.toml
+1
View File
@@ -15,6 +15,7 @@ readme.workspace = true
[dependencies]
anyhow = { workspace = true }
async-trait.workspace = true
axum = { workspace = true, features = ["tokio"] }
chrono = { workspace = true }
clap = { workspace = true, features = ["cargo", "derive", "env"] }
nyx-chain-watcher/migrations/003_create_transactions_table.sql
+12
View File
@@ -0,0 +1,12 @@
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tx_hash TEXT NOT NULL,
height INTEGER NOT NULL,
message_index INTEGER NOT NULL,
sender TEXT NOT NULL,
recipient TEXT NOT NULL,
amount TEXT NOT NULL,
memo TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(tx_hash, message_index)
);
nyx-chain-watcher/src/chain_scraper/mod.rs
+139 -2
View File
@@ -1,13 +1,20 @@
use crate::config::PaymentWatcherConfig;
use crate::env::vars::{
NYXD_SCRAPER_START_HEIGHT, NYXD_SCRAPER_UNSAFE_NUKE_DB,
NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT,
};
use nyxd_scraper::{NyxdScraper, PruningOptions};
use async_trait::async_trait;
use nyxd_scraper::{
error::ScraperError, storage::StorageTransaction, NyxdScraper, ParsedTransactionResponse,
PruningOptions, TxModule,
};
use sqlx::SqlitePool;
use std::fs;
use tracing::{info, warn};
pub(crate) async fn run_chain_scraper(
config: &crate::config::Config,
db_pool: SqlitePool,
) -> anyhow::Result<NyxdScraper> {
let websocket_url = std::env::var("NYXD_WS").expect("NYXD_WS not defined");
@@ -40,6 +47,10 @@ pub(crate) async fn run_chain_scraper(
fs::remove_file(config.chain_scraper_database_path())?;
}
if config.payment_watcher_config.is_none() {
anyhow::bail!("No payment watcher config found, not running chain scraper");
}
let scraper = NyxdScraper::builder(nyxd_scraper::Config {
websocket_url,
rpc_url,
@@ -50,7 +61,11 @@ pub(crate) async fn run_chain_scraper(
start_block_height,
use_best_effort_start_height,
},
});
})
.with_tx_module(EventScraperModule::new(
db_pool,
config.payment_watcher_config.clone().unwrap_or_default(),
));
let instance = scraper.build_and_start().await?;
@@ -59,3 +74,125 @@ pub(crate) async fn run_chain_scraper(
Ok(instance)
}
pub struct EventScraperModule {
db_pool: SqlitePool,
payment_config: PaymentWatcherConfig,
}
impl EventScraperModule {
pub fn new(db_pool: SqlitePool, payment_config: PaymentWatcherConfig) -> Self {
Self {
db_pool,
payment_config,
}
}
#[allow(clippy::too_many_arguments)]
async fn store_transfer_event(
&self,
tx_hash: &str,
height: i64,
message_index: i64,
sender: String,
recipient: String,
amount: String,
memo: Option<String>,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO transactions (tx_hash, height, message_index, sender, recipient, amount, memo)
VALUES (?, ?, ?, ?, ?, ?, ?)
"#,
tx_hash,
height,
message_index,
sender,
recipient,
amount,
memo
)
.execute(&self.db_pool)
.await?;
Ok(())
}
}
#[async_trait]
impl TxModule for EventScraperModule {
async fn handle_tx(
&mut self,
tx: &ParsedTransactionResponse,
_: &mut StorageTransaction,
) -> Result<(), ScraperError> {
let events = &tx.tx_result.events;
let height = tx.height.value() as i64;
let tx_hash = tx.hash.to_string();
let memo = tx.tx.body.memo.clone();
// Don't process failed transactions
if !tx.tx_result.code.is_ok() {
return Ok(());
}
// Process each event
for event in events {
// Only process transfer events
if event.kind == "transfer" {
let mut recipient = None;
let mut sender = None;
let mut amount = None;
// TODO: get message index from event
let message_index = 0;
// Extract transfer event attributes
for attr in &event.attributes {
if let (Ok(key), Ok(value)) = (attr.key_str(), attr.value_str()) {
match key {
"recipient" => recipient = Some(value.to_string()),
"sender" => sender = Some(value.to_string()),
"amount" => amount = Some(value.to_string()),
_ => continue,
}
}
}
// If we have all required fields, check if recipient is watched and store
if let (Some(recipient), Some(sender), Some(amount)) = (recipient, sender, amount) {
// Check if any watcher is watching this recipient
let is_watched = self.payment_config.watchers.iter().any(|watcher| {
if let Some(watched_accounts) =
&watcher.watch_for_transfer_recipient_accounts
{
watched_accounts
.iter()
.any(|account| account.to_string() == recipient)
} else {
false
}
});
if is_watched {
if let Err(e) = self
.store_transfer_event(
&tx_hash,
height,
message_index,
sender,
recipient,
amount,
Some(memo.clone()),
)
.await
{
warn!("Failed to store transfer event: {}", e);
}
}
}
}
}
Ok(())
}
}
nyx-chain-watcher/src/cli/commands/run/mod.rs
+6 -9
View File
@@ -47,18 +47,15 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa
// Spawn the payment listener task
let payment_listener_handle = tokio::spawn({
let obs_pool = watcher_pool.clone();
let chain_scraper = run_chain_scraper(&config).await?;
let chain_storage = chain_scraper.storage();
let price_scraper_pool = storage.pool_owned().await;
let scraper_pool = storage.pool_owned().await;
run_chain_scraper(&config, scraper_pool).await?;
let payment_watcher_config = config.payment_watcher_config.unwrap_or_default();
async move {
if let Err(e) = payment_listener::run_payment_listener(
payment_watcher_config,
obs_pool,
chain_storage,
)
.await
if let Err(e) =
payment_listener::run_payment_listener(payment_watcher_config, price_scraper_pool)
.await
{
error!("Payment listener error: {}", e);
}
nyx-chain-watcher/src/payment_listener/mod.rs
+73 -175
View File
@@ -2,33 +2,19 @@ use crate::config::payments_watcher::HttpAuthenticationOptions;
use crate::config::PaymentWatcherConfig;
use crate::db::queries;
use crate::models::WebhookPayload;
use nym_validator_client::nyxd::{AccountId, Coin};
use nyxd_scraper::storage::ScraperStorage;
use nym_validator_client::nyxd::Coin;
use reqwest::Client;
use rocket::form::validate::Contains;
use serde_json::Value;
use sqlx::SqlitePool;
use std::str::FromStr;
use tokio::time::{self, Duration};
use tracing::{error, info, trace};
#[derive(Debug)]
struct TransferEvent {
recipient: AccountId,
sender: AccountId,
amount: String,
message_index: u64,
}
use tracing::{error, info};
pub(crate) async fn run_payment_listener(
payment_watcher_config: PaymentWatcherConfig,
watcher_pool: SqlitePool,
chain_storage: ScraperStorage,
) -> anyhow::Result<()> {
let client = Client::new();
let default_message_types = vec!["/cosmos.bank.v1beta1.MsgSend".to_string()];
loop {
// 1. get the last height this watcher ran at
let last_checked_height = queries::payments::get_last_checked_height(&watcher_pool).await?;
@@ -36,107 +22,82 @@ pub(crate) async fn run_payment_listener(
// 2. iterate through watchers
for watcher in &payment_watcher_config.watchers {
let watch_for_chain_message_types = watcher
.watch_for_chain_message_types
.as_ref()
.unwrap_or(&default_message_types);
if watcher.watch_for_transfer_recipient_accounts.is_some() {
// 3. Query new transactions for this watcher's recipient accounts
let transactions = sqlx::query!(
r#"
SELECT * FROM transactions
WHERE height > ?
ORDER BY height ASC, message_index ASC
"#,
last_checked_height
)
.fetch_all(&watcher_pool)
.await?;
// 3. build up transactions that match the message types we are looking for
let mut transactions = vec![];
for message_type in watch_for_chain_message_types {
match chain_storage
.get_transactions_after_height(
last_checked_height,
Some(message_type),
if !transactions.is_empty() {
info!(
"[watcher = {}] Processing {} transactions",
watcher.id,
transactions.len()
);
}
for tx in transactions {
let funds = Coin::from_str(&tx.amount)?;
let amount: f64 = funds.amount as f64 / 1e6f64; // convert to major value, there will be precision loss
// Store transaction hash for later use
let tx_hash = tx.tx_hash.clone();
let message_index = tx.message_index;
queries::payments::insert_payment(
&watcher_pool,
tx.tx_hash,
tx.sender.clone(),
tx.recipient.clone(),
amount,
tx.height,
tx.memo.clone(),
)
.await {
Ok(txs) => {
for t in txs {
transactions.push(t);
.await?;
let webhook_data = WebhookPayload {
transaction_hash: tx_hash.clone(),
message_index: message_index as u64,
sender_address: tx.sender,
receiver_address: tx.recipient,
funds: funds.into(),
height: tx.height as u128,
memo: tx.memo,
};
let mut request_builder = client.post(&watcher.webhook_url).json(&webhook_data);
if let Some(auth) = &watcher.authentication {
match auth {
HttpAuthenticationOptions::AuthorizationBearerToken { token } => {
request_builder = request_builder.bearer_auth(token);
}
}
}
Err(e) => error!("Failed to get transactions (message_type = {message_type}) from scraper database: {e}")
}
}
for tx in transactions {
if let Some(raw_log) = tx.raw_log.as_deref() {
if let Some(watch_for_transfer_recipient_accounts) =
&watcher.watch_for_transfer_recipient_accounts
{
// 4. match recipient accounts we are looking for
match parse_transfer_from_raw_log(
raw_log,
watch_for_transfer_recipient_accounts,
) {
Ok(transfer_events) => {
if !transfer_events.is_empty() {
info!(
"[watcher = {}] Processing transaction: {} - {} payment events found",
watcher.id, tx.hash, transfer_events.len()
);
}
for transfer in transfer_events {
let funds = Coin::from_str(&transfer.amount)?;
let amount: f64 = funds.amount as f64 / 1e6f64; // convert to major value, there will be precision loss
queries::payments::insert_payment(
&watcher_pool,
tx.hash.clone(),
transfer.sender.clone().to_string(),
transfer.recipient.clone().to_string(),
amount,
tx.height,
tx.memo.clone(),
)
.await?;
let webhook_data = WebhookPayload {
transaction_hash: tx.hash.clone(),
message_index: transfer.message_index,
sender_address: transfer.sender.to_string(),
receiver_address: transfer.recipient.to_string(),
funds: funds.into(),
height: tx.height as u128,
memo: tx.memo.clone(),
};
let mut request_builder =
client.post(&watcher.webhook_url).json(&webhook_data);
if let Some(auth) = &watcher.authentication {
match auth {
HttpAuthenticationOptions::AuthorizationBearerToken { token } => {
request_builder = request_builder.bearer_auth(token);
}
}
}
match request_builder.send().await {
Ok(res) => info!(
"[watcher = {}] ✅ Webhook {} {} - tx {}, index {}",
watcher.id,
res.status(),
res.url(),
tx.hash,
transfer.message_index,
),
Err(e) => error!(
"[watcher = {}] ❌ Webhook {:?} {:?} error = {}",
watcher.id,
e.status(),
e.url(),
e,
),
}
}
}
Err(e) => error!(
"[watcher = {}] ❌ Parse logs for tx {} failed, error = {}",
watcher.id, tx.hash, e,
),
}
match request_builder.send().await {
Ok(res) => info!(
"[watcher = {}] ✅ Webhook {} {} - tx {}, index {}",
watcher.id,
res.status(),
res.url(),
tx_hash,
message_index,
),
Err(e) => error!(
"[watcher = {}] ❌ Webhook {:?} {:?} error = {}",
watcher.id,
e.status(),
e.url(),
e,
),
}
}
}
@@ -145,66 +106,3 @@ pub(crate) async fn run_payment_listener(
time::sleep(Duration::from_secs(10)).await;
}
}
fn parse_transfer_from_raw_log(
raw_log: &str,
watch_for_transfer_recipient_accounts: &Vec<AccountId>,
) -> anyhow::Result<Vec<TransferEvent>> {
let log_value: Value = serde_json::from_str(raw_log)?;
let mut transfers: Vec<TransferEvent> = vec![];
let default_value = vec![];
let log_entries: &Vec<Value> = log_value.as_array().unwrap_or(&default_value);
trace!("contains {} log entries", log_entries.len());
for log_entry in log_entries {
let message_index = log_entry["msg_index"].as_u64().unwrap_or_default();
trace!("entry - {message_index}...");
if let Some(events) = log_entry["events"].as_array() {
for transfer_event in events.iter().filter(|e| e["type"] == "transfer") {
if let Some(attrs) = transfer_event["attributes"].as_array() {
let mut recipient: Option<AccountId> = None;
let mut sender: Option<AccountId> = None;
let mut amount: Option<String> = None;
for attr in attrs {
match attr["key"].as_str() {
Some("recipient") => {
recipient =
AccountId::from_str(attr["value"].as_str().unwrap_or("")).ok();
}
Some("sender") => {
sender =
AccountId::from_str(attr["value"].as_str().unwrap_or("")).ok();
}
Some("amount") => {
amount = Some(attr["value"].as_str().unwrap_or("").to_string())
}
// TODO: parse message index
_ => continue,
}
}
if let (Some(recipient), Some(sender), Some(amount)) =
(recipient, sender, amount)
{
if watch_for_transfer_recipient_accounts.contains(&recipient) {
transfers.push(TransferEvent {
recipient,
sender,
amount,
message_index,
});
}
}
}
}
}
}
Ok(transfers)
}