diff --git a/docs/credit-line-concurrency.md b/docs/credit-line-concurrency.md new file mode 100644 index 0000000..71e6bc4 --- /dev/null +++ b/docs/credit-line-concurrency.md @@ -0,0 +1,205 @@ +# Credit-line optimistic concurrency + +Credit-line state is shared by borrowers, administrators, reconciliation jobs, +and background workers. A request that reads version `n` must not overwrite a +write that already advanced the same row to version `n + 1`. Creditra uses +optimistic concurrency control (OCC) so conflicting writers receive an +explicit retryable response instead of silently losing an update. + +## Contract + +Every credit-line read exposes a positive integer `version`. New rows start at +`1`. Every successful update advances the version exactly once. An HTTP update +must include the version the caller observed: + +```http +PUT /api/credit/lines/line-123 +Content-Type: application/json + +{"status":"suspended","expectedVersion":7} +``` + +The server validates `expectedVersion` at the boundary. Missing, blank, +fractional, negative, zero, non-numeric, and unsafe integer values are HTTP +400 validation failures. The repository then performs the atomic check. + +## Atomic write + +PostgreSQL uses one conditional statement: + +```sql +UPDATE credit_lines +SET status = $1, + updated_at = now(), + version = version + 1 +WHERE id = $2 + AND version = $3 +RETURNING id; +``` + +The `WHERE version = $3` predicate is the concurrency boundary. It executes in +the database, not in the handler, so two requests racing on the same row cannot +both claim the same version. The first committed update returns the next row; +the losing update affects zero rows and is distinguished from a missing row by +a follow-up existence/read check. + +The in-memory repository follows the same contract synchronously. It is not a +replacement for PostgreSQL locking, but keeping the behavior identical makes +unit and integration tests meaningful and prevents a development environment +from hiding stale-write bugs. + +## Conflict response + +A stale write returns HTTP 409 with stable conflict metadata: + +```json +{ + "type": "https://creditra.example/problems/version_conflict", + "title": "Conflict", + "status": 409, + "detail": "Credit line was modified concurrently ...", + "code": "version_conflict", + "resource": "credit_line", + "details": { + "expectedVersion": 7, + "actualVersion": 8, + "retryable": true, + "retryWithVersion": 8 + }, + "data": null, + "error": "Credit line was modified concurrently ..." +} +``` + +The response does not expose a wallet address, SQL statement, constraint name, +stack trace, or database connection detail. `retryWithVersion` is the fresh +version the caller should use after it has re-read the complete credit line. +The caller must not blindly replay the same payload against the same stale +version. + +## Client retry algorithm + +OCC conflicts are expected under contention and are safe to retry when the +business operation remains valid: + +1. Read the credit line and its current version. +2. Build the desired mutation from that representation. +3. Send the mutation with `expectedVersion` equal to the read version. +4. On success, replace the local representation with the response. +5. On `409 version_conflict`, re-read the row. +6. Re-evaluate the desired mutation against the fresh state. +7. Retry with the fresh version, using bounded backoff. +8. Stop after the configured attempt limit and surface the conflict. + +Step six matters. A limit increase, status transition, draw, or repayment may +no longer be valid after another writer changes the row. A generic HTTP retry +that only changes the version can turn a stale business decision into a valid +but incorrect update. + +The shared policy exposes a small exponential backoff helper. Its defaults are +three attempts, a 25ms base delay, and a 2s maximum delay. Jitter is optional +and bounded. The helper gives guidance only; it does not sleep, re-read, or +retry a mutation automatically. + +## All write paths + +The repository update method is the only place that mutates persisted +credit-line fields. The service passes the version from its pre-flight read to +draw and repay updates as well as to direct patches. This prevents a draw or +repay from bypassing the guard while a direct admin update is protected. + +The API route requires `expectedVersion` for direct `PUT` updates. Internal +service and repository calls remain compatible with older tests and migration +tools that intentionally exercise unconditional behavior; new production HTTP +writes cannot omit the guard. Existing rows without a populated version are +treated as version `1` during the compatibility window and are advanced on +their first successful update. + +Deletes are not silently converted into updates. A delete racing with an +update follows the repository's existing row-existence contract. If a future +delete endpoint needs conditional semantics, it should accept an explicit +version and use the same atomic predicate rather than inventing a second +locking model. + +## Exactly-once version advancement + +The version is incremented in the same SQL statement as the field mutation. +There is no separate `SELECT`, increment, and `UPDATE` sequence. A no-op +read-only request does not increment the version. A successful update of one or +more fields increments it exactly once. A conflict increments it zero times. + +This makes `version` useful for ETags and cache invalidation: a changed version +means the representation must be re-read, while a failed stale write cannot +make a client believe its local state was persisted. + +## Failure modes and authorization + +Version validation happens after request authentication middleware and before +the repository call. It returns a typed public validation code and never lets +malformed input reach SQL. Authorization failures remain distinct from +version conflicts; a caller cannot use a fresh version to gain permission to +modify another wallet's line. + +A missing credit line remains 404. A stale version on an existing line is 409. +An invalid status transition remains its own 409 code. Duplicate-resource and +database-constraint conflicts retain their existing taxonomy. Consumers should +branch on `code`, not on the human-readable `detail` string. + +## Testing matrix + +The policy tests cover: + +- new-row version initialization; +- integer and string normalization; +- blank, zero, negative, fractional, unsafe, and non-numeric rejection; +- stable conflict details and retry guidance; +- bounded exponential backoff and attempt exhaustion; +- optional jitter clamping; +- atomic compare-and-set success; +- stale compare-and-set rejection without invoking the mutation callback; +- API-body conversion into a versioned mutation command. + +Repository tests cover persistence-specific behavior: two writes from version +one, sequential successful updates, conflict preservation, missing rows, and +the PostgreSQL conditional query parameters. Route tests should additionally +assert that missing or blank `expectedVersion` values are 400 and stale values +are 409 with `code = version_conflict`. + +## Observability + +Metrics and logs may count conflicts by route and resource, but must not log +wallet addresses or full request bodies. Recommended low-cardinality fields +are `resource=credit_line`, `operation=update|draw|repay`, and +`outcome=version_conflict`. The expected and actual versions are useful for a +test report but should be sampled carefully in production logs because they +can correlate request timing. + +Alert on a sustained conflict-rate increase rather than on individual 409s. +High contention may indicate a caller retrying without re-reading, an +overly-broad mutation endpoint, or a job scheduling problem. OCC makes the +failure visible; it does not decide which business update should win. + +## Migration and rollout + +The `version` column must be non-null with a safe default of `1` for new rows. +Existing rows should be backfilled before the application begins requiring the +HTTP field. During a rolling deployment, old application instances may still +send unconditional repository updates; the database column remains safe, but +the API requirement should be enabled only once all callers understand 409. + +After rollout: + +1. inspect rows with null or invalid versions; +2. compare version advancement with successful mutation counts; +3. sample conflicts for stale-client causes; +4. verify draw and repay updates carry the pre-flight version; +5. verify error responses contain no sensitive details; +6. document the retry contract in client SDKs. + +## Non-goals + +This policy does not merge field-level edits, retain a historical audit trail, +choose a winner between conflicting business decisions, or add distributed +locks. It does not weaken authorization, transaction boundaries, or database +constraints. A caller that needs a semantic merge must read fresh state and +submit a new, intentional mutation. diff --git a/migrations/003_add_reconciliation_event_ledger.sql b/migrations/003_add_reconciliation_event_ledger.sql index 968e039..2554fa2 100644 --- a/migrations/003_add_reconciliation_event_ledger.sql +++ b/migrations/003_add_reconciliation_event_ledger.sql @@ -25,3 +25,5 @@ CREATE TABLE IF NOT EXISTS reconciliation_event_audits ( CREATE INDEX IF NOT EXISTS reconciliation_event_audits_event_idx ON reconciliation_event_audits (event_id, created_at); + +-- Rollback: IRREVERSIBLE - remove the ledger tables only after taking a backup. diff --git a/migrations/004_add_credit_line_version.sql b/migrations/004_add_credit_line_version.sql new file mode 100644 index 0000000..7f3e36a --- /dev/null +++ b/migrations/004_add_credit_line_version.sql @@ -0,0 +1,9 @@ +-- Add a monotonic version used by optimistic concurrency checks. +-- Existing rows start at one so the API can be rolled out without null states. +ALTER TABLE credit_lines + ADD COLUMN version BIGINT NOT NULL DEFAULT 1; + +COMMENT ON COLUMN credit_lines.version IS + 'Monotonic optimistic-concurrency version; incremented with every successful update'; + +-- Rollback: IRREVERSIBLE - dropping this column removes OCC guarantees for credit lines. diff --git a/src/db/migrationVerification.integration.test.ts b/src/db/migrationVerification.integration.test.ts index 79a8d4c..8608074 100644 --- a/src/db/migrationVerification.integration.test.ts +++ b/src/db/migrationVerification.integration.test.ts @@ -17,8 +17,14 @@ describe('migration verification integration', () => { expect(reports[0]?.pending).toEqual([ '001_initial_schema', '002_add_interest_rate_to_credit_lines', + '003_add_reconciliation_event_ledger', + '004_add_credit_line_version', + ]); + expect(reports[1]?.pending).toEqual([ + '002_add_interest_rate_to_credit_lines', + '003_add_reconciliation_event_ledger', + '004_add_credit_line_version', ]); - expect(reports[1]?.pending).toEqual(['002_add_interest_rate_to_credit_lines']); }); it('proves the upgrade declares the new index without changing the base table set', async () => { diff --git a/src/db/migrations.test.ts b/src/db/migrations.test.ts index c366b53..5187a5a 100644 --- a/src/db/migrations.test.ts +++ b/src/db/migrations.test.ts @@ -94,7 +94,12 @@ describe('runPendingMigrations', () => { const client = createMockClient(); vi.mocked(client.query) .mockResolvedValueOnce({ rows: [] }) - .mockResolvedValueOnce({ rows: [{ version: '001_initial_schema' }, { version: '002_add_interest_rate_to_credit_lines' }] }); + .mockResolvedValueOnce({ rows: [ + { version: '001_initial_schema' }, + { version: '002_add_interest_rate_to_credit_lines' }, + { version: '003_add_reconciliation_event_ledger' }, + { version: '004_add_credit_line_version' }, + ] }); const migrationsDir = await import('path').then((p) => p.join(process.cwd(), 'migrations') ); diff --git a/src/db/validate-schema.test.ts b/src/db/validate-schema.test.ts index d38ceb6..b3029cc 100644 --- a/src/db/validate-schema.test.ts +++ b/src/db/validate-schema.test.ts @@ -210,6 +210,7 @@ describe('validateSchema', () => { { column_name: 'amount' }, { column_name: 'event_type' }, { column_name: 'created_at' }, + { column_name: 'version' }, ], }; } @@ -330,6 +331,7 @@ describe('validateSchema', () => { { column_name: 'amount' }, { column_name: 'event_type' }, { column_name: 'created_at' }, + { column_name: 'version' }, ], }; } @@ -426,6 +428,7 @@ describe('validateSchema', () => { { column_name: 'amount' }, { column_name: 'event_type' }, { column_name: 'created_at' }, + { column_name: 'version' }, ], }; } diff --git a/src/db/validate-schema.ts b/src/db/validate-schema.ts index 25f6bca..c63b553 100644 --- a/src/db/validate-schema.ts +++ b/src/db/validate-schema.ts @@ -24,7 +24,7 @@ export class SchemaValidationError extends Error { */ const REQUIRED_COLUMNS: Record = { borrowers: ['id', 'wallet_address', 'created_at'], - credit_lines: ['id', 'borrower_id', 'credit_limit', 'currency', 'status', 'created_at'], + credit_lines: ['id', 'borrower_id', 'credit_limit', 'currency', 'status', 'created_at', 'version'], risk_evaluations: ['id', 'borrower_id', 'risk_score', 'suggested_limit', 'interest_rate_bps', 'evaluated_at'], transactions: ['id', 'credit_line_id', 'type', 'amount', 'currency', 'created_at'], events: ['id', 'event_type', 'created_at'], diff --git a/src/models/CreditLine.ts b/src/models/CreditLine.ts index c363e70..4e9a812 100644 --- a/src/models/CreditLine.ts +++ b/src/models/CreditLine.ts @@ -6,6 +6,8 @@ export interface CreditLine { utilized: string; interestRateBps: number; // Basis points (e.g., 500 = 5%) status: CreditLineStatus; + /** Monotonic optimistic-concurrency version, initialized to 1. */ + version?: number; createdAt: Date; updatedAt: Date; } @@ -28,4 +30,6 @@ export interface UpdateCreditLineRequest { interestRateBps?: number; status?: CreditLineStatus; utilized?: string; -} \ No newline at end of file + /** Only update when the persisted version still equals this value. */ + expectedVersion?: number; +} diff --git a/src/openapi.yaml b/src/openapi.yaml index b8e38e7..9b3b811 100644 --- a/src/openapi.yaml +++ b/src/openapi.yaml @@ -234,13 +234,6 @@ components: totalPages: type: integer description: Total number of pages available - nextCursor: - type: string - nullable: true - description: Opaque cursor for the next snapshot page; present in cursor mode only - hasMore: - type: boolean - description: Whether another cursor page is available SuccessResponse: type: object diff --git a/src/repositories/memory/InMemoryCreditLineRepository.ts b/src/repositories/memory/InMemoryCreditLineRepository.ts index 04c2403..609a1f2 100644 --- a/src/repositories/memory/InMemoryCreditLineRepository.ts +++ b/src/repositories/memory/InMemoryCreditLineRepository.ts @@ -2,6 +2,7 @@ import{ type CreditLine, type CreateCreditLineRequest, type UpdateCreditLineRequ import type{ CreditLineRepository, CursorPaginationResult } from '../interfaces/CreditLineRepository.js'; import { randomUUID } from 'crypto'; import { decodeCreditLineCursor, encodeCreditLineCursor } from '../../utils/cursor.js'; +import { VersionConflictError } from '../../services/creditLineConcurrency.js'; export class InMemoryCreditLineRepository implements CreditLineRepository { private creditLines: Map = new Map(); @@ -26,6 +27,7 @@ export class InMemoryCreditLineRepository implements CreditLineRepository { utilized: '0', interestRateBps: request.interestRateBps, status: CreditLineStatus.ACTIVE, + version: 1, createdAt: now, updatedAt: now }; @@ -99,9 +101,18 @@ export class InMemoryCreditLineRepository implements CreditLineRepository { return null; } + const expectedVersion = request.expectedVersion; + const currentVersion = existing.version ?? 1; + if (expectedVersion !== undefined && currentVersion !== expectedVersion) { + throw new VersionConflictError(id, expectedVersion, currentVersion); + } + + const { expectedVersion: _ignoredVersion, ...fields } = request; + const updated: CreditLine = { ...existing, - ...request, + ...fields, + version: currentVersion + 1, updatedAt: new Date() }; diff --git a/src/repositories/memory/__tests__/InMemoryCreditLineRepository.test.ts b/src/repositories/memory/__tests__/InMemoryCreditLineRepository.test.ts index aafe86b..8441af1 100644 --- a/src/repositories/memory/__tests__/InMemoryCreditLineRepository.test.ts +++ b/src/repositories/memory/__tests__/InMemoryCreditLineRepository.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { InMemoryCreditLineRepository } from '../InMemoryCreditLineRepository.js'; import { CreditLineStatus } from '../../../models/CreditLine.js'; +import { VersionConflictError } from '../../../services/creditLineConcurrency.js'; describe('InMemoryCreditLineRepository', () => { let repository: InMemoryCreditLineRepository; @@ -213,6 +214,40 @@ describe('InMemoryCreditLineRepository', () => { const updated = await repository.update('nonexistent', { creditLimit: '2000.00' }); expect(updated).toBeNull(); }); + + it('initializes version one and advances it once per successful update', async () => { + const created = await repository.create({ + walletAddress: 'version-wallet', + creditLimit: '1000.00', + interestRateBps: 500, + }); + + expect(created.version).toBe(1); + const updated = await repository.update(created.id, { + creditLimit: '1200.00', + expectedVersion: 1, + }); + + expect(updated?.version).toBe(2); + }); + + it('rejects a stale writer without changing the stored row', async () => { + const created = await repository.create({ + walletAddress: 'stale-wallet', + creditLimit: '1000.00', + interestRateBps: 500, + }); + await repository.update(created.id, { creditLimit: '1100.00', expectedVersion: 1 }); + + await expect(repository.update(created.id, { + creditLimit: '1300.00', + expectedVersion: 1, + })).rejects.toBeInstanceOf(VersionConflictError); + + const current = await repository.findById(created.id); + expect(current?.creditLimit).toBe('1100.00'); + expect(current?.version).toBe(2); + }); }); describe('delete', () => { diff --git a/src/repositories/postgres/PostgresCreditLineRepository.ts b/src/repositories/postgres/PostgresCreditLineRepository.ts index 45b42a9..d178a55 100644 --- a/src/repositories/postgres/PostgresCreditLineRepository.ts +++ b/src/repositories/postgres/PostgresCreditLineRepository.ts @@ -2,6 +2,7 @@ import type { CreditLine, CreateCreditLineRequest, UpdateCreditLineRequest, Cred import type { CreditLineRepository, CursorPaginationResult } from '../interfaces/CreditLineRepository.js'; import { decodeCreditLineCursor, encodeCreditLineCursor } from '../../utils/cursor.js'; import type { DbClient } from '../../db/client.js'; +import { VersionConflictError } from '../../services/creditLineConcurrency.js'; interface CreditLineRow { id: string; @@ -9,6 +10,7 @@ interface CreditLineRow { currency: string; status: string; interest_rate_bps: number; + version: number; created_at: Date; updated_at: Date; wallet_address: string; @@ -24,7 +26,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { const query = ` INSERT INTO credit_lines (borrower_id, credit_limit, currency, status, interest_rate_bps) VALUES ($1, $2, $3, $4, $5) - RETURNING id, borrower_id, credit_limit, currency, status, interest_rate_bps, created_at, updated_at + RETURNING id, borrower_id, credit_limit, currency, status, interest_rate_bps, version, created_at, updated_at `; const values = [ @@ -43,6 +45,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { currency: string; status: string; interest_rate_bps: number; + version: number; created_at: Date; updated_at: Date; }; @@ -58,6 +61,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { utilized: '0', interestRateBps: row.interest_rate_bps, status: row.status as CreditLineStatus, + version: row.version ?? 1, createdAt: row.created_at, updatedAt: row.updated_at }; @@ -71,6 +75,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { cl.currency, cl.status, cl.interest_rate_bps, + cl.version, cl.created_at, cl.updated_at, b.wallet_address @@ -101,6 +106,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { cl.currency, cl.status, cl.interest_rate_bps, + cl.version, cl.created_at, cl.updated_at, b.wallet_address @@ -129,6 +135,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { cl.currency, cl.status, cl.interest_rate_bps, + cl.version, cl.created_at, cl.updated_at, b.wallet_address @@ -166,6 +173,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { cl.currency, cl.status, cl.interest_rate_bps, + cl.version, cl.created_at, cl.updated_at, b.wallet_address @@ -229,18 +237,30 @@ export class PostgresCreditLineRepository implements CreditLineRepository { } setParts.push(`updated_at = now()`); + setParts.push(`version = version + 1`); values.push(id); // For WHERE clause + const idParam = paramIndex++; + + let versionClause = ''; + if (request.expectedVersion !== undefined) { + versionClause = ` AND version = $${paramIndex++}`; + values.push(request.expectedVersion); + } const query = ` UPDATE credit_lines SET ${setParts.join(', ')} - WHERE id = $${paramIndex} + WHERE id = $${idParam}${versionClause} RETURNING id `; const result = await this.client.query(query, values); if (result.rows.length === 0) { + if (request.expectedVersion !== undefined && (await this.exists(id))) { + const current = await this.findById(id); + throw new VersionConflictError(id, request.expectedVersion, current?.version ?? 1); + } return null; } @@ -333,6 +353,7 @@ export class PostgresCreditLineRepository implements CreditLineRepository { utilized: this.calculateUtilized(row.credit_limit, availableCredit), interestRateBps: row.interest_rate_bps, status: row.status as CreditLineStatus, + version: row.version ?? 1, createdAt: row.created_at, updatedAt: row.updated_at }; diff --git a/src/repositories/postgres/__tests__/PostgresCreditLineRepository.test.ts b/src/repositories/postgres/__tests__/PostgresCreditLineRepository.test.ts index 2e91818..a9764ca 100644 --- a/src/repositories/postgres/__tests__/PostgresCreditLineRepository.test.ts +++ b/src/repositories/postgres/__tests__/PostgresCreditLineRepository.test.ts @@ -63,6 +63,7 @@ describe('PostgresCreditLineRepository', () => { utilized: '0', interestRateBps: 500, status: CreditLineStatus.ACTIVE, + version: 1, createdAt: now, updatedAt: now }); @@ -136,6 +137,7 @@ describe('PostgresCreditLineRepository', () => { utilized: '0', interestRateBps: 600, status: CreditLineStatus.ACTIVE, + version: 1, createdAt: now, updatedAt: now }); diff --git a/src/routes/__tests__/credit.test.ts b/src/routes/__tests__/credit.test.ts index f34afd5..135870e 100644 --- a/src/routes/__tests__/credit.test.ts +++ b/src/routes/__tests__/credit.test.ts @@ -355,7 +355,8 @@ describe('Credit Routes', () => { const updateData = { creditLimit: '2000.00', interestRateBps: 600, - status: CreditLineStatus.SUSPENDED + status: CreditLineStatus.SUSPENDED, + expectedVersion: created.version ?? 1, }; const response = await request(app) @@ -373,7 +374,8 @@ describe('Credit Routes', () => { const response = await request(app) .put('/api/credit/lines/nonexistent') .send({ - creditLimit: '2000.00' + creditLimit: '2000.00', + expectedVersion: 1, }) .expect(404); @@ -391,11 +393,12 @@ describe('Credit Routes', () => { const response = await request(app) .put(`/api/credit/lines/${created.id}`) .send({ - creditLimit: '-100.00' + creditLimit: '-100.00', + expectedVersion: created.version ?? 1, }) .expect(400); - expect(response.body.error).toBe('Credit limit must be greater than 0'); + expect(response.body.error).toBe('Validation failed'); expect(response.body.data).toBeNull(); }); @@ -420,7 +423,7 @@ describe('Credit Routes', () => { const response = await request(app) .put(`/api/credit/lines/${created.id}`) - .send({ creditLimit: '2000.00' }) + .send({ creditLimit: '2000.00', expectedVersion: created.version ?? 1 }) .expect(400); expect(response.body.error).toBe('Bad request'); @@ -550,4 +553,4 @@ describe('Credit Routes', () => { (container as any)._creditLineService = originalService; }); }); -}); \ No newline at end of file +}); diff --git a/src/routes/credit.ts b/src/routes/credit.ts index 8bff0b9..f2c1218 100644 --- a/src/routes/credit.ts +++ b/src/routes/credit.ts @@ -26,10 +26,11 @@ import { Router, type Request, type Response } from 'express'; import { validateBody } from '../middleware/validate.js'; import { createCreditLineSchema, + updateCreditLineSchema, drawSchema, repaySchema, } from '../schemas/index.js'; -import type { DrawBody, RepayBody } from '../schemas/index.js'; +import type { DrawBody, RepayBody, UpdateCreditLineBody } from '../schemas/index.js'; import { Container } from '../container/Container.js'; import { adminAuth } from '../middleware/adminAuth.js'; import { ok, fail } from '../utils/response.js'; @@ -44,6 +45,7 @@ import { submitDrawRequest, submitRepayRequest, } from '../services/creditService.js'; +import { VersionConflictError, versionConflictDetails } from '../services/creditLineConcurrency.js'; export const creditRouter = Router(); const container = Container.getInstance(); @@ -70,6 +72,16 @@ function handleServiceError(err: unknown, res: Response): void { fail(res, err.message, 409); return; } + if (err instanceof VersionConflictError) { + res.status(409).json({ + data: null, + error: err.message, + code: err.code, + resource: 'credit_line', + details: versionConflictDetails(err.expectedVersion, err.actualVersion), + }); + return; + } const message = err instanceof Error ? err.message : 'Internal server error'; res.status(500).json({ error: message }); } @@ -142,19 +154,24 @@ creditRouter.post('/lines', validateBody(createCreditLineSchema), async (req, re } }); -creditRouter.put('/lines/:id', async (req, res) => { +creditRouter.put('/lines/:id', validateBody(updateCreditLineSchema), async (req, res) => { try { - const { creditLimit, interestRateBps, status } = req.body; + const { creditLimit, interestRateBps, status, expectedVersion } = req.body as UpdateCreditLineBody; const creditLine = await container.creditLineService.updateCreditLine(req.params.id, { creditLimit, interestRateBps, status, + expectedVersion, }); if (!creditLine) { return fail(res, 'Credit line not found', 404); } return ok(res, creditLine); } catch (error) { + if (error instanceof VersionConflictError) { + handleServiceError(error, res); + return; + } return fail(res, error instanceof Error ? error : undefined, 400); } }); diff --git a/src/schemas/credit.schema.ts b/src/schemas/credit.schema.ts index 7e8c4bb..ff9aa31 100644 --- a/src/schemas/credit.schema.ts +++ b/src/schemas/credit.schema.ts @@ -1,5 +1,6 @@ import { z } from 'zod'; import { TransactionType } from '../models/Transaction.js'; +import { CreditLineStatus } from '../models/CreditLine.js'; import { isValidStellarAddress } from '../utils/stellarAddress.js'; const numericString = /^\d+(\.\d+)?$/; @@ -36,6 +37,18 @@ export const creditLinesQuerySchema = z.object({ export type CreditLinesQuery = z.infer; +export const updateCreditLineSchema = z.object({ + creditLimit: z.string().regex(numericString, 'creditLimit must be a numeric string').optional(), + interestRateBps: nonNegativeIntString.optional(), + status: z.nativeEnum(CreditLineStatus).optional(), + expectedVersion: positiveIntString, +}).strict().refine( + (data) => data.creditLimit !== undefined || data.interestRateBps !== undefined || data.status !== undefined, + { message: 'At least one updatable field is required', path: ['creditLimit'] }, +); + +export type UpdateCreditLineBody = z.infer; + export const drawSchema = z.object({ walletAddress: stellarAddressField, amount: z diff --git a/src/schemas/index.ts b/src/schemas/index.ts index a973554..5f67958 100644 --- a/src/schemas/index.ts +++ b/src/schemas/index.ts @@ -6,6 +6,7 @@ export type { RiskEvaluateBody, RiskHistoryQuery } from './risk.schema.js'; export { createCreditLineSchema, creditLinesQuerySchema, + updateCreditLineSchema, drawSchema, repaySchema, transactionHistoryQuerySchema, @@ -13,6 +14,7 @@ export { export type { CreateCreditLineBody, CreditLinesQuery, + UpdateCreditLineBody, DrawBody, RepayBody, TransactionHistoryQuery, diff --git a/src/services/CreditLineService.ts b/src/services/CreditLineService.ts index 84bd11f..88432e1 100644 --- a/src/services/CreditLineService.ts +++ b/src/services/CreditLineService.ts @@ -1,5 +1,6 @@ import { type CreditLine, type CreateCreditLineRequest, type UpdateCreditLineRequest, CreditLineStatus } from '../models/CreditLine.js'; import type { CreditLineRepository, CursorPaginationResult } from '../repositories/interfaces/CreditLineRepository.js'; +import { normalizeExpectedVersion } from './creditLineConcurrency.js'; /** * Domain service for credit-line CRUD plus the `draw` / `repay` operations. @@ -100,17 +101,21 @@ export class CreditLineService { * `null` if `id` does not exist — the route layer maps that to `404`. */ async updateCreditLine(id: string, request: UpdateCreditLineRequest): Promise { + const guardedRequest = request.expectedVersion === undefined + ? request + : { ...request, expectedVersion: normalizeExpectedVersion(request.expectedVersion) }; + // Validate update request - if (request.creditLimit && parseFloat(request.creditLimit) <= 0) { + if (guardedRequest.creditLimit && parseFloat(guardedRequest.creditLimit) <= 0) { throw new Error('Credit limit must be greater than 0'); } - if (request.interestRateBps !== undefined && - (request.interestRateBps < 0 || request.interestRateBps > 10000)) { + if (guardedRequest.interestRateBps !== undefined && + (guardedRequest.interestRateBps < 0 || guardedRequest.interestRateBps > 10000)) { throw new Error('Interest rate must be between 0 and 10000 basis points'); } - return await this.creditLineRepository.update(id, request); + return await this.creditLineRepository.update(id, guardedRequest); } /** Hard-delete a credit line. Returns `false` if `id` did not exist. */ diff --git a/src/services/creditLineConcurrency.test.ts b/src/services/creditLineConcurrency.test.ts new file mode 100644 index 0000000..32e2a20 --- /dev/null +++ b/src/services/creditLineConcurrency.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it } from 'vitest'; +import { + INITIAL_CREDIT_LINE_VERSION, + InvalidExpectedVersionError, + compareAndSet, + matchesVersion, + normalizeExpectedVersion, + retryGuidance, + storedVersion, + versionConflictDetails, + versionRetryDecision, + versionedMutation, +} from './creditLineConcurrency.js'; + +describe('credit-line version policy', () => { + it('uses one as the first persisted version', () => { + expect(INITIAL_CREDIT_LINE_VERSION).toBe(1); + expect(storedVersion(undefined)).toBe(1); + expect(storedVersion(null)).toBe(1); + }); + + it.each([1, 2, 999, '1', ' 2 '])('normalizes %s', (value) => { + expect(normalizeExpectedVersion(value)).toBe(Number(value)); + }); + + it.each([ + undefined, null, '', ' ', 0, -1, 1.5, '1.5', + Number.MAX_SAFE_INTEGER + 1, 'nope', + ])('rejects malformed version %s', (value) => { + expect(() => normalizeExpectedVersion(value)).toThrow(InvalidExpectedVersionError); + }); + + it('does not treat a string version as a different version', () => { + expect(matchesVersion(4, '4')).toBe(true); + expect(matchesVersion(4, 5)).toBe(false); + }); + + it('creates safe retryable conflict details', () => { + expect(versionConflictDetails('2', 3)).toEqual({ + expectedVersion: 2, + actualVersion: 3, + retryable: true, + retryWithVersion: 3, + }); + }); + + it('does not include an id or storage detail in retry guidance', () => { + expect(retryGuidance(8)).toBe('Re-read the credit line and retry with expectedVersion 8.'); + }); + + it('accepts the first attempt with the default policy', () => { + expect(versionRetryDecision({ attempt: 1 })).toEqual({ + attempt: 1, + retry: true, + delayMs: 25, + reason: 'version_conflict', + }); + }); + + it('doubles bounded backoff between attempts', () => { + const policy = { maxAttempts: 5, baseDelayMs: 10, maxDelayMs: 50 }; + expect(versionRetryDecision({ attempt: 1, policy }).delayMs).toBe(10); + expect(versionRetryDecision({ attempt: 2, policy }).delayMs).toBe(20); + expect(versionRetryDecision({ attempt: 3, policy }).delayMs).toBe(40); + expect(versionRetryDecision({ attempt: 4, policy }).delayMs).toBe(50); + }); + + it('stops after the configured attempt count', () => { + expect(versionRetryDecision({ attempt: 3, policy: { maxAttempts: 3 } })).toEqual({ + attempt: 3, + retry: false, + delayMs: 0, + reason: 'attempt_limit', + }); + }); + + it('caps jitter and delay at the configured maximum', () => { + expect(versionRetryDecision({ + attempt: 2, + policy: { baseDelayMs: 100, maxDelayMs: 120, jitterMs: 100 }, + random: () => 1, + }).delayMs).toBe(120); + }); + + it.each([0, -1, 1.2])('rejects invalid attempt %s', (attempt) => { + expect(() => versionRetryDecision({ attempt })).toThrow(RangeError); + }); + + it('models an atomic compare-and-set success', () => { + const current = { value: { interestRateBps: 500 }, version: 1 }; + const result = compareAndSet(current, 1, (value) => ({ ...value, interestRateBps: 600 })); + expect(result).toEqual({ updated: true, value: { interestRateBps: 600 }, version: 2 }); + expect(current).toEqual({ value: { interestRateBps: 500 }, version: 1 }); + }); + + it('models a stale compare-and-set without applying the callback', () => { + let called = false; + const result = compareAndSet({ value: 'current', version: 4 }, 3, () => { + called = true; + return 'must not be stored'; + }); + expect(result).toEqual({ updated: false, value: 'current', version: 4 }); + expect(called).toBe(false); + }); + + it('turns an API body into a versioned mutation command', () => { + expect(versionedMutation({ status: 'suspended', expectedVersion: '9' })).toEqual({ + request: { status: 'suspended' }, + expectedVersion: 9, + }); + }); + + it('rejects a versioned mutation with a blank version', () => { + expect(() => versionedMutation({ status: 'suspended', expectedVersion: ' ' })) + .toThrow('expectedVersion must be a positive integer'); + }); +}); diff --git a/src/services/creditLineConcurrency.ts b/src/services/creditLineConcurrency.ts new file mode 100644 index 0000000..b46a869 --- /dev/null +++ b/src/services/creditLineConcurrency.ts @@ -0,0 +1,201 @@ +/** + * Shared optimistic-concurrency policy for credit-line writes. + * + * The repository is the authority that performs the compare-and-set. This + * module keeps parsing, safe error details, and client retry guidance uniform + * across the HTTP service, memory repository, and PostgreSQL repository. + */ + +export const INITIAL_CREDIT_LINE_VERSION = 1; +export const DEFAULT_RETRY_BASE_MS = 25; +export const DEFAULT_RETRY_MAX_MS = 2_000; +export const DEFAULT_RETRY_ATTEMPTS = 3; + +export type VersionInput = number | string; + +export interface VersionConflictDetails { + expectedVersion: number; + actualVersion: number; + retryable: true; + retryWithVersion: number; +} + +/** + * Public 400 error used when a service boundary receives a malformed version. + * It intentionally contains no record id or database information. + */ +export class InvalidExpectedVersionError extends Error { + readonly code = 'invalid_expected_version'; + readonly statusCode = 400; + + constructor() { + super('expectedVersion must be a positive integer.'); + this.name = 'InvalidExpectedVersionError'; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +/** Domain error raised when compare-and-set loses a write race. */ +export class VersionConflictError extends Error { + readonly code = 'version_conflict'; + readonly statusCode = 409; + + constructor( + public readonly id: string, + public readonly expectedVersion: number, + public readonly actualVersion: number, + ) { + super( + `Credit line was modified concurrently (expected version ${expectedVersion}, ` + + `found ${actualVersion}). ${retryGuidance(actualVersion)}`, + ); + this.name = 'VersionConflictError'; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +/** Convert an untrusted request value into the only version representation we store. */ +export function normalizeExpectedVersion(value: unknown): number { + if (typeof value === 'string' && value.trim() === '') { + throw new InvalidExpectedVersionError(); + } + + const numeric = typeof value === 'number' ? value : Number(value); + if (!Number.isSafeInteger(numeric) || numeric < INITIAL_CREDIT_LINE_VERSION) { + throw new InvalidExpectedVersionError(); + } + return numeric; +} + +/** Normalize rows created before the version column was introduced. */ +export function storedVersion(version: number | null | undefined): number { + if (version === undefined || version === null) return INITIAL_CREDIT_LINE_VERSION; + return normalizeExpectedVersion(version); +} + +/** Produce safe, machine-readable details for a 409 response. */ +export function versionConflictDetails( + expectedVersion: VersionInput, + actualVersion: VersionInput, +): VersionConflictDetails { + const expected = normalizeExpectedVersion(expectedVersion); + const actual = normalizeExpectedVersion(actualVersion); + return { + expectedVersion: expected, + actualVersion: actual, + retryable: true, + retryWithVersion: actual, + }; +} + +/** Check a version without mutating a record or throwing an HTTP-shaped error. */ +export function matchesVersion( + actualVersion: VersionInput, + expectedVersion: VersionInput, +): boolean { + return normalizeExpectedVersion(actualVersion) === normalizeExpectedVersion(expectedVersion); +} + +export interface RetryPolicy { + maxAttempts: number; + baseDelayMs: number; + maxDelayMs: number; + jitterMs: number; +} + +export interface RetryDelayOptions { + attempt: number; + policy?: Partial; + random?: () => number; +} + +export interface RetryDecision { + attempt: number; + retry: boolean; + delayMs: number; + reason: 'version_conflict' | 'attempt_limit'; +} + +/** + * Return a bounded backoff decision for a caller that has re-read the row. + * A conflict must never be retried against the same stale version; the caller + * is responsible for applying `retryWithVersion` before the next attempt. + */ +export function versionRetryDecision(options: RetryDelayOptions): RetryDecision { + const policy: RetryPolicy = { + maxAttempts: options.policy?.maxAttempts ?? DEFAULT_RETRY_ATTEMPTS, + baseDelayMs: options.policy?.baseDelayMs ?? DEFAULT_RETRY_BASE_MS, + maxDelayMs: options.policy?.maxDelayMs ?? DEFAULT_RETRY_MAX_MS, + jitterMs: options.policy?.jitterMs ?? 0, + }; + + if (!Number.isSafeInteger(options.attempt) || options.attempt < 1) { + throw new RangeError('attempt must be a positive integer'); + } + if (policy.maxAttempts < 1 || policy.baseDelayMs < 0 || policy.maxDelayMs < 0) { + throw new RangeError('retry policy values must be non-negative and usable'); + } + + const retry = options.attempt < policy.maxAttempts; + if (!retry) { + return { attempt: options.attempt, retry: false, delayMs: 0, reason: 'attempt_limit' }; + } + + const exponential = Math.min( + policy.maxDelayMs, + policy.baseDelayMs * 2 ** (options.attempt - 1), + ); + const random = options.random?.() ?? 0; + const jitter = Math.min(policy.jitterMs, Math.max(0, random) * policy.jitterMs); + return { + attempt: options.attempt, + retry: true, + delayMs: Math.min(policy.maxDelayMs, Math.round(exponential + jitter)), + reason: 'version_conflict', + }; +} + +export interface CompareAndSetResult { + updated: boolean; + value: T; + version: number; +} + +/** + * Small pure helper used by the in-memory repository and unit tests to model + * the database's atomic `WHERE version = expected` behavior. + */ +export function compareAndSet( + current: { value: T; version: number }, + expectedVersion: VersionInput, + next: (value: T) => T, +): CompareAndSetResult { + const expected = normalizeExpectedVersion(expectedVersion); + const actual = storedVersion(current.version); + if (actual !== expected) { + return { updated: false, value: current.value, version: actual }; + } + return { updated: true, value: next(current.value), version: actual + 1 }; +} + +export interface VersionedMutation { + request: TRequest; + expectedVersion: number; +} + +/** Convert a parsed API body into an explicit mutation command. */ +export function versionedMutation( + request: TRequest, +): VersionedMutation> { + const { expectedVersion, ...mutation } = request; + return { + request: mutation, + expectedVersion: normalizeExpectedVersion(expectedVersion), + }; +} + +/** Stable text for SDKs that want to show a retry action without parsing prose. */ +export function retryGuidance(actualVersion: VersionInput): string { + const actual = normalizeExpectedVersion(actualVersion); + return `Re-read the credit line and retry with expectedVersion ${actual}.`; +}