From 8e3ae7cb8bf779068a12812bb27cb23f2029748f Mon Sep 17 00:00:00 2001 From: titankrypt-gif <331549701+titankrypt-gif@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:23:42 +0100 Subject: [PATCH] feat(db): verify schema at startup and expose /ready (#352) --- docs/database-schema.md | 75 ++++++++++++ openapi.json | 94 ++++++++++++++ openapi.yaml | 65 ++++++++++ src/__tests__/ready.test.ts | 118 ++++++++++++++++++ src/__tests__/schemaVerification.test.ts | 87 +++++++++++++ src/api/rest.ts | 37 ++++++ src/app.ts | 2 +- src/db/schemaVerify.ts | 149 +++++++++++++++++++++++ src/index.ts | 6 + 9 files changed, 632 insertions(+), 1 deletion(-) create mode 100644 docs/database-schema.md create mode 100644 src/__tests__/ready.test.ts create mode 100644 src/__tests__/schemaVerification.test.ts create mode 100644 src/db/schemaVerify.ts diff --git a/docs/database-schema.md b/docs/database-schema.md new file mode 100644 index 00000000..4270285e --- /dev/null +++ b/docs/database-schema.md @@ -0,0 +1,75 @@ +# Database schema and startup verification + +## Current state: `prisma db push` + +`prisma/` contains `schema.prisma` but no `migrations/` directory. Schema +changes are applied with `prisma db push`; the container entrypoint in +`src/index.ts` also runs: + +```bash +prisma db push --accept-data-loss +``` + +before the API starts listening. That keeps the deployed schema in step with +the Prisma model, but it leaves no versioned record of *which* schema a +database is running. A push that silently fails, is blocked by permissions, or +runs against the wrong database would leave the process starting against a +schema the code does not understand. + +## Startup guard (#352) + +`src/db/schemaVerify.ts` owns the expected physical schema (Prisma model names +mapped through `@@map` / `@map`) and checks it with two cheap queries: + +1. `information_schema.tables` for every expected table. +2. `information_schema.columns` for the columns of the tables that exist. + +`assertSchemaValid()` runs after the boot-time push and before +`loadPersistedPairs()` and `app.listen()`. On a mismatch it throws +`DatabaseSchemaMismatchError`, which names the missing tables and columns, and +the existing fatal-startup handler exits the process. A connection failure is +allowed to propagate as a connection failure, not reported as a schema +mismatch. + +The successful result is cached, so `/ready` is a memory read rather than a +second pair of queries on every probe: + +```json +{ "ready": true, "checks": { "schema": { "ok": true } }, "checkedAt": "..." } +``` + +A stale schema returns HTTP 503 with the missing entities: + +```json +{ + "ready": false, + "checks": { + "schema": { + "ok": false, + "missingTables": ["pool_snapshots"], + "missingColumns": ["price_points.network"] + } + } +} +``` + +`/status` remains the data-freshness endpoint (last indexed ledger and ingest +lag); `/ready` answers only "can this instance serve traffic against the +schema it was built for?". + +## Why there is no migrations directory (and what comes next) + +The project has used `db push` for rapid iteration while the ingestion, +partitioning, and x402 schemas were still moving. Adopting Prisma Migrate is a +deliberate operational change, not a drive-by refactor: + +1. Produce a baseline from the current database: + `npx prisma migrate diff --from-empty --to-schema-datamodel prisma/schema.prisma --script > prisma/migrations/0_init/migration.sql` +2. Mark it applied: `npx prisma migrate resolve --applied 0_init`. +3. Replace the boot-time `db push` with `prisma migrate deploy` in the deploy + pipeline, where an operator can see and roll back a failed migration. +4. Keep this startup guard as the runtime barrier: migrations own the change, + and the guard refuses to serve traffic if the result is still incomplete. + +Until that migration work happens, the guard is the safety net that turns a +missing column from a request-time crash into a named startup failure. diff --git a/openapi.json b/openapi.json index b889418f..2d56e02d 100644 --- a/openapi.json +++ b/openapi.json @@ -168,6 +168,100 @@ } } }, + "/ready": { + "get": { + "operationId": "getReady", + "summary": "Readiness probe (database schema)", + "description": "Returns 200 when this process booted against a database schema that matches the build, and 503 with the missing tables/columns otherwise. The result is cached from startup, so the probe does not re-query the database. Public and rate-limit exempt like `/status`.", + "security": [], + "responses": { + "200": { + "description": "Instance is ready to serve traffic", + "headers": { + "Cache-Control": { + "$ref": "#/components/headers/CacheControl" + } + }, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "ready": { + "type": "boolean" + }, + "checks": { + "type": "object", + "properties": { + "schema": { + "type": "object", + "properties": { + "ok": { + "type": "boolean" + } + } + } + } + }, + "checkedAt": { + "type": "string", + "format": "date-time" + } + } + } + } + } + }, + "503": { + "description": "Schema is missing tables or columns, or the check could not reach the database", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "ready": { + "type": "boolean" + }, + "checks": { + "type": "object", + "properties": { + "schema": { + "type": "object", + "properties": { + "ok": { + "type": "boolean" + }, + "missingTables": { + "type": "array", + "items": { + "type": "string" + } + }, + "missingColumns": { + "type": "array", + "items": { + "type": "string" + } + }, + "error": { + "type": "string" + } + } + } + } + }, + "checkedAt": { + "type": "string", + "format": "date-time" + } + } + } + } + } + } + } + } + }, "/snapshot": { "get": { "operationId": "getSnapshot", diff --git a/openapi.yaml b/openapi.yaml index c96dd10a..d79ceb94 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -117,6 +117,71 @@ paths: '429': $ref: '#/components/responses/TooManyRequests' + /ready: + get: + operationId: getReady + summary: Readiness probe (database schema) + description: >- + Returns 200 when this process booted against a database schema that + matches the build, and 503 with the missing tables/columns otherwise. + The result is cached from startup, so the probe does not re-query the + database. Public and rate-limit exempt like `/status`. + security: [] + responses: + '200': + description: Instance is ready to serve traffic + headers: + Cache-Control: + $ref: '#/components/headers/CacheControl' + content: + application/json: + schema: + type: object + properties: + ready: + type: boolean + checks: + type: object + properties: + schema: + type: object + properties: + ok: + type: boolean + checkedAt: + type: string + format: date-time + '503': + description: Schema is missing tables or columns, or the check could not reach the database + content: + application/json: + schema: + type: object + properties: + ready: + type: boolean + checks: + type: object + properties: + schema: + type: object + properties: + ok: + type: boolean + missingTables: + type: array + items: + type: string + missingColumns: + type: array + items: + type: string + error: + type: string + checkedAt: + type: string + format: date-time + /snapshot: get: operationId: getSnapshot diff --git a/src/__tests__/ready.test.ts b/src/__tests__/ready.test.ts new file mode 100644 index 00000000..19e24791 --- /dev/null +++ b/src/__tests__/ready.test.ts @@ -0,0 +1,118 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' + +const { mockQuery, mockGetCachedPrice, mockGetBestRoute } = vi.hoisted(() => ({ + mockQuery: vi.fn(), + mockGetCachedPrice: vi.fn(), + mockGetBestRoute: vi.fn(), +})) + +vi.mock('../db', () => ({ + pgPool: { query: mockQuery }, +})) +vi.mock('../redis', () => ({ + getCachedPrice: mockGetCachedPrice, + setCachedPrice: vi.fn(), +})) +vi.mock('../aggregator/bestRoute', () => ({ + getBestRoute: mockGetBestRoute, +})) +vi.mock('../config', () => ({ + config: { pairs: [], cache: { priceTtl: 10 } }, + activeNetwork: 'testnet', + getNetworkConfig: () => ({ pairs: [] }), +})) + +import { registerRESTRoutes } from '../api/rest' +import { registerNetworkSelector } from '../middleware/network' +import { EXPECTED_SCHEMA, resetSchemaVerificationCache } from '../db/schemaVerify' + +type Schema = Record + +function completeSchema(): Schema { + return Object.fromEntries( + Object.entries(EXPECTED_SCHEMA).map(([table, columns]) => [table, [...columns]]), + ) +} + +function schemaQuery(schema: Schema) { + return async (sql: string) => { + if (sql.includes('information_schema.tables')) { + return { rows: Object.keys(schema).map(table_name => ({ table_name })) } + } + if (sql.includes('information_schema.columns')) { + return { + rows: Object.entries(schema).flatMap(([table_name, columns]) => + columns.map(column_name => ({ table_name, column_name })), + ), + } + } + return { rows: [] } + } +} + +async function buildApp() { + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + await registerRESTRoutes(app) + await app.ready() + return app +} + +beforeEach(() => { + vi.clearAllMocks() + resetSchemaVerificationCache() +}) + +describe('GET /ready (#352)', () => { + it('reports ready when the database schema matches', async () => { + mockQuery.mockImplementation(schemaQuery(completeSchema())) + const app = await buildApp() + + const res = await app.inject({ method: 'GET', url: '/ready' }) + + expect(res.statusCode).toBe(200) + expect(res.json()).toMatchObject({ ready: true, checks: { schema: { ok: true } } }) + }) + + it('returns 503 and names the missing table and column', async () => { + const schema = { + ...completeSchema(), + price_points: EXPECTED_SCHEMA.price_points.filter(column => column !== 'network'), + } + delete schema.pool_snapshots + mockQuery.mockImplementation(schemaQuery(schema)) + const app = await buildApp() + + const res = await app.inject({ method: 'GET', url: '/ready' }) + + expect(res.statusCode).toBe(503) + expect(res.json().ready).toBe(false) + expect(res.json().checks.schema.missingTables).toContain('pool_snapshots') + expect(res.json().checks.schema.missingColumns).toContain('price_points.network') + }) + + it('caches the check so repeated probes do not re-query', async () => { + mockQuery.mockImplementation(schemaQuery(completeSchema())) + const app = await buildApp() + + await app.inject({ method: 'GET', url: '/ready' }) + const callsAfterFirst = mockQuery.mock.calls.length + await app.inject({ method: 'GET', url: '/ready' }) + + expect(mockQuery.mock.calls.length).toBe(callsAfterFirst) + }) + + it('returns 503 instead of crashing when the schema check cannot reach the database', async () => { + mockQuery.mockRejectedValue(new Error('connection refused')) + const app = await buildApp() + + const res = await app.inject({ method: 'GET', url: '/ready' }) + + expect(res.statusCode).toBe(503) + expect(res.json()).toMatchObject({ + ready: false, + checks: { schema: { ok: false, error: 'database_unavailable' } }, + }) + }) +}) diff --git a/src/__tests__/schemaVerification.test.ts b/src/__tests__/schemaVerification.test.ts new file mode 100644 index 00000000..1ed6af65 --- /dev/null +++ b/src/__tests__/schemaVerification.test.ts @@ -0,0 +1,87 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { + EXPECTED_SCHEMA, + verifyDatabaseSchema, + assertSchemaValid, + DatabaseSchemaMismatchError, + resetSchemaVerificationCache, +} from '../db/schemaVerify' + +type Schema = Record + +/** A pg-like pool whose information_schema answers describe `schema`. */ +function poolFor(schema: Schema) { + const query = vi.fn(async (sql: string) => { + if (sql.includes('information_schema.tables')) { + return { rows: Object.keys(schema).map(table_name => ({ table_name })) } + } + if (sql.includes('information_schema.columns')) { + return { + rows: Object.entries(schema).flatMap(([table_name, columns]) => + columns.map(column_name => ({ table_name, column_name })), + ), + } + } + return { rows: [] } + }) + return { query } as any +} + +function completeSchema(): Schema { + return Object.fromEntries( + Object.entries(EXPECTED_SCHEMA).map(([table, columns]) => [table, [...columns]]), + ) +} + +beforeEach(() => { + resetSchemaVerificationCache() +}) + +describe('database schema verification (#352)', () => { + it('passes when every expected table and column exists', async () => { + const result = await verifyDatabaseSchema(poolFor(completeSchema())) + + expect(result.valid).toBe(true) + expect(result.missingTables).toEqual([]) + expect(result.missingColumns).toEqual([]) + expect(result.checkedAt).toEqual(expect.any(String)) + }) + + it('names a missing table and does not query its columns', async () => { + const schema = completeSchema() + delete schema.pool_snapshots + + const result = await verifyDatabaseSchema(poolFor(schema)) + + expect(result.valid).toBe(false) + expect(result.missingTables).toEqual(['pool_snapshots']) + expect(result.missingColumns).toEqual([]) + }) + + it('names missing columns inside tables that do exist', async () => { + const schema = { + ...completeSchema(), + price_points: EXPECTED_SCHEMA.price_points.filter(column => column !== 'network' && column !== 'source'), + } + + const result = await verifyDatabaseSchema(poolFor(schema)) + + expect(result.valid).toBe(false) + expect(result.missingTables).toEqual([]) + expect(result.missingColumns).toEqual(['price_points.network', 'price_points.source']) + }) + + it('throws a descriptive mismatch error at startup', async () => { + const pool = poolFor({ price_points: ['id'] }) + + await expect(assertSchemaValid(pool)).rejects.toBeInstanceOf(DatabaseSchemaMismatchError) + await expect(assertSchemaValid(pool)).rejects.toThrow(/missing tables \[.*pool_snapshots/) + await expect(assertSchemaValid(pool)).rejects.toThrow(/missing columns \[price_points\.network/) + }) + + it('lets connection failures propagate instead of reporting a schema mismatch', async () => { + const pool = { query: vi.fn().mockRejectedValue(new Error('connection refused')) } as any + + await expect(verifyDatabaseSchema(pool)).rejects.toThrow('connection refused') + }) +}) diff --git a/src/api/rest.ts b/src/api/rest.ts index 2f5fbdce..267c4a08 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -4,6 +4,7 @@ import { getCachedPrice, setCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' +import { getCachedSchemaVerification, verifyDatabaseSchema } from '../db/schemaVerify' import { config, getNetworkConfig, activeNetwork, type NetworkName } from '../config' import '../middleware/network' // declares req.network on the FastifyRequest type import { @@ -96,6 +97,42 @@ export async function registerRESTRoutes(app: FastifyInstance) { }) + // GET /ready — public readiness probe (no API key required). Reports the + // schema verification captured at startup; if boot has not run it yet (for + // example in a test harness), verify now so the answer is still live. + app.get('/ready', { config: { public: true } }, async (req, reply) => { + let schema = getCachedSchemaVerification() + if (!schema) { + try { + schema = await verifyDatabaseSchema(pgPool) + } catch (err) { + req.log.error({ err }, 'readiness schema check failed') + return reply.status(503).send({ + ready: false, + checks: { schema: { ok: false, error: 'database_unavailable' } }, + }) + } + } + if (!schema.valid) { + return reply.status(503).send({ + ready: false, + checks: { + schema: { + ok: false, + missingTables: schema.missingTables, + missingColumns: schema.missingColumns, + }, + }, + checkedAt: schema.checkedAt, + }) + } + return { + ready: true, + checks: { schema: { ok: true } }, + checkedAt: schema.checkedAt, + } + }) + // GET /price/:assetA/:assetB app.get<{ Params: { assetA: string; assetB: string } }>( '/price/:assetA/:assetB', diff --git a/src/app.ts b/src/app.ts index 32dac100..8dff56ac 100644 --- a/src/app.ts +++ b/src/app.ts @@ -103,7 +103,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise req.apiKey?.ratePerMin ?? ipRateLimitMax, timeWindow: '1 minute', keyGenerator: (req) => req.apiKey?.id ?? req.ip, - allowList: (req) => req.url === '/status', + allowList: (req) => req.url === '/status' || req.url === '/ready', errorResponseBuilder: (req, context) => ({ statusCode: 429, error: 'Too Many Requests', diff --git a/src/db/schemaVerify.ts b/src/db/schemaVerify.ts new file mode 100644 index 00000000..916d7852 --- /dev/null +++ b/src/db/schemaVerify.ts @@ -0,0 +1,149 @@ +import type { Pool } from 'pg' + +export interface SchemaVerificationResult { + valid: boolean + missingTables: string[] + missingColumns: string[] + checkedAt: string +} + +/** + * The physical tables and columns the code depends on, using the names from + * `prisma/schema.prisma`'s `@@map` / `@map` annotations. + * + * This is deliberately a small, explicit list instead of a Prisma DMMF walk: + * the check must work in the deployed container even if the generated client + * is rebuilt, and it must be cheap (two `information_schema` queries). + */ +export const EXPECTED_SCHEMA: Record = { + price_points: [ + 'id', 'network', 'asset_a', 'asset_b', 'pair_key', 'source', 'pool_id', + 'price', 'base_volume', 'counter_volume', 'ledger', 'timestamp', 'event_id', + ], + pool_snapshots: [ + 'id', 'network', 'pool_id', 'asset_a', 'asset_b', 'reserve_a', 'reserve_b', + 'spot_price', 'total_shares', 'fee_bp', 'ledger', 'timestamp', + ], + price_aggregates: [ + 'pair_key', 'network', 'window', 'bucket', 'vwap', 'sdex_vwap', 'amm_vwap', + 'volume', 'sdex_volume', 'amm_volume', 'trade_count', 'open_price', + 'close_price', 'high_price', 'low_price', + ], + price_snapshots: ['pair', 'network', 'ts', 'price', 'volume'], + indexer_state: ['id', 'network', 'last_cursor', 'last_ledger', 'last_processed_at', 'updated_at'], + pair_configs: [ + 'pair_key', 'network', 'asset_a_code', 'asset_a_issuer', 'asset_b_code', + 'asset_b_issuer', 'added_at', + ], + bazaar_resources: [ + 'id', 'type', 'network', 'url', 'description', 'mime_type', 'service_name', + 'tags', 'icon_url', 'mcp_tool_name', 'http_method', 'accepts', 'pay_to', + 'bazaar_info', 'bazaar_schema', 'route_template', 'extension_keys', + 'created_at', 'updated_at', + ], + settlement_attempts: [ + 'id', 'network', 'tx_hash', 'state', 'payer', 'pay_to', 'asset', 'amount', + 'error_reason', 'error_message', 'response', 'created_at', 'updated_at', + ], + webhooks: [ + 'id', 'network', 'url', 'asset_a', 'asset_b', 'threshold', 'direction', + 'secret', 'created_at', + ], + api_keys: [ + 'id', 'hash', 'label', 'rate_per_min', 'rate_per_day', 'revoked_at', + 'created_at', 'monthly_quota_cents', 'daily_quota_cents', 'overage_policy', + ], +} + +let cachedVerification: SchemaVerificationResult | null = null + +/** + * Verify the connected database contains every expected table and column. + * + * Query failures (the database being unreachable, for example) are allowed to + * propagate: an outage is not a schema mismatch, and callers need to tell the + * two apart. On success the result is cached so `/ready` is a memory read. + */ +export async function verifyDatabaseSchema( + pool: Pick, + expected: Record = EXPECTED_SCHEMA, +): Promise { + const tables = Object.keys(expected) + const tableResult = await pool.query( + `SELECT table_name + FROM information_schema.tables + WHERE table_schema = 'public' + AND table_name = ANY($1::text[])`, + [tables], + ) + const existingTables = new Set(tableResult.rows.map((row: { table_name: string }) => row.table_name)) + const missingTables = tables.filter((table) => !existingTables.has(table)) + + const tablesToInspect = tables.filter((table) => existingTables.has(table)) + const existingColumns = new Set() + if (tablesToInspect.length > 0) { + const columnResult = await pool.query( + `SELECT table_name, column_name + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = ANY($1::text[])`, + [tablesToInspect], + ) + for (const row of columnResult.rows as { table_name: string; column_name: string }[]) { + existingColumns.add(`${row.table_name}.${row.column_name}`) + } + } + + const missingColumns: string[] = [] + for (const table of tablesToInspect) { + for (const column of expected[table] ?? []) { + if (!existingColumns.has(`${table}.${column}`)) { + missingColumns.push(`${table}.${column}`) + } + } + } + + const result: SchemaVerificationResult = { + valid: missingTables.length === 0 && missingColumns.length === 0, + missingTables, + missingColumns, + checkedAt: new Date().toISOString(), + } + cachedVerification = result + return result +} + +/** The result captured at startup (or by the most recent `/ready` check). */ +export function getCachedSchemaVerification(): SchemaVerificationResult | null { + return cachedVerification +} + +/** Test helper: forget the cached result so the next check queries again. */ +export function resetSchemaVerificationCache(): void { + cachedVerification = null +} + +export class DatabaseSchemaMismatchError extends Error { + constructor(public readonly result: SchemaVerificationResult) { + const details = [ + result.missingTables.length > 0 ? `missing tables [${result.missingTables.join(', ')}]` : '', + result.missingColumns.length > 0 ? `missing columns [${result.missingColumns.join(', ')}]` : '', + ].filter(Boolean).join('; ') + super(`Database schema mismatch: ${details}`) + this.name = 'DatabaseSchemaMismatchError' + } +} + +/** + * Throws `DatabaseSchemaMismatchError` when the database is missing any table + * or column the code expects. Used at startup so a stale instance exits before + * serving a request that would fail on a missing column. + */ +export async function assertSchemaValid( + pool: Pick, + expected: Record = EXPECTED_SCHEMA, +): Promise { + const result = await verifyDatabaseSchema(pool, expected) + if (!result.valid) throw new DatabaseSchemaMismatchError(result) + return result +} diff --git a/src/index.ts b/src/index.ts index 55ed4c2b..f7f9176b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,6 +9,7 @@ import { getEnabledNetworks } from './network/enabledNetworks' import { createLogger } from './logger' import { redis } from './redis' import { pgPool } from './db' +import { assertSchemaValid } from './db/schemaVerify' import { buildApp } from './app' import { fanOutManager } from './ws/fanout' @@ -52,6 +53,11 @@ async function main() { await pgPool.connect() log.info('postgresql connected') + // Fail before listening if the database does not have the tables/columns + // this build expects. The guard is cached, so /ready is a memory read. + await assertSchemaValid(pgPool) + log.info('database schema verified') + // ── Load persisted runtime pairs ────────────────────────────────────────── await loadPersistedPairs()