fix(credit): commit draw and repay atomically with ledger and audit

.github/workflows/load-test.yml

@@ -37,6 +37,13 @@ jobs:
           sudo apt-get install k6
 
       - name: Start server
+        env:
+          PORT: '3000'
+          # 10 VUs × 3 endpoints over ~2 min exceeds the production defaults
+          # (100 req/min general, 10 req/min evaluate) and turns the smoke
+          # into a rate-limit test. CI needs the paths themselves to pass.
+          RATE_LIMIT_MAX_REQUESTS: '100000'
+          RATE_LIMIT_MAX_EVALUATE: '100000'
         run: |
           npm start &
           echo $! > server.pid

migrations/003_add_utilized_to_credit_lines.sql

@@ -0,0 +1,9 @@
+-- Persist the off-chain utilized balance on credit_lines so draw/repay can
+-- update state, ledger, and audit in one transaction without deriving the
+-- figure solely from an eventually-consistent SUM of transactions.
+
+ALTER TABLE credit_lines
+ADD COLUMN utilized NUMERIC(28,8) NOT NULL DEFAULT 0;
+
+COMMENT ON COLUMN credit_lines.utilized IS
+  'Current utilized credit; mutated atomically with ledger and audit writes on draw/repay';

scripts/load/smoke.js

@@ -20,19 +20,28 @@ export const options = {
 };
 
 const BASE_URL = __ENV.BASE_URL || 'http://localhost:3000';
