Accrue revenue only after on-chain confirmation

docs/revenue-ledger-indexing.md

@@ -0,0 +1,42 @@
+# Revenue ledger indexing
+
+Revenue accrual is written only when `usage_events.stellar_tx_hash` is present. On
+schemas with `deduction_status`, the insert also requires `deduction_status =
+'confirmed'`; this condition is enforced in the insert itself to cover state
+changes between scanning and writing.
+
+The indexer stores a high-water mark in `indexer_checkpoints` and retries
+pending events and unresolved APIs from `revenue_ledger_skips`. PostgreSQL
+instances hold a session advisory lock while a run is active. Event inserts
+remain idempotent through the usage-event unique index and `ON CONFLICT` guard.
+
+## Reversal model
+
+When a usage event becomes voided, an unsettled ledger row is marked with
+`reversed_at` and its amount is set to zero. If its original row was already
+settled, the original is retained as audit evidence and a separate negative
+row is added with `reversal_of_usage_event_id`; this makes the developer's
+ledger net to zero while preserving the original settlement record. Reversed
+rows are filtered from usage and unsettled reads. The compensation row has no
+`usage_event_id`, so it cannot be offered as a usage payout event.
+
+## Incremental scan plan
+
+The scan uses an ID range, an anti-join on `revenue_ledger.usage_event_id`, and
+a retry-index lookup. To capture the deployed plan after migration, run this
+against the target PostgreSQL database with a representative checkpoint and
+limit:
+
+```sql
+EXPLAIN (ANALYZE, BUFFERS)
+SELECT ue.id
+FROM usage_events ue
+LEFT JOIN revenue_ledger rl ON rl.usage_event_id = ue.id
+LEFT JOIN revenue_ledger_skips rs ON rs.usage_event_id = ue.id
+WHERE (ue.id > 0 OR (rs.usage_event_id IS NOT NULL AND rs.permanent = FALSE AND rs.next_retry_at <= NOW()))
+  AND rl.usage_event_id IS NULL
+  AND ue.deduction_status IN ('confirmed', 'pending', 'voided')
+  AND (rs.usage_event_id IS NULL OR rs.permanent = FALSE AND rs.next_retry_at <= NOW())
+ORDER BY ue.id ASC
+LIMIT 100;
+```

migrations/0028_revenue_ledger_indexer.down.sql

@@ -0,0 +1,9 @@
+-- destructive-approved: #1498
+-- Rollback removes the indexer's retry/checkpoint state and reversal metadata.
+DROP INDEX IF EXISTS idx_usage_events_revenue_scan;
+DROP INDEX IF EXISTS idx_revenue_ledger_reversal_event;
+DROP INDEX IF EXISTS idx_revenue_ledger_skips_retry;
+ALTER TABLE revenue_ledger DROP COLUMN IF EXISTS reversal_of_usage_event_id;
+ALTER TABLE revenue_ledger DROP COLUMN IF EXISTS reversed_at;
+DROP TABLE IF EXISTS revenue_ledger_skips;
+DROP TABLE IF EXISTS indexer_checkpoints;

migrations/0028_revenue_ledger_indexer.sql

