feat(observability): add business payment metrics and alerts (#243)

Closes #214

Author: mofejo1

backend/Cargo.lock

@@ -40,3 +40,27 @@ groups:
           severity: critical
         annotations:
           summary: BLINKS backend Prometheus scrape target is down
+
+      - alert: PaymentSuccessRateLow
+        expr: (sum(rate(payment_transactions_total{status="completed"}[10m])) / clamp_min(sum(rate(payment_transactions_total{status="created"}[10m])), 1)) < 0.9
+        for: 10m
+        labels:
+          severity: critical
+        annotations:
+          summary: BLINKS payment success rate is below 90% over 10m
+
+      - alert: MerchantPaymentSuccessRateLow
+        expr: (sum(rate(payment_transactions_total{status="completed"}[10m])) by (merchant_id) / clamp_min(sum(rate(payment_transactions_total{status="created"}[10m])) by (merchant_id), 1)) < 0.9
+        for: 10m
+        labels:
+          severity: warning
+        annotations:
+          summary: BLINKS merchant payment success rate is below 90% over 10m
+
+      - alert: NoPaymentsCreated
+        expr: increase(payment_transactions_total{status="created"}[15m]) == 0
+        for: 15m
+        labels:
+          severity: warning
+        annotations:
+          summary: No payments created in the last 15 minutes

backend/monitoring/alerts.yml

@@ -75,6 +75,42 @@
           "expr": "db_pool_connections"
         }
       ]
+    },
+    {
+      "type": "timeseries",
+      "title": "Payment volume / sec (by currency)",
+      "targets": [
+        {
+          "expr": "sum(rate(payment_volume_total[5m])) by (currency)"
+        }
+      ]
+    },
+    {
+      "type": "stat",
+      "title": "Payment success rate (global, 10m)",
+      "targets": [
+        {
+          "expr": "100 * (sum(rate(payment_transactions_total{status=\"completed\"}[10m])) / clamp_min(sum(rate(payment_transactions_total{status=\"created\"}[10m])), 1))"
+        }
+      ]
+    },
+    {
+      "type": "timeseries",
+      "title": "Merchant payment success rate (10m)",
+      "targets": [
+        {
+          "expr": "100 * (sum(rate(payment_transactions_total{status=\"completed\"}[10m])) by (merchant_id) / clamp_min(sum(rate(payment_transactions_total{status=\"created\"}[10m])) by (merchant_id), 1))"
+        }
+      ]
+    },
+    {
+      "type": "timeseries",
+      "title": "Payment processing duration p95",
+      "targets": [
+        {
+          "expr": "histogram_quantile(0.95, sum(rate(payment_processing_duration_seconds_bucket[5m])) by (le, payment_method, status))"
+        }
+      ]
     }
   ]
 }

backend/monitoring/grafana-dashboard.json

@@ -3,7 +3,6 @@ use std::str::FromStr;
 use tokio_postgres::NoTls;
 use tokio::time::{sleep, Duration};
 use crate::service::MetricsService;
-use std::sync::Arc;
 use std::cmp;
 
 pub type DbPool = Pool;
