feat: reconcile pending settlements against Horizon

.env.example

@@ -81,6 +81,8 @@ SOROBAN_BILLING_RPC_TIMEOUT_MS=5000
 HORIZON_ENABLED=false
 HORIZON_URL=https://horizon-testnet.stellar.org
 HORIZON_TIMEOUT=2000
+SETTLEMENT_STATUS_SYNC_INTERVAL_MS=60000
+SETTLEMENT_STATUS_SYNC_TIMEOUT_MS=5000
 
 # -----------------------------------------------------------------------------
 # Stellar / Soroban network selection

README.md

@@ -117,7 +117,7 @@ When refreshing it:
 1. Keep settlement IDs globally unique.
 2. Keep each settlement under the matching developer key and `developerId`.
 3. Use non-negative finite amounts and valid ISO-8601 `created_at` timestamps.
-4. Keep `tx_hash` as `null` for `pending` settlements and non-empty for `completed` settlements.
+4. Keep `tx_hash` as either `null` or a non-empty transaction hash for `pending` settlements, and non-empty for `completed` settlements.
 5. Update usage revenue so fixture summaries stay aligned with the live route semantics: `total_earned = completed + pending + usage` and `available_to_withdraw = usage`.
 
 Run `npm run lint`, `npm run typecheck`, and `npm test` after editing the fixture.
