added basic payment listener information to status api

Cargo.lock

@@ -7237,7 +7237,6 @@ dependencies = [
  "nym-validator-client",
  "nyxd-scraper",
  "reqwest 0.12.4",
- "rocket",
  "schemars",
  "serde",
  "serde_json",

nyx-chain-watcher/.sqlx/query-2e0fd886f2c68b105f5e609032bfb3b637383b370c24566497bcc5ea625e82db.json

@@ -1,6 +1,6 @@
 {
   "db_name": "SQLite",
-  "query": "\n                    SELECT * FROM transactions\n                    WHERE height > ? \n                    ORDER BY height ASC, message_index ASC\n                    ",
+  "query": "\n                SELECT * FROM transactions\n                WHERE height > ?\n                ORDER BY height ASC, message_index ASC\n            ",
   "describe": {
     "columns": [
       {
@@ -64,5 +64,5 @@
       true
     ]
   },
-  "hash": "7b9abf4ff422b8d7a942955dc4fba380e7d5f0127f4745705b8ac9af6c170d19"
+  "hash": "2e0fd886f2c68b105f5e609032bfb3b637383b370c24566497bcc5ea625e82db"
 }

nyx-chain-watcher/Cargo.toml

@@ -29,7 +29,6 @@ nym-node-requests = { path = "../nym-node/nym-node-requests", features = [
 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 }

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

@@ -1,4 +1,4 @@
-use crate::config::PaymentWatcherConfig;
+use crate::config::PaymentWatchersConfig;
 use crate::env::vars::{
     NYXD_SCRAPER_START_HEIGHT, NYXD_SCRAPER_UNSAFE_NUKE_DB,
     NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT,
@@ -10,7 +10,7 @@ use nyxd_scraper::{
 };
 use sqlx::SqlitePool;
 use std::fs;
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 
 pub(crate) async fn run_chain_scraper(
     config: &crate::config::Config,
@@ -73,11 +73,11 @@ pub(crate) async fn run_chain_scraper(
 
 pub struct EventScraperModule {
     db_pool: SqlitePool,
-    payment_config: PaymentWatcherConfig,
+    payment_config: PaymentWatchersConfig,
 }
 
 impl EventScraperModule {
-    pub fn new(db_pool: SqlitePool, payment_config: PaymentWatcherConfig) -> Self {
+    pub fn new(db_pool: SqlitePool, payment_config: PaymentWatchersConfig) -> Self {
         Self {
             db_pool,
             payment_config,
@@ -132,6 +132,12 @@ impl TxModule for EventScraperModule {
             return Ok(());
         }
 
+        if tx.tx.body.messages.len() > 1 {
+            error!(
+                "this transaction has more than 1 message in it - payment information will be lost"
+            );
+        }
+
         // Process each event
         for event in events {
             // Only process transfer events
@@ -157,17 +163,7 @@ impl TxModule for EventScraperModule {
                 // If we have all required fields, check if recipient is watched and store
                 if let (Some(recipient), Some(sender), Some(amount)) = (recipient, sender, amount) {
                     // Check if any watcher is watching this recipient
-                    let is_watched = self.payment_config.watchers.iter().any(|watcher| {
-                        if let Some(watched_accounts) =
-                            &watcher.watch_for_transfer_recipient_accounts
-                        {
-                            watched_accounts
-                                .iter()
-                                .any(|account| account.to_string() == recipient)
-                        } else {
-                            false
-                        }
-                    });
+                    let is_watched = self.payment_config.is_being_watched(&recipient);
 
                     if is_watched {
                         if let Err(e) = self

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

@@ -3,8 +3,8 @@
 
 use crate::cli::DEFAULT_NYX_CHAIN_WATCHER_ID;
 use crate::config::payments_watcher::HttpAuthenticationOptions::AuthorizationBearerToken;
-use crate::config::payments_watcher::PaymentWatcherEntry;
-use crate::config::{default_config_filepath, Config, ConfigBuilder, PaymentWatcherConfig};
+use crate::config::payments_watcher::PaymentWatcherConfig;
+use crate::config::{default_config_filepath, Config, ConfigBuilder, PaymentWatchersConfig};
 use crate::error::NyxChainWatcherError;
 use nym_config::save_unformatted_config_to_file;
 use nym_validator_client::nyxd::AccountId;
@@ -18,22 +18,22 @@ pub(crate) async fn execute(_args: Args) -> Result<(), NyxChainWatcherError> {
     let data_dir = Config::default_data_directory(&config_path)?;
 
     let builder = ConfigBuilder::new(config_path.clone(), data_dir).with_payment_watcher_config(
-        PaymentWatcherConfig {
-            watchers: vec![PaymentWatcherEntry {
+        PaymentWatchersConfig {
+            watchers: vec![PaymentWatcherConfig {
                 id: DEFAULT_NYX_CHAIN_WATCHER_ID.to_string(),
                 webhook_url: "https://webhook.site".to_string(),
-                watch_for_transfer_recipient_accounts: Some(vec![AccountId::from_str(
+                watch_for_transfer_recipient_accounts: vec![AccountId::from_str(
                     "n17g9a2pwwkg8m60wf59pq6mv0c2wusg9ukparkz",
                 )
-                .unwrap()]),
+                .unwrap()],
                 authentication: Some(AuthorizationBearerToken {
                     token: "1234".to_string(),
                 }),
                 description: None,
-                watch_for_chain_message_types: Some(vec![
+                watch_for_chain_message_types: vec![
                     "/cosmos.bank.v1beta1.MsgSend".to_string(),
                     "/ibc.applications.transfer.v1.MsgTransfer".to_string(),
-                ]),
+                ],
             }],
         },
     );

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

@@ -20,7 +20,7 @@ pub(crate) struct Args {
         value_delimiter = ',',
         env = NYX_CHAIN_WATCHER_WATCH_ACCOUNTS
     )]
-    pub watch_for_transfer_recipient_accounts: Option<Vec<AccountId>>,
+    pub watch_for_transfer_recipient_accounts: Vec<AccountId>,
 
     /// (Override) Watch for chain messages of these types
     #[clap(
@@ -28,7 +28,7 @@ pub(crate) struct Args {
         value_delimiter = ',',
         env = NYX_CHAIN_WATCHER_WATCH_CHAIN_MESSAGE_TYPES
     )]
-    pub watch_for_chain_message_types: Option<Vec<String>>,
+    pub watch_for_chain_message_types: Vec<String>,
 
     /// (Override) The webhook to call when we find something
     #[clap(

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

@@ -1,7 +1,7 @@
 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::config::payments_watcher::{HttpAuthenticationOptions, PaymentWatcherConfig};
+use crate::config::{default_config_filepath, Config, ConfigBuilder, PaymentWatchersConfig};
 use crate::error::NyxChainWatcherError;
 use tracing::{info, warn};
 
@@ -18,8 +18,8 @@ pub(crate) fn get_run_config(args: Args) -> Result<Config, NyxChainWatcherError>
     } = 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()
+    if args.watch_for_transfer_recipient_accounts.is_empty()
+        && args.watch_for_transfer_recipient_accounts.is_empty()
         && args.chain_watcher_db_path.is_none()
     {
         info!("Loading default config file...");
@@ -27,12 +27,12 @@ pub(crate) fn get_run_config(args: Args) -> Result<Config, NyxChainWatcherError>
     }
 
     // set default messages
-    if watch_for_chain_message_types.is_none() {
-        watch_for_chain_message_types = Some(vec!["/cosmos.bank.v1beta1.MsgSend".to_string()]);
+    if watch_for_chain_message_types.is_empty() {
+        watch_for_chain_message_types = vec!["/cosmos.bank.v1beta1.MsgSend".to_string()];
     }
 
     // warn if no accounts set
-    if watch_for_transfer_recipient_accounts.is_none() {
+    if watch_for_transfer_recipient_accounts.is_empty() {
         warn!(
             "You did not specify any accounts to watch in {}. Only chain data will be stored.",
             crate::env::vars::NYX_CHAIN_WATCHER_WATCH_ACCOUNTS
@@ -58,8 +58,8 @@ pub(crate) fn get_run_config(args: Args) -> Result<Config, NyxChainWatcherError>
         let authentication =
             webhook_auth.map(|token| HttpAuthenticationOptions::AuthorizationBearerToken { token });
 
-        let watcher_config = PaymentWatcherConfig {
-            watchers: vec![PaymentWatcherEntry {
+        let watcher_config = PaymentWatchersConfig {
+            watchers: vec![PaymentWatcherConfig {
                 id: DEFAULT_NYX_CHAIN_WATCHER_ID.to_string(),
                 description: None,
                 watch_for_transfer_recipient_accounts: watch_for_transfer_recipient_accounts

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

@@ -14,7 +14,9 @@ mod config;
 
 use crate::chain_scraper::run_chain_scraper;
 use crate::db::DbPool;
-use crate::{db, http, payment_listener, price_scraper};
+use crate::http::state::PaymentListenerState;
+use crate::payment_listener::PaymentListener;
+use crate::{db, http, price_scraper};
 pub(crate) use args::Args;
 use nym_task::signal::wait_for_signal;
 
@@ -141,6 +143,9 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa
     let scraper_pool = storage.pool_owned();
     let shutdown_pool = storage.pool_owned();
 
+    // construct shared state
+    let payment_listener_shared_state = PaymentListenerState::new();
+
     // spawn all the tasks
 
     // 1. chain scraper (note: this doesn't really spawn the full scraper on this task, but we don't want to be blocking waiting for its startup)
@@ -156,16 +161,17 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa
     // 2. payment listener
     let token = cancellation_token.clone();
     let payment_watcher_config = config.payment_watcher_config.clone();
+    let payment_listener = PaymentListener::new(
+        price_scraper_pool,
+        payment_watcher_config,
+        payment_listener_shared_state.clone(),
+    )?;
     {
         tasks.spawn(async move {
             token
                 .run_until_cancelled(async move {
-                    payment_listener::run_payment_listener(
-                        payment_watcher_config,
-                        price_scraper_pool,
-                    )
-                    .await
-                    .inspect_err(|err| error!("Payment listener error: {err}"))
+                    payment_listener.run().await;
+                    Ok(())
                 })
                 .await
         });
@@ -185,8 +191,13 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa
     }
 
     // 4. http api
-    let http_server =
-        http::server::build_http_api(storage.pool_owned(), &config, http_port).await?;
+    let http_server = http::server::build_http_api(
+        storage.pool_owned(),
+        &config,
+        http_port,
+        payment_listener_shared_state,
+    )
+    .await?;
     {
         let token = cancellation_token.clone();
         tasks.spawn(async move {

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

@@ -14,7 +14,7 @@ use tracing::{debug, error};
 pub(crate) mod payments_watcher;
 mod template;
 
-pub use crate::config::payments_watcher::PaymentWatcherConfig;
+pub use crate::config::payments_watcher::PaymentWatchersConfig;
 use crate::error::NyxChainWatcherError;
 
 const DEFAULT_NYM_CHAIN_WATCHER_DIR: &str = "nym-chain-watcher";
@@ -46,7 +46,7 @@ pub struct ConfigBuilder {
 
     pub chain_scraper_db_path: Option<String>,
 
-    pub payment_watcher_config: Option<PaymentWatcherConfig>,
+    pub payment_watcher_config: Option<PaymentWatchersConfig>,
 
     pub logging: Option<LoggingSettings>,
 }
@@ -76,7 +76,7 @@ impl ConfigBuilder {
     #[allow(dead_code)]
     pub fn with_payment_watcher_config(
         mut self,
-        payment_watcher_config: impl Into<PaymentWatcherConfig>,
+        payment_watcher_config: impl Into<PaymentWatchersConfig>,
     ) -> Self {
         self.payment_watcher_config = Some(payment_watcher_config.into());
         self
@@ -117,7 +117,7 @@ pub struct Config {
     chain_scraper_db_path: Option<String>,
 
     #[serde(default)]
-    pub payment_watcher_config: PaymentWatcherConfig,
+    pub payment_watcher_config: PaymentWatchersConfig,
 
     #[serde(default)]
     pub logging: LoggingSettings,

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

@@ -2,27 +2,28 @@ use nym_validator_client::nyxd::AccountId;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
-pub struct PaymentWatcherConfig {
-    pub watchers: Vec<PaymentWatcherEntry>,
+pub struct PaymentWatchersConfig {
+    pub watchers: Vec<PaymentWatcherConfig>,
 }
 
-impl PaymentWatcherConfig {
-    pub fn watched_transfer_accounts(&self) -> Vec<&AccountId> {
-        self.watchers
-            .iter()
-            .filter_map(|e| e.watch_for_transfer_recipient_accounts.as_ref())
-            .flat_map(|a| a)
-            .collect()
+impl PaymentWatchersConfig {
+    pub fn is_being_watched(&self, account: &str) -> bool {
+        self.watchers.iter().any(|watcher| {
+            watcher
+                .watch_for_transfer_recipient_accounts
+                .iter()
+                .any(|acc| acc.as_ref() == account)
+        })
     }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PaymentWatcherEntry {
+pub struct PaymentWatcherConfig {
     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 watch_for_transfer_recipient_accounts: Vec<AccountId>,
+    pub watch_for_chain_message_types: Vec<String>,
     pub authentication: Option<HttpAuthenticationOptions>,
 }
 

nyx-chain-watcher/src/db/models.rs

@@ -1,4 +1,8 @@
+use anyhow::Context;
+use nym_validator_client::nyxd::Coin;
 use serde::{Deserialize, Serialize};
+use sqlx::FromRow;
+use time::OffsetDateTime;
 use utoipa::ToSchema;
 
 #[derive(Clone, Deserialize, Debug, ToSchema)]
@@ -41,3 +45,25 @@ pub(crate) struct PaymentRecord {
     pub(crate) timestamp: i64,
     pub(crate) height: i64,
 }
+
+#[derive(Serialize, Deserialize, Debug, FromRow)]
+pub(crate) struct Transaction {
+    pub(crate) id: i64,
+    pub(crate) tx_hash: String,
+    pub(crate) height: i64,
+    pub(crate) message_index: i64,
+    pub(crate) sender: String,
+    pub(crate) recipient: String,
+    pub(crate) amount: String,
+    pub(crate) memo: Option<String>,
+    pub(crate) created_at: Option<OffsetDateTime>,
+}
+
+impl Transaction {
+    pub(crate) fn funds(&self) -> anyhow::Result<Coin> {
+        self.amount
+            .as_str()
+            .parse()
+            .context("failed to parse transaction amount")
+    }
+}

nyx-chain-watcher/src/helpers.rs

@@ -0,0 +1,55 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use serde::{Deserialize, Serialize};
+use std::collections::vec_deque::{IntoIter, Iter};
+use std::collections::VecDeque;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct RingBuffer<T> {
+    #[serde(flatten)]
+    inner: VecDeque<T>,
+}
+
+impl<T> RingBuffer<T> {
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            inner: VecDeque::with_capacity(capacity),
+        }
+    }
+
+    pub fn push(&mut self, item: T) {
+        if self.inner.len() == self.inner.capacity() {
+            self.inner.pop_front();
+            self.inner.push_back(item);
+            debug_assert!(self.inner.len() == self.inner.capacity());
+        } else {
+            self.inner.push_back(item);
+        }
+    }
+
+    pub fn iter(&self) -> Iter<'_, T> {
+        self.inner.iter()
+    }
+}
+
+impl<T> From<RingBuffer<T>> for VecDeque<T> {
+    fn from(value: RingBuffer<T>) -> Self {
+        value.inner
+    }
+}
+
+impl<T> From<RingBuffer<T>> for Vec<T> {
+    fn from(value: RingBuffer<T>) -> Self {
+        value.inner.into()
+    }
+}
+
+impl<T> IntoIterator for RingBuffer<T> {
+    type Item = T;
+    type IntoIter = IntoIter<T>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.inner.into_iter()
+    }
+}

nyx-chain-watcher/src/http/api/mod.rs

@@ -8,6 +8,7 @@ use utoipa_swagger_ui::SwaggerUi;
 use crate::http::{api_docs, server::HttpServer, state::AppState};
 
 pub(crate) mod price;
+pub(crate) mod status;
 pub(crate) mod watcher;
 
 pub(crate) struct RouterBuilder {
@@ -25,8 +26,13 @@ impl RouterBuilder {
                 "/",
                 axum::routing::get(|| async { Redirect::permanent("/swagger") }),
             )
-            .nest("/v1", Router::new().nest("/price", price::routes()))
-            .nest("/v1", Router::new().nest("/watcher", watcher::routes()));
+            .nest(
+                "/v1",
+                Router::new()
+                    .nest("/status", status::routes())
+                    .nest("/price", price::routes())
+                    .nest("/watcher", watcher::routes()),
+            );
 
         Self {
             unfinished_router: router,

nyx-chain-watcher/src/http/api/status.rs

@@ -0,0 +1,98 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::http::models::status::{
+    ActivePaymentWatchersResponse, PaymentListenerFailureDetails, PaymentListenerStatusResponse,
+    ProcessedPayment, WatcherFailureDetails, WatcherState,
+};
+use crate::http::state::{AppState, PaymentListenerState};
+use axum::extract::State;
+use axum::routing::get;
+use axum::{Json, Router};
+use std::ops::Deref;
+
+pub(crate) fn routes() -> Router<AppState> {
+    Router::new()
+        .route("/active-payment-watchers", get(active_payment_watchers))
+        .route("/payment-listener", get(payment_listener_status))
+}
+
+#[utoipa::path(
+    tag = "Status",
+    get,
+    path = "/active-payment-watchers",
+    context_path = "/v1/status",
+    responses(
+        (status = 200, body = ActivePaymentWatchersResponse)
+    )
+)]
+pub(crate) async fn active_payment_watchers(
+    State(state): State<AppState>,
+) -> Json<ActivePaymentWatchersResponse> {
+    Json(ActivePaymentWatchersResponse {
+        watchers: state.registered_payment_watchers.deref().clone(),
+    })
+}
+
+#[utoipa::path(
+    tag = "Status",
+    get,
+    path = "/payment-listener",
+    context_path = "/v1/status",
+    responses(
+        (status = 200, body = PaymentListenerStatusResponse)
+    )
+)]
+pub(crate) async fn payment_listener_status(
+    State(state): State<PaymentListenerState>,
+) -> Json<PaymentListenerStatusResponse> {
+    let guard = state.inner.read().await;
+
+    // sorry for the nasty conversion code here, run out of time : (
+    Json(PaymentListenerStatusResponse {
+        last_checked: guard.last_checked,
+        processed_payments_since_startup: guard.processed_payments_since_startup,
+        watcher_errors_since_startup: guard.watcher_errors_since_startup,
+        payment_listener_errors_since_startup: guard.payment_listener_errors_since_startup,
+        last_processed_payment: guard
+            .last_processed_payment
+            .as_ref()
+            .map(|p| ProcessedPayment {
+                processed_at: p.processed_at,
+                tx_hash: p.tx_hash.clone(),
+                message_index: p.message_index,
+                height: p.height,
+                sender: p.sender.clone(),
+                receiver: p.receiver.clone(),
+                funds: p.funds.clone(),
+                memo: p.memo.clone(),
+            }),
+        latest_failures: guard
+            .latest_failures
+            .iter()
+            .map(|f| PaymentListenerFailureDetails {
+                timestamp: f.timestamp,
+                error: f.error.clone(),
+            })
+            .collect(),
+        watchers: guard
+            .watchers
+            .iter()
+            .map(|(w, state)| {
+                (
+                    w.clone(),
+                    WatcherState {
+                        latest_failures: state
+                            .latest_failures
+                            .iter()
+                            .map(|f| WatcherFailureDetails {
+                                timestamp: f.timestamp,
+                                error: f.error.clone(),
+                            })
+                            .collect(),
+                    },
+                )
+            })
+            .collect(),
+    })
+}

nyx-chain-watcher/src/http/api/watcher.rs

@@ -18,5 +18,5 @@ pub(crate) fn routes() -> Router<AppState> {
 
 /// Fetch the addresses being watched by the chain watcher
 async fn get_addresses(State(state): State<AppState>) -> HttpResult<Json<Vec<String>>> {
-    Ok(Json(state.watched_addresses.clone()))
+    Ok(Json(state.watched_accounts()))
 }

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

@@ -1,5 +1,6 @@
 pub(crate) mod api;
 pub(crate) mod api_docs;
 pub(crate) mod error;
+pub(crate) mod models;
 pub(crate) mod server;
 pub(crate) mod state;

nyx-chain-watcher/src/http/models.rs

@@ -0,0 +1,92 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+// if we ever create some sort of chain watcher client, those would need to be extracted
+
+pub mod status {
+    use crate::config::payments_watcher::PaymentWatcherConfig;
+    use crate::models::openapi_schema;
+    use nym_validator_client::nyxd::Coin;
+    use serde::{Deserialize, Serialize};
+    use std::collections::HashMap;
+    use time::OffsetDateTime;
+    use utoipa::ToSchema;
+
+    #[derive(Debug, Serialize, Deserialize, ToSchema)]
+    pub struct ActivePaymentWatchersResponse {
+        pub watchers: Vec<PaymentWatcher>,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    pub struct PaymentWatcher {
+        pub id: String,
+        pub description: String,
+        pub webhook_url: String,
+        pub watched_accounts: Vec<String>,
+        pub watched_message_types: Vec<String>,
+    }
+
+    impl From<&PaymentWatcherConfig> for PaymentWatcher {
+        fn from(value: &PaymentWatcherConfig) -> Self {
+            PaymentWatcher {
+                id: value.id.clone(),
+                description: value.description.clone().unwrap_or_default(),
+                webhook_url: value.webhook_url.clone(),
+                watched_accounts: value
+                    .watch_for_transfer_recipient_accounts
+                    .iter()
+                    .map(|a| a.to_string())
+                    .collect(),
+                watched_message_types: value.watch_for_chain_message_types.clone(),
+            }
+        }
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    pub struct PaymentListenerStatusResponse {
+        #[serde(with = "time::serde::rfc3339")]
+        pub last_checked: OffsetDateTime,
+
+        pub processed_payments_since_startup: u64,
+        pub watcher_errors_since_startup: u64,
+        pub payment_listener_errors_since_startup: u64,
+
+        pub last_processed_payment: Option<ProcessedPayment>,
+
+        pub latest_failures: Vec<PaymentListenerFailureDetails>,
+        pub watchers: HashMap<String, WatcherState>,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    pub(crate) struct ProcessedPayment {
+        pub processed_at: OffsetDateTime,
+
+        pub tx_hash: String,
+        pub message_index: u64,
+        pub height: u64,
+        pub sender: String,
+        pub receiver: String,
+
+        #[schema(value_type = openapi_schema::Coin)]
+        pub funds: Coin,
+
+        pub memo: String,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    pub(crate) struct PaymentListenerFailureDetails {
+        pub(crate) timestamp: OffsetDateTime,
+        pub(crate) error: String,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    pub(crate) struct WatcherState {
+        pub(crate) latest_failures: Vec<WatcherFailureDetails>,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+    pub(crate) struct WatcherFailureDetails {
+        pub(crate) timestamp: OffsetDateTime,
+        pub(crate) error: String,
+    }
+}

nyx-chain-watcher/src/http/server.rs

@@ -4,6 +4,7 @@ use tokio::net::TcpListener;
 use tokio_util::sync::WaitForCancellationFutureOwned;
 
 use crate::config::Config;
+use crate::http::state::PaymentListenerState;
 use crate::{
     db::DbPool,
     http::{api::RouterBuilder, state::AppState},
@@ -13,17 +14,20 @@ pub(crate) async fn build_http_api(
     db_pool: DbPool,
     config: &Config,
     http_port: u16,
+    payment_listener_state: PaymentListenerState,
 ) -> anyhow::Result<HttpServer> {
     let router_builder = RouterBuilder::with_default_routes();
 
-    let watched_accounts = config
-        .payment_watcher_config
-        .watched_transfer_accounts()
-        .iter()
-        .map(|a| a.to_string())
-        .collect();
-
-    let state = AppState::new(db_pool, watched_accounts);
+    let state = AppState::new(
+        db_pool,
+        config
+            .payment_watcher_config
+            .watchers
+            .iter()
+            .map(Into::into)
+            .collect(),
+        payment_listener_state,
+    );
     let router = router_builder.with_state(state);
 
     let bind_addr = format!("0.0.0.0:{}", http_port);

nyx-chain-watcher/src/http/state.rs

@@ -1,20 +1,183 @@
 use crate::db::DbPool;
+use crate::helpers::RingBuffer;
+use crate::http::models::status::PaymentWatcher;
+use crate::models::WebhookPayload;
+use axum::extract::FromRef;
+use nym_validator_client::nyxd::Coin;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::sync::Arc;
+use time::OffsetDateTime;
+use tokio::sync::RwLock;
 
 #[derive(Debug, Clone)]
 pub(crate) struct AppState {
     db_pool: DbPool,
-    pub(crate) watched_addresses: Vec<String>,
+    pub(crate) registered_payment_watchers: Arc<Vec<PaymentWatcher>>,
+    pub(crate) payment_listener_state: PaymentListenerState,
 }
 
 impl AppState {
-    pub(crate) fn new(db_pool: DbPool, watched_addresses: Vec<String>) -> Self {
+    pub(crate) fn new(
+        db_pool: DbPool,
+        registered_payment_watchers: Vec<PaymentWatcher>,
+        payment_listener_state: PaymentListenerState,
+    ) -> Self {
         Self {
             db_pool,
-            watched_addresses,
+            registered_payment_watchers: Arc::new(registered_payment_watchers),
+            payment_listener_state,
         }
     }
 
     pub(crate) fn db_pool(&self) -> &DbPool {
         &self.db_pool
     }
+
+    pub(crate) fn watched_accounts(&self) -> Vec<String> {
+        self.registered_payment_watchers
+            .iter()
+            .flat_map(|w| w.watched_accounts.iter())
+            .cloned()
+            .collect()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct PaymentListenerState {
+    pub(crate) inner: Arc<RwLock<PaymentListenerStateInner>>,
+}
+
+impl PaymentListenerState {
+    // TODO: make those configurable
+    const MAX_WATCHER_FAILURES: usize = 20;
+    const MAX_PAYMENT_LISTENER_FAILURES: usize = 50;
+
+    pub(crate) fn new() -> Self {
+        PaymentListenerState {
+            inner: Arc::new(RwLock::new(PaymentListenerStateInner {
+                last_checked: OffsetDateTime::UNIX_EPOCH,
+                processed_payments_since_startup: 0,
+                watcher_errors_since_startup: 0,
+                payment_listener_errors_since_startup: 0,
+                last_processed_payment: None,
+                latest_failures: RingBuffer::new(Self::MAX_PAYMENT_LISTENER_FAILURES),
+                watchers: Default::default(),
+            })),
+        }
+    }
+
+    pub(crate) async fn insert_listener_failure(&self, failure: PaymentListenerFailureDetails) {
+        let mut guard = self.inner.write().await;
+
+        guard.payment_listener_errors_since_startup += 1;
+        guard.latest_failures.push(failure);
+    }
+
+    pub(crate) async fn insert_watcher_failure(&self, id: &str, failure: WatcherFailureDetails) {
+        self.inner
+            .write()
+            .await
+            .watchers
+            .entry(id.to_string())
+            .or_insert(WatcherState {
+                latest_failures: RingBuffer::new(Self::MAX_WATCHER_FAILURES),
+            })
+            .latest_failures
+            .push(failure);
+    }
+
+    pub(crate) async fn processed_payment_transaction(&self, payment: ProcessedPayment) {
+        let mut guard = self.inner.write().await;
+
+        guard.processed_payments_since_startup += 1;
+        guard.last_processed_payment = Some(payment)
+    }
+
+    pub(crate) async fn update_last_checked(&self) {
+        self.inner.write().await.last_checked = OffsetDateTime::now_utc();
+    }
+}
+
+impl FromRef<AppState> for PaymentListenerState {
+    fn from_ref(input: &AppState) -> Self {
+        input.payment_listener_state.clone()
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct PaymentListenerStateInner {
+    pub(crate) last_checked: OffsetDateTime,
+
+    pub(crate) processed_payments_since_startup: u64,
+    pub(crate) watcher_errors_since_startup: u64,
+    pub(crate) payment_listener_errors_since_startup: u64,
+
+    pub(crate) last_processed_payment: Option<ProcessedPayment>,
+
+    pub(crate) latest_failures: RingBuffer<PaymentListenerFailureDetails>,
+    pub(crate) watchers: HashMap<String, WatcherState>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub(crate) struct ProcessedPayment {
+    pub processed_at: OffsetDateTime,
+
+    pub tx_hash: String,
+    pub message_index: u64,
+    pub height: u64,
+    pub sender: String,
+    pub receiver: String,
+    pub funds: Coin,
+    pub memo: String,
+}
+
+impl From<WebhookPayload> for ProcessedPayment {
+    fn from(payload: WebhookPayload) -> Self {
+        ProcessedPayment {
+            processed_at: OffsetDateTime::now_utc(),
+            tx_hash: payload.transaction_hash,
+            message_index: payload.message_index,
+            height: payload.height as u64,
+            sender: payload.sender_address,
+            receiver: payload.receiver_address,
+            funds: payload.funds.into(),
+            memo: payload.memo.unwrap_or_default(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct PaymentListenerFailureDetails {
+    pub(crate) timestamp: OffsetDateTime,
+    pub(crate) error: String,
+}
+
+impl PaymentListenerFailureDetails {
+    pub(crate) fn new(error: String) -> Self {
+        PaymentListenerFailureDetails {
+            timestamp: OffsetDateTime::now_utc(),
+            error,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct WatcherState {
+    pub(crate) latest_failures: RingBuffer<WatcherFailureDetails>,
+}
+
+#[derive(Debug)]
+pub(crate) struct WatcherFailureDetails {
+    pub(crate) timestamp: OffsetDateTime,
+    pub(crate) error: String,
+}
+
+impl WatcherFailureDetails {
+    pub(crate) fn new(error: String) -> Self {
+        WatcherFailureDetails {
+            timestamp: OffsetDateTime::now_utc(),
+            error,
+        }
+    }
 }

nyx-chain-watcher/src/main.rs

@@ -10,6 +10,7 @@ mod config;
 mod db;
 mod env;
 mod error;
+pub(crate) mod helpers;
 mod http;
 mod logging;
 pub mod models;

nyx-chain-watcher/src/models.rs

@@ -1,6 +1,6 @@
 use nym_validator_client::nyxd::CosmWasmCoin;
-use rocket::serde::{Deserialize, Serialize};
 use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
 #[derive(Serialize, Deserialize, Clone, JsonSchema, ToSchema)]

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

@@ -1,108 +1,151 @@
-use crate::config::payments_watcher::HttpAuthenticationOptions;
-use crate::config::PaymentWatcherConfig;
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::config::PaymentWatchersConfig;
+use crate::db::models::Transaction;
 use crate::db::queries;
+use crate::http::state::{
+    PaymentListenerFailureDetails, PaymentListenerState, ProcessedPayment, WatcherFailureDetails,
+};
 use crate::models::WebhookPayload;
-use nym_validator_client::nyxd::Coin;
-use reqwest::Client;
+use crate::payment_listener::watcher::PaymentWatcher;
+use anyhow::Context;
 use sqlx::SqlitePool;
-use std::str::FromStr;
 use tokio::time::{self, Duration};
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
-pub(crate) async fn run_payment_listener(
-    payment_watcher_config: PaymentWatcherConfig,
-    watcher_pool: SqlitePool,
-) -> anyhow::Result<()> {
-    let client = Client::new();
+pub(crate) mod watcher;
 
-    loop {
-        // 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);
+pub(crate) struct PaymentListener {
+    db_pool: SqlitePool,
+    payment_watchers: Vec<PaymentWatcher>,
+    shared_state: PaymentListenerState,
+}
 
-        // 2. iterate through watchers
-        for watcher in &payment_watcher_config.watchers {
-            if watcher.watch_for_transfer_recipient_accounts.is_some() {
-                // 3. Query new transactions for this watcher's recipient accounts
-                let transactions = sqlx::query!(
-                    r#"
-                    SELECT * FROM transactions
-                    WHERE height > ? 
-                    ORDER BY height ASC, message_index ASC
-                    "#,
-                    last_checked_height
-                )
-                .fetch_all(&watcher_pool)
-                .await?;
+impl PaymentListener {
+    pub(crate) fn new(
+        db_pool: SqlitePool,
+        config: PaymentWatchersConfig,
+        shared_state: PaymentListenerState,
+    ) -> anyhow::Result<Self> {
+        Ok(PaymentListener {
+            db_pool,
+            payment_watchers: config
+                .watchers
+                .iter()
+                .map(|watcher_cfg| PaymentWatcher::new(watcher_cfg.clone()))
+                .collect::<anyhow::Result<Vec<_>>>()?,
+            shared_state,
+        })
+    }
 
-                if !transactions.is_empty() {
-                    info!(
-                        "[watcher = {}] Processing {} transactions",
-                        watcher.id,
-                        transactions.len()
-                    );
-                }
+    async fn retrieve_unprocessed_transactions(&self) -> anyhow::Result<Vec<Transaction>> {
+        let last_checked_height = queries::payments::get_last_checked_height(&self.db_pool).await?;
+        let txs = sqlx::query_as!(
+            Transaction,
+            r#"
+                SELECT * FROM transactions
+                WHERE height > ?
+                ORDER BY height ASC, message_index ASC
+            "#,
+            last_checked_height
+        )
+        .fetch_all(&self.db_pool)
+        .await?;
 
-                for tx in transactions {
-                    let funds = Coin::from_str(&tx.amount)?;
-                    let amount: f64 = funds.amount as f64 / 1e6f64; // convert to major value, there will be precision loss
+        Ok(txs)
+    }
 
-                    // Store transaction hash for later use
-                    let tx_hash = tx.tx_hash.clone();
-                    let message_index = tx.message_index;
+    async fn process_transaction(&self, tx: Transaction) -> anyhow::Result<()> {
+        // 3.1 process any payments
+        let funds = tx.funds()?;
+        let amount: f64 = funds.amount as f64 / 1e6f64; // convert to major value, there will be precision loss
 
-                    queries::payments::insert_payment(
-                        &watcher_pool,
-                        tx.tx_hash,
-                        tx.sender.clone(),
-                        tx.recipient.clone(),
-                        amount,
-                        tx.height,
-                        tx.memo.clone(),
+        // TODO: FIXME: it may happen that we insert a payment but fail to invoke all webhooks
+
+        queries::payments::insert_payment(
+            &self.db_pool,
+            tx.tx_hash.clone(),
+            tx.sender.clone(),
+            tx.recipient.clone(),
+            amount,
+            tx.height,
+            tx.memo.clone(),
+        )
+        .await?;
+
+        // 3.1. invoke all relevant webhooks for all registered watchers
+        let webhook_data = WebhookPayload {
+            transaction_hash: tx.tx_hash,
+            message_index: tx.message_index as u64,
+            sender_address: tx.sender,
+            receiver_address: tx.recipient,
+            funds: funds.into(),
+            height: tx.height as u128,
+            memo: tx.memo,
+        };
+
+        for watcher in &self.payment_watchers {
+            if let Err(err) = watcher.invoke_webhook(&webhook_data).await {
+                error!("watcher {} failure: {err:#}", watcher.id());
+                self.shared_state
+                    .insert_watcher_failure(
+                        watcher.id(),
+                        WatcherFailureDetails::new(err.to_string()),
                     )
-                    .await?;
-
-                    let webhook_data = WebhookPayload {
-                        transaction_hash: tx_hash.clone(),
-                        message_index: message_index as u64,
-                        sender_address: tx.sender,
-                        receiver_address: tx.recipient,
-                        funds: funds.into(),
-                        height: tx.height as u128,
-                        memo: tx.memo,
-                    };
-
-                    let mut request_builder = client.post(&watcher.webhook_url).json(&webhook_data);
-
-                    if let Some(auth) = &watcher.authentication {
-                        match auth {
-                            HttpAuthenticationOptions::AuthorizationBearerToken { token } => {
-                                request_builder = request_builder.bearer_auth(token);
-                            }
-                        }
-                    }
-
-                    match request_builder.send().await {
-                        Ok(res) => info!(
-                            "[watcher = {}] ✅ Webhook {} {} - tx {}, index {}",
-                            watcher.id,
-                            res.status(),
-                            res.url(),
-                            tx_hash,
-                            message_index,
-                        ),
-                        Err(e) => error!(
-                            "[watcher = {}] ❌ Webhook {:?} {:?} error = {}",
-                            watcher.id,
-                            e.status(),
-                            e.url(),
-                            e,
-                        ),
-                    }
-                }
+                    .await
             }
         }
 
-        time::sleep(Duration::from_secs(10)).await;
+        self.shared_state
+            .processed_payment_transaction(ProcessedPayment::from(webhook_data))
+            .await;
+
+        Ok(())
+    }
+
+    async fn check_for_unprocessed_payments(&self) -> anyhow::Result<()> {
+        // 1. retrieve any unprocessed transactions
+        let unprocessed_transactions = self
+            .retrieve_unprocessed_transactions()
+            .await
+            .context("failed to retrieve unprocessed transactions")?;
+
+        if unprocessed_transactions.is_empty() {
+            debug!("no payment transactions to process.");
+            return Ok(());
+        } else {
+            info!(
+                "processing {} payment transactions",
+                unprocessed_transactions.len()
+            );
+        }
+
+        // 2. attempt to process them
+        for tx in unprocessed_transactions {
+            let hash = tx.tx_hash.clone();
+            let height = tx.height;
+            self.process_transaction(tx).await.with_context(|| {
+                format!("failed to process transaction {hash} at height {height}")
+            })?;
+        }
+
+        Ok(())
+    }
+
+    pub(crate) async fn run(&self) {
+        loop {
+            time::sleep(Duration::from_secs(10)).await;
+
+            if let Err(err) = self.check_for_unprocessed_payments().await {
+                error!("failed to fully process payments: {err:#}");
+                self.shared_state
+                    .insert_listener_failure(PaymentListenerFailureDetails::new(err.to_string()))
+                    .await;
+                continue;
+            }
+
+            self.shared_state.update_last_checked().await;
+        }
     }
 }

nyx-chain-watcher/src/payment_listener/watcher.rs

@@ -0,0 +1,73 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::config::payments_watcher::{HttpAuthenticationOptions, PaymentWatcherConfig};
+use crate::models::WebhookPayload;
+use anyhow::Context;
+use reqwest::{Client, Url};
+use tracing::{error, info};
+
+pub(crate) struct PaymentWatcher {
+    webhook_url: Url,
+    config: PaymentWatcherConfig,
+}
+
+impl PaymentWatcher {
+    pub(crate) fn new(config: PaymentWatcherConfig) -> anyhow::Result<Self> {
+        Ok(PaymentWatcher {
+            webhook_url: config
+                .webhook_url
+                .as_str()
+                .parse()
+                .context("couldn't create payment watcher: provided webhook URL is malformed")?,
+            config,
+        })
+    }
+
+    pub(super) fn id(&self) -> &str {
+        &self.config.id
+    }
+
+    pub(crate) async fn invoke_webhook(&self, payload: &WebhookPayload) -> anyhow::Result<()> {
+        let client = Client::builder()
+            .user_agent(format!(
+                "nyx-chain-watcher/{}/watcher-{}",
+                env!("CARGO_PKG_VERSION"),
+                self.config.id
+            ))
+            .build()
+            .context("failed to build reqwest client")?;
+
+        let mut request_builder = client.post(self.webhook_url.clone()).json(payload);
+
+        if let Some(auth) = &self.config.authentication {
+            match auth {
+                HttpAuthenticationOptions::AuthorizationBearerToken { token } => {
+                    request_builder = request_builder.bearer_auth(token);
+                }
+            }
+        }
+
+        match request_builder.send().await {
+            Ok(res) => info!(
+                "[watcher = {}] ✅ Webhook {} {} - tx {}, index {}",
+                self.config.id,
+                res.status(),
+                res.url(),
+                payload.transaction_hash,
+                payload.message_index,
+            ),
+            Err(err) => {
+                error!(
+                    "[watcher = {}] ❌ Webhook {:?} {:?} error = {err}",
+                    self.config.id,
+                    err.status(),
+                    err.url(),
+                );
+                return Err(err.into());
+            }
+        }
+
+        Ok(())
+    }
+}