From 91f03214796cd9698514ef876e0620b46a8034a3 Mon Sep 17 00:00:00 2001 From: Sachin Kamath Date: Thu, 12 Dec 2024 01:09:18 +0530 Subject: [PATCH] chain-scraper : use tx module for parsing transactions --- Cargo.lock | 1 + common/nyxd-scraper/src/lib.rs | 1 + common/nyxd-scraper/src/storage/manager.rs | 52 ---- common/nyxd-scraper/src/storage/mod.rs | 12 - nyx-chain-watcher/Cargo.toml | 1 + .../003_create_transactions_table.sql | 12 + nyx-chain-watcher/src/chain_scraper/mod.rs | 141 +++++++++- nyx-chain-watcher/src/cli/commands/run/mod.rs | 15 +- nyx-chain-watcher/src/payment_listener/mod.rs | 248 ++++++------------ 9 files changed, 233 insertions(+), 250 deletions(-) create mode 100644 nyx-chain-watcher/migrations/003_create_transactions_table.sql diff --git a/Cargo.lock b/Cargo.lock index 8041a00e7f..70fc1c10d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/common/nyxd-scraper/src/lib.rs b/common/nyxd-scraper/src/lib.rs index 806de871dc..7d22921a3e 100644 --- a/common/nyxd-scraper/src/lib.rs +++ b/common/nyxd-scraper/src/lib.rs @@ -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; diff --git a/common/nyxd-scraper/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs index 5475f5d355..fb40a065b8 100644 --- a/common/nyxd-scraper/src/storage/manager.rs +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -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, 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, - pub raw_log: Option, } // make those generic over executor so that they could be performed over connection pool and a tx diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index 3b29438bbb..2095bde1e3 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -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 { Ok(self.manager.get_pruned_height().await?) } - - pub async fn get_transactions_after_height( - &self, - min_height: i64, - message_type: Option<&str>, - ) -> Result, ScraperError> { - Ok(self - .manager - .get_transactions_after_height(min_height, message_type) - .await?) - } } pub async fn persist_block( diff --git a/nyx-chain-watcher/Cargo.toml b/nyx-chain-watcher/Cargo.toml index 3b671ad14e..5c6fe60719 100644 --- a/nyx-chain-watcher/Cargo.toml +++ b/nyx-chain-watcher/Cargo.toml @@ -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"] } diff --git a/nyx-chain-watcher/migrations/003_create_transactions_table.sql b/nyx-chain-watcher/migrations/003_create_transactions_table.sql new file mode 100644 index 0000000000..e707ddaf85 --- /dev/null +++ b/nyx-chain-watcher/migrations/003_create_transactions_table.sql @@ -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) +); \ No newline at end of file diff --git a/nyx-chain-watcher/src/chain_scraper/mod.rs b/nyx-chain-watcher/src/chain_scraper/mod.rs index 701dd467dc..2835a8d105 100644 --- a/nyx-chain-watcher/src/chain_scraper/mod.rs +++ b/nyx-chain-watcher/src/chain_scraper/mod.rs @@ -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 { 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, + ) -> 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(()) + } +} diff --git a/nyx-chain-watcher/src/cli/commands/run/mod.rs b/nyx-chain-watcher/src/cli/commands/run/mod.rs index 96344a4014..ef2a553995 100644 --- a/nyx-chain-watcher/src/cli/commands/run/mod.rs +++ b/nyx-chain-watcher/src/cli/commands/run/mod.rs @@ -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); } diff --git a/nyx-chain-watcher/src/payment_listener/mod.rs b/nyx-chain-watcher/src/payment_listener/mod.rs index 30a7765589..826f23dfd8 100644 --- a/nyx-chain-watcher/src/payment_listener/mod.rs +++ b/nyx-chain-watcher/src/payment_listener/mod.rs @@ -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, -) -> anyhow::Result> { - let log_value: Value = serde_json::from_str(raw_log)?; - - let mut transfers: Vec = vec![]; - - let default_value = vec![]; - let log_entries: &Vec = 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 = None; - let mut sender: Option = None; - let mut amount: Option = 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) -}