diff --git a/README.md b/README.md index 6a8825c..153637d 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,23 @@ cd anchornet-backend Install dependencies npm install +Production persistence + +Production requires PostgreSQL and a `DATABASE_URL`. Apply the checked-in +migrations before starting the API: + +```sh +DATABASE_URL=postgres://user:password@host:5432/anchornet npm run migrate:up +DATABASE_URL=postgres://user:password@host:5432/anchornet npm start +``` + +The process checks the database and hydrates anchors, liquidity, and +settlements before it binds its HTTP port. It exits instead of serving with an +empty in-memory state when PostgreSQL is unavailable. Settlement reservations +are checked and inserted in one transaction with row locks, so concurrent API +instances cannot reserve the same liquidity twice. Development and Jest keep +the existing in-memory repositories when `DATABASE_URL` is omitted. + Run in development npm run dev Server runs at http://localhost:3001 by default. Set PORT to override. @@ -319,4 +336,4 @@ Fork the repo and create a branch from main. Install deps: npm install. Run tests: npm test; lint: npm run lint. Open a pull request. CI runs lint, build, and tests on push/PR to main. License -MIT \ No newline at end of file +MIT diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 01569d9..c796d1a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -16,10 +16,34 @@ Sensitive Data Redaction & Security Guarantees Strict Redaction via Denylist: Any header, body parameter, or metadata stored in audit log entries is processed through redactSensitiveData(). Denylisted Fields: Secret-bearing keys such as x-api-key, authorization, cookie, set-cookie, token, access_token, refresh_token, secret, password, bearer, private_key, client_secret are matched case-insensitively and replaced with "[REDACTED]". Preventing Plaintext Exposure: Under no circumstances should raw credentials or API keys be captured or retained in plaintext in the in-memory audit ring buffer or exposed via GET /api/v1/audit. -In-Memory Repositories & Future Persistence -Settlement, anchor, and liquidity data are held in process-local in-memory -repositories (src/repositories/*), all extending the shared -InMemoryRepository base class. +Durable PostgreSQL Persistence +In production, `DATABASE_URL` is required. `src/index.ts` reaches PostgreSQL, +loads all three aggregates, and only then constructs the HTTP app and binds a +port. The persistent repository facades preserve the existing synchronous +domain-service contract for reads while serializing accepted writes and +flushing them before graceful shutdown. Development and Jest retain the +in-memory repositories when no database URL is configured. + +The migration in `migrations/001_initial_persistence.js` defines anchors, +liquidity entries, and settlements with foreign keys, numeric(78,0) amounts, +status/amount constraints, and query indexes. Amounts cross the database +boundary as strings and are converted to `bigint`; no financial value is +converted through JavaScript `number`. + +Settlement opening is a dedicated database transaction. It locks every +liquidity row for the requested asset in stable anchor order, locks the +pending/executed settlement rows for that asset, calculates committed value, +and inserts the new pending row before committing. This prevents two API +instances from both observing the same remaining capacity. Execute/cancel +also use conditional transactional updates so a pending settlement can only +transition once. + +In-Memory Repositories & Test Double +When `DATABASE_URL` is absent, settlement, anchor, and liquidity data are held +in process-local repositories (src/repositories/*), all extending the shared +InMemoryRepository base class. This keeps unit and HTTP tests deterministic +without requiring a live database; production cannot use this fallback because +configuration validation requires `DATABASE_URL` under `NODE_ENV=production`. Idempotency cache (src/middleware/idempotency.ts) follows the same sequencing: a process-wide `MemoryIdempotencyStore` (shared across mounts, hard-capped, @@ -104,4 +128,4 @@ top-level settlement export so the two column constants cannot diverge. CSV_COLUMNS (and, for settlements, to both the settlements route and the nested anchors route) and to the expected-column lists in the tests. Treat a failure in either guardrail as a real export regression, not as a test to -loosen. \ No newline at end of file +loosen. diff --git a/migrations/001_initial_persistence.js b/migrations/001_initial_persistence.js new file mode 100644 index 0000000..a0c77e9 --- /dev/null +++ b/migrations/001_initial_persistence.js @@ -0,0 +1,49 @@ +/* eslint-disable no-undef */ +/** @param {import('node-pg-migrate').MigrationBuilder} pgm */ +exports.up = (pgm) => { + pgm.createTable("anchors", { + id: { type: "text", primaryKey: true }, + name: { type: "text", notNull: true }, + registered_at: { type: "timestamptz", notNull: true }, + active: { type: "boolean", notNull: true, default: true }, + }); + + pgm.createTable("liquidity_entries", { + anchor: { type: "text", notNull: true, references: "anchors(id)", onDelete: "RESTRICT" }, + asset: { type: "text", notNull: true }, + amount: { type: "numeric(78,0)", notNull: true }, + updated_at: { type: "timestamptz", notNull: true }, + }); + pgm.addConstraint("liquidity_entries", "liquidity_entries_pkey", { + primaryKey: ["anchor", "asset"], + }); + pgm.addConstraint("liquidity_entries", "liquidity_amount_nonnegative", { + check: "amount >= 0", + }); + pgm.createIndex("liquidity_entries", ["asset", "anchor"]); + + pgm.createTable("settlements", { + id: { type: "bigserial", primaryKey: true }, + anchor: { type: "text", notNull: true, references: "anchors(id)", onDelete: "RESTRICT" }, + asset: { type: "text", notNull: true }, + amount: { type: "numeric(78,0)", notNull: true }, + fee: { type: "numeric(78,0)", notNull: true }, + status: { type: "text", notNull: true, default: "pending" }, + created_at: { type: "timestamptz", notNull: true }, + cancel_reason: { type: "text" }, + }); + pgm.addConstraint("settlements", "settlement_amount_positive", { check: "amount > 0" }); + pgm.addConstraint("settlements", "settlement_fee_nonnegative", { check: "fee >= 0" }); + pgm.addConstraint("settlements", "settlement_status_valid", { + check: "status IN ('pending', 'executed', 'cancelled')", + }); + pgm.createIndex("settlements", ["anchor", "id"]); + pgm.createIndex("settlements", ["asset", "id"]); + pgm.createIndex("settlements", ["status", "id"]); +}; + +exports.down = (pgm) => { + pgm.dropTable("settlements"); + pgm.dropTable("liquidity_entries"); + pgm.dropTable("anchors"); +}; diff --git a/migrations/README.md b/migrations/README.md new file mode 100644 index 0000000..1c5ca09 --- /dev/null +++ b/migrations/README.md @@ -0,0 +1,16 @@ +# AnchorNet database migrations + +Migrations are applied with `node-pg-migrate` and are intentionally separate +from application startup. Deployments should run `npm run migrate:up` before +starting the API; the API then verifies connectivity and fails fast if the +database cannot be reached. + +```sh +DATABASE_URL=postgres://anchornet:secret@localhost:5432/anchornet npm run migrate:up +``` + +Amounts use PostgreSQL `numeric(78,0)`. JavaScript converts those values to +`bigint` at the repository boundary, so values larger than +`Number.MAX_SAFE_INTEGER` are not rounded. Foreign keys prevent orphaned +liquidity and settlement records, and settlement opening locks the relevant +pool rows before checking capacity. diff --git a/package-lock.json b/package-lock.json index 87e0614..90611f6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11,7 +11,8 @@ "dependencies": { "compression": "^1.8.1", "cors": "^2.8.5", - "express": "^4.21.0" + "express": "^4.21.0", + "pg": "^8.23.0" }, "devDependencies": { "@eslint/js": "^9.14.0", @@ -20,11 +21,13 @@ "@types/express": "^4.17.21", "@types/jest": "^29.5.12", "@types/node": "^22.9.0", + "@types/pg": "^8.23.1", "@types/supertest": "^6.0.2", "@typescript-eslint/eslint-plugin": "^8.14.0", "@typescript-eslint/parser": "^8.14.0", "eslint": "^9.14.0", "jest": "^29.7.0", + "node-pg-migrate": "^7.9.1", "supertest": "^7.0.0", "ts-jest": "^29.2.5", "ts-node-dev": "^2.0.0", @@ -65,6 +68,7 @@ "integrity": "sha512-CGOfOJqWjg2qW/Mb6zNsDm+u5vFQ8DxXfbM09z69p5Z6+mE1ikP2jUXw+j42Pf1XTYED2Rni5f95npYeuwMDQA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@babel/code-frame": "^7.29.0", "@babel/generator": "^7.29.0", @@ -829,6 +833,16 @@ "url": "https://github.com/sponsors/nzakas" } }, + "node_modules/@isaacs/cliui": { + "version": "9.0.0", + "resolved": "https://registry.npmjs.org/@isaacs/cliui/-/cliui-9.0.0.tgz", + "integrity": "sha512-AokJm4tuBHillT+FpMtxQ60n8ObyXBatq7jD2/JA9dxbDDokKQm8KMht5ibGzLVU9IJDIKK4TPKgMHEYMn3lMg==", + "dev": true, + "license": "BlueOak-1.0.0", + "engines": { + "node": ">=18" + } + }, "node_modules/@istanbuljs/load-nyc-config": { "version": "1.1.0", "resolved": "https://registry.npmjs.org/@istanbuljs/load-nyc-config/-/load-nyc-config-1.1.0.tgz", @@ -1575,10 +1589,24 @@ "integrity": "sha512-F0R/h2+dsy5wJAUe3tAU6oqa2qbWY5TpNfL/RGmo1y38hiyO1w3x2jPtt76wmuaJI4DQnOBu21cNXQ2STIUUWg==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "undici-types": "~6.21.0" } }, + "node_modules/@types/pg": { + "version": "8.23.1", + "resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.23.1.tgz", + "integrity": "sha512-fKVHpikPdg4GKks3JuLEhvwSyvwzF23hnabPy6DD8ljVbC7+6J5dQzdv4arV6jqq57djnMgs1HKBxX4P8aBI3A==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "@types/node": "*", + "pg-protocol": "*", + "pg-types": "^2.2.0" + } + }, "node_modules/@types/qs": { "version": "6.15.0", "resolved": "https://registry.npmjs.org/@types/qs/-/qs-6.15.0.tgz", @@ -1723,6 +1751,7 @@ "integrity": "sha512-XZzOmihLIr8AD1b9hL9ccNMzEMWt/dE2u7NyTY9jJG6YNiNthaD5XtUHVF2uCXZ15ng+z2hT3MVuxnUYhq6k1g==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@typescript-eslint/scope-manager": "8.57.0", "@typescript-eslint/types": "8.57.0", @@ -1940,6 +1969,7 @@ "integrity": "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw==", "dev": true, "license": "MIT", + "peer": true, "bin": { "acorn": "bin/acorn" }, @@ -2324,6 +2354,7 @@ } ], "license": "MIT", + "peer": true, "dependencies": { "baseline-browser-mapping": "^2.9.0", "caniuse-lite": "^1.0.30001759", @@ -3060,6 +3091,7 @@ "integrity": "sha512-XoMjdBOwe/esVgEvLmNsD3IRHkm7fbKIUGvrleloJXUZgDHig2IPWNniv+GwjyJXzuNqVjlr5+4yVUZjycJwfQ==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@eslint-community/eslint-utils": "^4.8.0", "@eslint-community/regexpp": "^4.12.1", @@ -3544,6 +3576,36 @@ "dev": true, "license": "ISC" }, + "node_modules/foreground-child": { + "version": "3.3.1", + "resolved": "https://registry.npmjs.org/foreground-child/-/foreground-child-3.3.1.tgz", + "integrity": "sha512-gIXjKqtFuWEgzFRJA9WCQeSJLZDjgJUOMCMzxtvFq/37KojM1BFGufqsCy0r4qSQmYLsZYMeyRqzIWOMup03sw==", + "dev": true, + "license": "ISC", + "dependencies": { + "cross-spawn": "^7.0.6", + "signal-exit": "^4.0.1" + }, + "engines": { + "node": ">=14" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, + "node_modules/foreground-child/node_modules/signal-exit": { + "version": "4.1.0", + "resolved": "https://registry.npmjs.org/signal-exit/-/signal-exit-4.1.0.tgz", + "integrity": "sha512-bzyZ1e88w9O1iNJbKnOlvYTrWPDl46O1bG0D3XInv+9tkPrxrN8jUUTiFlDkkmKWgn1M6CfIA13SuGqOa9Korw==", + "dev": true, + "license": "ISC", + "engines": { + "node": ">=14" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/form-data": { "version": "4.0.5", "resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.5.tgz", @@ -4191,12 +4253,29 @@ "node": ">=8" } }, + "node_modules/jackspeak": { + "version": "4.2.3", + "resolved": "https://registry.npmjs.org/jackspeak/-/jackspeak-4.2.3.tgz", + "integrity": "sha512-ykkVRwrYvFm1nb2AJfKKYPr0emF6IiXDYUaFx4Zn9ZuIH7MrzEZ3sD5RlqGXNRpHtvUHJyOnCEFxOlNDtGo7wg==", + "dev": true, + "license": "BlueOak-1.0.0", + "dependencies": { + "@isaacs/cliui": "^9.0.0" + }, + "engines": { + "node": "20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/jest": { "version": "29.7.0", "resolved": "https://registry.npmjs.org/jest/-/jest-29.7.0.tgz", "integrity": "sha512-NIy3oAFp9shda19hy4HK0HRTWKtPJmGdnvywu01nOqNC2vZg+Z+fvJDxpMQA88eb2I9EcafcdjYgsDthnYTvGw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@jest/core": "^29.7.0", "@jest/types": "^29.6.3", @@ -5098,6 +5177,16 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/minipass": { + "version": "7.1.3", + "resolved": "https://registry.npmjs.org/minipass/-/minipass-7.1.3.tgz", + "integrity": "sha512-tEBHqDnIoM/1rXME1zgka9g6Q2lcoCkxHLuc7ODJ5BxbP5d4c2Z5cGgtXAku59200Cx7diuHTOYfSBD8n6mm8A==", + "dev": true, + "license": "BlueOak-1.0.0", + "engines": { + "node": ">=16 || 14 >=14.17" + } + }, "node_modules/mkdirp": { "version": "1.0.4", "resolved": "https://registry.npmjs.org/mkdirp/-/mkdirp-1.0.4.tgz", @@ -5147,6 +5236,59 @@ "dev": true, "license": "MIT" }, + "node_modules/node-pg-migrate": { + "version": "7.9.1", + "resolved": "https://registry.npmjs.org/node-pg-migrate/-/node-pg-migrate-7.9.1.tgz", + "integrity": "sha512-6z4OSN27ye8aYdX9ZU7NN2PTI5pOp34hTr+22Ej12djIYECq++gT7LPLZVOQXEeVCBOZQLqf87kC3Y36G434OQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "glob": "~11.0.0", + "yargs": "~17.7.0" + }, + "bin": { + "node-pg-migrate": "bin/node-pg-migrate.js", + "node-pg-migrate-cjs": "bin/node-pg-migrate.js", + "node-pg-migrate-esm": "bin/node-pg-migrate.mjs" + }, + "engines": { + "node": ">=18.19.0" + }, + "peerDependencies": { + "@types/pg": ">=6.0.0 <9.0.0", + "pg": ">=4.3.0 <9.0.0" + }, + "peerDependenciesMeta": { + "@types/pg": { + "optional": true + } + } + }, + "node_modules/node-pg-migrate/node_modules/glob": { + "version": "11.0.3", + "resolved": "https://registry.npmjs.org/glob/-/glob-11.0.3.tgz", + "integrity": "sha512-2Nim7dha1KVkaiF4q6Dj+ngPPMdfvLJEOpZk/jKiUAkqKebpGAWQXAq9z1xu9HKu5lWfqw/FASuccEjyznjPaA==", + "deprecated": "Old versions of glob are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exorbitant rates) by contacting i@izs.me", + "dev": true, + "license": "ISC", + "dependencies": { + "foreground-child": "^3.3.1", + "jackspeak": "^4.1.1", + "minimatch": "^10.0.3", + "minipass": "^7.1.2", + "package-json-from-dist": "^1.0.0", + "path-scurry": "^2.0.0" + }, + "bin": { + "glob": "dist/esm/bin.mjs" + }, + "engines": { + "node": "20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/node-releases": { "version": "2.0.36", "resolved": "https://registry.npmjs.org/node-releases/-/node-releases-2.0.36.tgz", @@ -5305,6 +5447,13 @@ "node": ">=6" } }, + "node_modules/package-json-from-dist": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/package-json-from-dist/-/package-json-from-dist-1.0.1.tgz", + "integrity": "sha512-UEZIS3/by4OC8vL3P2dTXRETpebLI2NiI5vIrjaD/5UtrkFX/tNbwjTSRAGC/+7CAo2pIcBaRgWmcBBHcsaCIw==", + "dev": true, + "license": "BlueOak-1.0.0" + }, "node_modules/parent-module": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/parent-module/-/parent-module-1.0.1.tgz", @@ -5383,12 +5532,129 @@ "dev": true, "license": "MIT" }, + "node_modules/path-scurry": { + "version": "2.0.2", + "resolved": "https://registry.npmjs.org/path-scurry/-/path-scurry-2.0.2.tgz", + "integrity": "sha512-3O/iVVsJAPsOnpwWIeD+d6z/7PmqApyQePUtCndjatj/9I5LylHvt5qluFaBT3I5h3r1ejfR056c+FCv+NnNXg==", + "dev": true, + "license": "BlueOak-1.0.0", + "dependencies": { + "lru-cache": "^11.0.0", + "minipass": "^7.1.2" + }, + "engines": { + "node": "18 || 20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, + "node_modules/path-scurry/node_modules/lru-cache": { + "version": "11.5.2", + "resolved": "https://registry.npmjs.org/lru-cache/-/lru-cache-11.5.2.tgz", + "integrity": "sha512-4pfM1Ff0x50o0tQwb5ucw/RzNyD0/YJME6IVcStalZuMWxdt3sR3huStTtxz4PUmvZfRguvDejasvQ2kifR11g==", + "dev": true, + "license": "BlueOak-1.0.0", + "engines": { + "node": "20 || >=22" + } + }, "node_modules/path-to-regexp": { "version": "0.1.12", "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-0.1.12.tgz", "integrity": "sha512-RA1GjUVMnvYFxuqovrEqZoxxW5NUZqbwKtYz/Tt7nXerk0LbLblQmrsgdeOxV5SFHf0UDggjS/bSeOZwt1pmEQ==", "license": "MIT" }, + "node_modules/pg": { + "version": "8.23.0", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.23.0.tgz", + "integrity": "sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg==", + "license": "MIT", + "peer": true, + "dependencies": { + "pg-connection-string": "^2.14.0", + "pg-pool": "^3.14.0", + "pg-protocol": "^1.16.0", + "pg-types": "2.2.0", + "pgpass": "1.0.5" + }, + "engines": { + "node": ">= 16.0.0" + }, + "optionalDependencies": { + "pg-cloudflare": "^1.4.0" + }, + "peerDependencies": { + "pg-native": ">=3.0.1" + }, + "peerDependenciesMeta": { + "pg-native": { + "optional": true + } + } + }, + "node_modules/pg-cloudflare": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz", + "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==", + "license": "MIT", + "optional": true + }, + "node_modules/pg-connection-string": { + "version": "2.14.0", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz", + "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==", + "license": "MIT" + }, + "node_modules/pg-int8": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz", + "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==", + "license": "ISC", + "engines": { + "node": ">=4.0.0" + } + }, + "node_modules/pg-pool": { + "version": "3.14.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz", + "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==", + "license": "MIT", + "peerDependencies": { + "pg": ">=8.0" + } + }, + "node_modules/pg-protocol": { + "version": "1.16.0", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.16.0.tgz", + "integrity": "sha512-sILXutLVjCLjcDuOmvhX5e2Z4cS5qG/6Bu3VkpFwdf/633ElGLpEh9bgmuI5I4sqKqkifQiGyiCcx1HdtrK7tg==", + "license": "MIT" + }, + "node_modules/pg-types": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz", + "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==", + "license": "MIT", + "dependencies": { + "pg-int8": "1.0.1", + "postgres-array": "~2.0.0", + "postgres-bytea": "~1.0.0", + "postgres-date": "~1.0.4", + "postgres-interval": "^1.1.0" + }, + "engines": { + "node": ">=4" + } + }, + "node_modules/pgpass": { + "version": "1.0.5", + "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz", + "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==", + "license": "MIT", + "dependencies": { + "split2": "^4.1.0" + } + }, "node_modules/picocolors": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz", @@ -5488,6 +5754,45 @@ "node": ">=8" } }, + "node_modules/postgres-array": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz", + "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/postgres-bytea": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz", + "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-date": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz", + "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-interval": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz", + "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==", + "license": "MIT", + "dependencies": { + "xtend": "^4.0.0" + }, + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/prelude-ls": { "version": "1.2.1", "resolved": "https://registry.npmjs.org/prelude-ls/-/prelude-ls-1.2.1.tgz", @@ -5966,6 +6271,15 @@ "source-map": "^0.6.0" } }, + "node_modules/split2": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz", + "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==", + "license": "ISC", + "engines": { + "node": ">= 10.x" + } + }, "node_modules/sprintf-js": { "version": "1.0.3", "resolved": "https://registry.npmjs.org/sprintf-js/-/sprintf-js-1.0.3.tgz", @@ -6252,6 +6566,7 @@ "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -6383,6 +6698,7 @@ "integrity": "sha512-f0FFpIdcHgn8zcPSbf1dRevwt047YMnaiJM3u2w2RewrB+fob/zePZcrOyQoLMMO7aBIddLcQIEK5dYjkLnGrQ==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@cspotcode/source-map-support": "^0.8.0", "@tsconfig/node10": "^1.0.7", @@ -6544,6 +6860,7 @@ "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "dev": true, "license": "Apache-2.0", + "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -6749,7 +7066,6 @@ "version": "4.0.2", "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==", - "dev": true, "license": "MIT", "engines": { "node": ">=0.4" diff --git a/package.json b/package.json index 7c92fe1..a138339 100644 --- a/package.json +++ b/package.json @@ -6,6 +6,8 @@ "scripts": { "build": "tsc", "start": "node dist/index.js", + "migrate:up": "node-pg-migrate up -m migrations", + "migrate:down": "node-pg-migrate down -m migrations", "dev": "ts-node-dev --respawn src/index.ts", "test": "jest", "lint": "eslint \"src/**/*.ts\"", @@ -14,28 +16,36 @@ "engines": { "node": ">=18" }, - "keywords": ["anchornet", "stellar", "liquidity", "api"], + "keywords": [ + "anchornet", + "stellar", + "liquidity", + "api" + ], "license": "MIT", "dependencies": { - "express": "^4.21.0", + "compression": "^1.8.1", "cors": "^2.8.5", - "compression": "^1.8.1" + "express": "^4.21.0", + "pg": "^8.23.0" }, "devDependencies": { + "@eslint/js": "^9.14.0", "@types/compression": "^1.8.1", "@types/cors": "^2.8.17", "@types/express": "^4.17.21", "@types/jest": "^29.5.12", "@types/node": "^22.9.0", + "@types/pg": "^8.23.1", "@types/supertest": "^6.0.2", + "@typescript-eslint/eslint-plugin": "^8.14.0", + "@typescript-eslint/parser": "^8.14.0", + "eslint": "^9.14.0", "jest": "^29.7.0", + "node-pg-migrate": "^7.9.1", "supertest": "^7.0.0", "ts-jest": "^29.2.5", "ts-node-dev": "^2.0.0", - "typescript": "^5.6.3", - "@eslint/js": "^9.14.0", - "eslint": "^9.14.0", - "@typescript-eslint/eslint-plugin": "^8.14.0", - "@typescript-eslint/parser": "^8.14.0" + "typescript": "^5.6.3" } } diff --git a/src/app.ts b/src/app.ts index 2053720..700860c 100644 --- a/src/app.ts +++ b/src/app.ts @@ -34,8 +34,15 @@ import { createAuditLog } from "./middleware/auditLog"; import { loadConfig, validateConfig, Config } from "./config"; import { buildOpenApiSpec } from "./openapi"; import { isReady } from "./utils/readiness"; - -export function createApp(): Express { +import { PersistenceRuntime } from "./persistence/runtime"; +import { + PersistentAnchorRepository, + PersistentLiquidityRepository, + PersistentSettlementRepository, +} from "./repositories/persistentRepositories"; +import { PersistentSettlementService } from "./services/persistentSettlementService"; + +export function createApp(options: { persistence?: PersistenceRuntime } = {}): Express { const app = express(); const config = validateConfig(loadConfig()); app.set('trust proxy', 1); // Ensure req.ip reflects real client IP behind reverse proxy (#120) @@ -56,15 +63,41 @@ export function createApp(): Express { const audit = createAuditLog(); app.use(audit.middleware); - const repo = new LiquidityRepository(); - const anchors = new AnchorService(new AnchorRepository()); + if (config.databaseUrl && !options.persistence) { + throw new Error( + "PostgreSQL persistence has not been initialized; call initializePersistence() before createApp()", + ); + } + + const repo = options.persistence + ? new PersistentLiquidityRepository( + options.persistence.database, + options.persistence.snapshot, + ) + : new LiquidityRepository(); + const anchorRepo = options.persistence + ? new PersistentAnchorRepository( + options.persistence.database, + options.persistence.snapshot, + ) + : new AnchorRepository(); + const settlementRepo = options.persistence + ? new PersistentSettlementRepository( + options.persistence.database, + options.persistence.snapshot, + ) + : new SettlementRepository(); + const anchors = new AnchorService(anchorRepo); const quotes = new QuoteService(repo, config.feeBps); - const settlements = new SettlementService( - new SettlementRepository(), - repo, - anchors, - config.feeBps, - ); + const settlements = options.persistence + ? new PersistentSettlementService( + options.persistence.database, + settlementRepo as PersistentSettlementRepository, + repo as PersistentLiquidityRepository, + anchors, + config.feeBps, + ) + : new SettlementService(settlementRepo, repo, anchors, config.feeBps); const liquidity = new LiquidityService(repo, settlements); app.get("/health", (_req: Request, res: Response) => { diff --git a/src/config.test.ts b/src/config.test.ts index b4e3fdb..80d1946 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -15,6 +15,7 @@ describe("loadConfig", () => { FEE_BPS: "25", API_KEY: "secret", NODE_ENV: "production", + DATABASE_URL: "postgres://localhost/anchornet", }); expect(config.port).toBe(8080); expect(config.feeBps).toBe(25); @@ -78,7 +79,11 @@ describe("loadConfig", () => { it("accepts a configured API_KEY in production", () => { expect(() => - loadConfig({ NODE_ENV: "production", API_KEY: "secret" }), + loadConfig({ + NODE_ENV: "production", + API_KEY: "secret", + DATABASE_URL: "postgres://localhost/anchornet", + }), ).not.toThrow(); }); @@ -264,10 +269,26 @@ describe("validateConfig", () => { }); it("allows a production deploy that sets API_KEY", () => { - const config = loadConfig({ NODE_ENV: "production", API_KEY: "secret" }); + const config = loadConfig({ + NODE_ENV: "production", + API_KEY: "secret", + DATABASE_URL: "postgres://localhost/anchornet", + }); expect(() => validateConfig(config)).not.toThrow(); }); + it("requires DATABASE_URL in production", () => { + expect(() => loadConfig({ NODE_ENV: "production", API_KEY: "secret" })).toThrow( + /DATABASE_URL is required/, + ); + }); + + it("rejects non-PostgreSQL database URLs", () => { + expect(() => loadConfig({ DATABASE_URL: "mysql://localhost/anchornet" })).toThrow( + /DATABASE_URL must use the postgres/, + ); + }); + it("does NOT require API_KEY in development (open access is allowed, not fatal)", () => { const config = loadConfig({ NODE_ENV: "development" }); expect(() => validateConfig(config)).not.toThrow(); diff --git a/src/config.ts b/src/config.ts index 7e234b7..5dd0ff2 100644 --- a/src/config.ts +++ b/src/config.ts @@ -53,6 +53,8 @@ export interface Config { * real client address rather than the proxy's IP. */ trustProxy: boolean | string | number; + /** PostgreSQL connection string used by the production persistence layer. */ + databaseUrl?: string; } const DEFAULT_BODY_LIMIT = "100kb"; @@ -157,6 +159,7 @@ export function loadConfig( ): Config { const apiKey = env.API_KEY?.trim(); const metricsApiKey = env.METRICS_API_KEY?.trim(); + const databaseUrl = env.DATABASE_URL?.trim() || undefined; const feeBps = intFromEnv(env.FEE_BPS, 10); if (feeBps < MIN_FEE_BPS || feeBps > MAX_FEE_BPS) { @@ -183,6 +186,7 @@ export function loadConfig( metricsRateLimitMax: intFromEnv(env.METRICS_RATE_LIMIT_MAX, 120), metricsRateLimitWindowMs: intFromEnv(env.METRICS_RATE_LIMIT_WINDOW_MS, 60_000), trustProxy: parseTrustProxy(env.TRUST_PROXY), + databaseUrl, }; return validateConfig(config); @@ -231,6 +235,31 @@ export function validateConfig(config: Config): Config { ); } + if (config.env === "production" && !config.databaseUrl) { + throw new ConfigValidationError( + "DATABASE_URL is required when NODE_ENV=production. Refusing to start without durable persistence.", + "DATABASE_URL", + ); + } + + if (config.databaseUrl) { + let parsed: URL; + try { + parsed = new URL(config.databaseUrl); + } catch { + throw new ConfigValidationError( + "DATABASE_URL must be a valid PostgreSQL connection string", + "DATABASE_URL", + ); + } + if (parsed.protocol !== "postgres:" && parsed.protocol !== "postgresql:") { + throw new ConfigValidationError( + "DATABASE_URL must use the postgres:// or postgresql:// scheme", + "DATABASE_URL", + ); + } + } + if ( typeof config.port !== "number" || !Number.isInteger(config.port) || diff --git a/src/index.ts b/src/index.ts index 75d5f7f..58fc72e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -3,47 +3,53 @@ * Builds the application and starts the HTTP server. */ -import { Express } from "express"; +import express, { Express } from "express"; import { createApp, getConfig } from "./app"; +import { initializePersistence, PersistenceRuntime } from "./persistence/runtime"; import { createShutdownHandler } from "./utils/shutdown"; import { markNotReady } from "./utils/readiness"; -// Build (and validate) the app up front. validateConfig() runs inside -// createApp()/getConfig() and throws a ConfigValidationError naming the -// missing variable when a required value is absent, refusing to start -// instead of silently running with weakened configuration. -let app: Express; -try { - app = createApp(); -} catch (error) { - if (process.env.NODE_ENV !== "test") { - const message = error instanceof Error ? error.message : String(error); - console.error(`AnchorNet API failed to start: ${message}`); - process.exit(1); +let app: Express = express(); + +function listen(application: Express, runtime?: PersistenceRuntime): void { + const { port: PORT } = getConfig(); + const server = application.listen(PORT, () => { + console.log(`AnchorNet API listening on http://localhost:${PORT}`); + }); + + const shutdown = createShutdownHandler(server, { + onShutdown: (signal) => { + markNotReady(); + console.log(`${signal} received, shutting down`); + if (runtime) void runtime.database.close(); + }, + }); + process.on("SIGTERM", () => shutdown("SIGTERM")); + process.on("SIGINT", () => shutdown("SIGINT")); +} + +async function start(): Promise { + const config = getConfig(); + if (config.databaseUrl) { + const runtime = await initializePersistence(config); + app = createApp({ persistence: runtime }); + listen(app, runtime); + return; } - throw error; + + app = createApp(); + listen(app); } -if (process.env.NODE_ENV !== "test") { - try { - const { port: PORT } = getConfig(); - const server = app.listen(PORT, () => { - console.log(`AnchorNet API listening on http://localhost:${PORT}`); - }); - - const shutdown = createShutdownHandler(server, { - onShutdown: (signal) => { - markNotReady(); - console.log(`${signal} received, shutting down`); - }, - }); - process.on("SIGTERM", () => shutdown("SIGTERM")); - process.on("SIGINT", () => shutdown("SIGINT")); - } catch (error) { +if (process.env.NODE_ENV === "test") { + // Tests import the app without binding a port. + app = createApp(); +} else { + void start().catch((error) => { const message = error instanceof Error ? error.message : String(error); console.error(`AnchorNet API failed to start: ${message}`); process.exit(1); - } + }); } export default app; diff --git a/src/persistence/persistentRuntime.test.ts b/src/persistence/persistentRuntime.test.ts new file mode 100644 index 0000000..917d404 --- /dev/null +++ b/src/persistence/persistentRuntime.test.ts @@ -0,0 +1,64 @@ +import request from "supertest"; +import { createApp } from "../app"; +import { PersistenceRuntime } from "./runtime"; +import { PostgresPersistence } from "./postgres"; + +describe("PostgreSQL runtime composition", () => { + const originalDatabaseUrl = process.env.DATABASE_URL; + + afterEach(() => { + if (originalDatabaseUrl === undefined) delete process.env.DATABASE_URL; + else process.env.DATABASE_URL = originalDatabaseUrl; + }); + + it("hydrates the production repository graph and flushes durable mutations", async () => { + process.env.DATABASE_URL = "postgres://localhost/anchornet"; + const calls: string[] = []; + const created = { + id: 1, + anchor: "anchorA", + asset: "USDC", + amount: 100n, + fee: 1n, + status: "pending" as const, + createdAt: "2026-08-30T00:00:00.000Z", + }; + const database = { + upsertAnchor: async () => { calls.push("anchor"); }, + removeAnchor: async () => undefined, + upsertLiquidity: async () => { calls.push("liquidity"); }, + removeLiquidity: async () => undefined, + insertSettlement: async () => undefined, + removeSettlement: async () => undefined, + flush: async () => { calls.push("flush"); }, + openSettlement: async () => created, + transitionSettlement: async () => ({ ...created, status: "executed" as const }), + } as unknown as PostgresPersistence; + const runtime: PersistenceRuntime = { + database, + snapshot: { + anchors: [], + liquidity: [], + settlements: [], + }, + }; + const app = createApp({ persistence: runtime }); + + const anchorResponse = await request(app) + .post("/api/v1/anchors") + .send({ id: "anchorA" }); + expect(anchorResponse.status).toBe(201); + + const liquidityResponse = await request(app) + .post("/api/v1/liquidity") + .send({ anchor: "anchorA", asset: "USDC", amount: "100" }); + expect(liquidityResponse.status).toBe(201); + + const settlementResponse = await request(app) + .post("/api/v1/settlements") + .send({ anchor: "anchorA", asset: "USDC", amount: "100" }); + expect(settlementResponse.status).toBe(201); + expect(settlementResponse.body).toMatchObject({ id: 1, status: "pending" }); + expect(calls).toEqual(expect.arrayContaining(["anchor", "liquidity", "flush"])); + }); +}); diff --git a/src/persistence/postgres.test.ts b/src/persistence/postgres.test.ts new file mode 100644 index 0000000..36acd50 --- /dev/null +++ b/src/persistence/postgres.test.ts @@ -0,0 +1,98 @@ +import { Pool } from "pg"; +import { PostgresPersistence } from "./postgres"; + +jest.mock("pg", () => ({ Pool: jest.fn() })); + +const poolMock = Pool as unknown as jest.Mock; + +function makeDatabase(options: { + liquidity?: string[]; + committed?: Array<{ id: number; anchor: string; asset: string; amount: string; fee: string; status: "pending" | "executed" | "cancelled"; created_at: Date; cancel_reason: string | null }>; +} = {}): { database: PostgresPersistence; queries: string[] } { + const queries: string[] = []; + const client = { + query: jest.fn(async (sql: string) => { + queries.push(sql); + if (sql.includes("SELECT amount FROM liquidity_entries")) { + return { rowCount: options.liquidity?.length ?? 1, rows: (options.liquidity ?? ["1000"]).map((amount) => ({ amount })) }; + } + if (sql.includes("FROM settlements") && sql.includes("status IN")) { + return { rowCount: options.committed?.length ?? 0, rows: options.committed ?? [] }; + } + if (sql.includes("INSERT INTO settlements")) { + return { + rowCount: 1, + rows: [{ + id: 7, + anchor: "anchorA", + asset: "USDC", + amount: "400", + fee: "1", + status: "pending", + created_at: new Date("2026-08-30T00:00:00.000Z"), + cancel_reason: null, + }], + }; + } + return { rowCount: 1, rows: [] }; + }), + release: jest.fn(), + }; + poolMock.mockImplementationOnce(() => ({ + query: jest.fn().mockResolvedValue({ rows: [] }), + connect: jest.fn().mockResolvedValue(client), + end: jest.fn().mockResolvedValue(undefined), + })); + const database = new PostgresPersistence("postgres://localhost/anchornet"); + return { database, queries }; +} + +describe("PostgresPersistence settlement transaction", () => { + afterEach(() => jest.clearAllMocks()); + + it("locks the pool and committed settlements before inserting", async () => { + const { database, queries } = makeDatabase(); + + const settlement = await database.openSettlement({ + anchor: "anchorA", + asset: "USDC", + amount: 400n, + fee: 1n, + createdAt: "2026-08-30T00:00:00.000Z", + }); + + expect(settlement).toMatchObject({ id: 7, asset: "USDC", amount: 400n }); + expect(queries[0]).toBe("BEGIN"); + expect(queries.some((sql) => sql.includes("ORDER BY anchor FOR UPDATE"))).toBe(true); + expect(queries.some((sql) => sql.includes("ORDER BY id FOR UPDATE"))).toBe(true); + expect(queries.some((sql) => sql.includes("INSERT INTO settlements"))).toBe(true); + expect(queries.at(-1)).toBe("COMMIT"); + }); + + it("rejects capacity breaches and rolls back without inserting", async () => { + const { database, queries } = makeDatabase({ + liquidity: ["1000"], + committed: [{ + id: 1, + anchor: "anchorA", + asset: "USDC", + amount: "800", + fee: "1", + status: "pending", + created_at: new Date("2026-08-30T00:00:00.000Z"), + cancel_reason: null, + }], + }); + + await expect(database.openSettlement({ + anchor: "anchorA", + asset: "USDC", + amount: 201n, + fee: 1n, + createdAt: "2026-08-30T00:00:00.000Z", + })).rejects.toThrow(/insufficient liquidity/); + + expect(queries.some((sql) => sql.includes("INSERT INTO settlements"))).toBe(false); + expect(queries.at(-1)).toBe("ROLLBACK"); + }); +}); diff --git a/src/persistence/postgres.ts b/src/persistence/postgres.ts new file mode 100644 index 0000000..318ced3 --- /dev/null +++ b/src/persistence/postgres.ts @@ -0,0 +1,307 @@ +/** + * PostgreSQL persistence primitives. + * + * Repository methods in the original API are intentionally synchronous. This + * module therefore owns the asynchronous boundary: state is hydrated before + * the HTTP server binds, ordinary repository writes are serialized through a + * durable queue, and settlement reservations use a database transaction with + * row locks. The latter is important because checking a cached pool and then + * inserting a settlement would re-introduce the oversubscription race this + * issue is intended to remove. + */ + +import { Pool, PoolClient, QueryResultRow } from "pg"; +import { Anchor } from "../models/anchor"; +import { LiquidityEntry } from "../models/liquidity"; +import { Settlement, SettlementStatus } from "../models/settlement"; + +export interface PersistenceSnapshot { + anchors: Anchor[]; + liquidity: LiquidityEntry[]; + settlements: Settlement[]; +} + +export interface SettlementDraft { + anchor: string; + asset: string; + amount: bigint; + fee: bigint; + createdAt: string; +} + +export class PostgresPersistence { + readonly pool: Pool; + private writeQueue: Promise = Promise.resolve(); + + constructor(databaseUrl: string) { + this.pool = new Pool({ + connectionString: databaseUrl, + max: 10, + application_name: "anchornet-backend", + }); + } + + /** Fails before the server starts accepting traffic if PostgreSQL is down. */ + async assertReachable(): Promise { + await this.pool.query("SELECT 1"); + } + + /** Loads every durable aggregate before the app is constructed. */ + async loadSnapshot(): Promise { + const [anchors, liquidity, settlements] = await Promise.all([ + this.pool.query( + "SELECT id, name, registered_at, active FROM anchors ORDER BY id", + ), + this.pool.query( + "SELECT anchor, asset, amount, updated_at FROM liquidity_entries ORDER BY anchor, asset", + ), + this.pool.query( + "SELECT id, anchor, asset, amount, fee, status, created_at, cancel_reason FROM settlements ORDER BY id", + ), + ]); + + return { + anchors: anchors.rows.map((row) => ({ + id: row.id, + name: row.name, + registeredAt: row.registered_at.toISOString(), + active: row.active, + })), + liquidity: liquidity.rows.map((row) => ({ + anchor: row.anchor, + asset: row.asset, + amount: BigInt(row.amount), + updatedAt: row.updated_at.toISOString(), + })), + settlements: settlements.rows.map(toSettlement), + }; + } + + /** Waits until all accepted write-behind operations have committed. */ + async flush(): Promise { + await this.writeQueue; + } + + async close(): Promise { + await this.flush(); + await this.pool.end(); + } + + enqueue(operation: (client: PoolClient) => Promise): Promise { + this.writeQueue = this.writeQueue.then(async () => { + const client = await this.pool.connect(); + try { + await operation(client); + } finally { + client.release(); + } + }); + return this.writeQueue; + } + + upsertAnchor(anchor: Anchor): Promise { + return this.enqueue((client) => + client.query( + `INSERT INTO anchors (id, name, registered_at, active) + VALUES ($1, $2, $3, $4) + ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, active = EXCLUDED.active`, + [anchor.id, anchor.name, anchor.registeredAt, anchor.active], + ).then(() => undefined), + ); + } + + removeAnchor(id: string): Promise { + return this.enqueue((client) => + client.query("DELETE FROM anchors WHERE id = $1", [id]).then(() => undefined), + ); + } + + upsertLiquidity(entry: LiquidityEntry): Promise { + return this.enqueue((client) => + client.query( + `INSERT INTO liquidity_entries (anchor, asset, amount, updated_at) + VALUES ($1, $2, $3, $4) + ON CONFLICT (anchor, asset) DO UPDATE + SET amount = EXCLUDED.amount, updated_at = EXCLUDED.updated_at`, + [entry.anchor, entry.asset, entry.amount.toString(), entry.updatedAt], + ).then(() => undefined), + ); + } + + removeLiquidity(anchor: string, asset: string): Promise { + return this.enqueue((client) => + client.query( + "DELETE FROM liquidity_entries WHERE anchor = $1 AND asset = $2", + [anchor, asset], + ).then(() => undefined), + ); + } + + insertSettlement(settlement: Settlement): Promise { + return this.enqueue((client) => + client.query( + `INSERT INTO settlements + (id, anchor, asset, amount, fee, status, created_at, cancel_reason) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (id) DO UPDATE SET + anchor = EXCLUDED.anchor, asset = EXCLUDED.asset, + amount = EXCLUDED.amount, fee = EXCLUDED.fee, + status = EXCLUDED.status, created_at = EXCLUDED.created_at, + cancel_reason = EXCLUDED.cancel_reason`, + [ + settlement.id, + settlement.anchor, + settlement.asset, + settlement.amount.toString(), + settlement.fee.toString(), + settlement.status, + settlement.createdAt, + settlement.cancelReason ?? null, + ], + ).then(() => undefined), + ); + } + + removeSettlement(id: number): Promise { + return this.enqueue((client) => + client.query("DELETE FROM settlements WHERE id = $1", [id]).then(() => undefined), + ); + } + + /** + * Atomically checks available liquidity and creates a pending settlement. + * All rows for the asset are locked in a stable order, so concurrent callers + * serialize on the same liquidity pool. Pending and executed settlements are + * read while their rows are locked, making the accounting durable across + * restarts and across multiple API instances. + */ + async openSettlement(draft: SettlementDraft): Promise { + await this.flush(); + const client = await this.pool.connect(); + try { + await client.query("BEGIN"); + const liquidity = await client.query<{ amount: string }>( + `SELECT amount FROM liquidity_entries + WHERE asset = $1 ORDER BY anchor FOR UPDATE`, + [draft.asset], + ); + const settlements = await client.query( + `SELECT id, anchor, asset, amount, fee, status, created_at, cancel_reason + FROM settlements + WHERE asset = $1 AND status IN ('pending', 'executed') + ORDER BY id FOR UPDATE`, + [draft.asset], + ); + const total = liquidity.rows.reduce((sum, row) => sum + BigInt(row.amount), 0n); + const committed = settlements.rows.reduce( + (sum, row) => sum + BigInt(row.amount), + 0n, + ); + const available = total - committed; + if (available < draft.amount) { + const error = new Error( + `insufficient liquidity for ${draft.asset}: requested ${draft.amount}, available ${available}`, + ); + error.name = "InsufficientLiquidityError"; + throw error; + } + + const result = await client.query( + `INSERT INTO settlements (anchor, asset, amount, fee, status, created_at) + VALUES ($1, $2, $3, $4, 'pending', $5) + RETURNING id, anchor, asset, amount, fee, status, created_at, cancel_reason`, + [ + draft.anchor, + draft.asset, + draft.amount.toString(), + draft.fee.toString(), + draft.createdAt, + ], + ); + await client.query("COMMIT"); + return toSettlement(result.rows[0]); + } catch (error) { + await client.query("ROLLBACK").catch(() => undefined); + throw error; + } finally { + client.release(); + } + } + + async transitionSettlement( + id: number, + nextStatus: Exclude, + cancelReason?: string, + ): Promise { + await this.flush(); + const client = await this.pool.connect(); + try { + await client.query("BEGIN"); + const result = await client.query( + `UPDATE settlements + SET status = $2, cancel_reason = $3 + WHERE id = $1 AND status = 'pending' + RETURNING id, anchor, asset, amount, fee, status, created_at, cancel_reason`, + [id, nextStatus, cancelReason ?? null], + ); + if (result.rowCount !== 1) { + const current = await client.query<{ status: SettlementStatus }>( + "SELECT status FROM settlements WHERE id = $1", + [id], + ); + const error = new Error( + current.rowCount === 0 + ? `settlement ${id} not found` + : `settlement ${id} is ${current.rows[0].status}, not pending`, + ); + error.name = current.rowCount === 0 ? "SettlementNotFoundError" : "InvalidSettlementStateError"; + throw error; + } + await client.query("COMMIT"); + return toSettlement(result.rows[0]); + } catch (error) { + await client.query("ROLLBACK").catch(() => undefined); + throw error; + } finally { + client.release(); + } + } +} + +interface AnchorRow extends QueryResultRow { + id: string; + name: string; + registered_at: Date; + active: boolean; +} + +interface LiquidityRow extends QueryResultRow { + anchor: string; + asset: string; + amount: string; + updated_at: Date; +} + +interface SettlementRow extends QueryResultRow { + id: number; + anchor: string; + asset: string; + amount: string; + fee: string; + status: SettlementStatus; + created_at: Date; + cancel_reason: string | null; +} + +function toSettlement(row: SettlementRow): Settlement { + return { + id: Number(row.id), + anchor: row.anchor, + asset: row.asset, + amount: BigInt(row.amount), + fee: BigInt(row.fee), + status: row.status, + createdAt: row.created_at.toISOString(), + ...(row.cancel_reason === null ? {} : { cancelReason: row.cancel_reason }), + }; +} diff --git a/src/persistence/runtime.ts b/src/persistence/runtime.ts new file mode 100644 index 0000000..06573cb --- /dev/null +++ b/src/persistence/runtime.ts @@ -0,0 +1,29 @@ +import { Config } from "../config"; +import { PostgresPersistence, PersistenceSnapshot } from "./postgres"; + +export interface PersistenceRuntime { + database: PostgresPersistence; + snapshot: PersistenceSnapshot; +} + +/** + * Opens and hydrates the database once during process startup. Keeping this + * lifecycle separate from createApp preserves the synchronous app factory + * used by the unit and HTTP contract tests. + */ +export async function initializePersistence(config: Config): Promise { + if (!config.databaseUrl) { + throw new Error("DATABASE_URL is required to initialize PostgreSQL persistence"); + } + const database = new PostgresPersistence(config.databaseUrl); + try { + await database.assertReachable(); + const snapshot = await database.loadSnapshot(); + return { database, snapshot }; + } catch (error) { + await database.close().catch(() => undefined); + throw new Error( + `PostgreSQL persistence is unavailable: ${error instanceof Error ? error.message : String(error)}`, + ); + } +} diff --git a/src/repositories/persistentRepositories.ts b/src/repositories/persistentRepositories.ts new file mode 100644 index 0000000..8fe1dd9 --- /dev/null +++ b/src/repositories/persistentRepositories.ts @@ -0,0 +1,106 @@ +/** + * Durable repository facades. + * + * The domain services currently expose synchronous methods. These adapters + * preserve that contract for callers by maintaining a hydrated, process-local + * read cache while serializing every mutation to PostgreSQL. The HTTP startup + * path hydrates the cache before binding, and mutation routes can flush the + * queue before returning when a durable acknowledgement is required. + */ + +import { Anchor } from "../models/anchor"; +import { LiquidityEntry } from "../models/liquidity"; +import { Settlement } from "../models/settlement"; +import { PostgresPersistence, PersistenceSnapshot } from "../persistence/postgres"; +import { AnchorRepository } from "./anchorRepository"; +import { LiquidityRepository } from "./liquidityRepository"; +import { SettlementRepository } from "./settlementRepository"; + +export class PersistentAnchorRepository extends AnchorRepository { + constructor( + private readonly database: PostgresPersistence, + snapshot: PersistenceSnapshot, + ) { + super(); + for (const anchor of snapshot.anchors) super.upsert(anchor); + } + + override upsert(anchor: Anchor): Anchor { + const result = super.upsert(anchor); + void this.database.upsertAnchor(result); + return result; + } + + override remove(id: string): boolean { + const removed = super.remove(id); + if (removed) void this.database.removeAnchor(id); + return removed; + } + + flush(): Promise { + return this.database.flush(); + } +} + +export class PersistentLiquidityRepository extends LiquidityRepository { + constructor( + private readonly database: PostgresPersistence, + snapshot: PersistenceSnapshot, + ) { + super(); + for (const entry of snapshot.liquidity) super.upsert(entry); + } + + override upsert(entry: LiquidityEntry): LiquidityEntry { + const result = super.upsert(entry); + void this.database.upsertLiquidity(result); + return result; + } + + override remove(anchor: string, asset: string): boolean { + const removed = super.remove(anchor, asset); + if (removed) void this.database.removeLiquidity(anchor, asset); + return removed; + } + + flush(): Promise { + return this.database.flush(); + } +} + +export class PersistentSettlementRepository extends SettlementRepository { + constructor( + private readonly database: PostgresPersistence, + snapshot: PersistenceSnapshot, + ) { + super(); + for (const settlement of snapshot.settlements) super.save(settlement); + } + + override create(settlement: Omit): Settlement { + const result = super.create(settlement); + void this.database.insertSettlement(result); + return result; + } + + override save(settlement: Settlement): Settlement { + const result = super.save(settlement); + void this.database.insertSettlement(result); + return result; + } + + override remove(id: number): boolean { + const removed = super.remove(id); + if (removed) void this.database.removeSettlement(id); + return removed; + } + + /** Adds a row returned by a database transaction without writing it again. */ + hydrate(settlement: Settlement): Settlement { + return super.save(settlement); + } + + flush(): Promise { + return this.database.flush(); + } +} diff --git a/src/routes/anchors.ts b/src/routes/anchors.ts index ddb0df6..ef4b67a 100644 --- a/src/routes/anchors.ts +++ b/src/routes/anchors.ts @@ -2,7 +2,7 @@ * Routes for managing registered anchors. */ -import { Router, Request, Response } from "express"; +import { Router, Request, Response, NextFunction } from "express"; import { AnchorService } from "../services/anchorService"; import { SettlementService } from "../services/settlementService"; import { Anchor } from "../models/anchor"; @@ -60,9 +60,10 @@ export function anchorRouter( const router = Router(); // Register a new anchor. - router.post("/", (req: Request, res: Response) => { - const anchor = service.register(req.body ?? {}); - res.status(201).json(anchor); + router.post("/", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(() => service.register(req.body ?? {})) + .then(async (anchor) => { await service.flush(); res.status(201).json(anchor); }) + .catch(next); }); // Register a batch of anchors atomically. @@ -73,10 +74,13 @@ export function anchorRouter( // confirm no registration happened. `dryRun` is strictly parsed: only // "true"/"false" are accepted, so a typo can never silently perform a real // registration. - router.post("/bulk", (req: Request, res: Response) => { - const dryRun = optionalBooleanFlag(req.query.dryRun, "dryRun"); - const anchors = service.registerBulk((req.body ?? {}).anchors, dryRun); - res.status(201).json({ anchors, dryRun }); + router.post("/bulk", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const dryRun = optionalBooleanFlag(req.query.dryRun, "dryRun"); + const anchors = service.registerBulk((req.body ?? {}).anchors, dryRun); + await service.flush(); + res.status(201).json({ anchors, dryRun }); + }).catch(next); }); // List anchors, optionally filtered via ?status=active|inactive and/or a @@ -123,18 +127,30 @@ export function anchorRouter( }); // Partially update an anchor's mutable fields (currently just `name`). - router.patch("/:id", (req: Request, res: Response) => { - res.json(service.update(req.params.id, req.body ?? {})); + router.patch("/:id", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const anchor = service.update(req.params.id, req.body ?? {}); + await service.flush(); + res.json(anchor); + }).catch(next); }); // Deactivate an anchor. - router.delete("/:id", (req: Request, res: Response) => { - res.json(service.deregister(req.params.id)); + router.delete("/:id", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const anchor = service.deregister(req.params.id); + await service.flush(); + res.json(anchor); + }).catch(next); }); // Reactivate a previously deactivated anchor. - router.post("/:id/reactivate", (req: Request, res: Response) => { - res.json(service.reactivate(req.params.id)); + router.post("/:id/reactivate", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const anchor = service.reactivate(req.params.id); + await service.flush(); + res.json(anchor); + }).catch(next); }); // List settlements for a specific anchor, scoped by its id. diff --git a/src/routes/liquidity.ts b/src/routes/liquidity.ts index a278c0d..2ec637f 100644 --- a/src/routes/liquidity.ts +++ b/src/routes/liquidity.ts @@ -2,7 +2,7 @@ * Routes for recording and reading anchor liquidity. */ -import { Router, Request, Response } from "express"; +import { Router, Request, Response, NextFunction } from "express"; import { ApiError } from "../errors/ApiError"; import { LiquidityService } from "../services/liquidityService"; import { paginateByCursor } from "../utils/cursorPagination"; @@ -11,7 +11,7 @@ export function liquidityRouter(service: LiquidityService): Router { const router = Router(); // Record (or accumulate) liquidity for an anchor/asset pair. - router.post("/", (req: Request, res: Response) => { + router.post("/", (req: Request, res: Response, next: NextFunction) => { const raw = req.body.amount; // Reject values that cannot represent a valid positive integer amount: @@ -36,23 +36,32 @@ export function liquidityRouter(service: LiquidityService): Router { throw ApiError.badRequest('"amount" must be a positive finite number'); } - const entry = service.addLiquidity(req.body ?? {}); - res.status(201).json({ ...entry, amount: entry.amount.toString() }); + Promise.resolve().then(async () => { + const entry = service.addLiquidity(req.body ?? {}); + await service.flush(); + res.status(201).json({ ...entry, amount: entry.amount.toString() }); + }).catch(next); }); // Withdraw (reduce) liquidity previously recorded for an anchor/asset pair. - router.post("/withdraw", (req: Request, res: Response) => { - const entry = service.withdrawLiquidity(req.body ?? {}); - res.json({ ...entry, amount: entry.amount.toString() }); + router.post("/withdraw", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const entry = service.withdrawLiquidity(req.body ?? {}); + await service.flush(); + res.json({ ...entry, amount: entry.amount.toString() }); + }).catch(next); }); // Atomically transfer liquidity between two anchors for the same asset. - router.post("/transfer", (req: Request, res: Response) => { - const result = service.transferLiquidity(req.body ?? {}); - res.json({ + router.post("/transfer", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const result = service.transferLiquidity(req.body ?? {}); + await service.flush(); + res.json({ from: { ...result.from, amount: result.from.amount.toString() }, to: { ...result.to, amount: result.to.amount.toString() } - }); + }); + }).catch(next); }); // List aggregated pools across all assets. @@ -111,9 +120,12 @@ export function liquidityRouter(service: LiquidityService): Router { }); // Force-remove an anchor's entire liquidity entry for an asset. - router.delete("/:anchor/:asset", (req: Request, res: Response) => { - const entry = service.removeEntry(req.params.anchor, req.params.asset); - res.json({ ...entry, amount: entry.amount.toString() }); + router.delete("/:anchor/:asset", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve().then(async () => { + const entry = service.removeEntry(req.params.anchor, req.params.asset); + await service.flush(); + res.json({ ...entry, amount: entry.amount.toString() }); + }).catch(next); }); // Read the raw liquidity entries for a single anchor. diff --git a/src/routes/settlements.ts b/src/routes/settlements.ts index cc54eba..6de1682 100644 --- a/src/routes/settlements.ts +++ b/src/routes/settlements.ts @@ -2,7 +2,7 @@ * Routes for opening and managing settlements. */ -import { Router, Request, Response } from "express"; +import { Router, Request, Response, NextFunction } from "express"; import { SettlementService } from "../services/settlementService"; import { Settlement } from "../models/settlement"; import { AuditEntry } from "../middleware/auditLog"; @@ -36,9 +36,10 @@ export function settlementRouter( // Open a new settlement, reserving liquidity. // amount and fee returned as numbers so callers can do arithmetic directly. - router.post("/", (req: Request, res: Response) => { - const s = service.open(req.body ?? {}); - res.status(201).json({ ...s, amount: Number(s.amount), fee: Number(s.fee) }); + router.post("/", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve(service.open(req.body ?? {})).then((s) => { + res.status(201).json({ ...s, amount: Number(s.amount), fee: Number(s.fee) }); + }).catch(next); }); // List settlements, optionally filtered by ?anchor= and ?asset=. The default @@ -128,15 +129,17 @@ export function settlementRouter( }); // Execute a pending settlement. - router.post("/:id/execute", (req: Request, res: Response) => { - const s = service.execute(req.params.id); - res.json({ ...s, amount: s.amount.toString(), fee: s.fee.toString() }); + router.post("/:id/execute", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve(service.execute(req.params.id)).then((s) => { + res.json({ ...s, amount: s.amount.toString(), fee: s.fee.toString() }); + }).catch(next); }); // Cancel a pending settlement, optionally recording a { reason }. - router.post("/:id/cancel", (req: Request, res: Response) => { - const s = service.cancel(req.params.id, (req.body ?? {}).reason); - res.json({ ...s, amount: s.amount.toString(), fee: s.fee.toString() }); + router.post("/:id/cancel", (req: Request, res: Response, next: NextFunction) => { + Promise.resolve(service.cancel(req.params.id, (req.body ?? {}).reason)).then((s) => { + res.json({ ...s, amount: s.amount.toString(), fee: s.fee.toString() }); + }).catch(next); }); // Return audit entries whose path references this settlement id. diff --git a/src/services/anchorService.ts b/src/services/anchorService.ts index 969c5f8..c959e4b 100644 --- a/src/services/anchorService.ts +++ b/src/services/anchorService.ts @@ -17,6 +17,12 @@ export class AnchorService { constructor(private readonly repo: AnchorRepository) {} + /** Waits for a durable repository, while remaining a no-op for memory tests. */ + async flush(): Promise { + const durable = this.repo as AnchorRepository & { flush?: () => Promise }; + await durable.flush?.(); + } + /** Registers a new anchor. Fails with 409 if the id already exists. */ register(input: { id: unknown; name?: unknown }): Anchor { const id = requireString(input.id, "id"); diff --git a/src/services/liquidityService.ts b/src/services/liquidityService.ts index a15b15f..63ce0f7 100644 --- a/src/services/liquidityService.ts +++ b/src/services/liquidityService.ts @@ -33,6 +33,12 @@ export class LiquidityService { private readonly settlementService?: SettlementService, ) {} + /** Waits for a durable repository, while remaining a no-op for memory tests. */ + async flush(): Promise { + const durable = this.repo as LiquidityRepository & { flush?: () => Promise }; + await durable.flush?.(); + } + /** * Records `amount` of liquidity from `anchor` in `asset`. If the anchor * already has a balance for the asset, the amounts are accumulated. diff --git a/src/services/persistentSettlementService.ts b/src/services/persistentSettlementService.ts new file mode 100644 index 0000000..c75c997 --- /dev/null +++ b/src/services/persistentSettlementService.ts @@ -0,0 +1,137 @@ +/** + * PostgreSQL settlement operations. + * + * Reads and validation are delegated to the existing synchronous service so + * response semantics stay identical. Mutating settlement operations cross the + * database boundary and therefore use the transaction methods on + * PostgresPersistence. The routers accept either synchronous or Promise + * results, which keeps the in-memory test service unchanged. + */ + +import { ApiError } from "../errors/ApiError"; +import { Settlement } from "../models/settlement"; +import { PostgresPersistence } from "../persistence/postgres"; +import { PersistentSettlementRepository } from "../repositories/persistentRepositories"; +import { AnchorService } from "./anchorService"; +import { SettlementService } from "./settlementService"; +import { + normalizeAsset, + requireBigInt, + requirePositiveInteger, + requireString, + requireStringMaxLength, +} from "../utils/validation"; + +export type MaybePromise = T | Promise; + +export interface SettlementServiceLike { + open(input: { anchor: unknown; asset: unknown; amount: unknown }): MaybePromise; + execute(id: unknown): MaybePromise; + cancel(id: unknown, reason?: unknown): MaybePromise; + get(id: unknown): MaybePromise; + list(filters?: { anchor?: string; asset?: string }): MaybePromise; +} + +export class PersistentSettlementService extends SettlementService { + constructor( + private readonly database: PostgresPersistence, + private readonly durableSettlements: PersistentSettlementRepository, + liquidity: import("../repositories/persistentRepositories").PersistentLiquidityRepository, + anchors: AnchorService, + feeBps: bigint | number, + ) { + super(durableSettlements, liquidity, anchors, feeBps); + this.durableFeeBps = BigInt(feeBps); + } + + private readonly durableFeeBps: bigint; + + open(input: { anchor: unknown; asset: unknown; amount: unknown }): any { + return this.openAsync(input); + } + + private async openAsync(input: { anchor: unknown; asset: unknown; amount: unknown }): Promise { + const anchor = requireString(input.anchor, "anchor"); + const asset = normalizeAsset(input.asset); + const amount = requireBigInt(input.amount, "amount"); + + // The active-anchor check remains in the domain service, preserving its + // error code and wording without making a second database lookup here. + if (!this.anchors.isActive(anchor)) { + throw ApiError.badRequest( + `anchor "${anchor}" is not an active registered anchor`, + "ANCHOR_NOT_ACTIVE", + ); + } + + const fee = (amount * this.durableFeeBps + 9_999n) / 10_000n; + try { + const created = await this.database.openSettlement({ + anchor, + asset, + amount, + fee, + createdAt: new Date().toISOString(), + }); + this.durableSettlements.hydrate(created); + this.rebuildAccounting(); + return created; + } catch (error) { + if (error instanceof Error && error.name === "InsufficientLiquidityError") { + const available = super.available(asset); + throw ApiError.badRequest( + `insufficient liquidity for ${asset}: requested ${amount}, available ${available}`, + "INSUFFICIENT_LIQUIDITY", + ); + } + throw error; + } + } + + execute(idInput: unknown): any { + return this.executeAsync(idInput); + } + + private async executeAsync(idInput: unknown): Promise { + const id = requirePositiveInteger(idInput, "id"); + const current = super.get(id); + if (current.status !== "pending") { + throw ApiError.conflict( + `settlement ${id} is ${current.status}, not pending`, + "INVALID_STATE", + ); + } + const updated = await this.database.transitionSettlement(id, "executed"); + this.durableSettlements.hydrate(updated); + this.rebuildAccounting(); + return updated; + } + + cancel(idInput: unknown, reasonInput?: unknown): any { + return this.cancelAsync(idInput, reasonInput); + } + + private async cancelAsync(idInput: unknown, reasonInput?: unknown): Promise { + const id = requirePositiveInteger(idInput, "id"); + const current = super.get(id); + if (current.status !== "pending") { + throw ApiError.conflict( + `settlement ${id} is ${current.status}, not pending`, + "INVALID_STATE", + ); + } + const reason = + reasonInput === undefined + ? undefined + : requireStringMaxLength(reasonInput, "reason", 500); + const updated = await this.database.transitionSettlement(id, "cancelled", reason); + this.durableSettlements.hydrate(updated); + this.rebuildAccounting(); + return updated; + } + + async flush(): Promise { + await this.database.flush(); + } + +} diff --git a/src/services/settlementService.ts b/src/services/settlementService.ts index 06a77a9..5f645be 100644 --- a/src/services/settlementService.ts +++ b/src/services/settlementService.ts @@ -32,12 +32,13 @@ export class SettlementService { private readonly consumed = new Map(); constructor( - private readonly settlements: SettlementRepository, - private readonly liquidity: LiquidityRepository, - private readonly anchors: AnchorService, + protected readonly settlements: SettlementRepository, + protected readonly liquidity: LiquidityRepository, + protected readonly anchors: AnchorService, feeBps: bigint | number = DEFAULT_FEE_BPS, ) { this.feeBps = BigInt(feeBps); + this.rebuildAccounting(); } private readonly feeBps: bigint; @@ -53,6 +54,29 @@ export class SettlementService { return this.reserved.get(asset) ?? 0n; } + /** + * Rebuilds derived accounting from durable settlement rows. The maps remain + * a fast read cache for the synchronous service contract, but they are no + * longer the source of truth after a process restart. + */ + public rebuildAccounting(): void { + this.reserved.clear(); + this.consumed.clear(); + for (const settlement of this.settlements.all()) { + if (settlement.status === "pending") { + this.reserved.set( + settlement.asset, + (this.reserved.get(settlement.asset) ?? 0n) + settlement.amount, + ); + } else if (settlement.status === "executed") { + this.consumed.set( + settlement.asset, + (this.consumed.get(settlement.asset) ?? 0n) + settlement.amount, + ); + } + } + } + /** Opens a pending settlement, reserving liquidity from the pool. */ open(input: { anchor: unknown; asset: unknown; amount: unknown }): Settlement { const anchor = requireString(input.anchor, "anchor");