feat(workers): schedule billing reconciliation workers alongside other jobs

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()

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,

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();

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<T = unknown>(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,
+  });
+});
+

src/services/billingReconciliationJob.ts

@@ -24,6 +24,35 @@ export interface ReconciliationStore {
   insertRun(run: ReconciliationRunInput): Promise<void>;
 }
 
+export function createPostgresReconciliationStore(
+  db: ReconciliationQueryable,
+): ReconciliationStore {
+  return {
+    async insertRun(run: ReconciliationRunInput): Promise<void> {
+      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<typeof logger, 'info' | 'warn' | 'error'>;
 }
 
@@ -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<typeof logger, 'info' | 'warn' | 'error'>;
 
   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.
    *