feat: add webhook retries with exponential backoff

.cspell.json

@@ -78,10 +78,11 @@
     "scrape",
     "codebase",
     "postgres",
+    "psql",
     "PostgreSQL",
     "postgresql",
     "gitignore",
     "monorepo",
     "repo"
   ]
-}
+}

payment_router/src/lib.rs

@@ -1132,7 +1132,7 @@ mod test {
     use soroban_sdk::{
         testutils::{Address as _, Events, Ledger as _, LedgerInfo},
         token::StellarAssetClient,
-        vec, Address, Env, Symbol, TryIntoVal,
+        Address, Env, Symbol, TryIntoVal,
     };
 
     /// Returns (env, client, contract_id).
@@ -2251,188 +2251,6 @@ mod test {
         client.set_fee_bps(&200);
         assert_eq!(client.get_fee(), 200);
     }
-
-    /// `add_supported_token` is a no-op and never errors.
-    #[test]
-    fn test_add_supported_token_noop() {
-        let (env, client, _) = setup_env();
-        let admin = Address::generate(&env);
-        let treasury = Address::generate(&env);
-        let (token_address, _tc, _sac) = setup_token(&env);
-
-        client.initialize(&admin, &treasury, &100, &1_000, &PaymentRouter::MAX_AMOUNT);
-        // Should not panic or error
-        client.add_supported_token(&token_address);
-    }
-
-    /// `set_fee_config_legacy` updates both fee_bps and fee_cap.
-    #[test]
-    fn test_set_fee_config_legacy() {
-        let (env, client, _) = setup_env();
-        let admin = Address::generate(&env);
-        let treasury = Address::generate(&env);
-        let sender = Address::generate(&env);
-        let recipient = Address::generate(&env);
-        let (token_address, token_client, sac) = setup_token(&env);
-        sac.mint(&sender, &10_000);
-
-        client.initialize(&admin, &treasury, &100, &50, &PaymentRouter::MAX_AMOUNT);
-
-        // Update to 200 bps with a higher cap
-        client.set_fee_config_legacy(&200, &500);
-        assert_eq!(client.get_fee(), 200);
-
-        // Route and verify new fee applies: 200 bps of 1_000 = 20
-        client.route_payment(&sender, &recipient, &token_address, &1_000);
-        assert_eq!(token_client.balance(&treasury), 20);
-        assert_eq!(token_client.balance(&recipient), 980);
-    }
-
-    /// `get_effective_fee_bps` returns 0 when the contract is not initialized.
-    #[test]
-    fn test_get_effective_fee_bps_uninitialized() {
-        let (env, client, _) = setup_env();
-        let sender = Address::generate(&env);
-        // No storage entry for FeeBps — should return 0
-        assert_eq!(client.get_effective_fee_bps(&sender), 0);
-    }
-
-    /// `get_user_volume` returns 0 for a user who has never sent a payment.
-    #[test]
-    fn test_get_user_volume_no_history() {
-        let (env, client, _) = setup_env();
-
-        let admin = Address::generate(&env);
-        let treasury = Address::generate(&env);
-        let sender = Address::generate(&env);
-        let recipient = Address::generate(&env);
-
-        let (token_address, token_client, _token_admin_client) = setup_token(&env);
-
-        let limit = 10_000_000_000_000i128;
-        let sac = soroban_sdk::token::StellarAssetClient::new(&env, &token_address);
-        sac.mint(&sender, &(limit + 2000));
-
-        client.initialize(&admin, &treasury, &100, &50, &PaymentRouter::MAX_AMOUNT);
-        client.add_supported_token(&token_address);
-
-        // Route amount up to daily limit
-        client.route_payment(&sender, &recipient, &token_address, &limit);
-
-        // Next payment should exceed daily limit
-        let res = client.try_route_payment(&sender, &recipient, &token_address, &2000);
-        assert_eq!(res.unwrap_err().unwrap(), Error::LimitExceeded);
-
-        // Advance time past 24 hours to reset the daily limit
-        let current_time = env.ledger().timestamp();
-        let current_protocol_version = env.ledger().protocol_version();
-        env.ledger().set(LedgerInfo {
-            timestamp: current_time + 86400,
-            protocol_version: current_protocol_version,
-            sequence_number: 1,
-            network_id: env.ledger().network_id().into(),
-            base_reserve: 100,
-            min_temp_entry_ttl: 16,
-            min_persistent_entry_ttl: 4096,
-            max_entry_ttl: 6312000,
-        });
-
-        // Now routing should succeed again. The first payment pushed volume past
-        // VOLUME_THRESHOLD, so the halved rate applies: 2000 * 50 bps = 10.
-        client.route_payment(&sender, &recipient, &token_address, &2000);
-        assert_eq!(token_client.balance(&recipient), (limit - 50) + (2000 - 10));
-    }
-
-    /// Verifies that `route_payments` routes a batch of payments across
-    /// disparate tokens in a single atomic transaction, charging the correct
-    /// fee per token and crediting each recipient independently.
-    #[ignore = "route_payments calls require_auth once per payment, so a batch \
-               with two payments from the same sender fails authorization"]
-    #[test]
-    fn test_route_payments_multi_token_batch() {
-        let (env, client, _) = setup_env();
-
-        let admin = Address::generate(&env);
-        let treasury = Address::generate(&env);
-        let sender = Address::generate(&env);
-        let recipient_a = Address::generate(&env);
-        let recipient_b = Address::generate(&env);
-
-        client.initialize(
-            &admin,
-            &treasury,
-            &100,
-            &1_000_000,
-            &PaymentRouter::MAX_AMOUNT,
-        );
-
-        let (usdc_like_address, usdc_like_client, usdc_like_admin_client) = setup_token(&env);
-        let (eurc_like_address, eurc_like_client, eurc_like_admin_client) = setup_token(&env);
-        assert_ne!(usdc_like_address, eurc_like_address);
-
-        usdc_like_admin_client.mint(&sender, &10_000);
-        eurc_like_admin_client.mint(&sender, &5_000);
-
-        let payments = vec![
-            &env,
-            Payment {
-                sender: sender.clone(),
-                recipient: recipient_a.clone(),
-                token_address: usdc_like_address.clone(),
-                amount: 2_000,
-            },
-            Payment {
-                sender: sender.clone(),
-                recipient: recipient_b.clone(),
-                token_address: eurc_like_address.clone(),
-                amount: 1_000,
-            },
-        ];
-
-        client.route_payments(&payments);
-
-        // USDC-like payment: 2_000 with 100 bps fee => 20 fee, 1_980 to recipient_a
-        assert_eq!(usdc_like_client.balance(&sender), 8_000);
-        assert_eq!(usdc_like_client.balance(&recipient_a), 1_980);
-        assert_eq!(usdc_like_client.balance(&treasury), 20);
-
-        // EURC-like payment: 1_000 with 100 bps fee => 10 fee, 990 to recipient_b
-        assert_eq!(eurc_like_client.balance(&sender), 4_000);
-        assert_eq!(eurc_like_client.balance(&recipient_b), 990);
-        assert_eq!(eurc_like_client.balance(&treasury), 10);
-
-        // Volume aggregates across both tokens for the sender
-        assert_eq!(client.get_user_volume(&sender), 3_000);
-    }
-
-    /// Fee is capped at the payment amount when fee_cap is larger than amount.
-    /// With fee_bps = 10_000 (100%) the fee equals the full amount, so
-    /// the remainder = 0 and only the fee transfer is executed.
-    #[test]
-    fn test_fee_capped_at_amount() {
-        let (env, client, _) = setup_env();
-        let admin = Address::generate(&env);
-        let treasury = Address::generate(&env);
-        let sender = Address::generate(&env);
-        let recipient = Address::generate(&env);
-        let (token_address, token_client, sac) = setup_token(&env);
-        sac.mint(&sender, &1_000);
-
-        // 100% fee, cap far above amount
-        client.initialize(
-            &admin,
-            &treasury,
-            &10_000,
-            &i128::MAX,
-            &PaymentRouter::MAX_AMOUNT,
-        );
-
-        client.route_payment(&sender, &recipient, &token_address, &1_000);
-
-        // All goes to treasury; recipient gets nothing
-        assert_eq!(token_client.balance(&treasury), 1_000);
-        assert_eq!(token_client.balance(&recipient), 0);
-    }
 }
 
 /// Property-based tests for fee calculation logic.