@@ -56,10 +55,9 @@ pub fn start_db_pool_monitoring(
 ) -> tokio::task::JoinHandle<()> {
     tokio::spawn(async move {
         loop {
-            match tokio_postgres::Config::from_str(&database_url)
-                .and_then(|cfg| cfg.connect(NoTls))
-            {
-                Ok((client, connection)) => {
+            match tokio_postgres::Config::from_str(&database_url) {
+                Ok(cfg) => match cfg.connect(NoTls).await {
+                    Ok((client, connection)) => {
                     // detach connection handling
                     tokio::spawn(async move {
                         if let Err(e) = connection.await {
@@ -83,9 +81,13 @@ pub fn start_db_pool_monitoring(
                         Err(e) => tracing::warn!(error = %e, "Failed to query pg_stat_activity"),
                     }
 
-                    let _ = client.close().await;
                 }
-                Err(e) => tracing::error!(error = %e, "Failed to connect to Postgres for monitoring"),
+                    Err(e) => tracing::error!(
+                        error = %e,
+                        "Failed to connect to Postgres for monitoring"
+                    ),
+                },
+                Err(e) => tracing::error!(error = %e, "Invalid Postgres config for monitoring"),
             }
 
             sleep(Duration::from_secs(check_interval_secs)).await;
@@ -95,20 +97,19 @@ pub fn start_db_pool_monitoring(
 
 /// Health check for database connectivity.
 pub async fn health_check_db(database_url: &str) -> bool {
-    match tokio_postgres::Config::from_str(database_url)
-        .and_then(|cfg| cfg.connect(NoTls))
-        .await
-    {
-        Ok((mut client, connection)) => {
+    match tokio_postgres::Config::from_str(database_url) {
+        Ok(cfg) => match cfg.connect(NoTls).await {
+            Ok((mut client, connection)) => {
             // drive connection
             tokio::spawn(async move {
                 let _ = connection.await;
             });
 
             let res = client.query_one("SELECT 1", &[]).await.is_ok();
-            let _ = client.close().await;
             res
         }
+            Err(_) => false,
+        },
         Err(_) => false,
     }
 }

backend/src/db.rs

@@ -64,7 +64,8 @@ pub async fn json_metrics(State(services): State<Arc<ServiceContainer>>) -> Json
     let status = services.db_pool.status();
     let db_pool_size = status.size;
     // active connections = size - available (deadpool status exposes `available`)
-    let active_connections = db_pool_size.saturating_sub(status.available);
+    let available = usize::try_from(status.available).unwrap_or(0);
+    let active_connections = db_pool_size.saturating_sub(available);
     MetricsService::update_db_pool_status(db_pool_size, active_connections);
 
     let detailed = MetricsService::get_detailed_metrics();

backend/src/http/metrics.rs

@@ -4,6 +4,7 @@ use axum::{
 };
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
+use std::time::Instant;
 use uuid::Uuid;
 
 use crate::{
@@ -77,6 +78,7 @@ pub async fn create_payment(
     State(services): State<Arc<ServiceContainer>>,
     Json(request): Json<CreatePaymentRequest>,
 ) -> Result<Json<PaymentResponse>, ApiError> {
+    let start = Instant::now();
     // Get user from auth context (would need to implement proper auth extraction)
     // For now, using a placeholder address
     let from_address = "GEXAMPLE_ADDRESS".to_string();
@@ -131,6 +133,14 @@ pub async fn create_payment(
         .await;
 
     MetricsService::record_business_event("payment", "created");
+    MetricsService::record_payment_transaction(
+        &payment.merchant_id,
+        "created",
+        "api",
+        &payment.send_asset,
+        payment.send_amount,
+    );
+    MetricsService::record_payment_processing_duration("api", "created", start.elapsed().as_secs_f64());
 
     Ok(Json(PaymentResponse {
         id: Uuid::parse_str(&payment.id).unwrap_or_default(),

backend/src/http/payments.rs

@@ -1,5 +1,6 @@
 use crate::api_error::ApiError;
 use crate::config::Config;
+use crate::service::MetricsService;
 use chrono::{DateTime, Utc};
 use deadpool_postgres::Pool;
 use reqwest::Client;
@@ -282,16 +283,37 @@ impl IndexerService {
 
             // Update corresponding payment record if it exists
             if let Some(_dest) = destination {
-                let update_result = client
-                    .execute(
-                        "UPDATE payments SET tx_hash = $1, status = 'completed', updated_at = NOW()
-                         WHERE from_address = $2 AND status = 'processing'
+                // Find the most recent matching payment and update it by id to avoid UPDATE ... LIMIT
+                let payment_row = client
+                    .query_opt(
+                        "SELECT id, merchant_id, send_asset, send_amount, created_at
+                         FROM payments
+                         WHERE from_address = $1 AND status IN ('pending','processing')
+                         ORDER BY created_at DESC
                          LIMIT 1",
-                        &[&tx_hash, &source],
+                        &[&source],
                     )
                     .await?;
 
-                if update_result > 0 {
+                if let Some(payment_row) = payment_row {
+                    let payment_id: String = payment_row.get("id");
+                    let merchant_id: String = payment_row.get("merchant_id");
+                    let send_asset: String = payment_row.get("send_asset");
+                    let send_amount: i64 = payment_row.get("send_amount");
+                    let created_at: chrono::DateTime<chrono::Utc> = payment_row.get("created_at");
+
+                    let update_result = client
+                        .execute(
+                            "UPDATE payments SET tx_hash = $1, status = 'completed', updated_at = NOW()
+                             WHERE id = $2",
+                            &[&tx_hash, &payment_id],
+                        )
+                        .await?;
+
+                    if update_result == 0 {
+                        continue;
+                    }
+
                     // Mark event as processed
                     client
                         .execute(
@@ -301,6 +323,20 @@ impl IndexerService {
                         .await?;
 
                     processed += 1;
+                    let duration_secs =
+                        (chrono::Utc::now() - created_at).num_milliseconds() as f64 / 1000.0;
+                    MetricsService::record_payment_transaction(
+                        &merchant_id,
+                        "completed",
+                        "stellar_indexer",
+                        &send_asset,
+                        send_amount,
+                    );
+                    MetricsService::record_payment_processing_duration(
+                        "stellar_indexer",
+                        "completed",
+                        duration_secs.max(0.0),
+                    );
                     info!(tx_hash = %tx_hash, "Payment event processed");
                 }
             }

backend/src/service/indexer_service.rs

@@ -4,6 +4,7 @@ use prometheus::{
     register_histogram_vec, CounterVec, Encoder, Gauge, GaugeVec, HistogramVec, TextEncoder,
 };
 use serde::Serialize;
+use dashmap::DashMap;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -144,13 +145,24 @@ lazy_static! {
     )
     .expect("Can't create payment_processing_duration_seconds metric");
 
+    /// Payment transaction lifecycle counter
+    pub static ref PAYMENT_TRANSACTIONS_TOTAL: CounterVec = register_counter_vec!(
+        "payment_transactions_total",
+        "Total payment transactions by merchant, status, method and currency",
+        &["merchant_id", "status", "payment_method", "currency"]
+    )
+    .expect("Can't create payment_transactions_total metric");
+
     /// Application start time (Unix timestamp)
     static ref APP_START_TIME: AtomicU64 = AtomicU64::new(
         SystemTime::now()
             .duration_since(UNIX_EPOCH)
             .unwrap()
             .as_secs()
     );
+
+    /// In-memory counters for computing success rate gauge (process lifetime).
+    static ref PAYMENT_OUTCOME_COUNTS: DashMap<String, (u64, u64, u64)> = DashMap::new();
 }
 
 /// Metrics payload as specified in the issue
@@ -275,6 +287,7 @@ pub struct AlertPayload {
         let _ = &*PAYMENT_VOLUME_TOTAL;
         let _ = &*PAYMENT_SUCCESS_RATE;
         let _ = &*PAYMENT_PROCESSING_DURATION_SECONDS;
+        let _ = &*PAYMENT_TRANSACTIONS_TOTAL;
 
         tracing::info!("Metrics service initialized");
     }
@@ -304,6 +317,55 @@ pub struct AlertPayload {
             .observe(duration_secs);
     }
 
+    fn currency_label_from_asset(asset: &str) -> String {
+        if asset.eq_ignore_ascii_case("XLM") {
+            return "XLM".to_string();
+        }
+
+        // Expected format: CODE:ISSUER
+        asset.split(':').next().unwrap_or(asset).to_string()
+    }
+
+    /// Record a payment lifecycle event and refresh derived gauges.
+    pub fn record_payment_transaction(
+        merchant_id: &str,
+        status: &str,
+        payment_method: &str,
+        send_asset: &str,
+        amount: i64,
+    ) {
+        let currency = Self::currency_label_from_asset(send_asset);
+
+        PAYMENT_TRANSACTIONS_TOTAL
+            .with_label_values(&[merchant_id, status, payment_method, &currency])
+            .inc();
+
+        // Treat volume as "requested" unless status indicates a finalized successful payment.
+        if status == "created" || status == "completed" {
+            Self::record_payment_volume(merchant_id, &currency, amount as f64);
+        }
+
+        let mut entry = PAYMENT_OUTCOME_COUNTS
+            .entry(merchant_id.to_string())
+            .or_insert((0, 0, 0));
+
+        match status {
+            "created" => entry.value_mut().0 = entry.value().0.saturating_add(1),
+            "completed" => entry.value_mut().1 = entry.value().1.saturating_add(1),
+            "failed" => entry.value_mut().2 = entry.value().2.saturating_add(1),
+            _ => {}
+        }
+
+        let (created, completed, failed) = *entry.value();
+        let denom = created.saturating_add(failed);
+        let success_rate_percent = if denom == 0 {
+            0.0
+        } else {
+            (completed as f64 / denom as f64) * 100.0
+        };
+        Self::update_payment_success_rate(merchant_id, success_rate_percent);
+    }
+
     /// Get application uptime in seconds
     pub fn get_uptime() -> u64 {
         let start_time = APP_START_TIME.load(Ordering::Relaxed);
@@ -607,6 +669,43 @@ pub struct AlertPayload {
     }
 }
 
+#[cfg(test)]
+mod payment_metrics_tests {
+    use super::*;
+
+    #[test]
+    fn record_payment_transaction_updates_counters_and_success_rate() {
+        MetricsService::init();
+
+        MetricsService::record_payment_transaction("m1", "created", "api", "XLM", 100);
+        MetricsService::record_payment_transaction("m1", "completed", "stellar_indexer", "XLM", 100);
+
+        // Counter exists and has been incremented.
+        let families = PAYMENT_TRANSACTIONS_TOTAL.collect();
+        let total: f64 = families
+            .first()
+            .unwrap()
+            .get_metric()
+            .iter()
+            .map(|m| m.get_counter().get_value())
+            .sum();
+        assert!(total >= 2.0);
+
+        // Derived gauge is updated for merchant.
+        let gauge_families = PAYMENT_SUCCESS_RATE.collect();
+        let merchant_gauge = gauge_families
+            .first()
+            .unwrap()
+            .get_metric()
+            .iter()
+            .find(|m| m.get_label().iter().any(|l| l.get_name() == "merchant_id" && l.get_value() == "m1"))
+            .unwrap()
+            .get_gauge()
+            .get_value();
+        assert!(merchant_gauge > 0.0);
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;