diff --git a/src/config/env.ts b/src/config/env.ts index 95523758..1a86d71a 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -242,6 +242,21 @@ export const envSchema = z .int() .positive() .default(86_400_000), + BILLING_RECON_INTERVAL_MS: z.coerce + .number() + .int() + .positive() + .default(86_400_000), + BILLING_RECON_BATCH_SIZE: z.coerce + .number() + .int() + .positive() + .default(500), + BILLING_RECON_GRACE_PERIOD_MS: z.coerce + .number() + .int() + .nonnegative() + .default(300_000), REVENUE_LEDGER_INDEXER_INTERVAL_MS: z.coerce .number() .int() diff --git a/src/config/index.ts b/src/config/index.ts index 9094b5fd..193b2d24 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -230,6 +230,11 @@ export const config = { settlementRecon: { intervalMs: env.SETTLEMENT_RECON_INTERVAL_MS, }, + billingRecon: { + intervalMs: env.BILLING_RECON_INTERVAL_MS, + batchSize: env.BILLING_RECON_BATCH_SIZE, + gracePeriodMs: env.BILLING_RECON_GRACE_PERIOD_MS, + }, revenueLedgerIndexer: { intervalMs: env.REVENUE_LEDGER_INDEXER_INTERVAL_MS, batchSize: env.REVENUE_LEDGER_INDEXER_BATCH_SIZE, diff --git a/src/index.ts b/src/index.ts index b5b90c95..c729f57d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -58,29 +58,16 @@ import { import { createSloAlertJob } from "./workers/sloAlertJob.js"; import { createMonthlyInvoiceJob } from "./workers/monthlyInvoiceJob.js"; import { createSettlementReconWorker } from "./workers/settlementRecon.js"; -import { createDeveloperRouter } from './routes/developerRoutes.js'; -import { createGatewayRouter } from './routes/gatewayRoutes.js'; -import { createProxyRouter } from './routes/proxyRoutes.js'; +import { + createBillingReconciliationJob, + createPostgresReconciliationStore, +} from "./services/billingReconciliationJob.js"; import { createRefreshTokenRouter } from './routes/refresh-token.js'; import { AuthController } from './controllers/authController.js'; import { RefreshTokenService } from './services/refreshTokenService.js'; import { DatabaseRefreshTokenRepository } from './repositories/refreshTokenRepository.js'; -import { defaultDeveloperRepository } from './repositories/developerRepository.js'; -import { createBillingService } from './services/billingService.js'; -import { createRateLimiter } from './services/rateLimiter.js'; -import { PgUsageEventsRepository } from './repositories/usageEventsRepository.pg.js'; -import { createRevenueLedgerIndexerJob } from './services/revenueLedgerIndexer.js'; -import { RevenueSettlementService } from './services/revenueSettlementService.js'; -import { createSettlementStatusSyncJob } from './services/settlementStatusSyncJob.js'; -import { createSettlementReconciliationJob } from './services/settlementReconciliationJob.js'; -import { createIdempotencySweeperJob } from './services/idempotencySweeper.js'; -import { createPostgresUsageStore } from './services/usageStore.js'; -import { createPostgresSettlementStore } from './services/settlementStore.js'; -import { createApiRegistry } from './data/apiRegistry.js'; -import { ApiKey } from './types/gateway.js'; -import { listingsCache } from './lib/listingsCache.js'; -import { createSlowQueryAlerterJob } from './workers/slowQueryAlerter.js'; -import { createAnomalyDetectorJob } from './workers/anomalyDetector.js'; +import { createApiKeyRouter } from './routes/apiKeyRoutes.js'; +import { defaultApiRepository } from './repositories/apiRepository.js'; // Helper for Jest/CommonJS compat const isDirectExecution = @@ -210,6 +197,16 @@ if (isDirectExecution) { horizonRequestTimeoutMs: config.settlementSync.timeoutMs, }); + const billingReconJob = createBillingReconciliationJob( + pool, + createPostgresReconciliationStore(pool), + { + intervalMs: config.billingRecon.intervalMs, + batchSize: config.billingRecon.batchSize, + gracePeriodMs: config.billingRecon.gracePeriodMs, + }, + ); + const idempotencySweeperJob = createIdempotencySweeperJob(pool, { intervalMs: config.idempotency.sweeperIntervalMs, }); @@ -357,6 +354,11 @@ if (isDirectExecution) { beginShutdown: () => settlementReconJob.beginShutdown(), awaitIdle: () => settlementReconJob.awaitIdle(), }, + { + name: "billing-reconciliation", + beginShutdown: () => billingReconJob.beginShutdown(), + awaitIdle: () => billingReconJob.awaitIdle(), + }, ]; // Mount the refresh-token router. The drain middleware is applied inside the @@ -381,6 +383,7 @@ if (isDirectExecution) { revenueLedgerIndexerJob.stop(); settlementStatusSyncJob.stop(); settlementReconJob.stop(); + billingReconJob.stop(); idempotencySweeperJob.stop(); slowQueryAlerterJob?.stop(); anomalyDetectorJob?.stop(); @@ -423,6 +426,7 @@ if (isDirectExecution) { revenueLedgerIndexerJob.start(); settlementStatusSyncJob.start(); settlementReconJob.start(); + billingReconJob.start(); idempotencySweeperJob.start(); slowQueryAlerterJob?.start(); anomalyDetectorJob?.start(); diff --git a/src/services/billingReconciliationJob.test.ts b/src/services/billingReconciliationJob.test.ts index 10c3beac..569852ec 100644 --- a/src/services/billingReconciliationJob.test.ts +++ b/src/services/billingReconciliationJob.test.ts @@ -283,3 +283,56 @@ test('beginShutdown prevents new ticks from starting', async () => { jest.useRealTimers(); } }); + +test('createPostgresReconciliationStore persists run rows via parameterized INSERT', async () => { + const { createPostgresReconciliationStore } = await import('./billingReconciliationJob.js'); + const queries: Array<{ sql: string; params: unknown[] | undefined }> = []; + const db: ReconciliationQueryable = { + async query(sql: string, params?: unknown[]): Promise<{ rows: T[] }> { + queries.push({ sql, params }); + return { rows: [] }; + }, + }; + + const store = createPostgresReconciliationStore(db); + const runAt = new Date('2026-03-01T00:00:00.000Z'); + await store.insertRun({ + run_at: runAt, + developer_id: 'dev-42', + usage_total_usdc: 1200n, + ledger_total_usdc: 1000n, + delta_usdc: 200n, + discrepancy_count: 1, + status: 'discrepancy', + }); + + assert.equal(queries.length, 1); + assert.match(queries[0]!.sql, /INSERT INTO reconciliation_runs/i); + assert.deepEqual(queries[0]!.params, [ + runAt, + 'dev-42', + '1200', + '1000', + '200', + 1, + 'discrepancy', + ]); +}); + +test('BillingReconciliationJob stores configured batchSize, gracePeriodMs, and threshold', () => { + const db = makeDb([], []); + const { store } = makeStore(); + + const job = new BillingReconciliationJob(db, store, { + batchSize: 250, + gracePeriodMs: 120_000, + discrepancyThresholdUsdc: 50n, + }); + + assert.deepEqual(job.getConfig(), { + batchSize: 250, + gracePeriodMs: 120_000, + threshold: 50n, + }); +}); + diff --git a/src/services/billingReconciliationJob.ts b/src/services/billingReconciliationJob.ts index 8f65cffb..772cae6f 100644 --- a/src/services/billingReconciliationJob.ts +++ b/src/services/billingReconciliationJob.ts @@ -24,6 +24,35 @@ export interface ReconciliationStore { insertRun(run: ReconciliationRunInput): Promise; } +export function createPostgresReconciliationStore( + db: ReconciliationQueryable, +): ReconciliationStore { + return { + async insertRun(run: ReconciliationRunInput): Promise { + await db.query( + `INSERT INTO reconciliation_runs ( + run_at, + developer_id, + usage_total_usdc, + ledger_total_usdc, + delta_usdc, + discrepancy_count, + status + ) VALUES ($1, $2, $3, $4, $5, $6, $7)`, + [ + run.run_at, + run.developer_id, + run.usage_total_usdc.toString(), + run.ledger_total_usdc.toString(), + run.delta_usdc.toString(), + run.discrepancy_count, + run.status, + ], + ); + }, + }; +} + export interface ReconciliationRunInput { run_at: Date; developer_id: string; @@ -61,6 +90,10 @@ export interface ReconciliationRunSummary { export interface BillingReconciliationJobOptions { /** Amount (in smallest USDC units) above which a delta is an error. Default 0 = any delta. */ discrepancyThresholdUsdc?: bigint; + /** Maximum number of developers/events processed per reconciliation pass. Default 500. */ + batchSize?: number; + /** Grace period (ms) before newly recorded usage events are eligible for reconciliation. Default 300_000. */ + gracePeriodMs?: number; logger?: Pick; } @@ -70,6 +103,8 @@ export interface BillingReconciliationJobOptions { export class BillingReconciliationJob { private readonly threshold: bigint; + private readonly batchSize: number; + private readonly gracePeriodMs: number; private readonly log: Pick; constructor( @@ -78,9 +113,19 @@ export class BillingReconciliationJob { options: BillingReconciliationJobOptions = {}, ) { this.threshold = options.discrepancyThresholdUsdc ?? 0n; + this.batchSize = options.batchSize ?? 500; + this.gracePeriodMs = options.gracePeriodMs ?? 300_000; this.log = options.logger ?? logger; } + getConfig(): { batchSize: number; gracePeriodMs: number; threshold: bigint } { + return { + batchSize: this.batchSize, + gracePeriodMs: this.gracePeriodMs, + threshold: this.threshold, + }; + } + /** * Run one full reconciliation pass. *