@@ -0,0 +1,30 @@
+-- Durable revenue ledger indexing, retry tracking, and reversal support.
+
+CREATE TABLE IF NOT EXISTS indexer_checkpoints (
+  name VARCHAR(128) PRIMARY KEY,
+  last_usage_event_id BIGINT NOT NULL DEFAULT 0,
+  updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
+  locked_until TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS revenue_ledger_skips (
+  usage_event_id BIGINT PRIMARY KEY,
+  reason VARCHAR(64) NOT NULL,
+  first_seen_at TIMESTAMP NOT NULL DEFAULT NOW(),
+  attempts INTEGER NOT NULL DEFAULT 1,
+  next_retry_at TIMESTAMP NOT NULL DEFAULT NOW(),
+  permanent BOOLEAN NOT NULL DEFAULT FALSE
+);
+CREATE INDEX IF NOT EXISTS idx_revenue_ledger_skips_retry
+  ON revenue_ledger_skips (permanent, next_retry_at, usage_event_id);
+
+ALTER TABLE revenue_ledger
+  ADD COLUMN IF NOT EXISTS reversed_at TIMESTAMP,
+  ADD COLUMN IF NOT EXISTS reversal_of_usage_event_id BIGINT;
+CREATE UNIQUE INDEX IF NOT EXISTS idx_revenue_ledger_reversal_event
+  ON revenue_ledger (reversal_of_usage_event_id);
+
+-- The indexer keyset scan uses the usage_events primary key and the existing
+-- unique idx_revenue_ledger_usage_event anti-join index from migration 005.
+CREATE INDEX IF NOT EXISTS idx_usage_events_revenue_scan
+  ON usage_events (id) WHERE stellar_tx_hash IS NOT NULL;

src/__tests__/persistentStores.test.ts

@@ -6,6 +6,7 @@ import { errorHandler } from '../middleware/errorHandler.js';
 import type { DeveloperRepository } from '../repositories/developerRepository.js';
 import { createPostgresSettlementStore } from '../services/settlementStore.js';
 import { createPostgresUsageStore } from '../services/usageStore.js';
+import { PgUsageEventsRepository, type UsageEventsRepositoryQueryable } from '../repositories/usageEventsRepository.pg.js';
 
 function createPersistentStoreHarness() {
   const db = newDb();
@@ -28,6 +29,7 @@ function createPersistentStoreHarness() {
       request_id VARCHAR(255) NOT NULL UNIQUE,
       status_code INTEGER NOT NULL DEFAULT 200,
       stellar_tx_hash VARCHAR(64),
+      deduction_status VARCHAR(16) NOT NULL DEFAULT 'pending',
       created_at TIMESTAMP NOT NULL DEFAULT NOW()
     );
 
@@ -52,9 +54,11 @@ function createPersistentStoreHarness() {
       api_id VARCHAR(255) NOT NULL,
       developer_id VARCHAR(255) NOT NULL,
       amount_usdc NUMERIC NOT NULL,
-      usage_event_id BIGINT NOT NULL UNIQUE REFERENCES usage_events(id),
+      usage_event_id BIGINT UNIQUE REFERENCES usage_events(id),
       settlement_id BIGINT REFERENCES settlements(id),
-      created_at TIMESTAMP NOT NULL DEFAULT NOW()
+      created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+      reversed_at TIMESTAMP,
+      reversal_of_usage_event_id BIGINT UNIQUE
     );
   `);
 
@@ -104,7 +108,7 @@ test('PostgresSettlementStore preserves ordering and status updates', async () =
   }
 });
 
-test('PostgresUsageStore records idempotently and marks events as settled', async () => {
+test('PostgresUsageStore records idempotently without accruing unconfirmed revenue', async () => {
   const { pool, settlementStore, usageStore } = createPersistentStoreHarness();
 
   try {
@@ -139,6 +143,11 @@ test('PostgresUsageStore records idempotently and marks events as settled', asyn
       timestamp: '2026-02-02T10:00:00.000Z',
     });
 
+    const unconfirmedLedger = await pool.query(
+      "SELECT COUNT(*)::int AS count FROM revenue_ledger rl JOIN usage_events ue ON ue.id = rl.usage_event_id WHERE ue.request_id = 'req-1'",
+    );
+    expect(unconfirmedLedger.rows[0]?.count).toBe(0);
+
     const events = await usageStore.getEvents('key-1');
     expect(firstInsert).toBe(true);
     expect(duplicateInsert).toBe(false);
@@ -152,6 +161,15 @@ test('PostgresUsageStore records idempotently and marks events as settled', asyn
       settlementId: undefined,
     });
 
+    // Simulate the revenue indexer accruing the row only after confirmation.
+    await pool.query("UPDATE usage_events SET stellar_tx_hash = 'confirmed-tx', deduction_status = 'confirmed' WHERE request_id = 'req-1'");
+    await pool.query(
+      `INSERT INTO revenue_ledger (api_id, developer_id, amount_usdc, usage_event_id, created_at)
+       SELECT ue.api_id, apis.developer_id, ue.amount_usdc, ue.id, ue.created_at
+       FROM usage_events ue JOIN apis ON apis.id = ue.api_id
+       WHERE ue.request_id = 'req-1'`,
+    );
+
     await settlementStore.create({
       id: 'stl_1',
       developerId: 'dev_1',
@@ -173,6 +191,70 @@ test('PostgresUsageStore records idempotently and marks events as settled', asyn
   }
 });
 
+test('voided unsettled revenue is reversed and settled revenue receives a negative compensation', async () => {
+  const { pool, settlementStore, usageStore } = createPersistentStoreHarness();
+  const repository = new PgUsageEventsRepository(pool as unknown as UsageEventsRepositoryQueryable);
+  try {
+    await pool.query('INSERT INTO apis (id, developer_id) VALUES ($1, $2)', ['api-void', 'dev-void']);
+    await usageStore.record({
+      id: 'void-unsettled', requestId: 'req-void-unsettled', apiKey: 'key-void', apiKeyId: 'key-void',
+      apiId: 'api-void', endpointId: 'ep-void', userId: 'consumer', amountUsdc: 7, statusCode: 200,
+      timestamp: '2026-02-01T00:00:00.000Z',
+    });
+    await pool.query("UPDATE usage_events SET stellar_tx_hash = 'tx-u', deduction_status = 'confirmed' WHERE request_id = 'req-void-unsettled'");
+    await pool.query(
+      `INSERT INTO revenue_ledger (api_id, developer_id, amount_usdc, usage_event_id, created_at)
+       SELECT api_id, 'dev-void', amount_usdc, id, created_at FROM usage_events WHERE request_id = 'req-void-unsettled'`,
+    );
+    await pool.query("UPDATE usage_events SET deduction_status = 'voided' WHERE request_id = 'req-void-unsettled'");
+
+    expect(await repository.reverseVoidedRevenueLedgerEvents()).toBe(1);
+    const unsettledOriginal = await pool.query("SELECT amount_usdc::text, reversed_at FROM revenue_ledger WHERE usage_event_id = 1");
+    expect(unsettledOriginal.rows[0]).toMatchObject({ amount_usdc: '0' });
+    expect(unsettledOriginal.rows[0]?.reversed_at).toBeTruthy();
+    expect(await usageStore.getUnsettledEvents()).toEqual([]);
+
+    await usageStore.record({
+      id: 'void-settled', requestId: 'req-void-settled', apiKey: 'key-void', apiKeyId: 'key-void',
+      apiId: 'api-void', endpointId: 'ep-void', userId: 'consumer', amountUsdc: 9, statusCode: 200,
+      timestamp: '2026-02-02T00:00:00.000Z',
+    });
+    await pool.query("UPDATE usage_events SET stellar_tx_hash = 'tx-s', deduction_status = 'confirmed' WHERE request_id = 'req-void-settled'");
+    await pool.query(
+      `INSERT INTO revenue_ledger (api_id, developer_id, amount_usdc, usage_event_id, created_at)
+       SELECT api_id, 'dev-void', amount_usdc, id, created_at FROM usage_events WHERE request_id = 'req-void-settled'`,
+    );
+    await settlementStore.create({
+      id: 'stl-void-settled', developerId: 'dev-void', amount: 9, status: 'completed', tx_hash: 'tx-settlement',
+      created_at: '2026-02-03T00:00:00.000Z',
+    });
+    const settledEvent = (await usageStore.getEvents('key-void')).find((event) => event.requestId === 'req-void-settled');
+    await usageStore.markAsSettled([settledEvent!.id], 'stl-void-settled');
+    await pool.query("UPDATE usage_events SET deduction_status = 'voided' WHERE request_id = 'req-void-settled'");
+
+    expect(await repository.reverseVoidedRevenueLedgerEvents()).toBe(1);
+    const rows = await pool.query(
+      `SELECT usage_event_id, amount_usdc::text, reversed_at, reversal_of_usage_event_id
+       FROM revenue_ledger ORDER BY id`,
+    );
+    const settled = rows.rows.find((row) => String(row.usage_event_id) === String(settledEvent?.id));
+    const compensation = rows.rows.find((row) => String(row.reversal_of_usage_event_id) === String(settledEvent?.id));
+    expect(settled?.amount_usdc).toBe('9');
+    expect(settled?.reversed_at).toBeTruthy();
+    expect(compensation?.amount_usdc).toBe('-9');
+    expect(await usageStore.getUnsettledEvents()).toEqual([]);
+    const invariant = await pool.query(`
+      SELECT COUNT(*)::int AS count
+      FROM revenue_ledger rl JOIN usage_events ue ON ue.id = rl.usage_event_id
+      WHERE (ue.stellar_tx_hash IS NULL OR ue.deduction_status = 'voided')
+        AND rl.reversed_at IS NULL
+    `);
+    expect(invariant.rows[0]?.count).toBe(0);
+  } finally {
+    await pool.end();
+  }
+});
+
 test('persistent stores survive new instances and keep developer revenue available after restart', async () => {
   const harness = createPersistentStoreHarness();
 
@@ -235,9 +317,9 @@ test('persistent stores survive new instances and keep developer revenue availab
 
     expect(res.status).toBe(200);
     expect(res.body.summary).toEqual({
-      total_earned: 16,
+      total_earned: 13,
       pending: 5,
-      available_to_withdraw: 3,
+      available_to_withdraw: 0,
     });
     expect(res.body.settlements.map((settlement: { id: string }) => settlement.id)).toEqual([
       'stl_pending',

src/metrics.ts

@@ -1007,3 +1007,14 @@ export function resetBillingPendingMetrics(): void {
   billingPendingDeductions.reset();
   billingPendingResolvedTotal.reset();
 }
+
+const revenueLedgerSkippedTotal = new client.Counter({
+  name: 'revenue_ledger_skipped_total',
+  help: 'Usage events skipped while indexing developer revenue',
+  labelNames: ['reason'] as const,
+});
+register.registerMetric(revenueLedgerSkippedTotal);
+
+export function recordRevenueLedgerSkipped(reason: 'unknown_api' | 'pending' | 'voided'): void {
+  revenueLedgerSkippedTotal.inc({ reason });
+}

src/repositories/usageEventsRepository.pg.test.ts

@@ -36,9 +36,15 @@ function createUsageEventsRepository() {
       developer_id VARCHAR(255) NOT NULL,
       amount_usdc NUMERIC(20, 0) NOT NULL,
       usage_event_id BIGINT UNIQUE REFERENCES usage_events(id),
-      created_at TIMESTAMP NOT NULL DEFAULT NOW()
+      created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+      settlement_id BIGINT,
+      reversed_at TIMESTAMP,
+      reversal_of_usage_event_id BIGINT UNIQUE
     );
 
+    CREATE TABLE indexer_checkpoints (name VARCHAR(128) PRIMARY KEY, last_usage_event_id BIGINT NOT NULL DEFAULT 0, updated_at TIMESTAMP NOT NULL DEFAULT NOW(), locked_until TIMESTAMP);
+    CREATE TABLE revenue_ledger_skips (usage_event_id BIGINT PRIMARY KEY REFERENCES usage_events(id), reason VARCHAR(64) NOT NULL, first_seen_at TIMESTAMP NOT NULL DEFAULT NOW(), attempts INTEGER NOT NULL DEFAULT 1, next_retry_at TIMESTAMP NOT NULL DEFAULT NOW(), permanent BOOLEAN NOT NULL DEFAULT FALSE);
+
     CREATE INDEX idx_usage_events_user_created ON usage_events(user_id, created_at);
     CREATE INDEX idx_usage_events_api_created ON usage_events(api_id, created_at);
   `);
@@ -514,6 +520,7 @@ test('findUnindexedRevenueLedgerEvents resolves developer ownership from apis an
       developerId: 'dev-1',
       amount: 100n,
       requestId: 'req-ledger-1',
+      stellarTxHash: 'tx-ledger-1',
       createdAt: new Date('2026-02-01T10:00:00.000Z'),
     });
     await repository.create({
@@ -524,6 +531,7 @@ test('findUnindexedRevenueLedgerEvents resolves developer ownership from apis an
       developerId: 'dev-1',
       amount: 250n,
       requestId: 'req-ledger-2',
+      stellarTxHash: 'tx-ledger-2',
       createdAt: new Date('2026-02-02T10:00:00.000Z'),
     });
     await repository.create({
@@ -534,6 +542,7 @@ test('findUnindexedRevenueLedgerEvents resolves developer ownership from apis an
       developerId: 'dev-1',
       amount: 999n,
       requestId: 'req-ledger-3',
+      stellarTxHash: 'tx-ledger-3',
       createdAt: new Date('2026-02-03T10:00:00.000Z'),
     });
 
@@ -584,6 +593,7 @@ test('indexRevenueLedgerEvent inserts idempotently by usageEventId', async () =>
       developerId: 'dev-1',
       amount: 1500n,
       requestId: 'req-ledger-insert',
+      stellarTxHash: 'tx-ledger-insert',
       createdAt: new Date('2026-02-05T10:00:00.000Z'),
     });