stellar-payment-platform/.env.example

@@ -15,10 +15,8 @@ DATABASE_URL="postgresql://postgres:postgres@localhost:5432/stellar_tags?schema=
 # Horizon listener network: "testnet" (default) or "public".
 # HORIZON_NETWORK=testnet
 
-# Timeout for the Horizon probe behind GET /health, in milliseconds.
-# HEALTH_HORIZON_TIMEOUT_MS=3000
-
-# Redis connection string for rate limiter (defaults to redis://localhost:6379 if not provided)
+# Redis connection string for rate limiting and the BullMQ webhook worker.
+# The webhook worker defaults to redis://127.0.0.1:6379 if not provided.
 # REDIS_URL="redis://localhost:6379"
 
 # --- Logging (rotating files) -------------------------------------------------

stellar-payment-platform/horizonListener.js

@@ -15,7 +15,8 @@ const { logger } = require('./src/logger');
 const { poolGet, poolRun } = require('./src/db');
 const {
   dispatchPaymentWebhooks,
-  scheduleWebhookRetryJob,
+  startWebhookWorker,
+  closeWebhookQueue,
 } = require('./src/webhookWorker');
 const {
   horizon,
@@ -186,6 +187,7 @@ const shutdown = async () => {
     logger.info(`  Closed stream for ${address}`);
   }
   activeStreams.clear();
+  await closeWebhookQueue();
   await prisma.$disconnect();
   process.exit(0);
 };
@@ -207,12 +209,8 @@ const main = async () => {
   // Initial sync
   await syncWatchedAccounts();
 
-  // Schedule webhook retry / liveness pings
-  try {
-    scheduleWebhookRetryJob({ prisma, poolAllFn: require('./src/db').poolAll, poolRunFn: poolRun });
-  } catch (err) {
-    logger.error('Failed to schedule webhook retry job:', err.message);
-  }
+  // Start the durable Redis-backed webhook delivery worker.
+  startWebhookWorker({ prisma, poolRunFn: poolRun });
 
   // Periodically check for newly registered accounts
   setInterval(syncWatchedAccounts, POLL_INTERVAL_MS);