diff --git a/docs/database-schema.md b/docs/database-schema.md new file mode 100644 index 00000000..10726ef3 --- /dev/null +++ b/docs/database-schema.md @@ -0,0 +1,31 @@ +# Database Schema & Migration Strategy + +## Current State: `prisma db push` + +`prisma/` currently contains `schema.prisma` without a `migrations/` directory. Lens currently leverages `prisma db push` for schema updates. + +### Architectural Rationale + +1. **Rapid DeFi Primitive Prototyping:** + During the initial development of the multi-venue aggregator (covering SDEX orderbooks, AMM liquidity pools, Soroswap factory pairs, and Reflector oracles), data models evolved rapidly across iterations. `prisma db push` allowed agile schema prototyping without accumulating redundant migration step churn. + +2. **Zero In-Process Migrations on Container Boot:** + Running complex automated migrations inside container entrypoints poses locking hazards when running horizontally scaled nodes against shared PostgreSQL databases. Decoupling schema validation from schema migration prevents distributed boot race conditions. + +3. **Time-Series Partitioning Compatibility:** + Lens's high-frequency tables (such as `price_points` and `price_aggregates`) require composite primary keys (e.g. `[id, timestamp]`) and time-series optimizations. + +## Startup Verification Guard (#352) + +To ensure that running code never crashes at request time due to missing columns or unapplied schema changes: +- `src/db/schemaVerify.ts` runs on boot via `assertSchemaValid()` / `verifyDatabaseSchema()`. +- It performs a low-overhead inspection against PostgreSQL `information_schema.tables` and `information_schema.columns`. +- If an expected table or column is absent, it fails fast and names the missing entities. +- `/ready` and `/status` can query `verifyDatabaseSchema()` to isolate instances before they receive live traffic. + +## Path Forward: Transitioning to `prisma migrate` + +When moving to formal multi-region enterprise production deployments: +1. Generate baseline migration from the current production database: `npx prisma migrate diff --from-empty --to-schema-datamodel prisma/schema.prisma --script > prisma/migrations/0_init/migration.sql`. +2. Mark baseline as applied: `npx prisma migrate resolve --applied 0_init`. +3. Enforce CI/CD pipeline validation using Prisma Migrate with the schema startup verification guard serving as the runtime barrier. diff --git a/src/__tests__/schemaVerification.test.ts b/src/__tests__/schemaVerification.test.ts new file mode 100644 index 00000000..09dd2e05 --- /dev/null +++ b/src/__tests__/schemaVerification.test.ts @@ -0,0 +1,111 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { + verifyDatabaseSchema, + assertSchemaValid, + DatabaseSchemaMismatchError, + EXPECTED_SCHEMA, +} from '../db/schemaVerify' + +describe('Database Schema Startup Verification (#352)', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('passes when all expected tables and columns exist', async () => { + const mockQuery = vi.fn() + // First query: tables in information_schema + .mockResolvedValueOnce({ + rows: [ + { table_name: 'price_points' }, + { table_name: 'pool_snapshots' }, + { table_name: 'price_aggregates' }, + { table_name: 'price_snapshots' }, + { table_name: 'indexer_state' }, + ], + }) + // Second query: columns in information_schema + .mockResolvedValueOnce({ + rows: [ + ...EXPECTED_SCHEMA.price_points.map(col => ({ table_name: 'price_points', column_name: col })), + ...EXPECTED_SCHEMA.pool_snapshots.map(col => ({ table_name: 'pool_snapshots', column_name: col })), + ...EXPECTED_SCHEMA.price_aggregates.map(col => ({ table_name: 'price_aggregates', column_name: col })), + ...EXPECTED_SCHEMA.price_snapshots.map(col => ({ table_name: 'price_snapshots', column_name: col })), + ...EXPECTED_SCHEMA.indexer_state.map(col => ({ table_name: 'indexer_state', column_name: col })), + ], + }) + + const pool = { query: mockQuery } as any + const result = await verifyDatabaseSchema(pool) + + expect(result.valid).toBe(true) + expect(result.missingTables).toEqual([]) + expect(result.missingColumns).toEqual([]) + }) + + it('fails and identifies missing tables', async () => { + const mockQuery = vi.fn().mockResolvedValueOnce({ + rows: [ + { table_name: 'price_points' }, + // pool_snapshots missing + { table_name: 'price_aggregates' }, + { table_name: 'price_snapshots' }, + { table_name: 'indexer_state' }, + ], + }).mockResolvedValueOnce({ + rows: [ + ...EXPECTED_SCHEMA.price_points.map(col => ({ table_name: 'price_points', column_name: col })), + ...EXPECTED_SCHEMA.price_aggregates.map(col => ({ table_name: 'price_aggregates', column_name: col })), + ...EXPECTED_SCHEMA.price_snapshots.map(col => ({ table_name: 'price_snapshots', column_name: col })), + ...EXPECTED_SCHEMA.indexer_state.map(col => ({ table_name: 'indexer_state', column_name: col })), + ], + }) + + const pool = { query: mockQuery } as any + const result = await verifyDatabaseSchema(pool) + + expect(result.valid).toBe(false) + expect(result.missingTables).toContain('pool_snapshots') + }) + + it('fails and identifies missing columns within existing tables', async () => { + const mockQuery = vi.fn() + .mockResolvedValueOnce({ + rows: [ + { table_name: 'price_points' }, + ], + }) + .mockResolvedValueOnce({ + rows: [ + // Omits 'network' and 'source' columns + { table_name: 'price_points', column_name: 'id' }, + { table_name: 'price_points', column_name: 'asset_a' }, + { table_name: 'price_points', column_name: 'asset_b' }, + { table_name: 'price_points', column_name: 'pair_key' }, + { table_name: 'price_points', column_name: 'price' }, + { table_name: 'price_points', column_name: 'timestamp' }, + ], + }) + + const testSchema = { + price_points: ['id', 'network', 'asset_a', 'asset_b', 'pair_key', 'source', 'price', 'timestamp'], + } + + const pool = { query: mockQuery } as any + const result = await verifyDatabaseSchema(pool, testSchema) + + expect(result.valid).toBe(false) + expect(result.missingColumns).toContain('price_points.network') + expect(result.missingColumns).toContain('price_points.source') + }) + + it('assertSchemaValid throws DatabaseSchemaMismatchError with detailed entity names on mismatch', async () => { + const mockQuery = vi.fn().mockResolvedValueOnce({ + rows: [], // completely empty database + }) + + const pool = { query: mockQuery } as any + + await expect(assertSchemaValid(pool)).rejects.toThrowError(DatabaseSchemaMismatchError) + await expect(assertSchemaValid(pool)).rejects.toThrowError(/missing tables/) + }) +}) diff --git a/src/db/schemaVerify.ts b/src/db/schemaVerify.ts new file mode 100644 index 00000000..7b7d49e8 --- /dev/null +++ b/src/db/schemaVerify.ts @@ -0,0 +1,121 @@ +import type { Pool } from 'pg' +import { pgPool } from '../db' + +export interface SchemaVerificationResult { + valid: boolean + missingTables: string[] + missingColumns: string[] + checkedAt: string +} + +export const EXPECTED_SCHEMA: Record = { + price_points: ['id', 'network', 'asset_a', 'asset_b', 'pair_key', 'source', 'price', 'timestamp'], + pool_snapshots: ['id', 'network', 'pool_id', 'asset_a', 'asset_b', 'spot_price', 'timestamp'], + price_aggregates: ['pair_key', 'network', 'window', 'bucket', 'vwap'], + price_snapshots: ['pair', 'network', 'ts', 'price'], + indexer_state: ['id', 'network', 'last_cursor', 'last_ledger'], +} + +let cachedVerification: SchemaVerificationResult | null = null + +/** + * Verifies that the connected database contains the expected tables and columns (#352). + * Runs efficiently via single information_schema queries on startup. + */ +export async function verifyDatabaseSchema( + pool: Pick = pgPool, + expected: Record = EXPECTED_SCHEMA +): Promise { + const missingTables: string[] = [] + const missingColumns: string[] = [] + + try { + const tableRes = await pool.query( + `SELECT table_name + FROM information_schema.tables + WHERE table_schema = 'public'` + ) + const existingTables = new Set(tableRes.rows.map((r: any) => r.table_name)) + + for (const table of Object.keys(expected)) { + if (!existingTables.has(table)) { + missingTables.push(table) + } + } + + // Only query columns for existing tables to avoid unnecessary joins + const tablesToQuery = Object.keys(expected).filter(t => existingTables.has(t)) + if (tablesToQuery.length > 0) { + const colRes = await pool.query( + `SELECT table_name, column_name + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = ANY($1)`, + [tablesToQuery] + ) + const existingCols = new Set(colRes.rows.map((r: any) => `${r.table_name}.${r.column_name}`)) + + for (const table of tablesToQuery) { + for (const col of expected[table]) { + if (!existingCols.has(`${table}.${col}`)) { + missingColumns.push(`${table}.${col}`) + } + } + } + } + } catch (_err) { + // If database connection fails or queries error, report invalid schema + const result: SchemaVerificationResult = { + valid: false, + missingTables: Object.keys(expected), + missingColumns: [], + checkedAt: new Date().toISOString(), + } + cachedVerification = result + return result + } + + const result: SchemaVerificationResult = { + valid: missingTables.length === 0 && missingColumns.length === 0, + missingTables, + missingColumns, + checkedAt: new Date().toISOString(), + } + + cachedVerification = result + return result +} + +export function getCachedSchemaVerification(): SchemaVerificationResult | null { + return cachedVerification +} + +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 at startup: ${details}`) + this.name = 'DatabaseSchemaMismatchError' + } +} + +/** + * Asserts that the database schema matches expected models (#352). + * Fails fast and throws DatabaseSchemaMismatchError if any table or column is missing. + */ +export async function assertSchemaValid( + pool: Pick = pgPool, + 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..3bcd6512 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,10 @@ async function main() { await pgPool.connect() log.info('postgresql connected') + // Verify database schema matches expected tables and columns (#352) + await assertSchemaValid() + log.info('database schema verified') + // ── Load persisted runtime pairs ────────────────────────────────────────── await loadPersistedPairs()