feat: add horizontal scaling for payment workers (#945)

backend/docs/PAYMENT_WORKER_POOL.md

@@ -0,0 +1,116 @@
+# Horizontal Scaling for Payment Workers (Issue #945)
+
+A distributed, horizontally scalable payment worker pool backed by BullMQ +
+Redis. Multiple worker instances connect to the same Redis queue and compete
+for payment jobs, giving automatic load distribution across processes and hosts
+with at-least-once delivery, exponential back-off retries, and dead-lettering.
+
+## Problem
+
+Payment processing was handled synchronously in the request path
+(`ProviderPaymentRouter.route()`), which blocks the caller and limits
+throughput to a single instance. Under load, slow downstream providers (Stripe,
+Stellar Horizon, EVM RPCs) exhaust the event loop and there is no way to
+scale payment processing independently of the API.
+
+## Solution
+
+`PaymentWorkerPool` (`src/services/payments/payment-worker-pool.ts`) enqueues
+payments as BullMQ jobs and a pool of workers processes them concurrently:
+
+```
+client ──► enqueue() ──► BullMQ queue (Redis) ◀──► Worker instance A
+                                          ▲           │
+                                          └──► Worker instance B ◀──┘
+                                          (competes for same jobs)
+```
+
+Each worker pulls jobs from the shared queue, so adding worker instances
+automatically increases throughput without rebalancing. Jobs survive process
+restarts and are retried with exponential back-off before being dead-lettered.
+
+## Architecture
+
+| Component | Responsibility |
+|-----------|----------------|
+| `PaymentWorkerPool` | Creates/manages BullMQ queue + workers, enqueues payments |
+| `Queue` (BullMQ) | Persisted job store; shared across all instances |
+| `Worker` (BullMQ) | Pulls jobs, processes via `providerRegistry`, reports completion/failure |
+| `providerRegistry` | Routes payments to soroban/evm/fiat/credit providers |
+| In-process fallback | When Redis is unavailable, jobs run locally on a 1s interval |
+
+## Usage
+
+### Enqueue a payment (fire-and-forget)
+
+```ts
+import { getPaymentWorkerPool } from '../services/payments/payment-worker-pool.js';
+
+const pool = getPaymentWorkerPool();
+await pool.start();
+
+const paymentId = await pool.enqueue({
+  amount: 100,
+  currency: 'USDC',
+  toAddress: 'GABC...',
+  network: 'stellar',
+  tenantId: 'tenant-1',
+}, 'soroban'); // optional preferred provider
+```
+
+### Get metrics (for autoscaling / observability)
+
+```ts
+const metrics = await pool.getMetrics();
+// { queued, active, completed, failed, delayed, concurrency, workerCount }
+```
+
+Use `queued` and `active` to drive horizontal autoscaling: add worker
+instances when queue depth grows, remove when idle.
+
+### Graceful shutdown
+
+```ts
+await pool.shutdown(); // closes workers and queue connection
+```
+
+## Configuration
+
+| Option | Env | Default | Meaning |
+|--------|-----|---------|---------|
+| `redisUrl` | `REDIS_URL` | undefined | Redis connection string |
+| `concurrency` | — | 4 | Parallel workers per instance |
+| `maxAttempts` | — | 5 | Max attempts per payment job |
+| `initialDelayMs` | — | 1000 | First retry delay (ms) |
+| `maxDelayMs` | — | 30000 | Max retry delay (ms) |
+| `queueName` | — | `agenticpay:payments` | BullMQ queue name |
+| `prefix` | — | `agenticpay` | BullMQ key prefix |
+
+Redis is required for true horizontal scaling. When `REDIS_ENABLED=true` and
+`REDIS_URL` are set, all instances share the same queue. Without Redis the pool
+falls back to an in-process queue (single instance only).
+
+## Testing
+
+```bash
+# Backend tests
+cd backend
+npx vitest run src/services/payments/__tests__/payment-worker-pool.test.ts
+```
+
+## Horizontal scaling guide
+
+1. Deploy `N` worker instances (containers/pods) all running the same code.
+2. Set `REDIS_ENABLED=true` and `REDIS_URL` to a shared Redis.
+3. Each instance calls `pool.start()` — BullMQ workers automatically compete
+   for jobs from the shared queue.
+4. Monitor `pool.getMetrics().queued` — scale `N` up/down based on depth.
+5. On deploy, call `pool.shutdown()` to drain in-flight jobs before exit.
+
+## Security
+
+- Payment payloads are job data in Redis — ensure Redis is network-isolated
+  and requires authentication (`REDIS_URL` with password).
+- No secrets are stored in job metadata.
+- Each provider is an isolated circuit; a failing provider does not block
+  others (see circuit breaker integration, issue #948).

backend/src/services/payments/__tests__/payment-worker-pool.test.ts

@@ -0,0 +1,163 @@
+import { describe, it, expect, beforeEach, vi } from 'vitest';
+import { PaymentWorkerPool, resetPaymentWorkerPoolForTesting } from '../payment-worker-pool.js';
+import { PaymentProviderRegistry } from '../provider-registry.js';
+import { MockPaymentProvider } from '../providers/mock.js';
+import type { PaymentInput } from '../providers/types.js';
+
+const baseInput: PaymentInput = {
+  amount: 10,
+  currency: 'XLM',
+  toAddress: 'GABC',
+  network: 'stellar',
+  tenantId: 'tenant-1',
+};
+
+describe('PaymentWorkerPool', () => {
+  beforeEach(() => {
+    resetPaymentWorkerPoolForTesting();
+  });
+
+  it('constructs with defaults', () => {
+    const pool = new PaymentWorkerPool();
+    expect(pool).toBeInstanceOf(PaymentWorkerPool);
+  });
+
+  it('selects provider by network', () => {
+    const pool = new PaymentWorkerPool();
+    // @ts-ignore - accessing private method for unit verification
+    expect(pool['selectProviderId']({ ...baseInput, network: 'stellar' })).toBe('soroban');
+    expect(pool['selectProviderId']({ ...baseInput, network: 'polygon' })).toBe('evm');
+    expect(pool['selectProviderId']({ ...baseInput, network: 'other', currency: 'USD' })).toBe('fiat');
+    expect(pool['selectProviderId']({ ...baseInput, network: 'other', currency: 'POINTS' })).toBe('credit');
+  });
+
+  it('rejects invalid amounts before enqueue', async () => {
+    const pool = new PaymentWorkerPool();
+    await expect(pool.processPayment({ ...baseInput, amount: 0 })).resolves.toMatchObject({ ok: false });
+    await expect(pool.processPayment({ ...baseInput, amount: -5 })).resolves.toMatchObject({ ok: false });
+    await expect(pool.processPayment({ ...baseInput, amount: Number.NaN })).resolves.toMatchObject({ ok: false });
+  });
+
+  it('processes a successful payment through the registry', async () => {
+    const registry = new PaymentProviderRegistry();
+    const provider = new MockPaymentProvider('test-provider');
+    registry.register(provider);
+    const pool = new PaymentWorkerPool({}, registry);
+
+    const result = await pool.processPayment(baseInput);
+    expect(result.ok).toBe(true);
+    if (result.ok) expect(result.value.providerId).toBe('test-provider');
+    expect(registry.getMetrics('test-provider').successCount).toBe(1);
+  });
+
+  it('falls back to another provider when primary fails', async () => {
+    const registry = new PaymentProviderRegistry();
+    const failing = new MockPaymentProvider('failing', { shouldFail: true });
+    const healthy = new MockPaymentProvider('healthy');
+    registry.register(failing);
+    registry.register(healthy);
+    const pool = new PaymentWorkerPool({}, registry);
+
+    const result = await pool.processPayment({ ...baseInput, network: 'stellar' });
+    expect(result.ok).toBe(true);
+    if (result.ok) expect(result.value.providerId).toBe('healthy');
+    expect(registry.getMetrics('failing').errorCount).toBe(1);
+  });
+
+  it('returns PROVIDER_UNAVAILABLE when all providers fail', async () => {
+    const registry = new PaymentProviderRegistry();
+    registry.register(new MockPaymentProvider('failing', { shouldFail: true }));
+    const pool = new PaymentWorkerPool({}, registry);
+
+    const result = await pool.processPayment(baseInput);
+    expect(result.ok).toBe(false);
+    if (!result.ok) expect(result.error.code).toBe('PROVIDER_UNAVAILABLE');
+  });
+
+  it('respects preferred provider override', async () => {
+    const registry = new PaymentProviderRegistry();
+    registry.register(new MockPaymentProvider('soroban'));
+    registry.register(new MockPaymentProvider('evm'));
+    const pool = new PaymentWorkerPool({}, registry);
+
+    const result = await pool.processPayment({ ...baseInput, network: 'stellar' }, 'evm');
+    expect(result.ok).toBe(true);
+    if (result.ok) expect(result.value.providerId).toBe('evm');
+  });
+
+  it('getMetrics returns serialisable shape without Redis', async () => {
+    const pool = new PaymentWorkerPool();
+    await pool.start();
+    const metrics = await pool.getMetrics();
+    expect(typeof metrics.queued).toBe('number');
+    expect(typeof metrics.active).toBe('number');
+    expect(typeof metrics.completed).toBe('number');
+    expect(typeof metrics.failed).toBe('number');
+    expect(typeof metrics.concurrency).toBe('number');
+    await pool.shutdown();
+  });
+
+  it('enqueue + drainLocal processes a job in-process', async () => {
+    const registry = new PaymentProviderRegistry();
+    const provider = new MockPaymentProvider('local');
+    registry.register(provider);
+    const pool = new PaymentWorkerPool({}, registry);
+    await pool.start();
+
+    const paymentId = await pool.enqueue(baseInput);
+    expect(typeof paymentId).toBe('string');
+
+    // Allow the local drain interval to pick up the job
+    await new Promise((r) => setTimeout(r, 150));
+
+    expect(registry.getMetrics('local').successCount).toBe(1);
+    await pool.shutdown();
+  });
+
+  it('shutdown stops workers and clears state', async () => {
+    const pool = new PaymentWorkerPool();
+    await pool.start();
+    await pool.shutdown();
+    // Should not throw on double shutdown
+    await pool.shutdown();
+  });
+
+  it('pause/resume without Redis does not throw', async () => {
+    const pool = new PaymentWorkerPool();
+    await pool.start();
+    await pool.pause();
+    await pool.resume();
+    await pool.shutdown();
+  });
+
+  it('getPaymentWorkerPool returns a singleton', () => {
+    resetPaymentWorkerPoolForTesting();
+    const a = getPaymentWorkerPool();
+    const b = getPaymentWorkerPool();
+    expect(a).toBe(b);
+    resetPaymentWorkerPoolForTesting();
+  });
+
+  it('registry integration: records success and error metrics', async () => {
+    const registry = new PaymentProviderRegistry();
+    const failing = new MockPaymentProvider('fail', { shouldFail: true });
+    const ok = new MockPaymentProvider('ok');
+    registry.register(failing);
+    registry.register(ok);
+    const pool = new PaymentWorkerPool({}, registry);
+
+    await pool.processPayment({ ...baseInput, network: 'stellar' });
+    expect(registry.getMetrics('fail').errorCount).toBe(1);
+    expect(registry.getMetrics('ok').successCount).toBe(1);
+  });
+
+  it('processPayment returns structured error for invalid amount', async () => {
+    const pool = new PaymentWorkerPool();
+    const result = await pool.processPayment({ ...baseInput, amount: -100 });
+    expect(result.ok).toBe(false);
+    if (!result.ok) {
+      expect(result.error.statusCode).toBe(400);
+      expect(result.error.code).toBe('INVALID_PAYMENT_AMOUNT');
+    }
+  });
+});