+// Valid Ed25519-strkey shape (G + 55 base32 chars). Used only as a payload;
+// the rules-engine provider does not talk to Horizon in CI.
+const VALID_WALLET = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA';
+
+function envelopeData(body) {
+  try {
+    const parsed = JSON.parse(body);
+    if (parsed && typeof parsed === 'object' && parsed.data != null) {
+      return parsed.data;
+    }
+    return parsed;
+  } catch {
+    return null;
+  }
+}
 
 export default function () {
   // Test 1: Health check
   let healthRes = http.get(`${BASE_URL}/health`);
   check(healthRes, {
     'health status is 200': (r) => r.status === 200,
-    'health response has status ok': (r) => {
-      try {
-        return JSON.parse(r.body).status === 'ok';
-      } catch {
-        return false;
-      }
-    },
+    'health response has status ok': (r) => envelopeData(r.body)?.status === 'ok',
   }) || errorRate.add(1);
 
   sleep(0.5);
@@ -41,21 +50,15 @@ export default function () {
   let listRes = http.get(`${BASE_URL}/api/credit/lines?offset=0&limit=10`);
   check(listRes, {
     'list credit lines status is 200': (r) => r.status === 200,
-    'list response has creditLines array': (r) => {
-      try {
-        const body = JSON.parse(r.body);
-        return Array.isArray(body.creditLines);
-      } catch {
-        return false;
-      }
-    },
+    'list response has creditLines array': (r) =>
+      Array.isArray(envelopeData(r.body)?.creditLines),
   }) || errorRate.add(1);
 
   sleep(0.5);
 
   // Test 3: Risk evaluation
   const riskPayload = JSON.stringify({
-    walletAddress: 'GABC1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ234567890ABCDE',
+    walletAddress: VALID_WALLET,
   });
 
   const riskParams = {
@@ -67,14 +70,8 @@ export default function () {
   let riskRes = http.post(`${BASE_URL}/api/risk/evaluate`, riskPayload, riskParams);
   check(riskRes, {
     'risk evaluate status is 200': (r) => r.status === 200,
-    'risk response has walletAddress': (r) => {
-      try {
-        const body = JSON.parse(r.body);
-        return body.walletAddress !== undefined;
-      } catch {
-        return false;
-      }
-    },
+    'risk response has walletAddress': (r) =>
+      envelopeData(r.body)?.walletAddress !== undefined,
   }) || errorRate.add(1);
 
   sleep(1);

src/container/Container.ts

@@ -19,11 +19,20 @@
 import { type CreditLineRepository } from "../repositories/interfaces/CreditLineRepository.js";
 import { type RiskEvaluationRepository } from "../repositories/interfaces/RiskEvaluationRepository.js";
 import { type TransactionRepository } from "../repositories/interfaces/TransactionRepository.js";
+import { type AuditEventRepository } from "../repositories/interfaces/AuditEventRepository.js";
 import { getConnection, type DbClient } from "../db/client.js";
+import {
+  createDbTransactionRunner,
+  passthroughTransactionRunner,
+  type TransactionRunner,
+} from "../db/transaction.js";
 import { InMemoryCreditLineRepository } from "../repositories/memory/InMemoryCreditLineRepository.js";
 import { InMemoryRiskEvaluationRepository } from "../repositories/memory/InMemoryRiskEvaluationRepository.js";
 import { InMemoryTransactionRepository } from "../repositories/memory/InMemoryTransactionRepository.js";
+import { InMemoryAuditEventRepository } from "../repositories/memory/InMemoryAuditEventRepository.js";
 import { PostgresCreditLineRepository } from "../repositories/postgres/PostgresCreditLineRepository.js";
+import { PostgresTransactionRepository } from "../repositories/postgres/PostgresTransactionRepository.js";
+import { PostgresAuditEventRepository } from "../repositories/postgres/PostgresAuditEventRepository.js";
 import { CreditLineService } from "../services/CreditLineService.js";
 import { RiskEvaluationService } from "../services/RiskEvaluationService.js";
 import { createRiskProvider } from "../services/providers/providerFactory.js";
@@ -42,9 +51,11 @@ export class Container {
   private _creditLineRepository!: CreditLineRepository;
   private _riskEvaluationRepository!: RiskEvaluationRepository;
   private _transactionRepository!: TransactionRepository;
+  private _auditEventRepository!: AuditEventRepository;
+  private _runInTransaction: TransactionRunner = passthroughTransactionRunner;
 
   // Services
-  private _creditLineService: CreditLineService;
+  private _creditLineService!: CreditLineService;
   private _riskEvaluationService: RiskEvaluationService;
   private _reconciliationService: ReconciliationService;
   private _reconciliationWorker: ReconciliationWorker;
@@ -53,8 +64,7 @@ export class Container {
     // Initialize repositories based on environment
     this.initializeRepositories();
 
-    // Initialize services
-    this._creditLineService = new CreditLineService(this._creditLineRepository);
+    this.rebuildCreditLineService();
     this._riskEvaluationService = new RiskEvaluationService(
       this._riskEvaluationRepository,
       createRiskProvider(),
@@ -78,20 +88,31 @@ export class Container {
     const useDatabase = process.env.DATABASE_URL && process.env.NODE_ENV !== 'test';
 
     if (useDatabase) {
-      // Use PostgreSQL repositories
+      // Use PostgreSQL repositories sharing one client so BEGIN/COMMIT covers
+      // credit-line state, ledger rows, and audit events together.
       this._dbClient = getConnection();
+      this._runInTransaction = createDbTransactionRunner(this._dbClient);
       this._creditLineRepository = new PostgresCreditLineRepository(this._dbClient);
-      // TODO: Implement PostgreSQL versions of other repositories
       this._riskEvaluationRepository = new InMemoryRiskEvaluationRepository();
-      this._transactionRepository = new InMemoryTransactionRepository();
+      this._transactionRepository = new PostgresTransactionRepository(this._dbClient);
+      this._auditEventRepository = new PostgresAuditEventRepository(this._dbClient);
     } else {
       // Use in-memory repositories (for development/testing)
       this._creditLineRepository = new InMemoryCreditLineRepository();
       this._riskEvaluationRepository = new InMemoryRiskEvaluationRepository();
       this._transactionRepository = new InMemoryTransactionRepository();
+      this._auditEventRepository = new InMemoryAuditEventRepository();
     }
   }
 
+  private rebuildCreditLineService(): void {
+    this._creditLineService = new CreditLineService(this._creditLineRepository, {
+      transactionRepository: this._transactionRepository,
+      auditEventRepository: this._auditEventRepository,
+      runInTransaction: this._runInTransaction,
+    });
+  }
+
   public static getInstance(): Container {
     if (!Container.instance) {
       Container.instance = new Container();
@@ -137,9 +158,6 @@ export class Container {
   }): void {
     if (repositories.creditLineRepository) {
       this._creditLineRepository = repositories.creditLineRepository;
-      this._creditLineService = new CreditLineService(
-        this._creditLineRepository,
-      );
     }
 
     if (repositories.riskEvaluationRepository) {
@@ -153,6 +171,10 @@ export class Container {
     if (repositories.transactionRepository) {
       this._transactionRepository = repositories.transactionRepository;
     }
+
+    if (repositories.creditLineRepository || repositories.transactionRepository) {
+      this.rebuildCreditLineService();
+    }
   }
 
   /**

src/db/migrations.test.ts

@@ -92,12 +92,14 @@ describe('applyMigration', () => {
 describe('runPendingMigrations', () => {
   it('skips already applied migrations', async () => {
     const client = createMockClient();
-    vi.mocked(client.query)
-      .mockResolvedValueOnce({ rows: [] })
-      .mockResolvedValueOnce({ rows: [{ version: '001_initial_schema' }, { version: '002_add_interest_rate_to_credit_lines' }] });
     const migrationsDir = await import('path').then((p) =>
       p.join(process.cwd(), 'migrations')
     );
+    const files = await listMigrationFiles(migrationsDir);
+    const applied = files.map((f) => versionFromFilename(f)).map((version) => ({ version }));
+    vi.mocked(client.query)
+      .mockResolvedValueOnce({ rows: [] })
+      .mockResolvedValueOnce({ rows: applied });
     const run = await runPendingMigrations(client, migrationsDir);
     expect(run).toEqual([]);
   });
@@ -113,7 +115,8 @@ describe('runPendingMigrations', () => {
     const run = await runPendingMigrations(client, migrationsDir);
     expect(run).toContain('001_initial_schema');
     expect(run).toContain('002_add_interest_rate_to_credit_lines');
-    expect(run.length).toBeGreaterThanOrEqual(2);
+    expect(run).toContain('003_add_utilized_to_credit_lines');
+    expect(run.length).toBeGreaterThanOrEqual(3);
   });
 
   it('applies only new migrations when some are already applied', async () => {

src/db/transaction.test.ts

@@ -0,0 +1,158 @@
+import { describe, it, expect, vi } from 'vitest';
+import type { DbClient } from './client.js';
+import {
+  withTransaction,
+  createDbTransactionRunner,
+  passthroughTransactionRunner,
+} from './transaction.js';
+
+function createRecordingClient(options?: {
+  /** Fail when this SQL fragment is seen (case-insensitive). */
+  failOn?: string | RegExp;
+  /** Fail on the Nth non-control query (1-based). Control = BEGIN/COMMIT/ROLLBACK. */
+  failOnDataQuery?: number;
+}): DbClient & { statements: string[] } {
+  const statements: string[] = [];
+  let dataQueryCount = 0;
+
+  const client: DbClient & { statements: string[] } = {
+    statements,
+    async query(text: string) {
+      statements.push(text);
+      const normalized = text.trim().toUpperCase();
+      const isControl =
+        normalized === 'BEGIN' ||
+        normalized === 'COMMIT' ||
+        normalized === 'ROLLBACK';
+
+      if (!isControl) {
+        dataQueryCount += 1;
+        if (
+          options?.failOnDataQuery !== undefined &&
+          dataQueryCount === options.failOnDataQuery
+        ) {
+          throw new Error(`injected failure on data query #${dataQueryCount}`);
+        }
+      }
+
+      if (options?.failOn) {
+        const pattern =
+          typeof options.failOn === 'string'
+            ? new RegExp(options.failOn, 'i')
+            : options.failOn;
+        if (pattern.test(text)) {
+          throw new Error(`injected failure on: ${text}`);
+        }
+      }
+
+      return { rows: [] };
+    },
+    async end() {
+      /* no-op */
+    },
+  };
+
+  return client;
+}
+
+describe('withTransaction', () => {
+  it('runs work without control statements when client is undefined', async () => {
+    const result = await withTransaction(undefined, async () => 42);
+    expect(result).toBe(42);
+  });
+
+  it('commits after successful work', async () => {
+    const client = createRecordingClient();
+    const result = await withTransaction(client, async () => {
+      await client.query('INSERT INTO t VALUES (1)');
+      return 'ok';
+    });
+
+    expect(result).toBe('ok');
+    expect(client.statements).toEqual([
+      'BEGIN',
+      'INSERT INTO t VALUES (1)',
+      'COMMIT',
+    ]);
+  });
+
+  it('rolls back and rethrows when work fails', async () => {
+    const client = createRecordingClient();
+
+    await expect(
+      withTransaction(client, async () => {
+        await client.query('INSERT INTO t VALUES (1)');
+        throw new Error('boom');
+      }),
+    ).rejects.toThrow('boom');
+
+    expect(client.statements).toEqual([
+      'BEGIN',
+      'INSERT INTO t VALUES (1)',
+      'ROLLBACK',
+    ]);
+    expect(client.statements).not.toContain('COMMIT');
+  });
+
+  it('rolls back when a mid-flow data query fails (failure injection)', async () => {
+    const client = createRecordingClient({ failOnDataQuery: 2 });
+
+    await expect(
+      withTransaction(client, async () => {
+        await client.query('UPDATE credit_lines SET utilized = $1');
+        await client.query('INSERT INTO transactions ...');
+        return 'unreachable';
+      }),
+    ).rejects.toThrow('injected failure on data query #2');
+
+    expect(client.statements[0]).toBe('BEGIN');
+    expect(client.statements.at(-1)).toBe('ROLLBACK');
+    expect(client.statements).not.toContain('COMMIT');
+  });
+
+  it('prefers the original error if ROLLBACK itself fails', async () => {
+    const statements: string[] = [];
+    const client: DbClient = {
+      async query(text: string) {
+        statements.push(text);
+        const n = text.trim().toUpperCase();
+        if (n === 'ROLLBACK') {
+          throw new Error('rollback failed');
+        }
+        return { rows: [] };
+      },
+      async end() {
+        /* no-op */
+      },
+    };
+
+    await expect(
+      withTransaction(client, async () => {
+        throw new Error('work failed');
+      }),
+    ).rejects.toThrow('work failed');
+
+    expect(statements).toEqual(['BEGIN', 'ROLLBACK']);
+  });
+});
+
+describe('createDbTransactionRunner', () => {
+  it('delegates to withTransaction on the given client', async () => {
+    const client = createRecordingClient();
+    const run = createDbTransactionRunner(client);
+    const value = await run(async () => {
+      await client.query('SELECT 1');
+      return 7;
+    });
+    expect(value).toBe(7);
+    expect(client.statements).toEqual(['BEGIN', 'SELECT 1', 'COMMIT']);
+  });
+});
+
+describe('passthroughTransactionRunner', () => {
+  it('executes work without a database client', async () => {
+    const spy = vi.fn(async () => 'done');
+    await expect(passthroughTransactionRunner(spy)).resolves.toBe('done');
+    expect(spy).toHaveBeenCalledOnce();
+  });
+});