Merge pull request kellymusk#146 from afurious/feat/stellar-confirmation-worker

feat: implement Stellar confirmation polling worker

Author: kelly-musk

Cargo.toml

@@ -19,6 +19,7 @@ crate-type = ["cdylib", "rlib"]
 # ---------------------------------------------------------------------------
 [features]
 default = ["database", "cache"]
+integration = []
 database = [ "dep:tokio", "dep:async-trait", "dep:uuid", "dep:chrono", "dep:serde", "dep:serde_json", "dep:tracing", "dep:tracing-subscriber", "dep:axum", "dep:tower", "dep:tower-http", "dep:regex", "dep:http", "dep:sqlx", "dep:hmac", "dep:sha2", "dep:hex", "dep:bigdecimal", "dep:rust_decimal", "dep:stellar-strkey", "dep:ed25519-dalek", "dep:stellar-xdr" ]
 cache = ["dep:redis", "dep:bb8", "dep:bb8-redis", "dep:moka", "dep:prometheus", "dep:tokio-util", "database"]
 database = [ "dep:tokio", "dep:async-trait", "dep:uuid", "dep:chrono", "dep:serde", "dep:serde_json", "dep:tracing", "dep:tracing-subscriber", "dep:axum", "dep:tower", "dep:tower-http", "dep:regex", "dep:http", "dep:sqlx", "dep:hmac", "dep:sha2", "dep:hex", "dep:bigdecimal", "dep:rust_decimal", "dep:stellar-strkey", "dep:ed25519-dalek", "dep:stellar-xdr", "dep:jsonwebtoken" ]
@@ -116,6 +117,7 @@ thiserror = "2.0.18"
 
 # --- Nigerian bank account validation ---
 nuban = "1.1.0"
+prometheus = { version = "0.13", features = ["process"] }
 csv = "1.3"
 jsonwebtoken = { version = "9.2.0", optional = true }
 serde_yaml = "0.9"

migrations/20260325000000_stellar_confirmation_worker.sql

@@ -0,0 +1,29 @@
+-- migrate:up
+-- Adds columns required by the Stellar Confirmation Polling Worker.
+
+-- stellar_tx_hash: the on-chain hash the worker polls Horizon for.
+-- stale_flagged_at: set when a transaction is stuck beyond the stale timeout.
+-- state_transitioned_at: timestamp of the last status change (audit trail).
+
+ALTER TABLE transactions
+    ADD COLUMN IF NOT EXISTS stellar_tx_hash       TEXT,
+    ADD COLUMN IF NOT EXISTS stale_flagged_at      TIMESTAMPTZ,
+    ADD COLUMN IF NOT EXISTS state_transitioned_at TIMESTAMPTZ;
+
+-- Index for the worker's primary query: active txns with a stellar hash.
+CREATE INDEX IF NOT EXISTS idx_transactions_stellar_polling
+    ON transactions (status, stellar_tx_hash)
+    WHERE status IN ('pending', 'processing') AND stellar_tx_hash IS NOT NULL;
+
+-- Index for stale-transaction detection.
+CREATE INDEX IF NOT EXISTS idx_transactions_stale_check
+    ON transactions (status, created_at)
+    WHERE status IN ('pending', 'processing');
+
+-- migrate:down
+DROP INDEX IF EXISTS idx_transactions_stale_check;
+DROP INDEX IF EXISTS idx_transactions_stellar_polling;
+ALTER TABLE transactions
+    DROP COLUMN IF EXISTS state_transitioned_at,
+    DROP COLUMN IF EXISTS stale_flagged_at,
+    DROP COLUMN IF EXISTS stellar_tx_hash;

src/main.rs

@@ -463,6 +463,41 @@ async fn main() -> anyhow::Result<()> {
         info!("Offramp processor worker disabled (OFFRAMP_PROCESSOR_ENABLED=false)");
     }
 
+    // Start Stellar Confirmation Polling Worker
+    let stellar_confirm_enabled = std::env::var("STELLAR_CONFIRM_WORKER_ENABLED")
+        .unwrap_or_else(|_| "true".to_string())
+        .to_lowercase()
+        != "false";
+    if stellar_confirm_enabled {
+        if let (Some(pool), Some(client)) = (db_pool.clone(), stellar_client.clone()) {
+            let confirm_config =
+                workers::stellar_confirmation_worker::StellarConfirmationConfig::from_env();
+            let registry = prometheus::default_registry().clone();
+            match workers::stellar_confirmation_worker::WorkerMetrics::new(&registry) {
+                Ok(metrics) => {
+                    info!(
+                        poll_interval_secs = confirm_config.poll_interval.as_secs(),
+                        confirmation_threshold = confirm_config.confirmation_threshold,
+                        stale_timeout_secs = confirm_config.stale_timeout.as_secs(),
+                        "Starting Stellar confirmation polling worker"
+                    );
+                    let worker = workers::stellar_confirmation_worker::StellarConfirmationWorker::new(
+                        pool,
+                        client,
+                        confirm_config,
+                        std::sync::Arc::new(metrics),
+                    );
+                    tokio::spawn(worker.run(worker_shutdown_rx.clone()));
+                }
+                Err(e) => {
+                    error!(error = %e, "Failed to register Prometheus metrics for Stellar confirmation worker — skipping");
+                }
+            }
+        } else {
+            info!("Skipping Stellar confirmation worker (missing db pool or stellar client)");
+        }
+    } else {
+        info!("Stellar confirmation worker disabled (STELLAR_CONFIRM_WORKER_ENABLED=false)");
     // Start Onramp Processor Worker
     let onramp_enabled = std::env::var("ONRAMP_PROCESSOR_ENABLED")
         .unwrap_or_else(|_| "true".to_string())

src/workers/mod.rs

@@ -17,6 +17,7 @@ pub struct TransactionMonitorWorker {
     pub http_client: reqwest::Client,
 pub mod maintenance;
 pub mod offramp_processor;
+pub mod stellar_confirmation_worker;
 pub mod onramp_processor;
 pub mod payment_poller;
 pub mod transaction_monitor;