From ade88af3cbf42bace7d050a570df2c6724320888 Mon Sep 17 00:00:00 2001 From: Mark Sinclair Date: Fri, 6 Dec 2024 17:31:25 +0000 Subject: [PATCH] Add config, overrides and CLI --- .gitignore | 3 +- Cargo.lock | 5 + common/config/Cargo.toml | 3 +- common/config/src/error.rs | 10 + common/config/src/lib.rs | 37 +++ common/nyxd-scraper/src/scraper/mod.rs | 12 +- nym-validator-rewarder/src/config/mod.rs | 1 + nym-wallet/Cargo.lock | 1 + nyx-chain-watcher/Cargo.toml | 10 +- nyx-chain-watcher/README.md | 15 +- nyx-chain-watcher/build.rs | 4 +- nyx-chain-watcher/docker-compose.yml | 31 --- nyx-chain-watcher/src/chain_scraper/mod.rs | 16 +- .../src/cli/commands/build_info.rs | 17 ++ nyx-chain-watcher/src/cli/commands/init.rs | 19 ++ nyx-chain-watcher/src/cli/commands/mod.rs | 3 + .../src/cli/commands/run/args.rs | 101 +++++++ .../src/cli/commands/run/config.rs | 84 ++++++ nyx-chain-watcher/src/cli/commands/run/mod.rs | 82 ++++++ nyx-chain-watcher/src/cli/mod.rs | 67 +++++ nyx-chain-watcher/src/config/mod.rs | 249 ++++++++++++++++++ .../src/config/payments_watcher.rs | 23 ++ nyx-chain-watcher/src/config/template.rs | 29 ++ nyx-chain-watcher/src/env.rs | 25 ++ nyx-chain-watcher/src/error.rs | 40 +++ nyx-chain-watcher/src/main.rs | 84 +----- nyx-chain-watcher/src/models.rs | 14 + nyx-chain-watcher/src/payment_listener/mod.rs | 191 ++++++++++---- 28 files changed, 992 insertions(+), 184 deletions(-) create mode 100644 common/config/src/error.rs delete mode 100644 nyx-chain-watcher/docker-compose.yml create mode 100644 nyx-chain-watcher/src/cli/commands/build_info.rs create mode 100644 nyx-chain-watcher/src/cli/commands/init.rs create mode 100644 nyx-chain-watcher/src/cli/commands/mod.rs create mode 100644 nyx-chain-watcher/src/cli/commands/run/args.rs create mode 100644 nyx-chain-watcher/src/cli/commands/run/config.rs create mode 100644 nyx-chain-watcher/src/cli/commands/run/mod.rs create mode 100644 nyx-chain-watcher/src/cli/mod.rs create mode 100644 nyx-chain-watcher/src/config/mod.rs create mode 100644 nyx-chain-watcher/src/config/payments_watcher.rs create mode 100644 nyx-chain-watcher/src/config/template.rs create mode 100644 nyx-chain-watcher/src/env.rs create mode 100644 nyx-chain-watcher/src/error.rs create mode 100644 nyx-chain-watcher/src/models.rs diff --git a/.gitignore b/.gitignore index 67448571a4..3f0187cddc 100644 --- a/.gitignore +++ b/.gitignore @@ -53,4 +53,5 @@ nym-network-monitor/topology.json nym-network-monitor/__pycache__ nym-network-monitor/*.key -*.sqlite \ No newline at end of file +*.sqlite +.build \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index a22608cd19..8c9f88ecd8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5064,6 +5064,7 @@ dependencies = [ "log", "nym-network-defaults", "serde", + "thiserror", "toml 0.8.14", "url", ] @@ -6941,15 +6942,19 @@ dependencies = [ "chrono", "clap 4.5.20", "nym-bin-common", + "nym-config", "nym-network-defaults", "nym-node-requests", "nym-task", + "nym-validator-client", "nyxd-scraper", "reqwest 0.12.4", "rocket", + "schemars", "serde", "serde_json", "sqlx", + "thiserror", "time", "tokio", "tokio-util", diff --git a/common/config/Cargo.toml b/common/config/Cargo.toml index 0723f2c381..d978fe4aee 100644 --- a/common/config/Cargo.toml +++ b/common/config/Cargo.toml @@ -12,7 +12,8 @@ dirs = { workspace = true, optional = true } handlebars = { workspace = true } log = { workspace = true } serde = { workspace = true, features = ["derive"] } -toml = { workspace = true } +thiserror = { workspace = true } +toml = { workspace = true, features = ["display"] } url = { workspace = true } nym-network-defaults = { path = "../network-defaults", features = ["utoipa"] } diff --git a/common/config/src/error.rs b/common/config/src/error.rs new file mode 100644 index 0000000000..9bbd3444f8 --- /dev/null +++ b/common/config/src/error.rs @@ -0,0 +1,10 @@ +use std::io; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum NymConfigTomlError { + #[error(transparent)] + FileIoFailure(#[from] io::Error), + #[error(transparent)] + TomlSerializeFailure(#[from] toml::ser::Error), +} diff --git a/common/config/src/lib.rs b/common/config/src/lib.rs index 932d7515a6..42dcea8610 100644 --- a/common/config/src/lib.rs +++ b/common/config/src/lib.rs @@ -13,6 +13,7 @@ pub use helpers::{parse_urls, OptionalSet}; pub use toml::de::Error as TomlDeError; pub mod defaults; +pub mod error; pub mod helpers; pub mod legacy_helpers; pub mod serde_helpers; @@ -95,6 +96,42 @@ where config.format_to_writer(file) } +pub fn save_unformatted_config_to_file( + config: &C, + path: P, +) -> Result<(), error::NymConfigTomlError> +where + C: Serialize + ?Sized, + P: AsRef, +{ + let path = path.as_ref(); + log::info!("saving config file to {}", path.display()); + + if let Some(parent) = path.parent() { + create_dir_all(parent)?; + } + + let mut file = File::create(path)?; + + // TODO: check for whether any of our configs store anything sensitive + // and change that to 0o644 instead + #[cfg(target_family = "unix")] + { + use std::os::unix::fs::PermissionsExt; + + let mut perms = fs::metadata(path)?.permissions(); + perms.set_mode(0o600); + fs::set_permissions(path, perms)?; + } + + // let serde format the TOML in whatever ugly way it chooses + // TODO: in https://docs.rs/toml/latest/toml/fn.to_string_pretty.html it recommends using + // https://docs.rs/toml_edit/latest/toml_edit/struct.DocumentMut.html to preserve formatting + let toml_string = toml::to_string_pretty(config)?; + + Ok(file.write_all(toml_string.as_bytes())?) +} + pub fn deserialize_config_from_toml_str(raw: &str) -> Result where C: DeserializeOwned, diff --git a/common/nyxd-scraper/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs index 1b616f3247..e4d133341e 100644 --- a/common/nyxd-scraper/src/scraper/mod.rs +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -36,6 +36,8 @@ pub struct Config { pub pruning_options: PruningOptions, pub store_precommits: bool, + + pub start_block_height: Option, } pub struct NyxdScraperBuilder { @@ -47,10 +49,9 @@ pub struct NyxdScraperBuilder { } impl NyxdScraperBuilder { - pub async fn build_and_start( - self, - start_block: Option, - ) -> Result { + pub async fn build_and_start(self) -> Result { + let start_block_height = self.config.start_block_height.clone(); + let scraper = NyxdScraper::new(self.config).await?; let (processing_tx, processing_rx) = unbounded_channel(); @@ -93,7 +94,8 @@ impl NyxdScraperBuilder { ) .await?; - if let Some(height) = start_block { + // TODO: decide if this should be removed? + if let Some(height) = start_block_height { scraper.process_block_range(Some(height), None).await?; } diff --git a/nym-validator-rewarder/src/config/mod.rs b/nym-validator-rewarder/src/config/mod.rs index f0369adde7..31c44b07e7 100644 --- a/nym-validator-rewarder/src/config/mod.rs +++ b/nym-validator-rewarder/src/config/mod.rs @@ -129,6 +129,7 @@ impl Config { database_path: self.storage_paths.nyxd_scraper.clone(), pruning_options: self.nyxd_scraper.pruning, store_precommits: self.nyxd_scraper.store_precommits, + start_block_height: None, } } diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index 81fa6e2a7b..af26e978fc 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -3163,6 +3163,7 @@ dependencies = [ "log", "nym-network-defaults", "serde", + "thiserror", "toml 0.8.19", "url", ] diff --git a/nyx-chain-watcher/Cargo.toml b/nyx-chain-watcher/Cargo.toml index 58564b3d12..1f9e87d4a2 100644 --- a/nyx-chain-watcher/Cargo.toml +++ b/nyx-chain-watcher/Cargo.toml @@ -1,5 +1,5 @@ # Copyright 2024 - Nym Technologies SA -# SPDX-License-Identifier: Apache-2.0 +# SPDX-License-Identifier: GPL-3.0-only [package] name = "nyx-chain-watcher" @@ -17,19 +17,23 @@ readme.workspace = true anyhow = { workspace = true } axum = { workspace = true, features = ["tokio"] } chrono = { workspace = true } -clap = { workspace = true, features = ["derive", "env"] } -nym-bin-common = { path = "../common/bin-common" } +clap = { workspace = true, features = ["cargo", "derive", "env"] } +nym-config = { path = "../common/config" } +nym-bin-common = { path = "../common/bin-common", features = ["output_format"] } nym-network-defaults = { path = "../common/network-defaults" } nym-task = { path = "../common/task" } nym-node-requests = { path = "../nym-node/nym-node-requests", features = [ "openapi", ] } +nym-validator-client = { path = "../common/client-libs/validator-client" } nyxd-scraper = {path = "../common/nyxd-scraper"} reqwest = {workspace= true, features = ["rustls-tls"]} rocket = { workspace = true } +schemars = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "time"] } +thiserror = { workspace = true } time = {version = "0.3.36"} tokio = { workspace = true, features = ["process", "rt-multi-thread"] } tokio-util = { workspace = true } diff --git a/nyx-chain-watcher/README.md b/nyx-chain-watcher/README.md index 4142c4ecee..7c62d08296 100644 --- a/nyx-chain-watcher/README.md +++ b/nyx-chain-watcher/README.md @@ -2,14 +2,17 @@ A simple binary to watch addresses on the Nyx chain and to call webhooks when particular message types are in a block. +Look in [env.rs](./src/env.rs) for the names of environment variables that can be overridden. + ## Running locally ``` -DATABASE_URL=nyx_chain_watcher.sqlite \ -NYXD_WEBSOCKET_URL=wss://rpc.nymtech.net:443/websocket \ -NYXD_RPC_URL=https://rpc.nymtech.net \ -PAYMENT_RECEIVE_ADDRESS=n1... \ -WEBHOOK_URL=https://webhook.site/... \ -cargo run +NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH=chain_history.sqlite \ +NYX_CHAIN_WATCHER_DATABASE_PATH=nyx_chain_watcher.sqlite \ +NYX_CHAIN_WATCHER_WATCH_ACCOUNTS=n1...,n1...,n1... \ +NYX_CHAIN_WATCHER_WEBHOOK_URL="https://webhook.site" \ +NYX_CHAIN_WATCHER_WEBHOOK_AUTH=1234 \ +cargo run -- run ``` + diff --git a/nyx-chain-watcher/build.rs b/nyx-chain-watcher/build.rs index eb28ed3bf5..c1af059445 100644 --- a/nyx-chain-watcher/build.rs +++ b/nyx-chain-watcher/build.rs @@ -5,7 +5,9 @@ use std::{collections::HashMap, fs::File, path::PathBuf, str::FromStr}; #[tokio::main] async fn main() -> Result<()> { - let db_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("nyx_chain_watcher.sqlite"); + let db_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join(".build") + .join("nyx_chain_watcher.sqlite"); // Create the database directory if it doesn't exist if let Some(parent) = db_path.parent() { diff --git a/nyx-chain-watcher/docker-compose.yml b/nyx-chain-watcher/docker-compose.yml deleted file mode 100644 index b06a988e97..0000000000 --- a/nyx-chain-watcher/docker-compose.yml +++ /dev/null @@ -1,31 +0,0 @@ -services: - postgres: - image: postgres:13 - container_name: nym-data-observatory-pg - environment: - POSTGRES_PASSWORD: password - ports: - - "5432:5432" - volumes: - - pgdata:/var/lib/postgresql/data - - data-observatory: - depends_on: - - postgres - image: nym-data-observatory:latest - build: - context: ../ - dockerfile: nym-data-observatory/Dockerfile - container_name: nym-data-observatory - environment: - NYM_DATA_OBSERVATORY_CONNECTION_USERNAME: "postgres" - NYM_DATA_OBSERVATORY_CONNECTION_PASSWORD: "password" - NYM_DATA_OBSERVATORY_CONNECTION_HOST: "postgres" - NYM_DATA_OBSERVATORY_CONNECTION_PORT: "5432" - NYM_DATA_OBSERVATORY_CONNECTION_DB: "" - NYM_DATA_OBSERVATORY_HTTP_PORT: 8000 - env_file: - - ../envs/qa.env - -volumes: - pgdata: diff --git a/nyx-chain-watcher/src/chain_scraper/mod.rs b/nyx-chain-watcher/src/chain_scraper/mod.rs index 650b3f0c34..36634a5964 100644 --- a/nyx-chain-watcher/src/chain_scraper/mod.rs +++ b/nyx-chain-watcher/src/chain_scraper/mod.rs @@ -1,10 +1,9 @@ -use nyxd_scraper::{storage::ScraperStorage, Config, NyxdScraper, PruningOptions}; +use nyxd_scraper::{storage::ScraperStorage, NyxdScraper, PruningOptions}; -pub(crate) async fn run_chain_scraper() -> anyhow::Result { - let websocket_url = - std::env::var("NYXD_WEBSOCKET_URL").expect("NYXD_WEBSOCKET_URL not defined"); +pub(crate) async fn run_chain_scraper(config: &crate::config::Config) -> anyhow::Result { + let websocket_url = std::env::var("NYXD_WS").expect("NYXD_WS not defined"); - let rpc_url = std::env::var("NYXD_RPC_URL").expect("NYXD_RPC_URL not defined"); + let rpc_url = std::env::var("NYXD").expect("NYXD not defined"); let websocket_url = reqwest::Url::parse(&websocket_url)?; let rpc_url = reqwest::Url::parse(&rpc_url)?; @@ -12,15 +11,16 @@ pub(crate) async fn run_chain_scraper() -> anyhow::Result { .ok() .and_then(|value| value.parse::().ok()); - let scraper = NyxdScraper::builder(Config { + let scraper = NyxdScraper::builder(nyxd_scraper::Config { websocket_url, rpc_url, - database_path: "chain_history.sqlite".into(), + database_path: config.chain_scraper_database_path().into(), pruning_options: PruningOptions::nothing(), store_precommits: false, + start_block_height, }); - let instance = scraper.build_and_start(start_block_height).await?; + let instance = scraper.build_and_start().await?; Ok(instance.storage) } diff --git a/nyx-chain-watcher/src/cli/commands/build_info.rs b/nyx-chain-watcher/src/cli/commands/build_info.rs new file mode 100644 index 0000000000..38e144743c --- /dev/null +++ b/nyx-chain-watcher/src/cli/commands/build_info.rs @@ -0,0 +1,17 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::error::NyxChainWatcherError; +use nym_bin_common::bin_info_owned; +use nym_bin_common::output_format::OutputFormat; + +#[derive(clap::Args, Debug)] +pub(crate) struct Args { + #[clap(short, long, default_value_t = OutputFormat::default())] + output: OutputFormat, +} + +pub(crate) fn execute(args: Args) -> Result<(), NyxChainWatcherError> { + println!("{}", args.output.format(&bin_info_owned!())); + Ok(()) +} diff --git a/nyx-chain-watcher/src/cli/commands/init.rs b/nyx-chain-watcher/src/cli/commands/init.rs new file mode 100644 index 0000000000..163c6e9f1a --- /dev/null +++ b/nyx-chain-watcher/src/cli/commands/init.rs @@ -0,0 +1,19 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::config::{default_config_filepath, Config, ConfigBuilder}; +use crate::error::NyxChainWatcherError; +use nym_config::save_unformatted_config_to_file; + +#[derive(clap::Args, Debug)] +pub(crate) struct Args {} + +pub(crate) async fn execute(_args: Args) -> Result<(), NyxChainWatcherError> { + let config_path = default_config_filepath(); + let data_dir = Config::default_data_directory(&config_path)?; + + let builder = ConfigBuilder::new(config_path.clone(), data_dir); + let config = builder.build(); + + Ok(save_unformatted_config_to_file(&config, &config_path)?) +} diff --git a/nyx-chain-watcher/src/cli/commands/mod.rs b/nyx-chain-watcher/src/cli/commands/mod.rs new file mode 100644 index 0000000000..b1f63f4ae5 --- /dev/null +++ b/nyx-chain-watcher/src/cli/commands/mod.rs @@ -0,0 +1,3 @@ +pub(crate) mod build_info; +pub(crate) mod init; +pub(crate) mod run; diff --git a/nyx-chain-watcher/src/cli/commands/run/args.rs b/nyx-chain-watcher/src/cli/commands/run/args.rs new file mode 100644 index 0000000000..cc20ae2bac --- /dev/null +++ b/nyx-chain-watcher/src/cli/commands/run/args.rs @@ -0,0 +1,101 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::env::vars::*; +use nym_validator_client::nyxd::AccountId; + +#[derive(clap::Args, Debug)] +pub(crate) struct Args { + /// (Override) SQLite database file path for chain watcher + #[arg(long, env = NYX_CHAIN_WATCHER_DATABASE_PATH)] + pub(crate) chain_watcher_db_path: Option, + + /// (Override) SQLite database file path for chain scraper history + #[arg(long, env = NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH)] + pub(crate) chain_history_db_path: Option, + + /// (Override) Watch for transfers to these recipient accounts + #[clap( + long, + value_delimiter = ',', + env = NYX_CHAIN_WATCHER_WATCH_ACCOUNTS + )] + pub watch_for_transfer_recipient_accounts: Option>, + + /// (Override) Watch for chain messages of these types + #[clap( + long, + value_delimiter = ',', + env = NYX_CHAIN_WATCHER_WATCH_CHAIN_MESSAGE_TYPES + )] + pub watch_for_chain_message_types: Option>, + + /// (Override) The webhook to call when we find something + #[clap( + long, + env = NYX_CHAIN_WATCHER_WEBHOOK_URL + )] + pub webhook_url: Option, + + /// (Override) Optionally, authenticate with the webhook + #[clap( + long, + env = NYX_CHAIN_WATCHER_WEBHOOK_AUTH + )] + pub webhook_auth: Option, +} + +/*impl Args { + pub(super) fn take_mnemonic(&mut self) -> Option> { + self.entry_gateway.mnemonic.take().map(Zeroizing::new) + } +} + +impl Args { + pub(crate) fn build_config(self) -> Result { + let config_path = self.config.config_path(); + let data_dir = Config::default_data_directory(&config_path)?; + + let id = self + .config + .id() + .clone() + .ok_or(NymNodeError::MissingInitArg { + section: "global".to_string(), + name: "id".to_string(), + })?; + + let config = ConfigBuilder::new(id, config_path.clone(), data_dir.clone()) + .with_mode(self.mode.unwrap_or_default()) + .with_host(self.host.build_config_section()) + .with_http(self.http.build_config_section()) + .with_mixnet(self.mixnet.build_config_section()) + .with_wireguard(self.wireguard.build_config_section(&data_dir)) + .with_storage_paths(NymNodePaths::new(&data_dir)) + .with_mixnode(self.mixnode.build_config_section()) + .with_entry_gateway(self.entry_gateway.build_config_section(&data_dir)) + .with_exit_gateway(self.exit_gateway.build_config_section(&data_dir)) + .build(); + + Ok(config) + } + + pub(crate) fn override_config(self, mut config: Config) -> Config { + if let Some(mode) = self.mode { + config.mode = mode; + } + config.host = self.host.override_config_section(config.host); + config.http = self.http.override_config_section(config.http); + config.mixnet = self.mixnet.override_config_section(config.mixnet); + config.wireguard = self.wireguard.override_config_section(config.wireguard); + config.mixnode = self.mixnode.override_config_section(config.mixnode); + config.entry_gateway = self + .entry_gateway + .override_config_section(config.entry_gateway); + config.exit_gateway = self + .exit_gateway + .override_config_section(config.exit_gateway); + config + } +} +*/ diff --git a/nyx-chain-watcher/src/cli/commands/run/config.rs b/nyx-chain-watcher/src/cli/commands/run/config.rs new file mode 100644 index 0000000000..02e5a2fadc --- /dev/null +++ b/nyx-chain-watcher/src/cli/commands/run/config.rs @@ -0,0 +1,84 @@ +use crate::cli::commands::run::args::Args; +use crate::cli::DEFAULT_NYX_CHAIN_WATCHER_ID; +use crate::config::payments_watcher::{HttpAuthenticationOptions, PaymentWatcherEntry}; +use crate::config::{default_config_filepath, Config, ConfigBuilder, PaymentWatcherConfig}; +use crate::error::NyxChainWatcherError; +use tracing::{info, warn}; + +pub(crate) fn get_run_config(args: Args) -> Result { + info!("{args:#?}"); + + let Args { + ref watch_for_transfer_recipient_accounts, + mut watch_for_chain_message_types, + webhook_auth, + ref chain_watcher_db_path, + ref chain_history_db_path, + webhook_url, + } = args; + + // if there are no args set, then try load the config + if args.watch_for_transfer_recipient_accounts.is_none() + && args.watch_for_transfer_recipient_accounts.is_none() + && args.chain_watcher_db_path.is_none() + { + info!("Loading default config file..."); + return Config::read_from_toml_file_in_default_location(); + } + + // set default messages + if watch_for_chain_message_types.is_none() { + watch_for_chain_message_types = Some(vec!["/cosmos.bank.v1beta1.MsgSend".to_string()]); + } + + // warn if no accounts set + if watch_for_transfer_recipient_accounts.is_none() { + warn!( + "You did not specify any accounts to watch in {}. Only chain data will be stored.", + crate::env::vars::NYX_CHAIN_WATCHER_WATCH_ACCOUNTS + ); + } + + let config_path = default_config_filepath(); + let data_dir = Config::default_data_directory(&config_path)?; + + let mut builder = ConfigBuilder::new(config_path, data_dir); + + if let Some(db_path) = chain_watcher_db_path { + info!("Overriding database url with '{db_path}'"); + builder = builder.with_db_path(db_path.clone()); + } + + if let Some(db_path) = chain_history_db_path { + info!("Overriding chain history database url with '{db_path}'"); + builder = builder.with_chain_scraper_db_path(db_path.clone()); + } + + if let Some(webhook_url) = webhook_url { + let authentication = + webhook_auth.map(|token| HttpAuthenticationOptions::AuthorizationBearerToken { token }); + + let watcher_config = PaymentWatcherConfig { + watchers: vec![PaymentWatcherEntry { + id: DEFAULT_NYX_CHAIN_WATCHER_ID.to_string(), + description: None, + watch_for_transfer_recipient_accounts: watch_for_transfer_recipient_accounts + .clone(), + watch_for_chain_message_types, + webhook_url, + authentication, + }], + }; + + info!("Overriding watcher config with env vars"); + + builder = builder.with_payment_watcher_config(watcher_config); + } else { + warn!( + "You did not specify a webhook in {}. Only database items will be stored.", + crate::env::vars::NYX_CHAIN_WATCHER_WEBHOOK_URL + ); + } + + Ok(builder.build()) +} diff --git a/nyx-chain-watcher/src/cli/commands/run/mod.rs b/nyx-chain-watcher/src/cli/commands/run/mod.rs new file mode 100644 index 0000000000..84d3d8e1e5 --- /dev/null +++ b/nyx-chain-watcher/src/cli/commands/run/mod.rs @@ -0,0 +1,82 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::error::NyxChainWatcherError; +use tokio::join; +use tracing::{error, info, trace}; + +mod args; +mod config; + +use crate::chain_scraper::run_chain_scraper; +use crate::{db, http, payment_listener, price_scraper}; +pub(crate) use args::Args; +use nym_task::signal::wait_for_signal; + +pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWatcherError> { + trace!("passed arguments: {args:#?}"); + + let config = config::get_run_config(args)?; + + let db_path = config.database_path(); + + info!("Config is {config:#?}"); + info!("Database path is {:?}", std::path::Path::new(&db_path).canonicalize().unwrap_or_default()); + info!("Chain History Database path is {:?}", std::path::Path::new(&config.chain_scraper_database_path()).canonicalize().unwrap_or_default()); + + // Ensure parent directory exists + if let Some(parent) = std::path::Path::new(&db_path).parent() { + std::fs::create_dir_all(parent)?; + } + + let connection_url = format!("sqlite://{}?mode=rwc", db_path); + let storage = db::Storage::init(connection_url).await?; + let watcher_pool = storage.pool_owned().await; + + // Spawn the chain scraper and get its storage + + // Spawn the payment listener task + let payment_listener_handle = tokio::spawn({ + let obs_pool = watcher_pool.clone(); + let chain_storage = run_chain_scraper(&config).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 + { + error!("Payment listener error: {}", e); + } + Ok::<_, anyhow::Error>(()) + } + }); + + // Clone pool for each task that needs it + //let background_pool = db_pool.clone(); + + let price_scraper_handle = tokio::spawn(async move { + price_scraper::run_price_scraper(&watcher_pool).await; + }); + + let shutdown_handles = http::server::start_http_api(storage.pool_owned().await, http_port) + .await + .expect("Failed to start server"); + + info!("Started HTTP server on port {}", http_port); + + // Wait for the short-lived tasks to complete + let _ = join!(price_scraper_handle, payment_listener_handle); + + // Wait for a signal to terminate the long-running task + wait_for_signal().await; + + if let Err(err) = shutdown_handles.shutdown().await { + error!("{err}"); + }; + + Ok(()) +} diff --git a/nyx-chain-watcher/src/cli/mod.rs b/nyx-chain-watcher/src/cli/mod.rs new file mode 100644 index 0000000000..d1f03551de --- /dev/null +++ b/nyx-chain-watcher/src/cli/mod.rs @@ -0,0 +1,67 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::cli::commands::{build_info, init, run}; +use crate::env::vars::*; +use crate::error::NyxChainWatcherError; +use clap::{Parser, Subcommand}; +use nym_bin_common::bin_info; +use std::sync::OnceLock; + +mod commands; + +pub const DEFAULT_NYX_CHAIN_WATCHER_ID: &str = "default-nyx-chain-watcher"; + +// Helper for passing LONG_VERSION to clap +fn pretty_build_info_static() -> &'static str { + static PRETTY_BUILD_INFORMATION: OnceLock = OnceLock::new(); + PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print()) +} + +#[derive(Parser, Debug)] +#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)] +pub(crate) struct Cli { + /// Path pointing to an env file that configures the nym-chain-watcher and overrides any preconfigured values. + #[clap( + short, + long, + env = NYX_CHAIN_WATCHER_CONFIG_ENV_FILE_ARG + )] + pub(crate) config_env_file: Option, + + /// Flag used for disabling the printed banner in tty. + #[clap( + long, + env = NYX_CHAIN_WATCHER_NO_BANNER_ARG + )] + pub(crate) no_banner: bool, + + /// Port to listen on + #[arg(long, default_value_t = 8000, env = "NYX_CHAIN_WATCHER_HTTP_PORT")] + pub(crate) http_port: u16, + + #[clap(subcommand)] + command: Commands, +} + +impl Cli { + pub(crate) async fn execute(self) -> Result<(), NyxChainWatcherError> { + match self.command { + Commands::BuildInfo(args) => build_info::execute(args), + Commands::Run(args) => run::execute(*args, self.http_port).await, + Commands::Init(args) => init::execute(args).await, + } + } +} + +#[derive(Subcommand, Debug)] +pub(crate) enum Commands { + /// Show build information of this binary + BuildInfo(build_info::Args), + + /// Start this nym-chain-watcher + Run(Box), + + /// Initialise config + Init(init::Args), +} diff --git a/nyx-chain-watcher/src/config/mod.rs b/nyx-chain-watcher/src/config/mod.rs new file mode 100644 index 0000000000..73419a2e9e --- /dev/null +++ b/nyx-chain-watcher/src/config/mod.rs @@ -0,0 +1,249 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::config::template::CONFIG_TEMPLATE; +use nym_bin_common::logging::LoggingSettings; +use nym_config::{ + must_get_home, read_config_from_toml_file, save_unformatted_config_to_file, NymConfigTemplate, + DEFAULT_CONFIG_DIR, DEFAULT_CONFIG_FILENAME, DEFAULT_DATA_DIR, NYM_DIR, +}; +use serde::{Deserialize, Serialize}; +use std::path::{Path, PathBuf}; +use tracing::{debug, error}; + +pub(crate) mod payments_watcher; +mod template; + +pub use crate::config::payments_watcher::PaymentWatcherConfig; +use crate::error::NyxChainWatcherError; + +const DEFAULT_NYM_CHAIN_WATCHER_DIR: &str = "nym-chain-watcher"; + +pub(crate) const DEFAULT_NYM_CHAIN_WATCHER_DB_FILENAME: &str = "nyx_chain_watcher.sqlite"; +pub(crate) const DEFAULT_NYM_CHAIN_SCRAPER_HISTORY_DB_FILENAME: &str = "chain_history.sqlite"; + +/// Derive default path to nym-chain-watcher's config directory. +/// It should get resolved to `$HOME/.nym/nym-chain-watcher/config` +pub fn default_config_directory() -> PathBuf { + must_get_home() + .join(NYM_DIR) + .join(DEFAULT_NYM_CHAIN_WATCHER_DIR) + .join(DEFAULT_CONFIG_DIR) +} + +/// Derive default path to nym-chain-watcher's config file. +/// It should get resolved to `$HOME/.nym/nym-chain-watcher/config/config.toml` +pub fn default_config_filepath() -> PathBuf { + default_config_directory().join(DEFAULT_CONFIG_FILENAME) +} + +pub struct ConfigBuilder { + pub config_path: PathBuf, + + pub data_dir: PathBuf, + + pub db_path: Option, + + pub chain_scraper_db_path: Option, + + pub payment_watcher_config: Option, + + pub logging: Option, +} + +impl ConfigBuilder { + pub fn new(config_path: PathBuf, data_dir: PathBuf) -> Self { + ConfigBuilder { + config_path, + data_dir, + payment_watcher_config: None, + logging: None, + db_path: None, + chain_scraper_db_path: None, + } + } + + pub fn with_db_path(mut self, db_path: String) -> Self { + self.db_path = Some(db_path); + self + } + + pub fn with_chain_scraper_db_path(mut self, chain_scraper_db_path: String) -> Self { + self.chain_scraper_db_path = Some(chain_scraper_db_path); + self + } + + #[allow(dead_code)] + pub fn with_payment_watcher_config( + mut self, + payment_watcher_config: impl Into, + ) -> Self { + self.payment_watcher_config = Some(payment_watcher_config.into()); + self + } + + #[allow(dead_code)] + pub fn with_logging(mut self, section: impl Into>) -> Self { + self.logging = section.into(); + self + } + + pub fn build(self) -> Config { + Config { + logging: self.logging.unwrap_or_default(), + save_path: Some(self.config_path), + payment_watcher_config: self.payment_watcher_config, + data_dir: self.data_dir, + db_path: self.db_path, + chain_scraper_db_path: self.chain_scraper_db_path, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Config { + // additional metadata holding on-disk location of this config file + #[serde(skip)] + pub(crate) save_path: Option, + + #[serde(skip)] + pub(crate) data_dir: PathBuf, + + #[serde(skip)] + db_path: Option, + + #[serde(skip)] + chain_scraper_db_path: Option, + + pub payment_watcher_config: Option, + + #[serde(default)] + pub logging: LoggingSettings, +} + +impl NymConfigTemplate for Config { + fn template(&self) -> &'static str { + CONFIG_TEMPLATE + } +} + +impl Config { + #[allow(unused)] + pub fn save(&self) -> Result<(), NyxChainWatcherError> { + let save_location = self.save_location(); + debug!( + "attempting to save config file to '{}'", + save_location.display() + ); + save_unformatted_config_to_file(self, &save_location).map_err(|source| { + NyxChainWatcherError::UnformattedConfigSaveFailure { + path: save_location, + source, + } + }) + } + + #[allow(unused)] + pub fn save_location(&self) -> PathBuf { + self.save_path + .clone() + .unwrap_or(self.default_save_location()) + } + + #[allow(unused)] + pub fn default_save_location(&self) -> PathBuf { + default_config_filepath() + } + + pub fn default_data_directory>( + config_path: P, + ) -> Result { + let config_path = config_path.as_ref(); + + // we got a proper path to the .toml file + let Some(config_dir) = config_path.parent() else { + error!( + "'{}' does not have a parent directory. Have you pointed to the fs root?", + config_path.display() + ); + return Err(NyxChainWatcherError::DataDirDerivationFailure); + }; + + let Some(config_dir_name) = config_dir.file_name() else { + error!( + "could not obtain parent directory name of '{}'. Have you used relative paths?", + config_path.display() + ); + return Err(NyxChainWatcherError::DataDirDerivationFailure); + }; + + if config_dir_name != DEFAULT_CONFIG_DIR { + error!( + "the parent directory of '{}' ({}) is not {DEFAULT_CONFIG_DIR}. currently this is not supported", + config_path.display(), config_dir_name.to_str().unwrap_or("UNKNOWN") + ); + return Err(NyxChainWatcherError::DataDirDerivationFailure); + } + + let Some(node_dir) = config_dir.parent() else { + error!( + "'{}' does not have a parent directory. Have you pointed to the fs root?", + config_dir.display() + ); + return Err(NyxChainWatcherError::DataDirDerivationFailure); + }; + + Ok(node_dir.join(DEFAULT_DATA_DIR)) + } + + pub fn database_path(&self) -> String { + self.db_path.clone().unwrap_or_else(|| { + let mut path = self.data_dir.clone().to_path_buf(); + path.push(DEFAULT_NYM_CHAIN_WATCHER_DB_FILENAME); + path.to_str() + .unwrap_or(DEFAULT_NYM_CHAIN_WATCHER_DB_FILENAME) + .to_string() + }) + } + + pub fn chain_scraper_database_path(&self) -> String { + self.chain_scraper_db_path.clone().unwrap_or_else(|| { + let mut path = self.data_dir.clone().to_path_buf(); + path.push(DEFAULT_NYM_CHAIN_SCRAPER_HISTORY_DB_FILENAME); + path.to_str() + .unwrap_or(DEFAULT_NYM_CHAIN_SCRAPER_HISTORY_DB_FILENAME) + .to_string() + }) + } + + // simple wrapper that reads config file and assigns path location + fn read_from_path>(path: P, data_dir: P) -> Result { + let path = path.as_ref(); + let data_dir = data_dir.as_ref(); + let mut loaded: Config = read_config_from_toml_file(path).map_err(|source| { + NyxChainWatcherError::ConfigLoadFailure { + path: path.to_path_buf(), + source, + } + })?; + loaded.data_dir = data_dir.to_path_buf(); + loaded.save_path = Some(path.to_path_buf()); + debug!("loaded config file from {}", path.display()); + Ok(loaded) + } + + #[allow(unused)] + pub fn read_from_toml_file>( + path: P, + data_dir: P, + ) -> Result { + Self::read_from_path(path, data_dir) + } + + pub fn read_from_toml_file_in_default_location() -> Result { + let config_path = default_config_filepath(); + let data_dir = Config::default_data_directory(&config_path)?; + Self::read_from_path(config_path, data_dir) + } +} diff --git a/nyx-chain-watcher/src/config/payments_watcher.rs b/nyx-chain-watcher/src/config/payments_watcher.rs new file mode 100644 index 0000000000..a335830a85 --- /dev/null +++ b/nyx-chain-watcher/src/config/payments_watcher.rs @@ -0,0 +1,23 @@ +use nym_validator_client::nyxd::AccountId; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PaymentWatcherConfig { + pub watchers: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PaymentWatcherEntry { + pub id: String, + pub description: Option, + pub webhook_url: String, + pub watch_for_transfer_recipient_accounts: Option>, + pub watch_for_chain_message_types: Option>, + pub authentication: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum HttpAuthenticationOptions { + AuthorizationBearerToken { token: String }, +} diff --git a/nyx-chain-watcher/src/config/template.rs b/nyx-chain-watcher/src/config/template.rs new file mode 100644 index 0000000000..f1dde4955c --- /dev/null +++ b/nyx-chain-watcher/src/config/template.rs @@ -0,0 +1,29 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +// While using normal toml marshalling would have been way simpler with less overhead, +// I think it's useful to have comments attached to the saved config file to explain behaviour of +// particular fields. +// Note: any changes to the template must be reflected in the appropriate structs. +pub(crate) const CONFIG_TEMPLATE: &str = r#" +# This is a TOML config file. +# For more information, see https://github.com/toml-lang/toml + +[payment_watcher_config] +{{#each payment_watcher_config.watchers }} +[[watchers]] +id={{this.id}} +description='{{this.description}}' +webhook_url='{{this.webhook_url}}' +{{/each}} + + + + +##### logging configuration options ##### + +[logging] + +# TODO + +"#; diff --git a/nyx-chain-watcher/src/env.rs b/nyx-chain-watcher/src/env.rs new file mode 100644 index 0000000000..5b66a63c92 --- /dev/null +++ b/nyx-chain-watcher/src/env.rs @@ -0,0 +1,25 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +#[allow(unused)] +pub mod vars { + pub const NYX_CHAIN_WATCHER_NO_BANNER_ARG: &str = "NYX_CHAIN_WATCHER_NO_BANNER"; + pub const NYX_CHAIN_WATCHER_CONFIG_ENV_FILE_ARG: &str = "NYX_CHAIN_WATCHER_CONFIG_ENV_FILE_ARG"; + + pub const NYX_CHAIN_WATCHER_DATABASE_PATH: &str = "NYX_CHAIN_WATCHER_DATABASE_PATH"; + pub const NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH: &str = + "NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH"; + + pub const NYXD_SCRAPER_START_HEIGHT: &str = "NYXD_SCRAPER_START_HEIGHT"; + + pub const NYX_CHAIN_WATCHER_ID_ARG: &str = "NYX_CHAIN_WATCHER_ID"; + pub const NYX_CHAIN_WATCHER_OUTPUT_ARG: &str = "NYX_CHAIN_WATCHER_OUTPUT"; + + pub const NYX_CHAIN_WATCHER_CONFIG_PATH_ARG: &str = "NYX_CHAIN_WATCHER_CONFIG"; + + pub const NYX_CHAIN_WATCHER_WATCH_ACCOUNTS: &str = "NYX_CHAIN_WATCHER_WATCH_ACCOUNTS"; + pub const NYX_CHAIN_WATCHER_WATCH_CHAIN_MESSAGE_TYPES: &str = + "NYX_CHAIN_WATCHER_WATCH_CHAIN_MESSAGE_TYPES"; + pub const NYX_CHAIN_WATCHER_WEBHOOK_URL: &str = "NYX_CHAIN_WATCHER_WEBHOOK_URL"; + pub const NYX_CHAIN_WATCHER_WEBHOOK_AUTH: &str = "NYX_CHAIN_WATCHER_WEBHOOK_AUTH"; +} diff --git a/nyx-chain-watcher/src/error.rs b/nyx-chain-watcher/src/error.rs new file mode 100644 index 0000000000..70230f50dc --- /dev/null +++ b/nyx-chain-watcher/src/error.rs @@ -0,0 +1,40 @@ +use std::io; +use std::path::PathBuf; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum NyxChainWatcherError { + // #[error("failed to save config file using path '{}'. detailed message: {source}", path.display())] + // ConfigSaveFailure { + // path: PathBuf, + // #[source] + // source: io::Error, + // }, + #[error("failed to save config file using path '{}'. detailed message: {source}", path.display())] + UnformattedConfigSaveFailure { + path: PathBuf, + #[source] + source: nym_config::error::NymConfigTomlError, + }, + + #[error("could not derive path to data directory of this nyx chain watcher")] + DataDirDerivationFailure, + + // #[error("could not derive path to config directory of this nyx chain watcher")] + // ConfigDirDerivationFailure, + #[error("failed to load config file using path '{}'. detailed message: {source}", path.display())] + ConfigLoadFailure { + path: PathBuf, + #[source] + source: io::Error, + }, + + #[error(transparent)] + FileIoFailure(#[from] io::Error), + + #[error(transparent)] + AnyhowFailure(#[from] anyhow::Error), + + #[error(transparent)] + NymConfigTomlE(#[from] nym_config::error::NymConfigTomlError), +} diff --git a/nyx-chain-watcher/src/main.rs b/nyx-chain-watcher/src/main.rs index 3eb513b889..b32649b994 100644 --- a/nyx-chain-watcher/src/main.rs +++ b/nyx-chain-watcher/src/main.rs @@ -1,90 +1,30 @@ -use chain_scraper::run_chain_scraper; -use clap::Parser; +use clap::{crate_name, crate_version, Parser}; +use nym_bin_common::logging::maybe_print_banner; use nym_network_defaults::setup_env; -use nym_task::signal::wait_for_signal; -use tokio::join; mod chain_scraper; +mod cli; +mod config; mod db; +mod env; +mod error; mod http; mod logging; +pub mod models; mod payment_listener; mod price_scraper; -#[derive(Parser, Debug)] -#[command(version, about, long_about = None)] -struct Args { - /// Port to listen on - #[arg(long, default_value_t = 8000, env = "NYX_CHAIN_WATCHER_HTTP_PORT")] - http_port: u16, - - /// Path to the environment variables file. If you don't provide one, variables for the mainnet will be used. - #[arg(short, long, default_value = None, env = "NYX_CHAIN_WATCHER_ENV_FILE")] - env_file: Option, - - /// SQLite database file path - #[arg( - long, - default_value = "nyx_chain_watcher.sqlite", - env = "DATABASE_URL" - )] - db_path: String, -} - #[tokio::main] async fn main() -> anyhow::Result<()> { + let cli = cli::Cli::parse(); + setup_env(cli.config_env_file.as_ref()); logging::setup_tracing_logger(); - let args = Args::parse(); - setup_env(args.env_file); // Defaults to mainnet if empty - - let db_path = args.db_path; - // Ensure parent directory exists - if let Some(parent) = std::path::Path::new(&db_path).parent() { - std::fs::create_dir_all(parent)?; + if !cli.no_banner { + maybe_print_banner(crate_name!(), crate_version!()); } - let connection_url = format!("sqlite://{}?mode=rwc", db_path); - let storage = db::Storage::init(connection_url).await?; - let watcher_pool = storage.pool_owned().await; - - // Spawn the chain scraper and get its storage - - // Spawn the payment listener task - let payment_listener_handle = tokio::spawn({ - let obs_pool = watcher_pool.clone(); - let chain_storage = run_chain_scraper().await?; - - async move { - if let Err(e) = payment_listener::run_payment_listener(obs_pool, chain_storage).await { - tracing::error!("Payment listener error: {}", e); - } - Ok::<_, anyhow::Error>(()) - } - }); - - // Clone pool for each task that needs it - //let background_pool = db_pool.clone(); - - let price_scraper_handle = tokio::spawn(async move { - price_scraper::run_price_scraper(&watcher_pool).await; - }); - - let shutdown_handles = http::server::start_http_api(storage.pool_owned().await, args.http_port) - .await - .expect("Failed to start server"); - - tracing::info!("Started HTTP server on port {}", args.http_port); - - // Wait for the short-lived tasks to complete - let _ = join!(price_scraper_handle, payment_listener_handle); - - // Wait for a signal to terminate the long-running task - wait_for_signal().await; - - if let Err(err) = shutdown_handles.shutdown().await { - tracing::error!("{err}"); - }; + cli.execute().await?; Ok(()) } diff --git a/nyx-chain-watcher/src/models.rs b/nyx-chain-watcher/src/models.rs new file mode 100644 index 0000000000..7972b1bdeb --- /dev/null +++ b/nyx-chain-watcher/src/models.rs @@ -0,0 +1,14 @@ +use rocket::serde::{Deserialize, Serialize}; +use schemars::JsonSchema; +use utoipa::ToSchema; + +#[derive(Serialize, Deserialize, Clone, JsonSchema, ToSchema)] +pub struct WebhookPayload { + pub transaction_hash: String, + pub message_index: u64, + pub sender_address: String, + pub receiver_address: String, + pub amount: String, + pub height: u128, + pub memo: Option, +} diff --git a/nyx-chain-watcher/src/payment_listener/mod.rs b/nyx-chain-watcher/src/payment_listener/mod.rs index bddfecbbda..e0ba3aa668 100644 --- a/nyx-chain-watcher/src/payment_listener/mod.rs +++ b/nyx-chain-watcher/src/payment_listener/mod.rs @@ -1,68 +1,130 @@ +use crate::config::PaymentWatcherConfig; use crate::db::queries; +use crate::models::WebhookPayload; +use nym_validator_client::nyxd::AccountId; use nyxd_scraper::storage::ScraperStorage; use reqwest::Client; -use serde_json::{json, Value}; +use rocket::form::validate::Contains; +use serde_json::Value; use sqlx::SqlitePool; -use std::env; +use std::str::FromStr; use tokio::time::{self, Duration}; +use tracing::{error, info}; #[derive(Debug)] struct TransferEvent { - recipient: String, - sender: String, + recipient: AccountId, + sender: AccountId, amount: String, + message_index: u64, } pub(crate) async fn run_payment_listener( + payment_watcher_config: PaymentWatcherConfig, watcher_pool: SqlitePool, chain_storage: ScraperStorage, ) -> anyhow::Result<()> { - let payment_receive_address = env::var("PAYMENT_RECEIVE_ADDRESS").map_err(|_| { - anyhow::anyhow!("Environment variable `PAYMENT_RECEIVE_ADDRESS` not defined") - })?; - let webhook_url = env::var("WEBHOOK_URL") - .map_err(|_| anyhow::anyhow!("Environment variable `WEBHOOK_URL` not defined"))?; - let client = Client::new(); + + let default_message_types = vec!["/cosmos.bank.v1beta1.MsgSend".to_string()]; + loop { - let last_checked_height = - queries::payments::get_last_checked_height(&watcher_pool).await?; - tracing::info!("Last checked height: {}", last_checked_height); + // 1. get the last height this watcher ran at + let last_checked_height = queries::payments::get_last_checked_height(&watcher_pool).await?; + info!("Last checked height: {}", last_checked_height); - let transactions = chain_storage - .get_transactions_after_height( - last_checked_height, - Some("/cosmos.bank.v1beta1.MsgSend"), - ) - .await?; + // 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); - for tx in transactions { - tracing::info!("Processing transaction: {}", tx.hash); - if let Some(raw_log) = tx.raw_log.as_deref() { - if let Some(transfer) = parse_transfer_from_raw_log(raw_log)? { - if transfer.recipient == payment_receive_address { - let amount: f64 = parse_unym_amount(&transfer.amount)?; + // 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), + ) + .await { + Ok(txs) => { + for t in txs { + transactions.push(t); + } + } + Err(e) => error!("Failed to get transactions (message_type = {message_type}) from scraper database: {e}") + } + } - queries::payments::insert_payment( - &watcher_pool, - tx.hash.clone(), - transfer.sender.clone(), - transfer.recipient.clone(), - amount, - tx.height, - tx.memo.clone(), - ) - .await?; + for tx in transactions { + info!( + "[watcher = {}] Processing transaction: {}", + watcher.id, tx.hash + ); + 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) => { + for transfer in transfer_events { + let amount: f64 = parse_unym_amount(&transfer.amount)?; - let webhook_data = json!({ - "transaction_hash": tx.hash, - "sender_address": transfer.sender, - "receiver_address": transfer.recipient, - "amount": amount, - "height": tx.height, - "memo": tx.memo, - }); - let _ = client.post(&webhook_url).json(&webhook_data).send().await; + 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(), + amount: transfer.amount, + height: tx.height as u128, + memo: tx.memo.clone(), + }; + match client + .post(&watcher.webhook_url) + .json(&webhook_data) + .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, + ), + } } } } @@ -72,39 +134,56 @@ pub(crate) async fn run_payment_listener( } } -fn parse_transfer_from_raw_log(raw_log: &str) -> anyhow::Result> { +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![]; + if let Some(events) = log_value[0]["events"].as_array() { - if let Some(transfer_event) = events.iter().find(|e| e["type"] == "transfer") { + for transfer_event in events.iter().filter(|e| e["type"] == "transfer") { if let Some(attrs) = transfer_event["attributes"].as_array() { - let mut transfer = TransferEvent { - recipient: String::new(), - sender: String::new(), - amount: String::new(), - }; + let mut recipient: Option = None; + let mut sender: Option = None; + let mut amount: Option = None; + let message_index: Option = Some(0u64); for attr in attrs { match attr["key"].as_str() { Some("recipient") => { - transfer.recipient = attr["value"].as_str().unwrap_or("").to_string() + recipient = + AccountId::from_str(attr["value"].as_str().unwrap_or("")).ok(); } Some("sender") => { - transfer.sender = attr["value"].as_str().unwrap_or("").to_string() + sender = AccountId::from_str(attr["value"].as_str().unwrap_or("")).ok(); } Some("amount") => { - transfer.amount = attr["value"].as_str().unwrap_or("").to_string() + amount = Some(attr["value"].as_str().unwrap_or("").to_string()) } + // TODO: parse message index _ => continue, } } - return Ok(Some(transfer)); + if let (Some(recipient), Some(sender), Some(amount), Some(message_index)) = + (recipient, sender, amount, message_index) + { + if watch_for_transfer_recipient_accounts.contains(&recipient) { + transfers.push(TransferEvent { + recipient, + sender, + amount, + message_index, + }); + } + } } } } - Ok(None) + Ok(transfers) } fn parse_unym_amount(amount: &str) -> anyhow::Result {