Add config, overrides and CLI

.gitignore

@@ -53,4 +53,5 @@ nym-network-monitor/topology.json
 nym-network-monitor/__pycache__
 nym-network-monitor/*.key
 
-*.sqlite
+*.sqlite
+.build

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",

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"] }

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),
+}

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<C, P>(
+    config: &C,
+    path: P,
+) -> Result<(), error::NymConfigTomlError>
+where
+    C: Serialize + ?Sized,
+    P: AsRef<Path>,
+{
+    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<C>(raw: &str) -> Result<C, TomlDeError>
 where
     C: DeserializeOwned,

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<u32>,
 }
 
 pub struct NyxdScraperBuilder {
@@ -47,10 +49,9 @@ pub struct NyxdScraperBuilder {
 }
 
 impl NyxdScraperBuilder {
-    pub async fn build_and_start(
-        self,
-        start_block: Option<u32>,
-    ) -> Result<NyxdScraper, ScraperError> {
+    pub async fn build_and_start(self) -> Result<NyxdScraper, ScraperError> {
+        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?;
         }
 

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,
         }
     }
 

nym-wallet/Cargo.lock

@@ -3163,6 +3163,7 @@ dependencies = [
  "log",
  "nym-network-defaults",
  "serde",
+ "thiserror",
  "toml 0.8.19",
  "url",
 ]

nyx-chain-watcher/Cargo.toml

@@ -1,5 +1,5 @@
 # Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
-# 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 }

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
 ```
 
+

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() {

nyx-chain-watcher/docker-compose.yml

@@ -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:

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<ScraperStorage> {
-    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<ScraperStorage> {
+    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<ScraperStorage> {
         .ok()
         .and_then(|value| value.parse::<u32>().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)
 }

nyx-chain-watcher/src/cli/commands/build_info.rs

@@ -0,0 +1,17 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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(())
+}

nyx-chain-watcher/src/cli/commands/init.rs

@@ -0,0 +1,19 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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)?)
+}

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;

nyx-chain-watcher/src/cli/commands/run/args.rs

@@ -0,0 +1,101 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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<String>,
+
+    /// (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<String>,
+
+    /// (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<Vec<AccountId>>,
+
+    /// (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<Vec<String>>,
+
+    /// (Override) The webhook to call when we find something
+    #[clap(
+        long,
+        env = NYX_CHAIN_WATCHER_WEBHOOK_URL
+    )]
+    pub webhook_url: Option<String>,
+
+    /// (Override) Optionally, authenticate with the webhook
+    #[clap(
+        long,
+        env = NYX_CHAIN_WATCHER_WEBHOOK_AUTH
+    )]
+    pub webhook_auth: Option<String>,
+}
+
+/*impl Args {
+    pub(super) fn take_mnemonic(&mut self) -> Option<Zeroizing<bip39::Mnemonic>> {
+        self.entry_gateway.mnemonic.take().map(Zeroizing::new)
+    }
+}
+
+impl Args {
+    pub(crate) fn build_config(self) -> Result<Config, NymNodeError> {
+        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
+    }
+}
+*/

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<Config, NyxChainWatcherError> {
+    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())
+}

nyx-chain-watcher/src/cli/commands/run/mod.rs

@@ -0,0 +1,82 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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(())
+}

nyx-chain-watcher/src/cli/mod.rs

@@ -0,0 +1,67 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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<String> = 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<std::path::PathBuf>,
+
+    /// 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<run::Args>),
+
+    /// Initialise config
+    Init(init::Args),
+}

nyx-chain-watcher/src/config/mod.rs

@@ -0,0 +1,249 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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<String>,
+
+    pub chain_scraper_db_path: Option<String>,
+
+    pub payment_watcher_config: Option<PaymentWatcherConfig>,
+
+    pub logging: Option<LoggingSettings>,
+}
+
+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<PaymentWatcherConfig>,
+    ) -> Self {
+        self.payment_watcher_config = Some(payment_watcher_config.into());
+        self
+    }
+
+    #[allow(dead_code)]
+    pub fn with_logging(mut self, section: impl Into<Option<LoggingSettings>>) -> 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<PathBuf>,
+
+    #[serde(skip)]
+    pub(crate) data_dir: PathBuf,
+
+    #[serde(skip)]
+    db_path: Option<String>,
+
+    #[serde(skip)]
+    chain_scraper_db_path: Option<String>,
+
+    pub payment_watcher_config: Option<PaymentWatcherConfig>,
+
+    #[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<P: AsRef<Path>>(
+        config_path: P,
+    ) -> Result<PathBuf, NyxChainWatcherError> {
+        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<P: AsRef<Path>>(path: P, data_dir: P) -> Result<Self, NyxChainWatcherError> {
+        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<P: AsRef<Path>>(
+        path: P,
+        data_dir: P,
+    ) -> Result<Self, NyxChainWatcherError> {
+        Self::read_from_path(path, data_dir)
+    }
+
+    pub fn read_from_toml_file_in_default_location() -> Result<Self, NyxChainWatcherError> {
+        let config_path = default_config_filepath();
+        let data_dir = Config::default_data_directory(&config_path)?;
+        Self::read_from_path(config_path, data_dir)
+    }
+}

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<PaymentWatcherEntry>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PaymentWatcherEntry {
+    pub id: String,
+    pub description: Option<String>,
+    pub webhook_url: String,
+    pub watch_for_transfer_recipient_accounts: Option<Vec<AccountId>>,
+    pub watch_for_chain_message_types: Option<Vec<String>>,
+    pub authentication: Option<HttpAuthenticationOptions>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum HttpAuthenticationOptions {
+    AuthorizationBearerToken { token: String },
+}

nyx-chain-watcher/src/config/template.rs

@@ -0,0 +1,29 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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
+
+"#;

nyx-chain-watcher/src/env.rs

@@ -0,0 +1,25 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// 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";
+}

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),
+}

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<String>,
-
-    /// 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(())
 }

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<String>,
+}

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<Option<TransferEvent>> {
+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![];
+
     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<AccountId> = None;
+                let mut sender: Option<AccountId> = None;
+                let mut amount: Option<String> = None;
+                let message_index: Option<u64> = 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<f64> {