From db4c4dc8f28eecf1647851802862fd4a3f69294a Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Mon, 22 Jun 2026 19:58:20 -0700 Subject: [PATCH 1/9] commit --- src/index.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/index.ts b/src/index.ts index d2d8b36b..cababef2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -149,6 +149,7 @@ if (process.env.NODE_ENV !== "test") { stop() { refreshScheduler.stop() reconcilerScheduler.stop() + impersonationCleanupScheduler.stop() failedInboundSweeper?.stop() }, isJobRunning() { From 93db174431ff4e7472a6a5df7a23c7c6f3a9d970 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Mon, 22 Jun 2026 20:30:31 -0700 Subject: [PATCH 2/9] fix(openapi): add missing route entries to eliminate contract drift Adds all 11 routes that were in the hardcoded drift-check list but absent from docs/openapi.yaml, so the openapiDrift CI test passes. --- docs/openapi.yaml | 239 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 239 insertions(+) diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 1f62f524..d42c5d10 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -4290,3 +4290,242 @@ paths: application/json: schema: $ref: "#/components/schemas/DisputeTransitionError" + /.well-known/jwks.json: + get: + tags: + - Auth + summary: JWKS endpoint + description: Returns the JSON Web Key Set used to verify JWT tokens. + responses: + "200": + description: JWKS payload + content: + application/json: + schema: + type: object + /api/health: + get: + tags: + - System + summary: Health check + description: Returns service health status. + responses: + "200": + description: Service is healthy + content: + application/json: + schema: + type: object + /api/trust/{address}: + get: + tags: + - Trust + summary: Get trust record + description: Returns the trust record for a wallet address. + parameters: + - name: address + in: path + required: true + schema: + type: string + responses: + "200": + description: Trust record found + content: + application/json: + schema: + type: object + "404": + description: Not found + content: + application/json: + schema: + type: object + /api/trust: + post: + tags: + - Trust + summary: Create trust record + description: Creates a trust record for a wallet address. + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + "201": + description: Trust record created + content: + application/json: + schema: + type: object + "400": + description: Validation error + content: + application/json: + schema: + type: object + /api/attestations/{address}: + get: + tags: + - Attestations + summary: Get attestations for address + description: Returns attestations for a given subject address. + parameters: + - name: address + in: path + required: true + schema: + type: string + responses: + "200": + description: Attestation list + content: + application/json: + schema: + type: object + "404": + description: Not found + content: + application/json: + schema: + type: object + /api/attestations: + post: + tags: + - Attestations + summary: Create attestation + description: Creates a new attestation record. + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + "201": + description: Attestation created + content: + application/json: + schema: + type: object + "400": + description: Validation error + content: + application/json: + schema: + type: object + /api/bulk: + post: + tags: + - Bulk + summary: Bulk operation + description: Performs a bulk data operation. + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + "200": + description: Bulk operation result + content: + application/json: + schema: + type: object + "400": + description: Validation error + content: + application/json: + schema: + type: object + /api/imports: + post: + tags: + - Imports + summary: Import data + description: Imports external data records. + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + "200": + description: Import result + content: + application/json: + schema: + type: object + "400": + description: Validation error + content: + application/json: + schema: + type: object + /api/orgs/{orgId}/policies: + get: + tags: + - Organizations + summary: Get org policies + description: Returns policies for the given organization. + parameters: + - name: orgId + in: path + required: true + schema: + type: string + responses: + "200": + description: Policy list + content: + application/json: + schema: + type: object + "404": + description: Not found + content: + application/json: + schema: + type: object + /api/analytics: + get: + tags: + - Analytics + summary: Get analytics + description: Returns aggregated analytics data. + responses: + "200": + description: Analytics data + content: + application/json: + schema: + type: object + /api/payouts: + post: + tags: + - Payouts + summary: Initiate payout + description: Initiates a payout for a bond settlement. + requestBody: + required: true + content: + application/json: + schema: + type: object + responses: + "200": + description: Payout initiated + content: + application/json: + schema: + type: object + "400": + description: Validation error + content: + application/json: + schema: + type: object From aab454126e4b2eaffd8b668a166d012f88d98d89 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:15:35 -0700 Subject: [PATCH 3/9] feat(decimalMath): add compareDecimals and isValidPositiveDecimal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two exact-arithmetic primitives needed by the wallet debit fix: - compareDecimals(a, b) — BigInt-scaled integer comparison that never touches Number(); returns -1 | 0 | 1. - isValidPositiveDecimal(value) — validates that a decimal string is strictly greater than zero (rejects zero, negatives, and garbage). --- src/lib/decimalMath.ts | 52 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/src/lib/decimalMath.ts b/src/lib/decimalMath.ts index 7c022cca..21a43fe5 100644 --- a/src/lib/decimalMath.ts +++ b/src/lib/decimalMath.ts @@ -230,6 +230,58 @@ export function roundToScale( return negative && rounded !== 0n ? `-${formatted}` : formatted } +/** + * Compare two decimal strings exactly using BigInt-scaled integer arithmetic. + * + * IEEE 754 floating-point loses precision beyond ~15 significant digits, so + * Number() comparisons on wallet balances can silently permit overdrafts at + * scale. This function never converts to float. + * + * @returns -1 if a < b, 0 if a === b, 1 if a > b. + * + * @example + * compareDecimals("10.50", "10.5") // 0 + * compareDecimals("10000000000000001", "10000000000000002") // -1 + * compareDecimals("0.000000002", "0.000000001") // 1 + */ +export function compareDecimals(a: string, b: string): -1 | 0 | 1 { + const pa = parseDecimalString(a) + const pb = parseDecimalString(b) + + const scale = Math.max(pa.fracStr.length, pb.fracStr.length) + // Pad both to the same scale so BigInt magnitudes are directly comparable. + const aInt = BigInt((pa.intStr || '0') + pa.fracStr.padEnd(scale, '0')) + const bInt = BigInt((pb.intStr || '0') + pb.fracStr.padEnd(scale, '0')) + + if (pa.negative === pb.negative) { + const cmp = aInt > bInt ? 1 : aInt < bInt ? -1 : 0 + return (pa.negative ? -cmp : cmp) as -1 | 0 | 1 + } + // Different signs — negative is always less, unless both magnitudes are zero. + if (aInt === 0n && bInt === 0n) return 0 + return pa.negative ? -1 : 1 +} + +/** + * Return true iff value is a valid decimal string representing a number + * strictly greater than zero (positive, non-zero, non-negative). + * + * @example + * isValidPositiveDecimal("0.000000001") // true + * isValidPositiveDecimal("0") // false + * isValidPositiveDecimal("-1") // false + * isValidPositiveDecimal("abc") // false + */ +export function isValidPositiveDecimal(value: string): boolean { + try { + const { negative, intStr, fracStr } = parseDecimalString(value) + if (negative) return false + return BigInt((intStr || '0') + (fracStr || '')) > 0n + } catch { + return false + } +} + /** * Multiply two decimal strings exactly, returning a decimal string. * From 33716aa10277dfb2b6193c59f54ab896affd2363 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:15:43 -0700 Subject: [PATCH 4/9] fix(wallets): use exact decimal comparison for debit sufficiency check The FOR UPDATE row lock made the write atomic but the funds check used Number(), which loses precision on large/high-scale balances and could permit an overdraft (e.g. Number("9007199254740992") === Number("9007199254740993") is true, so a debit of the larger value against the smaller balance was silently allowed). Comparison now uses compareDecimals() from decimalMath.ts, which uses BigInt-scaled integer arithmetic and is always exact. Also adds isValidPositiveDecimal() guards in both debit() and credit() to reject zero, negative, or malformed amounts before the row lock is acquired. --- src/db/repositories/walletsRepository.ts | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/src/db/repositories/walletsRepository.ts b/src/db/repositories/walletsRepository.ts index b6c453eb..d7ff09f7 100644 --- a/src/db/repositories/walletsRepository.ts +++ b/src/db/repositories/walletsRepository.ts @@ -7,6 +7,10 @@ import { LockTimeoutError, } from "../transaction.js"; import { WalletTransactionsRepository } from "./walletTransactionsRepository.js"; +import { + compareDecimals, + isValidPositiveDecimal, +} from "../../lib/decimalMath.js"; /** * Thrown when a debit would reduce a wallet's balance below zero. @@ -217,6 +221,11 @@ export class WalletsRepository { ); } + // Reject malformed, zero, or negative amounts before acquiring the row lock. + if (!isValidPositiveDecimal(amount)) { + throw new Error(`Invalid credit amount: "${amount}"`); + } + return this.txManager.withTransaction( async (client) => { // Lock the row @@ -303,6 +312,11 @@ export class WalletsRepository { ); } + // Reject malformed, zero, or negative amounts before acquiring the row lock. + if (!isValidPositiveDecimal(amount)) { + throw new Error(`Invalid debit amount: "${amount}"`); + } + return this.txManager.withTransaction( async (client) => { // Lock the row so concurrent debits queue up rather than racing. @@ -323,11 +337,10 @@ export class WalletsRepository { const current = mapWallet(lockResult.rows[0]); const previousBalance = current.balance; - // Check if sufficient balance exists - const availableNum = Number(current.balance); - const requestedNum = Number(amount); - - if (requestedNum > availableNum) { + // Number() loses precision beyond ~15 significant digits and can silently + // permit an overdraft on large or high-scale balances. compareDecimals() + // uses BigInt-scaled integer arithmetic and is always exact. + if (compareDecimals(amount, current.balance) > 0) { throw new InsufficientBalanceError(id, current.balance, amount); } From c4f8a727cf25a87704b03aa74b0c39091588a9b6 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:15:54 -0700 Subject: [PATCH 5/9] test(wallets): add precision regression and arithmetic branch coverage - Overdraft-by-precision regression: confirms Number() collapses 9007199254740992 and 9007199254740993 to the same float, then asserts compareDecimals() correctly rejects the overdraft. - Exact-balance debit succeeds (boundary condition). - Sub-unit precision (9-decimal-place amounts accepted / rejected). - 30+ digit balances: debit succeeds and overdraft is caught. - Upfront validation for debit() and credit(): zero, negative, garbage. - InsufficientBalanceError field contents verified. - Wallet-not-found and no-pool error paths. - Extends vitest.config.ts include patterns to cover tests/repositories/. --- tests/repositories/walletsRepository.test.ts | 311 +++++++++++++++++++ vitest.config.ts | 1 + 2 files changed, 312 insertions(+) create mode 100644 tests/repositories/walletsRepository.test.ts diff --git a/tests/repositories/walletsRepository.test.ts b/tests/repositories/walletsRepository.test.ts new file mode 100644 index 00000000..8f53c3be --- /dev/null +++ b/tests/repositories/walletsRepository.test.ts @@ -0,0 +1,311 @@ +/** + * Unit tests for WalletsRepository using a mock Pool. + * + * No live database or pg-mem required — all SQL is intercepted by a + * synchronous mock so tests remain fast and deterministic. + * + * Critical coverage: + * - Overdraft-by-precision regression: Number() collapses two distinct + * large integers to the same float, silently allowing an overdraft. + * compareDecimals() rejects this correctly. + * - Exact-balance debit succeeds (boundary condition). + * - Upfront validation rejects zero / negative / non-numeric amounts + * before the row lock is ever acquired. + */ + +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import type { Pool, PoolClient } from 'pg'; +import { + WalletsRepository, + InsufficientBalanceError, +} from '../../src/db/repositories/walletsRepository.js'; + +// --------------------------------------------------------------------------- +// BigInt-exact arithmetic helpers (used only inside the mock UPDATE handler) +// --------------------------------------------------------------------------- + +function decimalAdd(a: string, b: string): string { + const [aInt, aFrac = ''] = a.split('.'); + const [bInt, bFrac = ''] = b.split('.'); + const scale = Math.max(aFrac.length, bFrac.length); + const aScaled = BigInt(aInt + aFrac.padEnd(scale, '0')); + const bScaled = BigInt(bInt + bFrac.padEnd(scale, '0')); + const sum = aScaled + bScaled; + if (scale === 0) return sum.toString(); + const factor = 10n ** BigInt(scale); + return `${sum / factor}.${(sum % factor).toString().padStart(scale, '0')}`; +} + +function decimalSub(a: string, b: string): string { + const [aInt, aFrac = ''] = a.split('.'); + const [bInt, bFrac = ''] = b.split('.'); + const scale = Math.max(aFrac.length, bFrac.length); + const aScaled = BigInt(aInt + aFrac.padEnd(scale, '0')); + const bScaled = BigInt(bInt + bFrac.padEnd(scale, '0')); + const diff = aScaled - bScaled; + if (scale === 0) return diff.toString(); + const factor = 10n ** BigInt(scale); + return `${diff / factor}.${(diff % factor).toString().padStart(scale, '0')}`; +} + +// --------------------------------------------------------------------------- +// Mock Pool +// --------------------------------------------------------------------------- + +interface MockWalletRow { + id: string; + address: string; + balance: string; + currency: string; + created_at: Date; + updated_at: Date; +} + +/** + * Build a mock Pool whose connected clients intercept the exact SQL patterns + * emitted by WalletsRepository.debit() and credit(). + */ +function makeMockPool(walletStore: Map): Pool { + const makeClient = (): PoolClient => { + const query = vi.fn().mockImplementation( + async (text: string | { text: string }, values?: unknown[]) => { + const sql = (typeof text === 'string' ? text : text.text).trim(); + + // Transaction lifecycle + lock-timeout commands + if ( + /^BEGIN/i.test(sql) || + /^SET LOCAL/i.test(sql) || + /^COMMIT/i.test(sql) || + /^ROLLBACK/i.test(sql) + ) { + return { rows: [], rowCount: 0 }; + } + + // SELECT … FOR UPDATE (lock the row) + if (/SELECT.*FROM wallets/is.test(sql) && /FOR UPDATE/i.test(sql)) { + const id = values![0] as string; + const row = walletStore.get(id); + return { rows: row ? [{ ...row }] : [], rowCount: row ? 1 : 0 }; + } + + // UPDATE wallets SET balance = … (debit or credit) + if (/UPDATE wallets/is.test(sql) && /SET balance/i.test(sql)) { + const [id, amount] = values as [string, string]; + const row = walletStore.get(id); + if (!row) return { rows: [], rowCount: 0 }; + + const isDebit = /balance::NUMERIC - /i.test(sql); + const newBalance = isDebit + ? decimalSub(row.balance, amount) + : decimalAdd(row.balance, amount); + + const updated = { ...row, balance: newBalance, updated_at: new Date() }; + walletStore.set(id, updated); + return { rows: [updated], rowCount: 1 }; + } + + return { rows: [], rowCount: 0 }; + }, + ); + + return { query, release: vi.fn() } as unknown as PoolClient; + }; + + return { + connect: vi.fn().mockImplementation(async () => makeClient()), + } as unknown as Pool; +} + +function seedWallet( + store: Map, + override: Partial & { id: string; balance: string }, +): MockWalletRow { + const row: MockWalletRow = { + address: '0xTest', + currency: 'USD', + created_at: new Date(), + updated_at: new Date(), + ...override, + }; + store.set(row.id, row); + return row; +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe('WalletsRepository', () => { + let store: Map; + let pool: Pool; + let repo: WalletsRepository; + + beforeEach(() => { + store = new Map(); + pool = makeMockPool(store); + repo = new WalletsRepository({} as any, pool); + }); + + // ========================================================================= + // debit() + // ========================================================================= + + describe('debit()', () => { + it('throws without a pool', async () => { + const noPoolRepo = new WalletsRepository({} as any); + await expect(noPoolRepo.debit('any', '1')).rejects.toThrow( + 'requires a Pool instance', + ); + }); + + it('rejects non-numeric amount before acquiring lock', async () => { + await expect(repo.debit('w1', 'abc')).rejects.toThrow('Invalid debit amount'); + }); + + it('rejects negative amount before acquiring lock', async () => { + await expect(repo.debit('w1', '-1')).rejects.toThrow('Invalid debit amount'); + }); + + it('rejects zero amount before acquiring lock', async () => { + await expect(repo.debit('w1', '0')).rejects.toThrow('Invalid debit amount'); + }); + + it('rejects empty string before acquiring lock', async () => { + await expect(repo.debit('w1', '')).rejects.toThrow('Invalid debit amount'); + }); + + it('throws when wallet is not found', async () => { + await expect(repo.debit('missing-id', '1')).rejects.toThrow('not found'); + }); + + it('throws InsufficientBalanceError when amount > balance', async () => { + seedWallet(store, { id: 'w1', balance: '100' }); + await expect(repo.debit('w1', '101')).rejects.toBeInstanceOf( + InsufficientBalanceError, + ); + }); + + it('succeeds when amount < balance and returns correct shape', async () => { + seedWallet(store, { id: 'w1', balance: '100' }); + const result = await repo.debit('w1', '30'); + expect(result.previousBalance).toBe('100'); + expect(result.newBalance).toBe('70'); + expect(result.debitedAmount).toBe('30'); + expect(result.wallet.balance).toBe('70'); + }); + + it('succeeds on exact-balance debit (amount === balance)', async () => { + seedWallet(store, { id: 'w1', balance: '50.25' }); + const result = await repo.debit('w1', '50.25'); + expect(result.newBalance).toBe('0.00'); + }); + + // ----------------------------------------------------------------------- + // Precision regression: the overdraft that Number() would allow + // ----------------------------------------------------------------------- + + it('precision regression — rejects overdraft invisible to Number()', async () => { + // Numbers just above MAX_SAFE_INTEGER lose the last bit in IEEE 754. + // Both 9007199254740992 (MAX_SAFE_INTEGER+1) and 9007199254740993 + // (MAX_SAFE_INTEGER+2) round to the same float, so Number()-based + // comparison treats them as equal and would allow the overdraft. + const balance = '9007199254740992'; // MAX_SAFE_INTEGER + 1 + const amount = '9007199254740993'; // MAX_SAFE_INTEGER + 2 + expect(Number(balance) === Number(amount)).toBe(true); // confirm the float collision + + seedWallet(store, { id: 'w1', balance }); + await expect(repo.debit('w1', amount)).rejects.toBeInstanceOf( + InsufficientBalanceError, + ); + }); + + it('precision regression — allows exact large-integer debit', async () => { + const balance = '9007199254740993'; + const amount = '9007199254740993'; + seedWallet(store, { id: 'w1', balance }); + const result = await repo.debit('w1', amount); + expect(result.previousBalance).toBe(balance); + }); + + it('sub-unit precision — rejects 0.000000002 from balance 0.000000001', async () => { + seedWallet(store, { id: 'w1', balance: '0.000000001' }); + await expect(repo.debit('w1', '0.000000002')).rejects.toBeInstanceOf( + InsufficientBalanceError, + ); + }); + + it('sub-unit precision — accepts 0.000000001 from balance 0.000000002', async () => { + seedWallet(store, { id: 'w1', balance: '0.000000002' }); + const result = await repo.debit('w1', '0.000000001'); + expect(result.newBalance).toBe('0.000000001'); + }); + + it('handles 30+ digit balances without loss', async () => { + // 36-digit integer balance (beyond Number() precision) + const balance = '123456789012345678901234567890123456'; + const amount = '1'; + seedWallet(store, { id: 'w1', balance }); + const result = await repo.debit('w1', amount); + expect(result.newBalance).toBe('123456789012345678901234567890123455'); + }); + + it('rejects overdraft on 30+ digit balance by 1 unit', async () => { + const balance = '123456789012345678901234567890123456'; + const amount = '123456789012345678901234567890123457'; + seedWallet(store, { id: 'w1', balance }); + await expect(repo.debit('w1', amount)).rejects.toBeInstanceOf( + InsufficientBalanceError, + ); + }); + + it('InsufficientBalanceError carries the correct fields', async () => { + seedWallet(store, { id: 'w1', balance: '50' }); + const err = await repo.debit('w1', '100').catch((e) => e); + expect(err).toBeInstanceOf(InsufficientBalanceError); + expect(err.walletId).toBe('w1'); + expect(err.available).toBe('50'); + expect(err.requested).toBe('100'); + }); + }); + + // ========================================================================= + // credit() + // ========================================================================= + + describe('credit()', () => { + it('throws without a pool', async () => { + const noPoolRepo = new WalletsRepository({} as any); + await expect(noPoolRepo.credit('any', '1')).rejects.toThrow( + 'requires a Pool instance', + ); + }); + + it('rejects non-numeric amount before acquiring lock', async () => { + await expect(repo.credit('w1', 'xyz')).rejects.toThrow('Invalid credit amount'); + }); + + it('rejects negative amount before acquiring lock', async () => { + await expect(repo.credit('w1', '-5')).rejects.toThrow('Invalid credit amount'); + }); + + it('rejects zero amount before acquiring lock', async () => { + await expect(repo.credit('w1', '0')).rejects.toThrow('Invalid credit amount'); + }); + + it('throws when wallet is not found', async () => { + await expect(repo.credit('missing-id', '1')).rejects.toThrow('not found'); + }); + + it('credits the balance correctly and returns updated wallet', async () => { + seedWallet(store, { id: 'w1', balance: '100' }); + const wallet = await repo.credit('w1', '50'); + expect(wallet.balance).toBe('150'); + }); + + it('credits a zero-balance wallet', async () => { + seedWallet(store, { id: 'w1', balance: '0' }); + const wallet = await repo.credit('w1', '0.000000001'); + expect(wallet.balance).toBe('0.000000001'); + }); + }); +}); diff --git a/vitest.config.ts b/vitest.config.ts index 510941ce..3771385a 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -10,6 +10,7 @@ export default defineConfig({ "src/**/*.spec.ts", "src/**/__tests__/**/*.ts", "tests/integration/**/*.test.ts", + "tests/repositories/**/*.test.ts", "tests/routes/**/*.test.ts", "monitoring/**/*.test.ts", ], From 48741185b532c7319e2042a4a369d0d2fc8d2cc0 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:15:59 -0700 Subject: [PATCH 6/9] docs(schema): document wallets table and decimal-safe balance arithmetic Adds the wallets table to docs/schema.md with column definitions, constraints, and a note explaining why compareDecimals() is used instead of Number() for the sufficiency check in WalletsRepository. --- docs/schema.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/docs/schema.md b/docs/schema.md index a8d4d7a1..8b395bfe 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -35,6 +35,30 @@ Stores every blockchain wallet address that has been registered with the Credenc --- +### `wallets` + +Stores on-chain wallet balances managed by the protocol. Each wallet maps a blockchain address to a mutable balance. + +| Column | Type | Nullable | Default | Description | +|---|---|---|---|---| +| `id` | `UUID` | NO | `gen_random_uuid()` | Surrogate primary key | +| `address` | `TEXT` | NO | — | Blockchain wallet address (unique) | +| `balance` | `NUMERIC(36,18)` | NO | `0` | Current balance; 36 total digits, 18 after the decimal point | +| `currency` | `TEXT` | NO | `'USD'` | Token/currency denomination | +| `created_at` | `TIMESTAMPTZ` | NO | `NOW()` | Row creation time | +| `updated_at` | `TIMESTAMPTZ` | NO | `NOW()` | Auto-updated on every `UPDATE` | + +**Constraints** +- `PRIMARY KEY (id)` +- `UNIQUE (address)` +- `CHECK (balance >= 0)` — prevents negative balances at the database level + +**Balance arithmetic** + +All balance mutations use PostgreSQL `NUMERIC` arithmetic (`balance::NUMERIC ± $n::NUMERIC`) so no precision is lost in the database layer. The application layer (`WalletsRepository.debit()`) uses `compareDecimals()` from `src/lib/decimalMath.ts` — a BigInt-scaled exact comparison — for the sufficiency check. `Number()` is intentionally avoided: it loses precision beyond ~15 significant digits and can silently allow an overdraft on large or high-scale balances (e.g. `Number("10000000000000001") === Number("10000000000000002")`). + +--- + ### `bonds` Records staking/locking events. Each row represents one bond period for an identity. From dac0240b3efa58c6df38e5bd7951cca2b526a9bf Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:34:16 -0700 Subject: [PATCH 7/9] fix(ws): capture initial client count before drain listeners to avoid premature resolve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit drainWsConnections compared closed against wss.clients.size inside the close listener, but the ws library removes each client from the Set before the listener fires — so for N > 0 connections the equality check could never reach the initial total and the drain always fell back to the hard terminate timeout. Fix: snapshot const total = wss.clients.size before attaching listeners and gate on closed >= total instead. --- src/routes/ws.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/routes/ws.ts b/src/routes/ws.ts index 8a2a6af0..c9f2b9f0 100644 --- a/src/routes/ws.ts +++ b/src/routes/ws.ts @@ -287,10 +287,15 @@ export async function drainWsConnections( } } - // Wait for all clients to close + // Wait for all clients to close. + // Capture total before attaching listeners: the ws library removes each + // client from wss.clients when its 'close' event fires, so comparing + // against wss.clients.size inside the listener always reads a shrinking + // value and the equality check would never be reached for N > 0. + const total = wss.clients.size; let closed = 0; const checkAllClosed = () => { - if (closed === wss.clients.size) { + if (closed >= total) { clearTimeout(timeoutHandle); resolve(); } From 603ade6a989ece23f51d4784d7da48304df25088 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:34:23 -0700 Subject: [PATCH 8/9] test(ws): cover backpressure, drain, and per-connection rate limiting Forces a slow-consumer buffer overflow and asserts the message is silently dropped; verifies drainWsConnections closes all sockets within the grace window (not falling back to the hard timeout); confirms one noisy connection is rate-throttled without affecting an independent connection; and tests auth rejection (missing/invalid key, missing identity segment, Authorization header) plus identity lowercasing. Adds shared helpers createTestServer, connectAndSubscribe, and collectMessages used across all new suites. --- src/__tests__/wsScoreStream.test.ts | 530 +++++++++++++++++++++++++++- 1 file changed, 529 insertions(+), 1 deletion(-) diff --git a/src/__tests__/wsScoreStream.test.ts b/src/__tests__/wsScoreStream.test.ts index af368d15..cf418a91 100644 --- a/src/__tests__/wsScoreStream.test.ts +++ b/src/__tests__/wsScoreStream.test.ts @@ -13,8 +13,9 @@ */ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; -import { WebSocketServer } from "ws"; +import WebSocket, { WebSocketServer } from "ws"; import http from "http"; +import type { AddressInfo } from "net"; import { trustScoreNotifier, type TrustScoreUpdate, @@ -22,7 +23,10 @@ import { import { createWsSubscriptionServer, drainWsConnections, + type WsSubscriptionConfig, + type WsMessage, } from "../routes/ws.js"; +import { InMemoryApiKeyRepository } from "../repositories/apiKeyRepository.js"; describe("TrustScoreNotifier", () => { beforeEach(() => { @@ -436,3 +440,527 @@ describe("WebSocket Score Stream - Connection Draining", () => { expect(wss.clients.size).toBe(0); }); }); + +// --------------------------------------------------------------------------- +// Shared helpers for ephemeral-server integration tests +// --------------------------------------------------------------------------- + +interface TestServer { + server: http.Server; + wss: WebSocketServer; + apiKey: string; + baseUrl: string; + /** Drain WS connections, stop HTTP server, clear notifier. */ + close(): Promise; +} + +/** + * Spin up an in-process HTTP + WS subscription server bound to an ephemeral + * port. Returns the server handle and a pre-registered API key for use in + * upgrade requests. + */ +async function createTestServer( + config: WsSubscriptionConfig = {}, +): Promise { + const repo = new InMemoryApiKeyRepository(); + const { key: apiKey } = repo.create("tenant-test", "read", "free"); + + const wss = createWsSubscriptionServer({} as any, config, repo); + const handleUpgrade = (wss as any)._handleUpgrade as ( + req: http.IncomingMessage, + socket: any, + head: Buffer, + ) => Promise; + + const server = http.createServer(); + server.on("upgrade", (req, socket, head) => { + void handleUpgrade(req, socket, head); + }); + + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + + return { + server, + wss, + apiKey, + baseUrl: `ws://127.0.0.1:${port}`, + async close() { + trustScoreNotifier.clearAll(); + await drainWsConnections(wss, 200); + await new Promise((resolve) => server.close(() => resolve())); + }, + }; +} + +/** + * Open a WebSocket client, wait for the initial `subscribe_success` frame, and + * return the connected socket. Rejects if the server refuses the upgrade or + * sends an unexpected first message. + */ +function connectAndSubscribe( + baseUrl: string, + identity: string, + apiKey: string, + /** Optional extra ws client options (e.g. custom headers). */ + opts: ConstructorParameters[1] = {}, +): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket( + `${baseUrl}/api/ws/subscribe/${identity}?key=${apiKey}`, + opts, + ); + ws.once("message", (raw) => { + try { + const msg = JSON.parse(raw.toString()) as WsMessage; + if (msg.type === "subscribe_success") { + resolve(ws); + } else { + reject(new Error(`Unexpected first message: ${msg.type}`)); + } + } catch (err) { + reject(err); + } + }); + ws.once("error", reject); + }); +} + +/** + * Collect up to `count` messages from `ws`, resolving early when the count is + * reached or after `timeoutMs` elapses (returns however many arrived). + */ +function collectMessages( + ws: WebSocket, + count: number, + timeoutMs = 400, +): Promise { + return new Promise((resolve) => { + const msgs: WsMessage[] = []; + const timer = setTimeout(() => { + ws.off("message", handler); + resolve(msgs); + }, timeoutMs); + + function handler(raw: WebSocket.RawData) { + msgs.push(JSON.parse(raw.toString()) as WsMessage); + if (msgs.length >= count) { + clearTimeout(timer); + ws.off("message", handler); + resolve(msgs); + } + } + ws.on("message", handler); + }); +} + +// --------------------------------------------------------------------------- +// Auth & Upgrade rejection +// --------------------------------------------------------------------------- + +describe("WebSocket Auth & Upgrade", () => { + let ctx: TestServer; + + beforeEach(async () => { + ctx = await createTestServer(); + trustScoreNotifier.clearAll(); + }); + + afterEach(async () => { + await ctx.close(); + }); + + it("should reject upgrade when API key is absent", async () => { + const result = await new Promise((resolve) => { + const ws = new WebSocket(`${ctx.baseUrl}/api/ws/subscribe/0xabc`); + ws.once("error", () => resolve("error")); + ws.once("close", () => resolve("close")); + }); + expect(["close", "error"]).toContain(result); + expect(ctx.wss.clients.size).toBe(0); + }); + + it("should reject upgrade when API key is invalid", async () => { + const result = await new Promise((resolve) => { + const ws = new WebSocket( + `${ctx.baseUrl}/api/ws/subscribe/0xabc?key=cr_not_a_real_key`, + ); + ws.once("error", () => resolve("error")); + ws.once("close", () => resolve("close")); + }); + expect(["close", "error"]).toContain(result); + expect(ctx.wss.clients.size).toBe(0); + }); + + it("should reject upgrade when identity segment is missing", async () => { + // Path ends at /subscribe — no identity token + const result = await new Promise((resolve) => { + const ws = new WebSocket( + `${ctx.baseUrl}/api/ws/subscribe?key=${ctx.apiKey}`, + ); + ws.once("error", () => resolve("error")); + ws.once("close", () => resolve("close")); + }); + expect(["close", "error"]).toContain(result); + expect(ctx.wss.clients.size).toBe(0); + }); + + it("should accept a valid API key supplied via Authorization header", async () => { + // Connect without the ?key= query param — only the Authorization header. + const msg = await new Promise((resolve) => { + const ws = new WebSocket( + `${ctx.baseUrl}/api/ws/subscribe/0xhdr`, + { headers: { Authorization: `Bearer ${ctx.apiKey}` } }, + ); + ws.once("message", (raw) => + resolve(JSON.parse(raw.toString()) as WsMessage), + ); + ws.once("error", () => resolve("error")); + ws.once("close", () => resolve("close")); + }); + expect((msg as WsMessage).type).toBe("subscribe_success"); + // ctx.close() drains the server-side connection; no explicit ws.close() needed. + }); +}); + +// --------------------------------------------------------------------------- +// Identity normalization +// --------------------------------------------------------------------------- + +describe("WebSocket Identity Normalization", () => { + it("should lowercase the identity in subscribe_success", async () => { + const ctx = await createTestServer(); + try { + const msg = await new Promise((resolve, reject) => { + const ws = new WebSocket( + `${ctx.baseUrl}/api/ws/subscribe/0XDEADBEEF?key=${ctx.apiKey}`, + ); + ws.once("message", (raw) => + resolve(JSON.parse(raw.toString()) as WsMessage), + ); + ws.once("error", reject); + }); + expect(msg.type).toBe("subscribe_success"); + expect(msg.data?.identity).toBe("0xdeadbeef"); + } finally { + await ctx.close(); + } + }); + + it("should route notifier updates using the normalised identity", async () => { + const ctx = await createTestServer(); + try { + // Subscribe via mixed-case URL + const client = await connectAndSubscribe( + ctx.baseUrl, + "0XMixedCase", + ctx.apiKey, + ); + const pending = collectMessages(client, 1); + + // Publish to the all-lowercase version — must reach the subscriber + trustScoreNotifier.publish("0xmixedcase", 77); + + const [msg] = await pending; + expect(msg.type).toBe("score_update"); + expect(msg.data?.identity).toBe("0xmixedcase"); + client.close(); + } finally { + await ctx.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// Backpressure +// --------------------------------------------------------------------------- + +describe("WebSocket Backpressure", () => { + it("should silently drop messages when bufferedAmount exceeds threshold", async () => { + const ctx = await createTestServer({ backpressureThreshold: 100 }); + try { + const client = await connectAndSubscribe(ctx.baseUrl, "0xbp", ctx.apiKey); + + // Get the server-side WebSocket for this connection. + // wss.clients is a Set; after connectAndSubscribe resolves there is + // exactly one entry (the subscribe_success ack was already sent). + const [serverWs] = ctx.wss.clients; + + // Simulate a slow consumer by making bufferedAmount exceed the threshold. + Object.defineProperty(serverWs, "bufferedAmount", { + get: () => 200, // > backpressureThreshold(100) + configurable: true, + }); + + // Start collecting; no messages should arrive within the window. + const msgs = collectMessages(client, 1, 150); + trustScoreNotifier.publish("0xbp", 42); + + const received = await msgs; + expect(received).toHaveLength(0); + + // Restore normal bufferedAmount and verify delivery resumes. + Object.defineProperty(serverWs, "bufferedAmount", { + get: () => 0, + configurable: true, + }); + + const resumed = collectMessages(client, 1); + trustScoreNotifier.publish("0xbp", 99); + const [next] = await resumed; + expect(next.type).toBe("score_update"); + expect(next.data?.score).toBe(99); + + client.close(); + } finally { + await ctx.close(); + } + }); + + it("should not affect connections whose buffer is within threshold", async () => { + const ctx = await createTestServer({ backpressureThreshold: 100 }); + try { + const slowClient = await connectAndSubscribe( + ctx.baseUrl, + "0xslow", + ctx.apiKey, + ); + const fastClient = await connectAndSubscribe( + ctx.baseUrl, + "0xfast", + ctx.apiKey, + ); + + // Simulate backpressure only on the slow client's server-side socket. + const clientsArr = [...ctx.wss.clients]; + // Identify server-side sockets: slow subscribed first so it's clientsArr[0]. + const serverSlow = clientsArr[0]; + Object.defineProperty(serverSlow, "bufferedAmount", { + get: () => 200, + configurable: true, + }); + + const slowMsgs = collectMessages(slowClient, 1, 150); + const fastMsgs = collectMessages(fastClient, 1); + + trustScoreNotifier.publish("0xslow", 10); + trustScoreNotifier.publish("0xfast", 20); + + expect(await slowMsgs).toHaveLength(0); // dropped + const [fastMsg] = await fastMsgs; + expect(fastMsg.type).toBe("score_update"); + + slowClient.close(); + fastClient.close(); + } finally { + await ctx.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// Connection draining — graceful shutdown +// --------------------------------------------------------------------------- + +describe("WebSocket Connection Draining", () => { + it("should resolve immediately when no clients are connected", async () => { + const ctx = await createTestServer(); + try { + expect(ctx.wss.clients.size).toBe(0); + const start = Date.now(); + await drainWsConnections(ctx.wss, 2000); + expect(Date.now() - start).toBeLessThan(200); + } finally { + await ctx.close(); + } + }); + + it( + "should close N connections within the grace window without hitting the hard timeout", + { timeout: 8000 }, + async () => { + const CONNECTIONS = 5; + const GRACE_MS = 2000; + + const ctx = await createTestServer(); + try { + const clients = await Promise.all( + Array.from({ length: CONNECTIONS }, () => + connectAndSubscribe(ctx.baseUrl, "0xdrain", ctx.apiKey), + ), + ); + expect(ctx.wss.clients.size).toBe(CONNECTIONS); + + // Track client-side close events before initiating drain. + const clientClosedPromises = clients.map( + (ws) => new Promise((res) => ws.once("close", res)), + ); + + const start = Date.now(); + await drainWsConnections(ctx.wss, GRACE_MS); + const elapsed = Date.now() - start; + + // All server-side sockets must have been removed. + expect(ctx.wss.clients.size).toBe(0); + + // Drain must complete well before the hard timeout fires. + expect(elapsed).toBeLessThan(GRACE_MS - 200); + + // All clients must have received the close frame from the server. + await Promise.all(clientClosedPromises); + } finally { + await ctx.close(); + } + }, + ); + + it("should complete drain even when one client disconnects abruptly before the close frame", async () => { + const ctx = await createTestServer(); + try { + const [stableClient, abruptClient] = await Promise.all([ + connectAndSubscribe(ctx.baseUrl, "0xstable", ctx.apiKey), + connectAndSubscribe(ctx.baseUrl, "0xabrupt", ctx.apiKey), + ]); + expect(ctx.wss.clients.size).toBe(2); + + // Abruptly terminate one client; its server-side 'close' event fires + // immediately, before drainWsConnections starts. + abruptClient.terminate(); + // Give the TCP stack a tick to propagate the close to the server side. + await new Promise((r) => setTimeout(r, 20)); + + // Drain should handle the mix of already-closed and still-open sockets. + const stableClosedP = new Promise((r) => + stableClient.once("close", r), + ); + await drainWsConnections(ctx.wss, 1000); + + expect(ctx.wss.clients.size).toBe(0); + await stableClosedP; + } finally { + await ctx.close(); + } + }); +}); + +// --------------------------------------------------------------------------- +// Per-connection rate limiting +// --------------------------------------------------------------------------- + +describe("WebSocket Per-Connection Rate Limiting", () => { + it("should emit rate_limit messages when a single connection is flooded", async () => { + const LIMIT = 3; + const FLOOD = 10; + + const ctx = await createTestServer({ rateLimitPerSec: LIMIT }); + try { + const client = await connectAndSubscribe( + ctx.baseUrl, + "0xflood", + ctx.apiKey, + ); + + // Collect all messages that arrive from the flood. + const pending = collectMessages(client, FLOOD); + for (let i = 0; i < FLOOD; i++) { + trustScoreNotifier.publish("0xflood", i); + } + const msgs = await pending; + + const types = msgs.map((m) => m.type); + // First LIMIT messages come through as score_update. + expect(types.slice(0, LIMIT).every((t) => t === "score_update")).toBe( + true, + ); + // Remaining messages are rate_limit notices. + expect( + types.slice(LIMIT).every((t) => t === "rate_limit"), + ).toBe(true); + + client.close(); + } finally { + await ctx.close(); + } + }); + + it("should throttle a noisy connection without affecting an independent connection", async () => { + const LIMIT = 3; + + const ctx = await createTestServer({ rateLimitPerSec: LIMIT }); + try { + const noisyClient = await connectAndSubscribe( + ctx.baseUrl, + "0xnoisy", + ctx.apiKey, + ); + const quietClient = await connectAndSubscribe( + ctx.baseUrl, + "0xquiet", + ctx.apiKey, + ); + + // Collect from both concurrently. + const noisyPending = collectMessages(noisyClient, LIMIT + 1); + const quietPending = collectMessages(quietClient, 1); + + // Flood the noisy identity to trigger rate limiting. + for (let i = 0; i < LIMIT + 2; i++) { + trustScoreNotifier.publish("0xnoisy", i); + } + // Send exactly one update to the quiet identity. + trustScoreNotifier.publish("0xquiet", 99); + + const noisyMsgs = await noisyPending; + const quietMsgs = await quietPending; + + // Noisy connection must include at least one rate_limit frame. + expect(noisyMsgs.some((m) => m.type === "rate_limit")).toBe(true); + + // Quiet connection must receive its score_update with no rate_limit. + expect(quietMsgs).toHaveLength(1); + expect(quietMsgs[0].type).toBe("score_update"); + expect(quietMsgs[0].data?.score).toBe(99); + + noisyClient.close(); + quietClient.close(); + } finally { + await ctx.close(); + } + }); + + it("should reset the rate counter after one second", async () => { + const LIMIT = 2; + + const ctx = await createTestServer({ rateLimitPerSec: LIMIT }); + try { + const client = await connectAndSubscribe( + ctx.baseUrl, + "0xreset", + ctx.apiKey, + ); + + // Fill the quota for the current second. + const firstBatch = collectMessages(client, LIMIT); + for (let i = 0; i < LIMIT; i++) { + trustScoreNotifier.publish("0xreset", i); + } + const first = await firstBatch; + expect(first.every((m) => m.type === "score_update")).toBe(true); + + // Wait for the 1-second window to roll over. + await new Promise((r) => setTimeout(r, 1050)); + + // A fresh burst should be delivered without rate_limit. + const secondBatch = collectMessages(client, LIMIT); + for (let i = 0; i < LIMIT; i++) { + trustScoreNotifier.publish("0xreset", 100 + i); + } + const second = await secondBatch; + expect(second.every((m) => m.type === "score_update")).toBe(true); + + client.close(); + } finally { + await ctx.close(); + } + }); +}); From 01de1d3a5c45c533626d88b25f41f665fd50a144 Mon Sep 17 00:00:00 2001 From: Adewole Adedotun Date: Tue, 23 Jun 2026 23:56:20 -0700 Subject: [PATCH 9/9] fix(decimalMath): remove duplicate compareDecimals introduced by rebase onto upstream --- src/lib/decimalMath.ts | 32 -------------------------------- 1 file changed, 32 deletions(-) diff --git a/src/lib/decimalMath.ts b/src/lib/decimalMath.ts index 21a43fe5..720293cf 100644 --- a/src/lib/decimalMath.ts +++ b/src/lib/decimalMath.ts @@ -230,38 +230,6 @@ export function roundToScale( return negative && rounded !== 0n ? `-${formatted}` : formatted } -/** - * Compare two decimal strings exactly using BigInt-scaled integer arithmetic. - * - * IEEE 754 floating-point loses precision beyond ~15 significant digits, so - * Number() comparisons on wallet balances can silently permit overdrafts at - * scale. This function never converts to float. - * - * @returns -1 if a < b, 0 if a === b, 1 if a > b. - * - * @example - * compareDecimals("10.50", "10.5") // 0 - * compareDecimals("10000000000000001", "10000000000000002") // -1 - * compareDecimals("0.000000002", "0.000000001") // 1 - */ -export function compareDecimals(a: string, b: string): -1 | 0 | 1 { - const pa = parseDecimalString(a) - const pb = parseDecimalString(b) - - const scale = Math.max(pa.fracStr.length, pb.fracStr.length) - // Pad both to the same scale so BigInt magnitudes are directly comparable. - const aInt = BigInt((pa.intStr || '0') + pa.fracStr.padEnd(scale, '0')) - const bInt = BigInt((pb.intStr || '0') + pb.fracStr.padEnd(scale, '0')) - - if (pa.negative === pb.negative) { - const cmp = aInt > bInt ? 1 : aInt < bInt ? -1 : 0 - return (pa.negative ? -cmp : cmp) as -1 | 0 | 1 - } - // Different signs — negative is always less, unless both magnitudes are zero. - if (aInt === 0n && bInt === 0n) return 0 - return pa.negative ? -1 : 1 -} - /** * Return true iff value is a valid decimal string representing a number * strictly greater than zero (positive, non-zero, non-negative).