feat(db): verify database schema matches code on startup (#352)

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.

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/)
+  })
+})

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<string, string[]> = {
+  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<Pool, 'query'> = pgPool,
+  expected: Record<string, string[]> = EXPECTED_SCHEMA
+): Promise<SchemaVerificationResult> {
+  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<Pool, 'query'> = pgPool,
+  expected: Record<string, string[]> = EXPECTED_SCHEMA
+): Promise<SchemaVerificationResult> {
+  const result = await verifyDatabaseSchema(pool, expected)
+  if (!result.valid) {
+    throw new DatabaseSchemaMismatchError(result)
+  }
+  return result
+}

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()