@@ -181,6 +181,8 @@ The app validates all environment variables at startup using [Zod](https://zod.d
 | `HORIZON_ENABLED` | No | `false` | Enable Horizon health check |
 | `HORIZON_URL` | If `HORIZON_ENABLED=true` | — | Horizon endpoint URL |
 | `HORIZON_TIMEOUT` | No | `2000` | Horizon timeout (ms) |
+| `SETTLEMENT_STATUS_SYNC_INTERVAL_MS` | No | `60000` | Settlement-status sync polling interval (ms) |
+| `SETTLEMENT_STATUS_SYNC_TIMEOUT_MS` | No | `5000` | Per-request Horizon timeout for settlement sync (ms) |
 | `HEALTH_CHECK_DB_TIMEOUT` | No | `2000` | DB health check timeout (ms) |
 | `APP_VERSION` | No | `1.0.0` | Reported in health check responses |
 | `LOG_LEVEL` | No | `info` | `trace` / `debug` / `info` / `warn` / `error` / `fatal` |
@@ -221,6 +223,8 @@ STELLAR_MAINNET_SETTLEMENT_CONTRACT_ID=CC...MAINNET_SETTLEMENT
 # Optional transaction builder overrides
 STELLAR_BASE_FEE=100
 STELLAR_TRANSACTION_TIMEOUT=300
+SETTLEMENT_STATUS_SYNC_INTERVAL_MS=60000
+SETTLEMENT_STATUS_SYNC_TIMEOUT_MS=5000
 ```
 
 Notes:

src/__tests__/revenueSettlementService.test.ts

@@ -67,8 +67,9 @@ describe('RevenueSettlementService', () => {
     expect(settlements[0]).toMatchObject({
       developerId: 'dev_1',
       amount: 6,
-      status: 'completed',
+      status: 'pending',
       tx_hash: '0xmocktx_dev_1',
+      completed_at: null,
     });
 
     expect(usageStore.getUnsettledEvents()).toHaveLength(0);
@@ -168,8 +169,9 @@ describe('RevenueSettlementService', () => {
     expect(result).toEqual({ processed: 1, settledAmount: 7, errors: 1 });
     expect(settlementStore.getDeveloperSettlements('dev_1')).toHaveLength(0);
     expect(settlementStore.getDeveloperSettlements('dev_2')[0]).toMatchObject({
-      status: 'completed',
+      status: 'pending',
       tx_hash: '0xmocktx_dev_2',
+      completed_at: null,
     });
     expect(usageStore.getUnsettledEvents().map((event) => event.id)).toEqual(['e1']);
 
@@ -233,13 +235,178 @@ describe('RevenueSettlementService', () => {
       tx_hash: null,
     });
     expect(settlementStore.getDeveloperSettlements('dev_2')[0]).toMatchObject({
-      status: 'completed',
+      status: 'pending',
       tx_hash: '0xmocktx_dev_2',
+      completed_at: null,
     });
     expect(usageStore.getUnsettledEvents().map((event) => event.id)).toEqual(['e1']);
 
     updateStatusSpy.mockRestore();
   });
+
+  it('reconciles pending settlements to completed when Horizon confirms the transaction', async () => {
+    settlementStore.create({
+      id: 'stl_1',
+      developerId: 'dev_1',
+      amount: 12,
+      status: 'pending',
+      tx_hash: 'tx-confirmed',
+      created_at: '2026-04-01T00:00:00.000Z',
+    });
+
+    service = new RevenueSettlementService(usageStore, settlementStore, apiRegistry, client, {
+      fetchImpl: jest.fn(async () => ({
+        ok: true,
+        status: 200,
+        json: async () => ({ successful: true }),
+      })) as unknown as typeof fetch,
+      horizonUrl: 'https://horizon-testnet.stellar.org/',
+    });
+
+    const result = await service.reconcilePendingSettlements();
+
+    expect(result).toEqual({ checked: 1, completed: 1, failed: 0, errors: 0 });
+    expect(settlementStore.getDeveloperSettlements('dev_1')[0]).toMatchObject({
+      status: 'completed',
+      tx_hash: 'tx-confirmed',
+    });
+    expect(settlementStore.getDeveloperSettlements('dev_1')[0]?.completed_at).toBeTruthy();
+  });
+
+  it('reconciles pending settlements to failed when Horizon reports an unsuccessful transaction', async () => {
+    settlementStore.create({
+      id: 'stl_1',
+      developerId: 'dev_1',
+      amount: 12,
+      status: 'pending',
+      tx_hash: 'tx-failed',
+      created_at: '2026-04-01T00:00:00.000Z',
+    });
+
+    service = new RevenueSettlementService(usageStore, settlementStore, apiRegistry, client, {
+      fetchImpl: jest.fn(async () => ({
+        ok: true,
+        status: 200,
+        json: async () => ({ successful: false }),
+      })) as unknown as typeof fetch,
+      horizonUrl: 'https://horizon-testnet.stellar.org/',
+    });
+
+    const result = await service.reconcilePendingSettlements();
+
+    expect(result).toEqual({ checked: 1, completed: 0, failed: 1, errors: 0 });
+    expect(settlementStore.getDeveloperSettlements('dev_1')[0]).toMatchObject({
+      status: 'failed',
+      tx_hash: 'tx-failed',
+      completed_at: null,
+    });
+  });
+
+  it('treats missing Horizon transactions as failed settlements', async () => {
+    settlementStore.create({
+      id: 'stl_1',
+      developerId: 'dev_1',
+      amount: 12,
+      status: 'pending',
+      tx_hash: 'tx-missing',
+      created_at: '2026-04-01T00:00:00.000Z',
+    });
+
+    service = new RevenueSettlementService(usageStore, settlementStore, apiRegistry, client, {
+      fetchImpl: jest.fn(async () => ({
+        ok: false,
+        status: 404,
+        json: async () => ({}),
+      })) as unknown as typeof fetch,
+      horizonUrl: 'https://horizon-testnet.stellar.org/',
+    });
+
+    const result = await service.reconcilePendingSettlements();
+
+    expect(result).toEqual({ checked: 1, completed: 0, failed: 1, errors: 0 });
+    expect(settlementStore.getDeveloperSettlements('dev_1')[0]?.status).toBe('failed');
+  });
+
+  it('retries transient Horizon errors with backoff and completes once Horizon succeeds', async () => {
+    settlementStore.create({
+      id: 'stl_1',
+      developerId: 'dev_1',
+      amount: 12,
+      status: 'pending',
+      tx_hash: 'tx-retry',
+      created_at: '2026-04-01T00:00:00.000Z',
+    });
+
+    const fetchMock = jest
+      .fn()
+      .mockResolvedValueOnce({
+        ok: false,
+        status: 503,
+        json: async () => ({}),
+      })
+      .mockResolvedValueOnce({
+        ok: true,
+        status: 200,
+        json: async () => ({ successful: true }),
+      });
+
+    const setTimeoutSpy = jest
+      .spyOn(global, 'setTimeout')
+      .mockImplementation(((fn: (...args: unknown[]) => void) => {
+        fn();
+        return 0 as unknown as NodeJS.Timeout;
+      }) as typeof setTimeout);
+
+    service = new RevenueSettlementService(usageStore, settlementStore, apiRegistry, client, {
+      fetchImpl: fetchMock as unknown as typeof fetch,
+      horizonUrl: 'https://horizon-testnet.stellar.org/',
+      horizonMaxRetries: 1,
+      horizonRetryBaseDelayMs: 1,
+    });
+
+    const result = await service.reconcilePendingSettlements();
+
+    expect(result).toEqual({ checked: 1, completed: 1, failed: 0, errors: 0 });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    setTimeoutSpy.mockRestore();
+  });
+
+  it('does not flip settlement status when Horizon transient errors exhaust retries', async () => {
+    settlementStore.create({
+      id: 'stl_1',
+      developerId: 'dev_1',
+      amount: 12,
+      status: 'pending',
+      tx_hash: 'tx-still-pending',
+      created_at: '2026-04-01T00:00:00.000Z',
+    });
+
+    const setTimeoutSpy = jest
+      .spyOn(global, 'setTimeout')
+      .mockImplementation(((fn: (...args: unknown[]) => void) => {
+        fn();
+        return 0 as unknown as NodeJS.Timeout;
+      }) as typeof setTimeout);
+
+    service = new RevenueSettlementService(usageStore, settlementStore, apiRegistry, client, {
+      fetchImpl: jest.fn(async () => {
+        throw new TypeError('fetch failed');
+      }) as unknown as typeof fetch,
+      horizonUrl: 'https://horizon-testnet.stellar.org/',
+      horizonMaxRetries: 1,
+      horizonRetryBaseDelayMs: 1,
+    });
+
+    const result = await service.reconcilePendingSettlements();
+
+    expect(result).toEqual({ checked: 1, completed: 0, failed: 0, errors: 1 });
+    expect(settlementStore.getDeveloperSettlements('dev_1')[0]).toMatchObject({
+      status: 'pending',
+      tx_hash: 'tx-still-pending',
+      completed_at: null,
+    });
+    setTimeoutSpy.mockRestore();
+  });
 });
 
 function recordUsageEvent(

src/config/env.ts

@@ -59,6 +59,8 @@ export const envSchema = z
       .default(false),
     HORIZON_URL: z.string().url().optional(),
     HORIZON_TIMEOUT: z.coerce.number().default(2_000),
+    SETTLEMENT_STATUS_SYNC_INTERVAL_MS: z.coerce.number().int().positive().default(60_000),
+    SETTLEMENT_STATUS_SYNC_TIMEOUT_MS: z.coerce.number().int().positive().default(5_000),
 
     // Stellar network configuration
     STELLAR_NETWORK: stellarNetworkSchema.optional(),

src/config/index.ts

@@ -139,6 +139,11 @@ export const config = {
         }
       : undefined,
 
+  settlementSync: {
+    intervalMs: env.SETTLEMENT_STATUS_SYNC_INTERVAL_MS,
+    timeoutMs: env.SETTLEMENT_STATUS_SYNC_TIMEOUT_MS,
+  },
+
   stellar: {
     network: selectedNetwork,
     baseFee: String(env.STELLAR_BASE_FEE),

src/data/developerData.ts

@@ -100,8 +100,12 @@ export function assertDeveloperDataIntegrity(): void {
         throw new Error(`Settlement ${settlement.id} has invalid created_at ${settlement.created_at}.`);
       }
 
-      if (settlement.status === 'pending' && settlement.tx_hash !== null) {
-        throw new Error(`Pending settlement ${settlement.id} must not include a transaction hash.`);
+      if (
+        settlement.status === 'pending' &&
+        settlement.tx_hash !== null &&
+        settlement.tx_hash.trim().length === 0
+      ) {
+        throw new Error(`Pending settlement ${settlement.id} cannot use an empty transaction hash.`);
       }
 
       if (

src/index.ts

@@ -121,6 +121,23 @@ if (isDirectExecution) {
   const usageStore = createPostgresUsageStore(pool);
   const settlementStore = createPostgresSettlementStore(pool);
   const registry = createApiRegistry();
+  const revenueSettlementService = new RevenueSettlementService(
+    usageStore,
+    settlementStore,
+    registry,
+    {
+      distribute: async () => ({
+        success: false,
+        error: 'Runtime settlement distribution is not configured in this process',
+      }),
+    },
+    {
+      horizonRequestTimeoutMs: config.settlementSync.timeoutMs,
+    },
+  );
+  const settlementStatusSyncJob = createSettlementStatusSyncJob(revenueSettlementService, {
+    intervalMs: config.settlementSync.intervalMs,
+  });
 
   const apiKeys = new Map<string, ApiKey>([
     ['test-key-1', { key: 'test-key-1', developerId: 'dev_001', apiId: 'api_001' }],
@@ -166,6 +183,7 @@ if (isDirectExecution) {
   const PORT = config.port;
 
   const closeAllDataResources = async () => {
+    settlementStatusSyncJob.stop();
     await closeDb();
     await Promise.allSettled([
       closePgPool(),
@@ -178,6 +196,7 @@ if (isDirectExecution) {
   async function startServer() {
     try {
       await initializeDb();
+      settlementStatusSyncJob.start();
 
       const server = app.listen(PORT, () => {
         console.log(`Callora backend listening on http://localhost:${PORT}`);

src/services/revenueSettlementService.ts

@@ -2,16 +2,33 @@ import { Settlement, SettlementStore } from '../types/developer.js';
 import { ApiRegistry, UsageEvent, UsageStore } from '../types/gateway.js';
 import { SorobanSettlementClient } from './sorobanSettlement.js';
 import { randomUUID } from 'node:crypto';
+import { config } from '../config/index.js';
+import {
+  RETRIABLE_HTTP_STATUSES,
+  TransientError,
+  isTransientNetworkError,
+  withRetry,
+} from '../lib/retry.js';
 
 export interface RevenueSettlementOptions {
   /** Minimum accumulated USDC to trigger a payout (default: 5.00) */
   minPayoutUsdc?: number;
   /** Maximum number of events to process per developer per batch (to avoid hitting transaction limits) */
   maxEventsPerBatch?: number;
+  horizonUrl?: string;
+  fetchImpl?: typeof fetch;
+  horizonRequestTimeoutMs?: number;
+  horizonMaxRetries?: number;
+  horizonRetryBaseDelayMs?: number;
+}
+
+interface HorizonTransactionResponse {
+  successful?: boolean;
 }
 
 export class RevenueSettlementService {
   private batchTail: Promise<void> = Promise.resolve();
+  private reconcileTail: Promise<void> = Promise.resolve();
 
   constructor(
     private usageStore: UsageStore,