diff --git a/docs/revenue-ledger-indexing.md b/docs/revenue-ledger-indexing.md new file mode 100644 index 00000000..b047549a --- /dev/null +++ b/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; +``` diff --git a/migrations/0028_revenue_ledger_indexer.down.sql b/migrations/0028_revenue_ledger_indexer.down.sql new file mode 100644 index 00000000..64645ff2 --- /dev/null +++ b/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; diff --git a/migrations/0028_revenue_ledger_indexer.sql b/migrations/0028_revenue_ledger_indexer.sql new file mode 100644 index 00000000..acb92130 --- /dev/null +++ b/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; diff --git a/src/__tests__/persistentStores.test.ts b/src/__tests__/persistentStores.test.ts index 06fca8d2..a86d46f1 100644 --- a/src/__tests__/persistentStores.test.ts +++ b/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', diff --git a/src/metrics.ts b/src/metrics.ts index 66821771..1688953a 100644 --- a/src/metrics.ts +++ b/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 }); +} diff --git a/src/repositories/usageEventsRepository.pg.test.ts b/src/repositories/usageEventsRepository.pg.test.ts index e5d03dfd..47e52bf9 100644 --- a/src/repositories/usageEventsRepository.pg.test.ts +++ b/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'), }); diff --git a/src/repositories/usageEventsRepository.pg.ts b/src/repositories/usageEventsRepository.pg.ts index 982c44ee..3d3c1f7c 100644 --- a/src/repositories/usageEventsRepository.pg.ts +++ b/src/repositories/usageEventsRepository.pg.ts @@ -5,7 +5,7 @@ import { } from './usageEventsRepository.js'; import { encodeCursor, type CursorPayload } from '../lib/cursorPagination.js'; import { generateCursor, decodeCursor } from '../lib/pagination.js'; -import { readQuery, writeQuery } from '../db.js'; +import { pool, readQuery, writeQuery } from '../db.js'; export interface CreateUsageEventInput { userId: string; @@ -42,6 +42,8 @@ export interface RevenueLedgerUsageEvent { apiId: string; amount: bigint; createdAt: Date; + stellarTxHash?: string | null; + deductionStatus?: 'pending' | 'confirmed' | 'voided' | null; } export interface UsageEventsPgRepository { @@ -52,6 +54,13 @@ export interface UsageEventsPgRepository { getTotalRevenueByApi(apiId: string, from?: Date, to?: Date): Promise; findUnindexedRevenueLedgerEvents(cursor?: string, limit?: number): Promise; indexRevenueLedgerEvent(event: RevenueLedgerUsageEvent, developerId: string): Promise; + getRevenueLedgerCheckpoint(): Promise; + advanceRevenueLedgerCheckpoint(id: string): Promise; + recordRevenueLedgerSkip(id: string, reason: string, maxAttempts: number): Promise<{ attempts: number; permanent: boolean }>; + clearRevenueLedgerSkip(id: string): Promise; + reverseVoidedRevenueLedgerEvents(): Promise; + acquireRevenueLedgerIndexerLock(): Promise; + releaseRevenueLedgerIndexerLock(): Promise; findByUserIdCursor(params: { userId: string; from?: Date; @@ -73,6 +82,10 @@ export interface UsageEventsPgRepository { export interface UsageEventsRepositoryQueryable { query(text: string, params?: unknown[]): Promise<{ rows: T[] }>; + connect?: () => Promise<{ + query(text: string, params?: unknown[]): Promise<{ rows: T[] }>; + release(): void; + }>; } interface UsageEventRow { @@ -98,6 +111,8 @@ interface RevenueLedgerUsageEventRow { api_id: string; amount_usdc: string | number | bigint; created_at: Date | string; + stellar_tx_hash?: string | null; + deduction_status?: 'pending' | 'confirmed' | 'voided' | null; } const assertNonEmpty = (value: string, fieldName: string): string => { @@ -208,6 +223,8 @@ const mapRevenueLedgerUsageEventRow = ( apiId: row.api_id, amount: toBigInt(row.amount_usdc, 'amount_usdc'), createdAt: row.created_at instanceof Date ? row.created_at : new Date(row.created_at), + stellarTxHash: row.stellar_tx_hash, + deductionStatus: row.deduction_status, }); const appendDateFilters = (params: unknown[], clauses: string[], from?: Date, to?: Date): void => { @@ -225,6 +242,7 @@ const appendDateFilters = (params: unknown[], clauses: string[], from?: Date, to export class PgUsageEventsRepository implements UsageEventsPgRepository { private readonly readDb: UsageEventsRepositoryQueryable; private readonly writeDb: UsageEventsRepositoryQueryable; + private advisoryLockClient: Awaited>> | undefined; /** * @param db - Optional injectable queryable used in tests. @@ -238,9 +256,9 @@ export class PgUsageEventsRepository implements UsageEventsPgRepository { this.readDb = db; this.writeDb = db; } else { - // Production: route reads to replicas and writes to primary. + // Production: route reads to replicas and writes, including session locks, to the primary. this.readDb = { query: readQuery }; - this.writeDb = { query: writeQuery }; + this.writeDb = { query: writeQuery, connect: () => pool.connect() }; } } @@ -477,18 +495,33 @@ export class PgUsageEventsRepository implements UsageEventsPgRepository { return []; } - const result = await this.readDb.query( + // The indexer is accounting work: always read from the primary. Inspect + // the catalog first so this remains compatible with deployments that have + // not yet applied the deduction_status migration. + const statusColumn = await this.writeDb.query<{ column_name: string }>( + `SELECT column_name FROM information_schema.columns WHERE table_name = 'usage_events' AND column_name = 'deduction_status' LIMIT 1`, + ); + const statusSelect = statusColumn.rows.length ? 'ue.deduction_status' : 'NULL::text'; + const statusPredicate = statusColumn.rows.length + ? `(ue.deduction_status = 'confirmed' OR ue.deduction_status = 'pending' OR ue.deduction_status = 'voided')` + : 'TRUE'; + const result = await this.writeDb.query( ` SELECT ue.id AS usage_event_id, ue.api_id, ue.amount_usdc, - ue.created_at + ue.created_at, + ue.stellar_tx_hash, + ${statusSelect} AS deduction_status FROM usage_events ue LEFT JOIN revenue_ledger rl ON rl.usage_event_id = ue.id - WHERE ue.id > $1 + LEFT JOIN revenue_ledger_skips rs ON rs.usage_event_id = ue.id + WHERE (ue.id > $1 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 ${statusPredicate} + AND (rs.usage_event_id IS NULL OR rs.permanent = FALSE AND rs.next_retry_at <= NOW()) ORDER BY ue.id ASC LIMIT $2 `, @@ -498,7 +531,146 @@ export class PgUsageEventsRepository implements UsageEventsPgRepository { return result.rows.map(mapRevenueLedgerUsageEventRow); } + async getRevenueLedgerCheckpoint(): Promise { + const result = await this.writeDb.query<{ last_usage_event_id: string | number | bigint }>( + `SELECT last_usage_event_id FROM indexer_checkpoints WHERE name = 'revenue_ledger'`, + ); + return String(result.rows[0]?.last_usage_event_id ?? '0'); + } + + async advanceRevenueLedgerCheckpoint(id: string): Promise { + await this.writeDb.query( + `INSERT INTO indexer_checkpoints (name, last_usage_event_id, updated_at) + VALUES ('revenue_ledger', $1::bigint, NOW()) + ON CONFLICT (name) DO UPDATE SET + last_usage_event_id = GREATEST(indexer_checkpoints.last_usage_event_id, EXCLUDED.last_usage_event_id), + updated_at = NOW(), + locked_until = GREATEST(COALESCE(indexer_checkpoints.locked_until, NOW()), NOW() + INTERVAL '15 minutes')`, [id], + ); + } + + async recordRevenueLedgerSkip(id: string, reason: string, maxAttempts: number): Promise<{ attempts: number; permanent: boolean }> { + const result = await this.writeDb.query<{ attempts: number; permanent: boolean }>( + `INSERT INTO revenue_ledger_skips (usage_event_id, reason, first_seen_at, attempts, next_retry_at, permanent) + VALUES ($1::bigint, $2, NOW(), 1, NOW() + INTERVAL '30 seconds', FALSE) + ON CONFLICT (usage_event_id) DO UPDATE SET + reason = EXCLUDED.reason, + attempts = revenue_ledger_skips.attempts + 1, + next_retry_at = NOW() + (LEAST(3600, 30 * POWER(2, LEAST(revenue_ledger_skips.attempts - 1, 7))) * INTERVAL '1 second'), + permanent = CASE WHEN EXCLUDED.reason = 'pending' THEN FALSE ELSE revenue_ledger_skips.attempts + 1 >= $3 END + RETURNING attempts, permanent`, [id, reason, maxAttempts], + ); + return result.rows[0] ?? { attempts: 1, permanent: false }; + } + + async clearRevenueLedgerSkip(id: string): Promise { + await this.writeDb.query(`DELETE FROM revenue_ledger_skips WHERE usage_event_id = $1::bigint`, [id]); + } + + async reverseVoidedRevenueLedgerEvents(): Promise { + const statusColumn = await this.writeDb.query<{ column_name: string }>( + `SELECT column_name FROM information_schema.columns WHERE table_name = 'usage_events' AND column_name = 'deduction_status' LIMIT 1`, + ); + if (statusColumn.rows.length === 0) return 0; + const client = this.writeDb.connect ? await this.writeDb.connect() : undefined; + const queryable = client ?? this.writeDb; + let count = 0; + try { + if (client) await client.query('BEGIN'); + const voided = await queryable.query<{ + id: string | number; + usage_event_id: string | number; + settlement_id: string | number | null; + }>( + `SELECT rl.id, rl.usage_event_id, rl.settlement_id + FROM revenue_ledger rl JOIN usage_events ue ON ue.id = rl.usage_event_id + WHERE ue.deduction_status = 'voided' AND rl.reversed_at IS NULL + ORDER BY rl.id ASC`, + ); + + for (const row of voided.rows) { + const updated = await queryable.query<{ + usage_event_id: string | number; + settlement_id: string | number | null; + amount_usdc: string | number; + api_id: string; + developer_id: string; + }>( + `UPDATE revenue_ledger + SET reversed_at = NOW(), amount_usdc = CASE WHEN settlement_id IS NULL THEN 0 ELSE amount_usdc END + WHERE id = $1 AND reversed_at IS NULL + RETURNING usage_event_id, settlement_id, amount_usdc, api_id, developer_id`, + [row.id], + ); + const original = updated.rows[0]; + if (!original) continue; + count += 1; + if (original.settlement_id !== null) { + await queryable.query( + `INSERT INTO revenue_ledger (api_id, developer_id, amount_usdc, reversal_of_usage_event_id, created_at, reversed_at) + VALUES ($1, $2, -$3::numeric, $4::bigint, NOW(), NOW()) + ON CONFLICT (reversal_of_usage_event_id) DO NOTHING`, + [original.api_id, original.developer_id, original.amount_usdc, original.usage_event_id], + ); + } + } + if (client) await client.query('COMMIT'); + return count; + } catch (error) { + if (client) await client.query('ROLLBACK'); + throw error; + } finally { + client?.release(); + } + } + + async acquireRevenueLedgerIndexerLock(): Promise { + if (this.writeDb.connect) { + const client = await this.writeDb.connect(); + try { + const lock = await client.query<{ locked: boolean }>( + `SELECT pg_try_advisory_lock(hashtext('callora:revenue-ledger-indexer')) AS locked`, + ); + if (lock.rows[0]?.locked) { + this.advisoryLockClient = client; + return true; + } + client.release(); + return false; + } catch { + // SQL emulators do not implement PostgreSQL advisory locks. + client.release(); + } + } + const result = await this.writeDb.query<{ name: string }>( + `INSERT INTO indexer_checkpoints (name, last_usage_event_id, updated_at, locked_until) + VALUES ('revenue_ledger', 0, NOW(), NOW() + INTERVAL '15 minutes') + ON CONFLICT (name) DO UPDATE SET locked_until = NOW() + INTERVAL '15 minutes' + WHERE indexer_checkpoints.locked_until IS NULL OR indexer_checkpoints.locked_until < NOW() + RETURNING name`, + ); + return result.rows.length > 0; + } + + async releaseRevenueLedgerIndexerLock(): Promise { + if (this.advisoryLockClient) { + const client = this.advisoryLockClient; + this.advisoryLockClient = undefined; + try { + await client.query(`SELECT pg_advisory_unlock(hashtext('callora:revenue-ledger-indexer'))`); + } finally { + client.release(); + } + return; + } + await this.writeDb.query(`UPDATE indexer_checkpoints SET locked_until = NULL WHERE name = 'revenue_ledger'`); + } + async indexRevenueLedgerEvent(event: RevenueLedgerUsageEvent, developerId: string): Promise { + const statusColumn = await this.writeDb.query<{ column_name: string }>( + `SELECT column_name FROM information_schema.columns WHERE table_name = 'usage_events' AND column_name = 'deduction_status' LIMIT 1`, + ); + const confirmationPredicate = statusColumn.rows.length ? `AND ue.deduction_status = 'confirmed'` : ''; const result = await this.writeDb.query<{ inserted: number }>( ` INSERT INTO revenue_ledger ( @@ -508,12 +680,14 @@ export class PgUsageEventsRepository implements UsageEventsPgRepository { usage_event_id, created_at ) - SELECT $1, $2, $3::numeric, $4::bigint, $5::timestamp - WHERE NOT EXISTS ( - SELECT 1 - FROM revenue_ledger - WHERE usage_event_id = $4::bigint - ) + SELECT $1, $2, $3::numeric, ue.id, $5::timestamp + FROM usage_events ue + WHERE ue.id = $4::bigint + AND ue.stellar_tx_hash IS NOT NULL + ${confirmationPredicate} + AND NOT EXISTS ( + SELECT 1 FROM revenue_ledger existing WHERE existing.usage_event_id = ue.id + ) ON CONFLICT (usage_event_id) DO NOTHING RETURNING 1 AS inserted `, diff --git a/src/services/revenueLedgerIndexer.test.ts b/src/services/revenueLedgerIndexer.test.ts index cd2b8cc1..ada4d539 100644 --- a/src/services/revenueLedgerIndexer.test.ts +++ b/src/services/revenueLedgerIndexer.test.ts @@ -45,8 +45,14 @@ function createIndexerHarness() { 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); `); const { Pool } = db.adapters.createPg(); @@ -73,6 +79,7 @@ test('RevenueLedgerIndexer backfills unindexed usage events in cursor-ordered ba developerId: 'dev-1', amount: 100n, requestId: 'req-1', + stellarTxHash: 'tx-1', createdAt: new Date('2026-02-01T10:00:00.000Z'), }); await repository.create({ @@ -83,6 +90,7 @@ test('RevenueLedgerIndexer backfills unindexed usage events in cursor-ordered ba developerId: 'dev-1', amount: 200n, requestId: 'req-2', + stellarTxHash: 'tx-2', createdAt: new Date('2026-02-02T10:00:00.000Z'), }); await repository.create({ @@ -93,6 +101,7 @@ test('RevenueLedgerIndexer backfills unindexed usage events in cursor-ordered ba developerId: 'dev-1', amount: 300n, requestId: 'req-3', + stellarTxHash: 'tx-3', createdAt: new Date('2026-02-03T10:00:00.000Z'), }); @@ -122,6 +131,75 @@ test('RevenueLedgerIndexer backfills unindexed usage events in cursor-ordered ba } }); +test('only confirmed on-chain charges are accrued, and a pending charge is picked up after confirmation', async () => { + const { pool, repository } = createIndexerHarness(); + try { + await pool.query("ALTER TABLE usage_events ADD COLUMN deduction_status VARCHAR(16) NOT NULL DEFAULT 'pending'"); + await pool.query('INSERT INTO apis (id, developer_id) VALUES ($1, $2)', ['api-1', 'dev-1']); + await repository.create({ userId: 'c1', apiId: 'api-1', endpointId: 'e1', apiKeyId: 'k1', amount: 10n, requestId: 'confirmed', stellarTxHash: 'tx-confirmed' }); + await repository.create({ userId: 'c2', apiId: 'api-1', endpointId: 'e2', apiKeyId: 'k2', amount: 20n, requestId: 'pending' }); + await pool.query("UPDATE usage_events SET deduction_status = 'confirmed' WHERE request_id = 'confirmed'"); + + const indexer = new RevenueLedgerIndexer(repository, { resolveDeveloperId: async () => 'dev-1' }); + const first = await indexer.runOnce(); + assert.deepEqual(first, { scanned: 2, inserted: 1 }); + assert.equal((await pool.query('SELECT COUNT(*)::int AS count FROM revenue_ledger')).rows[0]?.count, 1); + + await pool.query("UPDATE usage_events SET stellar_tx_hash = 'tx-pending', deduction_status = 'confirmed' WHERE request_id = 'pending'"); + await pool.query("UPDATE revenue_ledger_skips SET next_retry_at = NOW() - INTERVAL '1 second' WHERE reason = 'pending'"); + const second = await indexer.runOnce(); + assert.equal(second.inserted, 1); + + 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 + `); + assert.equal(invariant.rows[0]?.count, 0); + assert.equal((await repository.getRevenueLedgerCheckpoint()), '2'); + } finally { + await pool.end(); + } +}); + +test('unknown APIs are persisted with retry backoff and do not block later events', async () => { + const { pool, repository } = createIndexerHarness(); + try { + await pool.query("INSERT INTO apis (id, developer_id) VALUES ('api-known', 'dev-1')"); + await repository.create({ userId: 'c1', apiId: 'api-missing', endpointId: 'e1', apiKeyId: 'k1', amount: 10n, requestId: 'poison', stellarTxHash: 'tx-poison' }); + await repository.create({ userId: 'c2', apiId: 'api-known', endpointId: 'e2', apiKeyId: 'k2', amount: 20n, requestId: 'healthy', stellarTxHash: 'tx-healthy' }); + const resolveDeveloperId = jest.fn(async (apiId: string) => apiId === 'api-known' ? 'dev-1' : undefined); + const indexer = new RevenueLedgerIndexer(repository, { resolveDeveloperId }); + + const first = await indexer.runOnce(); + const second = await indexer.runOnce(); + const skip = await pool.query('SELECT reason, attempts, permanent FROM revenue_ledger_skips WHERE usage_event_id = 1'); + const ledger = await pool.query('SELECT usage_event_id::text FROM revenue_ledger'); + assert.equal(first.inserted, 1); + assert.equal(second.scanned, 0); + expect(resolveDeveloperId).toHaveBeenCalledTimes(2); + assert.equal(skip.rows[0]?.reason, 'unknown_api'); + assert.equal(skip.rows[0]?.attempts, 1); + assert.deepEqual(ledger.rows, [{ usage_event_id: '2' }]); + assert.equal(await repository.getRevenueLedgerCheckpoint(), '2'); + } finally { + await pool.end(); + } +}); + +test('database lease prevents concurrent indexer instances from scanning together', async () => { + const { pool, repository } = createIndexerHarness(); + try { + assert.equal(await repository.acquireRevenueLedgerIndexerLock(), true); + assert.equal(await repository.acquireRevenueLedgerIndexerLock(), false); + await repository.releaseRevenueLedgerIndexerLock(); + assert.equal(await repository.acquireRevenueLedgerIndexerLock(), true); + } finally { + await pool.end(); + } +}); + test('RevenueLedgerIndexerJob drains in-flight work during shutdown', async () => { jest.useFakeTimers(); diff --git a/src/services/revenueLedgerIndexer.ts b/src/services/revenueLedgerIndexer.ts index c4a4ced6..91fdeab2 100644 --- a/src/services/revenueLedgerIndexer.ts +++ b/src/services/revenueLedgerIndexer.ts @@ -1,5 +1,6 @@ import { eq } from 'drizzle-orm'; import { db, schema } from '../db/index.js'; +import { recordRevenueLedgerSkipped } from '../metrics.js'; import type { RevenueLedgerUsageEvent, UsageEventsPgRepository, @@ -63,35 +64,72 @@ export class RevenueLedgerIndexer { } private async runOnceInternal(): Promise { - let cursor: string | undefined; + const lockable = this.usageEventsRepository as UsageEventsPgRepository & { + getRevenueLedgerCheckpoint?: () => Promise; + advanceRevenueLedgerCheckpoint?: (id: string) => Promise; + acquireRevenueLedgerIndexerLock?: () => Promise; + releaseRevenueLedgerIndexerLock?: () => Promise; + reverseVoidedRevenueLedgerEvents?: () => Promise; + recordRevenueLedgerSkip?: (id: string, reason: string, cap: number) => Promise<{ attempts: number; permanent: boolean }>; + clearRevenueLedgerSkip?: (id: string) => Promise; + }; + const locked = lockable.acquireRevenueLedgerIndexerLock + ? await lockable.acquireRevenueLedgerIndexerLock() + : true; + if (!locked) return { scanned: 0, inserted: 0 }; + let scanned = 0; let inserted = 0; - while (true) { - const events = await this.usageEventsRepository.findUnindexedRevenueLedgerEvents( - cursor, - this.batchSize, - ); - if (events.length === 0) { - return { scanned, inserted }; + try { + let cursor: string | undefined = lockable.getRevenueLedgerCheckpoint + ? await lockable.getRevenueLedgerCheckpoint() + : undefined; + const reversed = await lockable.reverseVoidedRevenueLedgerEvents?.(); + if (reversed) { + this.logger.error('Revenue ledger voided events were reversed', { count: reversed }); } - - scanned += events.length; - for (const event of events) { - inserted += await this.insertEvent(event); + while (true) { + const events = await this.usageEventsRepository.findUnindexedRevenueLedgerEvents(cursor, this.batchSize); + if (events.length === 0) return { scanned, inserted }; + + scanned += events.length; + for (const event of events) { + const explicitStatus = event.deductionStatus; + const confirmed = event.stellarTxHash !== null && event.stellarTxHash !== undefined && explicitStatus !== 'pending' && explicitStatus !== 'voided'; + if (explicitStatus === 'voided') { + await lockable.recordRevenueLedgerSkip?.(event.usageEventId, 'voided', 1); + recordRevenueLedgerSkipped('voided'); + } else if (!confirmed) { + await lockable.recordRevenueLedgerSkip?.(event.usageEventId, 'pending', 2147483647); + recordRevenueLedgerSkipped('pending'); + } else { + const added = await this.insertEvent(event); + inserted += added ?? 0; + if (added === null) { + // A failed developer/API lookup is durably retried with exponential backoff. + await lockable.recordRevenueLedgerSkip?.(event.usageEventId, 'unknown_api', 8); + recordRevenueLedgerSkipped('unknown_api'); + } else { + await lockable.clearRevenueLedgerSkip?.(event.usageEventId); + } + } + cursor = event.usageEventId; + await lockable.advanceRevenueLedgerCheckpoint?.(cursor); + } } - - cursor = events[events.length - 1]?.usageEventId; + } finally { + await lockable.releaseRevenueLedgerIndexerLock?.(); } } - private async insertEvent(event: RevenueLedgerUsageEvent): Promise { + private async insertEvent(event: RevenueLedgerUsageEvent): Promise { try { const developerId = await this.resolveDeveloperId(event.apiId); if (!developerId) { this.logger.error('Revenue ledger indexing failed: API or developer not found', { apiId: event.apiId }); - return 0; + return null; } return (await this.usageEventsRepository.indexRevenueLedgerEvent(event, developerId)) ? 1 : 0; diff --git a/src/services/usageStore.ts b/src/services/usageStore.ts index 008fb7ea..b9aac507 100644 --- a/src/services/usageStore.ts +++ b/src/services/usageStore.ts @@ -207,14 +207,16 @@ const usageEventSelect = ` ue.api_key_id, ue.api_id, ue.endpoint_id, - rl.developer_id AS user_id, + COALESCE(rl.developer_id, a.developer_id::text, ue.user_id::text) AS user_id, ue.amount_usdc, ue.status_code, ue.created_at, s.external_id AS settlement_external_id FROM usage_events ue - INNER JOIN revenue_ledger rl - ON rl.usage_event_id = ue.id + LEFT JOIN revenue_ledger rl + ON rl.usage_event_id = ue.id AND rl.reversed_at IS NULL + LEFT JOIN apis a + ON a.id = ue.api_id LEFT JOIN settlements s ON s.id = rl.settlement_id `; @@ -270,33 +272,6 @@ export class PostgresUsageStore implements UsageAdminStore { return false; } - await client.query( - ` - INSERT INTO revenue_ledger ( - api_id, - developer_id, - amount_usdc, - usage_event_id, - created_at - ) - SELECT - $1, - a.developer_id, - $2::numeric, - $3::bigint, - $4::timestamp - FROM apis a - WHERE a.id = $1 - ON CONFLICT (usage_event_id) DO NOTHING - `, - [ - event.apiId, - event.amountUsdc, - inserted.id, - event.timestamp, - ], - ); - await client.query('COMMIT'); return true; } catch (error) { @@ -339,6 +314,8 @@ export class PostgresUsageStore implements UsageAdminStore { ` ${usageEventSelect} WHERE rl.settlement_id IS NULL + AND rl.id IS NOT NULL + AND rl.reversed_at IS NULL AND ue.amount_usdc > 0 ORDER BY ue.created_at ASC, ue.id ASC `, @@ -366,6 +343,7 @@ export class PostgresUsageStore implements UsageAdminStore { INNER JOIN usage_events ue ON ue.id = rl.usage_event_id WHERE rl.developer_id = $1 + AND rl.reversed_at IS NULL `, [developerId], ), @@ -376,6 +354,7 @@ export class PostgresUsageStore implements UsageAdminStore { INNER JOIN usage_events ue ON ue.id = rl.usage_event_id WHERE rl.developer_id = $1 + AND rl.reversed_at IS NULL GROUP BY ue.status_code ORDER BY ue.status_code ASC `,