diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8c5a5e78..d402e4c7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,6 +27,21 @@ jobs: run: working-directory: backend + services: + postgres: + image: postgres:16 + env: + POSTGRES_DB: quittance_test + POSTGRES_USER: quittance + POSTGRES_PASSWORD: password + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 10s + --health-timeout 5s + --health-retries 5 + env: # The integration test starts its own Horizon stub on a loopback port and # overrides this at runtime. It is set here so the suite never reaches the @@ -34,6 +49,8 @@ jobs: STELLAR_NETWORK: TESTNET STELLAR_HORIZON_URL: http://127.0.0.1:1 NODE_ENV: test + DATABASE_URL: postgresql://quittance:password@localhost:5432/quittance_test + INVOICE_STORAGE: postgres steps: - uses: actions/checkout@v4 @@ -58,7 +75,13 @@ jobs: - name: Typecheck run: npm run typecheck - - name: Test + - name: Migrate database (run 1) + run: npm run db:migrate + + - name: Migrate database (run 2 — idempotency check) + run: npm run db:migrate + + - name: Test (including Postgres integration suite) run: npm test frontend: diff --git a/backend/scripts/cutover.ts b/backend/scripts/cutover.ts index 40018822..e7c83a83 100644 --- a/backend/scripts/cutover.ts +++ b/backend/scripts/cutover.ts @@ -19,6 +19,7 @@ import { importSnapshotToPostgres, CutoverSnapshot, } from '../src/services/cutover.service'; +import { checkMigrationStatus, runMigrations } from '../src/db/migration-runner'; async function main() { const options = { @@ -78,6 +79,17 @@ Environment Variables: console.log(`[Cutover] Snapshot validation PASSED (${snapshot.count} invoices).`); const isDryRun = Boolean(values['dry-run']); + console.log(`[Cutover] Checking database migration status...`); + const status = await checkMigrationStatus(pool); + if (!status.ready) { + if (isDryRun) { + console.log(`[Cutover] Pending migrations detected: ${status.pendingVersions.join(', ')}`); + } else { + console.log(`[Cutover] Applying pending migrations before import...`); + await runMigrations(pool); + } + } + console.log(`[Cutover] Executing transactional import (dry-run: ${isDryRun})...`); const result = await importSnapshotToPostgres(pool, snapshot, { dryRun: isDryRun }); diff --git a/backend/src/config/runtime.ts b/backend/src/config/runtime.ts index 4a3081c3..8b624d54 100644 --- a/backend/src/config/runtime.ts +++ b/backend/src/config/runtime.ts @@ -48,9 +48,12 @@ export function simulationAllowed(env: RuntimeEnvironment = process.env): boolea return env.NODE_ENV !== 'production' && env.ALLOW_SIMULATE === 'true'; } +import { pool } from './database'; +import { checkMigrationStatus } from '../db/migration-runner'; + export function deploymentReadiness( env: RuntimeEnvironment = process.env, - options: { storage?: string } = {} + options: { storage?: string; storageReady?: boolean } = {} ): ReadinessCheck { const network = (env.STELLAR_NETWORK || 'TESTNET').toUpperCase(); const horizonUrl = env.STELLAR_HORIZON_URL || @@ -59,12 +62,24 @@ export function deploymentReadiness( : 'https://horizon.stellar.org'); const isHttps = /^https:\/\//i.test(horizonUrl); const isDevOrTestLocal = env.NODE_ENV !== 'production' && /^http:\/\/(localhost|127\.0\.0\.1)(:\d+)?/i.test(horizonUrl); + + const isPostgresSync = Boolean( + (options.storage && /postgres/i.test(options.storage)) || + (!options.storage && env.INVOICE_STORAGE === 'postgres') + ); + + const storageReadySync = options.storageReady !== undefined + ? options.storageReady + : isPostgresSync + ? Boolean(env.DATABASE_URL) + : true; + const checks: ReadinessCheck['checks'] = { frontendOrigins: configuredFrontendOrigins(env).length > 0, simulationDisabled: env.ALLOW_SIMULATE !== 'true', stellarNetwork: network === 'TESTNET' || network === 'PUBLIC', horizonUrl: isHttps || isDevOrTestLocal, - storageReady: true, // In-memory MVP is immediately ready without Postgres + storageReady: storageReadySync, }; const reasons: string[] = []; const missing: string[] = []; @@ -85,6 +100,10 @@ export function deploymentReadiness( reasons.push('STELLAR_HORIZON_URL must use HTTPS'); missing.push('STELLAR_HORIZON_URL'); } + if (!checks.storageReady && isPostgresSync && !env.DATABASE_URL) { + reasons.push('DATABASE_URL is required for PostgreSQL storage'); + missing.push('DATABASE_URL'); + } return { ready: Object.values(checks).every(Boolean), checks, reasons, missing }; } @@ -145,6 +164,8 @@ export interface DeploymentReadinessOptions { storage?: string; pingHorizon?: boolean; timeoutMs?: number; + pool?: any; + migrationsDir?: string; } export async function deploymentReadinessAsync( @@ -173,6 +194,50 @@ export async function deploymentReadinessAsync( } } + const isExplicitMemory = + options.storage === 'memory' || + options.storage === 'in-memory' || + env.INVOICE_STORAGE === 'memory'; + + const isPostgres = Boolean( + (options.storage && /postgres/i.test(options.storage)) || + (!options.storage && env.INVOICE_STORAGE === 'postgres') || + (!options.storage && !env.INVOICE_STORAGE && env.DATABASE_URL) + ); + + if (isPostgres && !isExplicitMemory) { + if (!env.DATABASE_URL && !options.pool) { + checks.storageReady = false; + if (!missing.includes('DATABASE_URL')) { + missing.push('DATABASE_URL'); + reasons.push('DATABASE_URL is required for PostgreSQL storage'); + } + } else { + try { + const dbPool = options.pool ?? pool; + const migrationStatus = await checkMigrationStatus(dbPool, { + migrationsDir: options.migrationsDir, + }); + + if (!migrationStatus.ready) { + checks.storageReady = false; + if (migrationStatus.pendingVersions.length > 0) { + reasons.push( + `Database schema is not up to date (pending migrations: ${migrationStatus.pendingVersions.join(', ')})` + ); + } else { + reasons.push('Database schema is not ready'); + } + } else { + checks.storageReady = true; + } + } catch (err: any) { + checks.storageReady = false; + reasons.push(`PostgreSQL connection failed: ${err.message || 'connection error'}`); + } + } + } + const allChecksPassed = Object.values(checks).every(Boolean); return { ready: allChecksPassed, diff --git a/backend/src/db/migrate.ts b/backend/src/db/migrate.ts index ea5915b8..905fae68 100644 --- a/backend/src/db/migrate.ts +++ b/backend/src/db/migrate.ts @@ -1,37 +1,30 @@ -import * as fs from 'fs'; -import * as path from 'path'; import { pool } from '../config/database'; +import { runMigrations, checkMigrationStatus } from './migration-runner'; -// SQL lives at the repo root (db/), the runner lives with the backend code so -// `npm run db:migrate` (tsx src/db/migrate.ts) resolves both the script and its -// dependencies. -// -// schema.sql applied here defines the full parity column set: seller name and -// email, asset_code + asset_issuer pair, customer name + email, all payer -// fields (public_key / name / email / tx_hash / paid_at), cancellation and -// settlement context fields, expires_at, and JSONB metadata. Any column added -// to StoredInvoice in invoice-storage.ts requires a matching ALTER or a re-run -// of this migrate against production; the shared interface + TypeScript will -// surface the mismatch at compile time if the two drift. -const SQL_DIR = path.join(__dirname, '../../../db'); - -async function migrate() { +export async function migrate() { console.log('šŸš€ Starting database migration...\n'); try { - const schemaPath = path.join(SQL_DIR, 'schema.sql'); - const schema = fs.readFileSync(schemaPath, 'utf-8'); + const result = await runMigrations(pool); - await pool.query(schema); + if (result.adopted.length > 0) { + console.log('šŸ“¦ Adopted baseline migrations for existing database:'); + for (const m of result.adopted) { + console.log(` - ${m.version}_${m.name} (adopted)`); + } + } - console.log('āœ… Database migration completed successfully!\n'); - console.log('šŸ“‹ Created tables:'); - console.log(' - invoices (keyed by seller_public_key)'); - console.log(' - transactions'); - console.log(' - payment_events'); - console.log('\nšŸ“Š Created views:'); - console.log(' - invoice_stats\n'); + if (result.applied.length > 0) { + console.log('āœ… Applied migrations:'); + for (const m of result.applied) { + console.log(` - ${m.version}_${m.name}`); + } + } else if (result.adopted.length === 0) { + console.log('✨ No pending migrations. Database schema is already up to date.'); + } + console.log(`\nšŸ“Œ Current schema version: ${result.latestVersion || 'none'}\n`); + return result; } catch (error) { console.error('āŒ Migration failed:', error); throw error; @@ -40,13 +33,15 @@ async function migrate() { } } -// Run migration -migrate() - .then(() => { - console.log('✨ All done!'); - process.exit(0); - }) - .catch((error) => { - console.error('Migration error:', error); - process.exit(1); - }); +// Run CLI when invoked directly +if (require.main === module || process.argv[1]?.endsWith('migrate.ts') || process.argv[1]?.endsWith('migrate.js')) { + migrate() + .then(() => { + console.log('✨ All done!'); + process.exit(0); + }) + .catch((error) => { + console.error('Migration error:', error); + process.exit(1); + }); +} diff --git a/backend/src/db/migration-runner.ts b/backend/src/db/migration-runner.ts new file mode 100644 index 00000000..5574ebf9 --- /dev/null +++ b/backend/src/db/migration-runner.ts @@ -0,0 +1,282 @@ +import * as fs from 'fs'; +import * as path from 'path'; +import * as crypto from 'crypto'; +import type { Pool, PoolClient } from 'pg'; + +export const MIGRATION_LOCK_ID = 742918451; +export const DEFAULT_MIGRATIONS_DIR = path.resolve(__dirname, '../../../db/migrations'); + +export interface MigrationFile { + version: string; + name: string; + filename: string; + filePath: string; + sql: string; + checksum: string; +} + +export interface AppliedMigration { + version: string; + name: string; + checksum: string; + applied_at: Date; +} + +export interface MigrationResult { + applied: Array<{ version: string; name: string }>; + adopted: Array<{ version: string; name: string }>; + latestVersion: string | null; +} + +export interface MigrationStatus { + ready: boolean; + latestAppliedVersion: string | null; + latestBundledVersion: string | null; + pendingVersions: string[]; +} + +export interface MigrationOptions { + migrationsDir?: string; +} + +export function computeChecksum(content: string): string { + return crypto.createHash('sha256').update(content, 'utf8').digest('hex'); +} + +export function loadMigrationFiles(migrationsDir: string = DEFAULT_MIGRATIONS_DIR): MigrationFile[] { + if (!fs.existsSync(migrationsDir)) { + return []; + } + + const entries = fs.readdirSync(migrationsDir, { withFileTypes: true }); + const migrationFiles: MigrationFile[] = []; + + for (const entry of entries) { + if (!entry.isFile() || !entry.name.endsWith('.sql')) { + continue; + } + + const match = /^(\d+)_([a-zA-Z0-9_\-]+)\.sql$/.exec(entry.name); + if (!match) { + continue; + } + + const [, version, name] = match; + const filePath = path.join(migrationsDir, entry.name); + const sql = fs.readFileSync(filePath, 'utf8'); + const checksum = computeChecksum(sql); + + migrationFiles.push({ + version, + name, + filename: entry.name, + filePath, + sql, + checksum, + }); + } + + return migrationFiles.sort((a, b) => + a.version.localeCompare(b.version, undefined, { numeric: true }) + ); +} + +export async function ensureMigrationTable(client: PoolClient | Pool): Promise { + await client.query(` + CREATE TABLE IF NOT EXISTS schema_migrations ( + version VARCHAR(50) PRIMARY KEY, + name VARCHAR(255) NOT NULL, + checksum VARCHAR(64) NOT NULL, + applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + `); +} + +/** + * Execute all pending migrations in forward-only order. + * + * Each migration runs in its own transaction under a global advisory lock. + * If a database built from the legacy schema is detected, baseline migration 0001 + * is adopted without replaying. If any previously applied migration's checksum has + * changed, execution aborts with a checksum error. + */ +export async function runMigrations( + poolOrClient: Pool | PoolClient, + options: MigrationOptions = {} +): Promise { + const migrationsDir = options.migrationsDir || DEFAULT_MIGRATIONS_DIR; + const files = loadMigrationFiles(migrationsDir); + + const isPool = 'connect' in poolOrClient && typeof (poolOrClient as Pool).connect === 'function'; + const client: PoolClient = isPool + ? await (poolOrClient as Pool).connect() + : (poolOrClient as PoolClient); + + try { + // Acquire exclusive advisory lock for migration execution + await client.query('SELECT pg_advisory_lock($1)', [MIGRATION_LOCK_ID]); + + try { + await ensureMigrationTable(client); + + const appliedRes = await client.query(` + SELECT version, name, checksum, applied_at + FROM schema_migrations + ORDER BY version ASC + `); + const appliedMap = new Map( + appliedRes.rows.map((row) => [row.version, row]) + ); + + // Verify checksums of all already applied migrations + for (const file of files) { + const applied = appliedMap.get(file.version); + if (applied && applied.checksum !== file.checksum) { + throw new Error( + `Migration checksum mismatch for version ${file.version} (${file.name}): database has ${applied.checksum}, file has ${file.checksum}` + ); + } + } + + const applied: Array<{ version: string; name: string }> = []; + const adopted: Array<{ version: string; name: string }> = []; + + // Check baseline adoption for 0001 on existing legacy databases + const baseline0001 = files.find((f) => f.version === '0001'); + if (baseline0001 && !appliedMap.has('0001')) { + const checkInvoicesTable = await client.query<{ exists: boolean }>(` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = current_schema() AND table_name = 'invoices' + ) AS exists + `); + + if (checkInvoicesTable.rows[0]?.exists) { + await client.query( + `INSERT INTO schema_migrations (version, name, checksum, applied_at) VALUES ($1, $2, $3, NOW())`, + [baseline0001.version, baseline0001.name, baseline0001.checksum] + ); + adopted.push({ version: baseline0001.version, name: baseline0001.name }); + appliedMap.set('0001', { + version: baseline0001.version, + name: baseline0001.name, + checksum: baseline0001.checksum, + applied_at: new Date(), + }); + } + } + + // Apply remaining pending migrations in sequence + for (const file of files) { + if (appliedMap.has(file.version)) { + continue; + } + + await client.query('BEGIN'); + try { + await client.query(file.sql); + await client.query( + `INSERT INTO schema_migrations (version, name, checksum, applied_at) VALUES ($1, $2, $3, NOW())`, + [file.version, file.name, file.checksum] + ); + await client.query('COMMIT'); + applied.push({ version: file.version, name: file.name }); + appliedMap.set(file.version, { + version: file.version, + name: file.name, + checksum: file.checksum, + applied_at: new Date(), + }); + } catch (error) { + await client.query('ROLLBACK'); + throw error; + } + } + + const latestVersion = + files.length > 0 ? files[files.length - 1].version : null; + + return { applied, adopted, latestVersion }; + } finally { + await client.query('SELECT pg_advisory_unlock($1)', [MIGRATION_LOCK_ID]); + } + } finally { + if (isPool) { + client.release(); + } + } +} + +/** + * Check if the database has applied all bundled migrations. + */ +export async function checkMigrationStatus( + poolOrClient: Pool | PoolClient, + options: MigrationOptions = {} +): Promise { + const migrationsDir = options.migrationsDir || DEFAULT_MIGRATIONS_DIR; + const files = loadMigrationFiles(migrationsDir); + + if (files.length === 0) { + return { + ready: true, + latestAppliedVersion: null, + latestBundledVersion: null, + pendingVersions: [], + }; + } + + const latestBundledVersion = files[files.length - 1].version; + + try { + const tableCheck = await poolOrClient.query<{ exists: boolean }>(` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = current_schema() AND table_name = 'schema_migrations' + ) AS exists + `); + + if (!tableCheck.rows[0]?.exists) { + return { + ready: false, + latestAppliedVersion: null, + latestBundledVersion, + pendingVersions: files.map((f) => f.version), + }; + } + + const appliedRes = await poolOrClient.query(` + SELECT version, name, checksum, applied_at + FROM schema_migrations + ORDER BY version ASC + `); + + const appliedSet = new Set(appliedRes.rows.map((r) => r.version)); + const pendingVersions = files + .map((f) => f.version) + .filter((v) => !appliedSet.has(v)); + + const latestAppliedVersion = + appliedRes.rows.length > 0 + ? appliedRes.rows[appliedRes.rows.length - 1].version + : null; + + const ready = + pendingVersions.length === 0 && + latestAppliedVersion === latestBundledVersion; + + return { + ready, + latestAppliedVersion, + latestBundledVersion, + pendingVersions, + }; + } catch { + return { + ready: false, + latestAppliedVersion: null, + latestBundledVersion, + pendingVersions: files.map((f) => f.version), + }; + } +} diff --git a/backend/src/db/seed.ts b/backend/src/db/seed.ts index f06991c8..4faebdb3 100644 --- a/backend/src/db/seed.ts +++ b/backend/src/db/seed.ts @@ -1,6 +1,7 @@ import * as fs from 'fs'; import * as path from 'path'; import { pool } from '../config/database'; +import { checkMigrationStatus, runMigrations } from './migration-runner'; // SQL lives at the repo root (db/), the runner lives with the backend code so // `npm run db:seed` (tsx src/db/seed.ts) resolves both the script and its @@ -20,6 +21,13 @@ async function seed() { console.log('🌱 Seeding sample invoices...\n'); try { + // Ensure migrations are up to date before seeding + const status = await checkMigrationStatus(pool); + if (!status.ready) { + console.log('⚔ Pending migrations detected before seed, applying migrations...'); + await runMigrations(pool); + } + const seedPath = path.join(SQL_DIR, 'seed.sql'); const seedSql = fs.readFileSync(seedPath, 'utf-8'); diff --git a/backend/tests/fixtures/legacy-schema.sql b/backend/tests/fixtures/legacy-schema.sql new file mode 100644 index 00000000..bf44e527 --- /dev/null +++ b/backend/tests/fixtures/legacy-schema.sql @@ -0,0 +1,194 @@ +-- Quittance Database Schema +-- PostgreSQL Database +-- +-- Identity is the connected Freighter wallet: every invoice is keyed by +-- seller_public_key. There is no user/email login table. +-- +-- Full parity column set (kept in sync with backend/types StoredInvoice via +-- db/migrate.ts + InvoiceService INSERT/SELECT column lists): +-- seller_name, seller_email +-- amount, asset_code, asset_issuer (credit assets always require both) +-- memo, description +-- customer_name, customer_email +-- status, payment_tx_hash +-- payer_public_key, payer_name, payer_email, paid_at, cancelled_at +-- settled_at, settlement_context, prior_status, late_payment_warning_code +-- created_at, expires_at (expires_at NOT NULL, default 7d) +-- metadata (JSONB) + +-- Invoices Table +CREATE TABLE IF NOT EXISTS invoices ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + seller_public_key VARCHAR(56) NOT NULL, + seller_name VARCHAR(255), + seller_email VARCHAR(255), + amount DECIMAL(20, 7) NOT NULL, + asset_code VARCHAR(12) DEFAULT 'XLM', + asset_issuer VARCHAR(56), + memo TEXT UNIQUE NOT NULL, + description TEXT, + customer_name VARCHAR(255), + customer_email VARCHAR(255), + status VARCHAR(20) DEFAULT 'PENDING' CHECK (status IN ('PENDING', 'PAID', 'EXPIRED', 'CANCELLED')), + payment_tx_hash VARCHAR(64), + payer_public_key VARCHAR(56), + payer_name VARCHAR(255), + payer_email VARCHAR(255), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + paid_at TIMESTAMPTZ, + cancelled_at TIMESTAMPTZ, + settled_at TIMESTAMPTZ, + settlement_context VARCHAR(20) CHECK (settlement_context IN ('ON_TIME', 'AFTER_EXPIRY', 'AFTER_CANCEL')), + prior_status VARCHAR(20) CHECK (prior_status IN ('PENDING', 'PAID', 'EXPIRED', 'CANCELLED')), + late_payment_warning_code VARCHAR(50) CHECK (late_payment_warning_code IN ('PAYMENT_RECEIVED_AFTER_EXPIRY', 'PAYMENT_RECEIVED_AFTER_CANCEL')), + expires_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + INTERVAL '7 days', + metadata JSONB +); + +-- One transaction settles at most one invoice (issue #501). The in-process +-- claim index covers the memory MVP; this partial unique index is the durable +-- form on Postgres, so a racing verify/monitor claim fails with 23505 instead +-- of double-settling. +CREATE UNIQUE INDEX IF NOT EXISTS uq_invoices_payment_tx_hash + ON invoices (payment_tx_hash) + WHERE payment_tx_hash IS NOT NULL; + +-- Transactions Table +CREATE TABLE IF NOT EXISTS transactions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + invoice_id UUID REFERENCES invoices(id) ON DELETE SET NULL, + from_address VARCHAR(56) NOT NULL, + to_address VARCHAR(56) NOT NULL, + amount DECIMAL(20, 7) NOT NULL, + asset_code VARCHAR(12) DEFAULT 'XLM', + asset_issuer VARCHAR(56), + tx_hash VARCHAR(64) UNIQUE NOT NULL, + memo TEXT, + ledger BIGINT, + processed_at TIMESTAMP DEFAULT NOW(), + metadata JSONB +); + +-- Payment Events Log +CREATE TABLE IF NOT EXISTS payment_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + invoice_id UUID REFERENCES invoices(id), + event_type VARCHAR(50) NOT NULL, + event_data JSONB, + created_at TIMESTAMP DEFAULT NOW() +); + +-- The Horizon paging token is committed after each operation is handled. A +-- restart therefore resumes after the last completed operation; replay after +-- a crash is safe because invoice settlement and tx_hash are idempotent. +CREATE TABLE IF NOT EXISTS payment_monitor_checkpoints ( + account VARCHAR(56) NOT NULL, + network VARCHAR(20) NOT NULL, + cursor TEXT NOT NULL, + ledger BIGINT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (account, network) +); + +-- Wallet alignment: databases created before wallet-scoped sellers still have +-- the unused users table and invoices.user_id column. Both are dropped here so +-- re-running the migration converges on the wallet-only schema. +ALTER TABLE invoices DROP COLUMN IF EXISTS user_id; +DROP TABLE IF EXISTS users CASCADE; + +-- Converge databases created before expiry became an enforced lifecycle. +UPDATE invoices +SET expires_at = COALESCE(created_at, NOW()) + INTERVAL '7 days' +WHERE expires_at IS NULL; +ALTER TABLE invoices ALTER COLUMN expires_at SET DEFAULT NOW() + INTERVAL '7 days'; +ALTER TABLE invoices ALTER COLUMN expires_at SET NOT NULL; + +-- Converge databases created before deterministic cancel/payment settlement. +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS cancelled_at TIMESTAMPTZ; +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS settled_at TIMESTAMPTZ; +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS settlement_context VARCHAR(20); +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS prior_status VARCHAR(20); +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS late_payment_warning_code VARCHAR(50); + +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_settlement_context_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_settlement_context_check + CHECK (settlement_context IS NULL OR settlement_context IN ('ON_TIME', 'AFTER_EXPIRY', 'AFTER_CANCEL')); + +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_prior_status_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_prior_status_check + CHECK (prior_status IS NULL OR prior_status IN ('PENDING', 'PAID', 'EXPIRED', 'CANCELLED')); + +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_late_payment_warning_code_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_late_payment_warning_code_check + CHECK ( + late_payment_warning_code IS NULL OR + late_payment_warning_code IN ('PAYMENT_RECEIVED_AFTER_EXPIRY', 'PAYMENT_RECEIVED_AFTER_CANCEL') + ); + +-- Unique constraint: one transaction hash may settle at most one invoice. +-- Added idempotently so re-running the migration on an upgraded database is safe. +-- The constraint is partial (WHERE payment_tx_hash IS NOT NULL) so unpaid rows +-- do not consume unique index space and NULL values never collide. +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'invoices_payment_tx_hash_unique' + AND conrelid = 'invoices'::regclass + ) THEN + -- Check for any existing duplicates before adding the constraint. + -- A duplicate means a bug in prior code; surface it rather than silently skip. + IF ( + SELECT COUNT(*) FROM ( + SELECT payment_tx_hash + FROM invoices + WHERE payment_tx_hash IS NOT NULL + GROUP BY payment_tx_hash + HAVING COUNT(*) > 1 + ) dupes + ) > 0 THEN + RAISE EXCEPTION + 'Cannot add payment_tx_hash uniqueness: duplicate hashes exist in invoices table. ' + 'Resolve conflicts before re-running the migration.'; + END IF; + + CREATE UNIQUE INDEX invoices_payment_tx_hash_unique + ON invoices(payment_tx_hash) + WHERE payment_tx_hash IS NOT NULL; + END IF; +END +$$; + +-- Indexes for performance +CREATE INDEX IF NOT EXISTS idx_invoices_seller ON invoices(seller_public_key); +CREATE INDEX IF NOT EXISTS idx_invoices_status ON invoices(status); +CREATE INDEX IF NOT EXISTS idx_invoices_asset_code ON invoices(asset_code); +CREATE INDEX IF NOT EXISTS idx_invoices_memo ON invoices(memo); +CREATE INDEX IF NOT EXISTS idx_invoices_created_at ON invoices(created_at DESC); +CREATE INDEX IF NOT EXISTS idx_invoices_seller_created_at ON invoices(seller_public_key, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_invoices_pending_expiry ON invoices(expires_at) WHERE status = 'PENDING'; +CREATE INDEX IF NOT EXISTS idx_transactions_tx_hash ON transactions(tx_hash); +CREATE INDEX IF NOT EXISTS idx_transactions_invoice_id ON transactions(invoice_id); + +-- Sample view for invoice statistics +CREATE OR REPLACE VIEW invoice_stats AS +SELECT + seller_public_key, + COUNT(*) as total_invoices, + SUM(CASE WHEN status = 'PAID' THEN 1 ELSE 0 END) as paid_invoices, + SUM(CASE WHEN status = 'PENDING' THEN 1 ELSE 0 END) as pending_invoices, + SUM(CASE WHEN status = 'PENDING' THEN 1 ELSE 0 END) as actionable_invoices, + SUM(CASE WHEN status = 'EXPIRED' THEN 1 ELSE 0 END) as expired_invoices, + SUM(CASE WHEN status = 'PAID' THEN amount ELSE 0 END) as total_revenue, + asset_code +FROM invoices +GROUP BY seller_public_key, asset_code; + +-- Issue #514: replayed creates collapse onto the original row instead of +-- minting a second memo + pay link. The partial unique index keeps NULL +-- (legacy/keyless) rows untouched while giving the service an atomic +-- ON CONFLICT arbiter for (seller, key) races. +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS idempotency_key VARCHAR(255); +CREATE UNIQUE INDEX IF NOT EXISTS idx_invoices_seller_idempotency + ON invoices (seller_public_key, idempotency_key) + WHERE idempotency_key IS NOT NULL; diff --git a/backend/tests/health-readiness.test.ts b/backend/tests/health-readiness.test.ts index 7f91bfc8..68243e69 100644 --- a/backend/tests/health-readiness.test.ts +++ b/backend/tests/health-readiness.test.ts @@ -321,5 +321,31 @@ describe('Liveness and Readiness Probes (#437)', () => { await new Promise((resolve) => server.close(() => resolve())); } }); + + it('returns HTTP 503 in postgres mode when DATABASE_URL is missing or migration pending', async () => { + const app = express(); + process.env.NODE_ENV = 'production'; + process.env.FRONTEND_URL = 'https://quittance.example.com'; + process.env.STELLAR_NETWORK = 'TESTNET'; + process.env.STELLAR_HORIZON_URL = 'https://horizon-testnet.stellar.org'; + process.env.ALLOW_SIMULATE = 'false'; + delete process.env.DATABASE_URL; + + app.get('/api/ready', readinessHandler('postgres')); + + const server = app.listen(0, '127.0.0.1'); + await new Promise((resolve) => server.once('listening', resolve)); + const port = (server.address() as AddressInfo).port; + + try { + const res = await fetch(`http://127.0.0.1:${port}/api/ready`); + assert.equal(res.status, 503); + const body = (await res.json()) as any; + assert.equal(body.ready, false); + assert.equal(body.checks.storageReady, false); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } + }); }); }); diff --git a/backend/tests/invoice-postgres.integration.test.ts b/backend/tests/invoice-postgres.integration.test.ts index 39f1caa0..d9af63ce 100644 --- a/backend/tests/invoice-postgres.integration.test.ts +++ b/backend/tests/invoice-postgres.integration.test.ts @@ -5,6 +5,8 @@ import * as path from 'node:path'; import { Pool } from 'pg'; import { Keypair } from '@stellar/stellar-sdk'; import { InvoiceService } from '../src/services/invoice.service.ts'; +import { PaymentClaimError } from '../src/domain/payment-attribution.ts'; +import { runMigrations } from '../src/db/migration-runner.ts'; import type { CreateInvoiceInput } from '../src/utils/validation.ts'; import { PostgresInvoiceStorage } from '../src/storage/postgres-invoice-storage.ts'; import { MemoryInvoiceStorage } from '../src/storage/memory-invoice-storage.ts'; @@ -13,8 +15,6 @@ import { MemoryStorage } from '../src/storage/memory-storage.ts'; import type { PayerInfo, StoredInvoice } from '../src/storage/invoice-storage.ts'; const DATABASE_URL = process.env.DATABASE_URL; -const SCHEMA_PATH = path.join(__dirname, '../../db/schema.sql'); -const SEED_PATH = path.join(__dirname, '../../db/seed.sql'); const SELLER_A = Keypair.random().publicKey(); const SELLER_B = Keypair.random().publicKey(); @@ -35,7 +35,7 @@ describe('Invoice persistence on Postgres', { skip: DATABASE_URL ? false : 'DATA before(async () => { adminPool = new Pool({ connectionString: DATABASE_URL }); - await adminPool.query(fs.readFileSync(SCHEMA_PATH, 'utf-8')); + await runMigrations(adminPool); }); after(async () => { @@ -227,6 +227,32 @@ describe('Invoice persistence on Postgres', { skip: DATABASE_URL ? false : 'DATA } }); + it('settling one tx hash on two invoices returns the typed claim error on a migrated database', async () => { + const pool = new Pool({ connectionString: DATABASE_URL }); + const service = new InvoiceService(pool); + + try { + const sharedTxHash = 'e'.repeat(64); + const inv1 = await service.createInvoice(createInput(SELLER_A, { amount: 10 })); + const inv2 = await service.createInvoice(createInput(SELLER_A, { amount: 20 })); + + await service.markAsPaid(inv1.id, sharedTxHash, PAYER, undefined, { settledAt: new Date() }); + + await assert.rejects( + () => service.markAsPaid(inv2.id, sharedTxHash, PAYER, undefined, { settledAt: new Date() }), + (err: any) => { + assert.ok(err instanceof PaymentClaimError, `Expected PaymentClaimError, got: ${err}`); + assert.equal(err.name, 'PaymentClaimError'); + assert.equal(err.txHash, sharedTxHash); + assert.equal(err.claimedByInvoiceId, inv1.id); + return true; + } + ); + } finally { + await pool.end(); + } + }); + it('cancels a PENDING invoice exactly once, then rejects second cancel', async () => { const pool = new Pool({ connectionString: DATABASE_URL }); const service = new InvoiceService(pool); diff --git a/backend/tests/migration-runner.test.ts b/backend/tests/migration-runner.test.ts new file mode 100644 index 00000000..8a17c646 --- /dev/null +++ b/backend/tests/migration-runner.test.ts @@ -0,0 +1,340 @@ +import assert from 'node:assert/strict'; +import { describe, it, beforeEach, afterEach } from 'node:test'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import express from 'express'; +import type { AddressInfo } from 'node:net'; +import { + runMigrations, + checkMigrationStatus, + loadMigrationFiles, + computeChecksum, + ensureMigrationTable, + MIGRATION_LOCK_ID, + type AppliedMigration, +} from '../src/db/migration-runner.ts'; +import { readinessHandler } from '../src/health.ts'; +import { deploymentReadinessAsync } from '../src/config/runtime.ts'; + +interface MockRow { + version: string; + name: string; + checksum: string; + applied_at: Date; +} + +function createMockPostgresDb(initialHasInvoices = false) { + const migrations: MockRow[] = []; + let hasInvoices = initialHasInvoices; + let hasSchemaMigrations = false; + let lockHeld = false; + const lockWaiters: Array<() => void> = []; + const executedSqls: string[] = []; + const invoicesData: any[] = []; + + const pool = { + connect: async () => client, + query: async (sql: string, params: any[] = []) => client.query(sql, params), + }; + + const client = { + query: async (text: string, params: any[] = []) => { + const sql = text.replace(/\s+/g, ' ').trim(); + + if (sql.includes('pg_advisory_lock')) { + if (lockHeld) { + await new Promise((resolve) => lockWaiters.push(resolve)); + } + lockHeld = true; + return { rows: [{ pg_advisory_lock: null }], rowCount: 1 }; + } + + if (sql.includes('pg_advisory_unlock')) { + lockHeld = false; + const next = lockWaiters.shift(); + if (next) { + next(); + } + return { rows: [{ pg_advisory_unlock: true }], rowCount: 1 }; + } + + if (sql.includes('CREATE TABLE IF NOT EXISTS schema_migrations')) { + hasSchemaMigrations = true; + return { rows: [], rowCount: 0 }; + } + + if (sql.includes("information_schema.tables") && sql.includes("'schema_migrations'")) { + return { rows: [{ exists: hasSchemaMigrations }], rowCount: 1 }; + } + + if (sql.includes("information_schema.tables") && sql.includes("'invoices'")) { + return { rows: [{ exists: hasInvoices }], rowCount: 1 }; + } + + if (sql.includes('SELECT version, name, checksum, applied_at FROM schema_migrations')) { + if (!hasSchemaMigrations) { + throw new Error('relation "schema_migrations" does not exist'); + } + return { rows: [...migrations], rowCount: migrations.length }; + } + + if (sql.startsWith('INSERT INTO schema_migrations')) { + const [version, name, checksum] = params; + if (migrations.some((m) => m.version === version)) { + throw new Error(`duplicate key value violates unique constraint "schema_migrations_pkey"`); + } + migrations.push({ + version: String(version), + name: String(name), + checksum: String(checksum), + applied_at: new Date(), + }); + return { rows: [], rowCount: 1 }; + } + + if (sql === 'BEGIN' || sql === 'COMMIT' || sql === 'ROLLBACK') { + return { rows: [], rowCount: 0 }; + } + + // Track general migration execution + executedSqls.push(sql); + if (sql.includes('CREATE TABLE IF NOT EXISTS invoices') || sql.includes('CREATE TABLE invoices')) { + hasInvoices = true; + } + + return { rows: [], rowCount: 0 }; + }, + release: () => {}, + }; + + return { + pool, + client, + migrations, + getHasInvoices: () => hasInvoices, + setHasInvoices: (val: boolean) => { + hasInvoices = val; + }, + invoicesData, + executedSqls, + isLockHeld: () => lockHeld, + }; +} + +describe('Database Migration Runner (#583)', () => { + let tempMigrationsDir: string; + + beforeEach(() => { + tempMigrationsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'quittance-migrations-test-')); + }); + + afterEach(() => { + if (fs.existsSync(tempMigrationsDir)) { + fs.rmSync(tempMigrationsDir, { recursive: true, force: true }); + } + }); + + it('runs migrations on an empty database and the second run applies nothing', async () => { + // Setup two migration files + const m1Path = path.join(tempMigrationsDir, '0001_baseline.sql'); + const m2Path = path.join(tempMigrationsDir, '0002_add_checks.sql'); + fs.writeFileSync(m1Path, 'CREATE TABLE IF NOT EXISTS invoices (id uuid primary key);'); + fs.writeFileSync(m2Path, 'ALTER TABLE invoices ADD COLUMN IF NOT EXISTS status text;'); + + const db = createMockPostgresDb(false); + + // Run 1: Should apply both 0001 and 0002 + const run1 = await runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }); + assert.equal(run1.applied.length, 2); + assert.equal(run1.applied[0].version, '0001'); + assert.equal(run1.applied[1].version, '0002'); + assert.equal(run1.adopted.length, 0); + assert.equal(run1.latestVersion, '0002'); + assert.equal(db.migrations.length, 2); + + // Run 2: Should apply nothing (idempotency check) + const run2 = await runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }); + assert.equal(run2.applied.length, 0); + assert.equal(run2.adopted.length, 0); + assert.equal(run2.latestVersion, '0002'); + assert.equal(db.migrations.length, 2); + }); + + it('adopts baseline 0001 on an existing database built from legacy schema without replaying it', async () => { + // Setup two migration files + const m1Path = path.join(tempMigrationsDir, '0001_baseline.sql'); + const m2Path = path.join(tempMigrationsDir, '0002_align_constraints.sql'); + fs.writeFileSync(m1Path, 'CREATE TABLE invoices (id uuid primary key);'); + fs.writeFileSync(m2Path, 'ALTER TABLE invoices ADD COLUMN status text;'); + + // Existing database already has invoices table, but no schema_migrations + const db = createMockPostgresDb(true); + + const result = await runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }); + + // Baseline 0001 should be adopted, 0002 should be applied + assert.equal(result.adopted.length, 1); + assert.equal(result.adopted[0].version, '0001'); + assert.equal(result.applied.length, 1); + assert.equal(result.applied[0].version, '0002'); + + // 0001 SQL should NOT have been executed + assert.equal(db.executedSqls.some((sql) => sql.includes('CREATE TABLE invoices')), false); + // 0002 SQL should have been executed + assert.equal(db.executedSqls.some((sql) => sql.includes('ALTER TABLE invoices ADD COLUMN status text')), true); + // schema_migrations should contain both + assert.equal(db.migrations.length, 2); + assert.equal(db.migrations[0].version, '0001'); + assert.equal(db.migrations[1].version, '0002'); + }); + + it('two concurrent runs under advisory lock never apply the same version twice', async () => { + const m1Path = path.join(tempMigrationsDir, '0001_baseline.sql'); + fs.writeFileSync(m1Path, 'CREATE TABLE IF NOT EXISTS invoices (id uuid primary key);'); + + const db = createMockPostgresDb(false); + + // Run two migration runners concurrently + const [res1, res2] = await Promise.all([ + runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }), + runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }), + ]); + + const totalApplied = res1.applied.length + res2.applied.length; + assert.equal(totalApplied, 1, 'Migration 0001 should only be applied once across concurrent runs'); + assert.equal(db.migrations.length, 1); + }); + + it('fails with a checksum error if an already-applied migration file is modified', async () => { + const m1Path = path.join(tempMigrationsDir, '0001_baseline.sql'); + fs.writeFileSync(m1Path, 'CREATE TABLE invoices (id uuid);'); + + const db = createMockPostgresDb(false); + await runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }); + + // Now tamper with the migration file + fs.writeFileSync(m1Path, 'CREATE TABLE invoices (id uuid, tampered text);'); + + await assert.rejects( + () => runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }), + (err: any) => { + assert.match(err.message, /Migration checksum mismatch for version 0001/); + return true; + } + ); + }); + + it('checkMigrationStatus correctly reports readiness', async () => { + const m1Path = path.join(tempMigrationsDir, '0001_baseline.sql'); + const m2Path = path.join(tempMigrationsDir, '0002_more.sql'); + fs.writeFileSync(m1Path, 'SELECT 1;'); + fs.writeFileSync(m2Path, 'SELECT 2;'); + + const db = createMockPostgresDb(false); + + // Initial state: not ready, both migrations pending + const statusBefore = await checkMigrationStatus(db.pool as any, { migrationsDir: tempMigrationsDir }); + assert.equal(statusBefore.ready, false); + assert.deepEqual(statusBefore.pendingVersions, ['0001', '0002']); + + // Run migrations + await runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }); + + // After migration: ready, 0 pending + const statusAfter = await checkMigrationStatus(db.pool as any, { migrationsDir: tempMigrationsDir }); + assert.equal(statusAfter.ready, true); + assert.equal(statusAfter.latestAppliedVersion, '0002'); + assert.equal(statusAfter.latestBundledVersion, '0002'); + assert.deepEqual(statusAfter.pendingVersions, []); + }); + + it('readiness check fails (HTTP 503) when database migrations are pending in postgres mode', async () => { + const m1Path = path.join(tempMigrationsDir, '0001_baseline.sql'); + fs.writeFileSync(m1Path, 'SELECT 1;'); + + const db = createMockPostgresDb(false); + + // Test deploymentReadinessAsync directly with unmigrated database + const readinessUnmigrated = await deploymentReadinessAsync( + { + NODE_ENV: 'production', + FRONTEND_URL: 'https://quittance.example.com', + STELLAR_NETWORK: 'TESTNET', + STELLAR_HORIZON_URL: 'https://horizon-testnet.stellar.org', + ALLOW_SIMULATE: 'false', + DATABASE_URL: 'postgresql://test:test@localhost:5432/test', + INVOICE_STORAGE: 'postgres', + }, + { + storage: 'postgres', + pool: db.pool, + migrationsDir: tempMigrationsDir, + } + ); + + assert.equal(readinessUnmigrated.ready, false); + assert.equal(readinessUnmigrated.checks.storageReady, false); + assert.ok(readinessUnmigrated.reasons.some((r) => r.includes('Database schema is not up to date'))); + + // Test through HTTP express probe + const app = express(); + app.get('/api/ready', async (req, res) => { + const result = await deploymentReadinessAsync( + { + NODE_ENV: 'production', + FRONTEND_URL: 'https://quittance.example.com', + STELLAR_NETWORK: 'TESTNET', + STELLAR_HORIZON_URL: 'https://horizon-testnet.stellar.org', + ALLOW_SIMULATE: 'false', + DATABASE_URL: 'postgresql://test:test@localhost:5432/test', + INVOICE_STORAGE: 'postgres', + }, + { + storage: 'postgres', + pool: db.pool, + migrationsDir: tempMigrationsDir, + } + ); + res.status(result.ready ? 200 : 503).json(result); + }); + + const server = app.listen(0, '127.0.0.1'); + await new Promise((resolve) => server.once('listening', resolve)); + const port = (server.address() as AddressInfo).port; + + try { + const res = await fetch(`http://127.0.0.1:${port}/api/ready`); + assert.equal(res.status, 503); + const json = (await res.json()) as any; + assert.equal(json.ready, false); + assert.equal(json.checks.storageReady, false); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } + + // Now migrate the database + await runMigrations(db.pool as any, { migrationsDir: tempMigrationsDir }); + + // Readiness check should now succeed (HTTP 200) + const readinessMigrated = await deploymentReadinessAsync( + { + NODE_ENV: 'production', + FRONTEND_URL: 'https://quittance.example.com', + STELLAR_NETWORK: 'TESTNET', + STELLAR_HORIZON_URL: 'https://horizon-testnet.stellar.org', + ALLOW_SIMULATE: 'false', + DATABASE_URL: 'postgresql://test:test@localhost:5432/test', + INVOICE_STORAGE: 'postgres', + }, + { + storage: 'postgres', + pool: db.pool, + migrationsDir: tempMigrationsDir, + } + ); + + assert.equal(readinessMigrated.ready, true); + assert.equal(readinessMigrated.checks.storageReady, true); + }); +}); diff --git a/db/migrations/0001_baseline.sql b/db/migrations/0001_baseline.sql new file mode 100644 index 00000000..bf44e527 --- /dev/null +++ b/db/migrations/0001_baseline.sql @@ -0,0 +1,194 @@ +-- Quittance Database Schema +-- PostgreSQL Database +-- +-- Identity is the connected Freighter wallet: every invoice is keyed by +-- seller_public_key. There is no user/email login table. +-- +-- Full parity column set (kept in sync with backend/types StoredInvoice via +-- db/migrate.ts + InvoiceService INSERT/SELECT column lists): +-- seller_name, seller_email +-- amount, asset_code, asset_issuer (credit assets always require both) +-- memo, description +-- customer_name, customer_email +-- status, payment_tx_hash +-- payer_public_key, payer_name, payer_email, paid_at, cancelled_at +-- settled_at, settlement_context, prior_status, late_payment_warning_code +-- created_at, expires_at (expires_at NOT NULL, default 7d) +-- metadata (JSONB) + +-- Invoices Table +CREATE TABLE IF NOT EXISTS invoices ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + seller_public_key VARCHAR(56) NOT NULL, + seller_name VARCHAR(255), + seller_email VARCHAR(255), + amount DECIMAL(20, 7) NOT NULL, + asset_code VARCHAR(12) DEFAULT 'XLM', + asset_issuer VARCHAR(56), + memo TEXT UNIQUE NOT NULL, + description TEXT, + customer_name VARCHAR(255), + customer_email VARCHAR(255), + status VARCHAR(20) DEFAULT 'PENDING' CHECK (status IN ('PENDING', 'PAID', 'EXPIRED', 'CANCELLED')), + payment_tx_hash VARCHAR(64), + payer_public_key VARCHAR(56), + payer_name VARCHAR(255), + payer_email VARCHAR(255), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + paid_at TIMESTAMPTZ, + cancelled_at TIMESTAMPTZ, + settled_at TIMESTAMPTZ, + settlement_context VARCHAR(20) CHECK (settlement_context IN ('ON_TIME', 'AFTER_EXPIRY', 'AFTER_CANCEL')), + prior_status VARCHAR(20) CHECK (prior_status IN ('PENDING', 'PAID', 'EXPIRED', 'CANCELLED')), + late_payment_warning_code VARCHAR(50) CHECK (late_payment_warning_code IN ('PAYMENT_RECEIVED_AFTER_EXPIRY', 'PAYMENT_RECEIVED_AFTER_CANCEL')), + expires_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + INTERVAL '7 days', + metadata JSONB +); + +-- One transaction settles at most one invoice (issue #501). The in-process +-- claim index covers the memory MVP; this partial unique index is the durable +-- form on Postgres, so a racing verify/monitor claim fails with 23505 instead +-- of double-settling. +CREATE UNIQUE INDEX IF NOT EXISTS uq_invoices_payment_tx_hash + ON invoices (payment_tx_hash) + WHERE payment_tx_hash IS NOT NULL; + +-- Transactions Table +CREATE TABLE IF NOT EXISTS transactions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + invoice_id UUID REFERENCES invoices(id) ON DELETE SET NULL, + from_address VARCHAR(56) NOT NULL, + to_address VARCHAR(56) NOT NULL, + amount DECIMAL(20, 7) NOT NULL, + asset_code VARCHAR(12) DEFAULT 'XLM', + asset_issuer VARCHAR(56), + tx_hash VARCHAR(64) UNIQUE NOT NULL, + memo TEXT, + ledger BIGINT, + processed_at TIMESTAMP DEFAULT NOW(), + metadata JSONB +); + +-- Payment Events Log +CREATE TABLE IF NOT EXISTS payment_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + invoice_id UUID REFERENCES invoices(id), + event_type VARCHAR(50) NOT NULL, + event_data JSONB, + created_at TIMESTAMP DEFAULT NOW() +); + +-- The Horizon paging token is committed after each operation is handled. A +-- restart therefore resumes after the last completed operation; replay after +-- a crash is safe because invoice settlement and tx_hash are idempotent. +CREATE TABLE IF NOT EXISTS payment_monitor_checkpoints ( + account VARCHAR(56) NOT NULL, + network VARCHAR(20) NOT NULL, + cursor TEXT NOT NULL, + ledger BIGINT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (account, network) +); + +-- Wallet alignment: databases created before wallet-scoped sellers still have +-- the unused users table and invoices.user_id column. Both are dropped here so +-- re-running the migration converges on the wallet-only schema. +ALTER TABLE invoices DROP COLUMN IF EXISTS user_id; +DROP TABLE IF EXISTS users CASCADE; + +-- Converge databases created before expiry became an enforced lifecycle. +UPDATE invoices +SET expires_at = COALESCE(created_at, NOW()) + INTERVAL '7 days' +WHERE expires_at IS NULL; +ALTER TABLE invoices ALTER COLUMN expires_at SET DEFAULT NOW() + INTERVAL '7 days'; +ALTER TABLE invoices ALTER COLUMN expires_at SET NOT NULL; + +-- Converge databases created before deterministic cancel/payment settlement. +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS cancelled_at TIMESTAMPTZ; +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS settled_at TIMESTAMPTZ; +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS settlement_context VARCHAR(20); +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS prior_status VARCHAR(20); +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS late_payment_warning_code VARCHAR(50); + +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_settlement_context_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_settlement_context_check + CHECK (settlement_context IS NULL OR settlement_context IN ('ON_TIME', 'AFTER_EXPIRY', 'AFTER_CANCEL')); + +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_prior_status_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_prior_status_check + CHECK (prior_status IS NULL OR prior_status IN ('PENDING', 'PAID', 'EXPIRED', 'CANCELLED')); + +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_late_payment_warning_code_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_late_payment_warning_code_check + CHECK ( + late_payment_warning_code IS NULL OR + late_payment_warning_code IN ('PAYMENT_RECEIVED_AFTER_EXPIRY', 'PAYMENT_RECEIVED_AFTER_CANCEL') + ); + +-- Unique constraint: one transaction hash may settle at most one invoice. +-- Added idempotently so re-running the migration on an upgraded database is safe. +-- The constraint is partial (WHERE payment_tx_hash IS NOT NULL) so unpaid rows +-- do not consume unique index space and NULL values never collide. +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'invoices_payment_tx_hash_unique' + AND conrelid = 'invoices'::regclass + ) THEN + -- Check for any existing duplicates before adding the constraint. + -- A duplicate means a bug in prior code; surface it rather than silently skip. + IF ( + SELECT COUNT(*) FROM ( + SELECT payment_tx_hash + FROM invoices + WHERE payment_tx_hash IS NOT NULL + GROUP BY payment_tx_hash + HAVING COUNT(*) > 1 + ) dupes + ) > 0 THEN + RAISE EXCEPTION + 'Cannot add payment_tx_hash uniqueness: duplicate hashes exist in invoices table. ' + 'Resolve conflicts before re-running the migration.'; + END IF; + + CREATE UNIQUE INDEX invoices_payment_tx_hash_unique + ON invoices(payment_tx_hash) + WHERE payment_tx_hash IS NOT NULL; + END IF; +END +$$; + +-- Indexes for performance +CREATE INDEX IF NOT EXISTS idx_invoices_seller ON invoices(seller_public_key); +CREATE INDEX IF NOT EXISTS idx_invoices_status ON invoices(status); +CREATE INDEX IF NOT EXISTS idx_invoices_asset_code ON invoices(asset_code); +CREATE INDEX IF NOT EXISTS idx_invoices_memo ON invoices(memo); +CREATE INDEX IF NOT EXISTS idx_invoices_created_at ON invoices(created_at DESC); +CREATE INDEX IF NOT EXISTS idx_invoices_seller_created_at ON invoices(seller_public_key, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_invoices_pending_expiry ON invoices(expires_at) WHERE status = 'PENDING'; +CREATE INDEX IF NOT EXISTS idx_transactions_tx_hash ON transactions(tx_hash); +CREATE INDEX IF NOT EXISTS idx_transactions_invoice_id ON transactions(invoice_id); + +-- Sample view for invoice statistics +CREATE OR REPLACE VIEW invoice_stats AS +SELECT + seller_public_key, + COUNT(*) as total_invoices, + SUM(CASE WHEN status = 'PAID' THEN 1 ELSE 0 END) as paid_invoices, + SUM(CASE WHEN status = 'PENDING' THEN 1 ELSE 0 END) as pending_invoices, + SUM(CASE WHEN status = 'PENDING' THEN 1 ELSE 0 END) as actionable_invoices, + SUM(CASE WHEN status = 'EXPIRED' THEN 1 ELSE 0 END) as expired_invoices, + SUM(CASE WHEN status = 'PAID' THEN amount ELSE 0 END) as total_revenue, + asset_code +FROM invoices +GROUP BY seller_public_key, asset_code; + +-- Issue #514: replayed creates collapse onto the original row instead of +-- minting a second memo + pay link. The partial unique index keeps NULL +-- (legacy/keyless) rows untouched while giving the service an atomic +-- ON CONFLICT arbiter for (seller, key) races. +ALTER TABLE invoices ADD COLUMN IF NOT EXISTS idempotency_key VARCHAR(255); +CREATE UNIQUE INDEX IF NOT EXISTS idx_invoices_seller_idempotency + ON invoices (seller_public_key, idempotency_key) + WHERE idempotency_key IS NOT NULL; diff --git a/db/migrations/0002_align_constraints_and_timestamps.sql b/db/migrations/0002_align_constraints_and_timestamps.sql new file mode 100644 index 00000000..c63b6bdf --- /dev/null +++ b/db/migrations/0002_align_constraints_and_timestamps.sql @@ -0,0 +1,47 @@ +-- Migration 0002: Align constraints, timestamps, and enforce PAID row invariant +-- +-- 1. Deduplicate payment_tx_hash uniqueness indexes: +-- Keep uq_invoices_payment_tx_hash (referenced by InvoiceService.markAsPaid) +-- and drop invoices_payment_tx_hash_unique. +DROP INDEX IF EXISTS invoices_payment_tx_hash_unique; + +-- Ensure uq_invoices_payment_tx_hash is present if it was somehow omitted +CREATE UNIQUE INDEX IF NOT EXISTS uq_invoices_payment_tx_hash + ON invoices (payment_tx_hash) + WHERE payment_tx_hash IS NOT NULL; + +-- 2. Convert naive timestamps to TIMESTAMPTZ, treating stored values as UTC. +-- Default is updated to NOW() (evaluates to timestamptz). +ALTER TABLE payment_events + ALTER COLUMN created_at TYPE TIMESTAMPTZ USING created_at AT TIME ZONE 'UTC', + ALTER COLUMN created_at SET DEFAULT NOW(); + +ALTER TABLE transactions + ALTER COLUMN processed_at TYPE TIMESTAMPTZ USING processed_at AT TIME ZONE 'UTC', + ALTER COLUMN processed_at SET DEFAULT NOW(); + +-- 3. Pre-check for PAID row invariant from docs/postgres-cutover-draft.sql. +-- Fail loudly if any existing row violates (status = 'PAID' requires payment_tx_hash and paid_at). +DO $$ +DECLARE + violating_count INTEGER; +BEGIN + SELECT COUNT(*) INTO violating_count + FROM invoices + WHERE status = 'PAID' AND (payment_tx_hash IS NULL OR paid_at IS NULL); + + IF violating_count > 0 THEN + RAISE EXCEPTION + 'Cannot add invoices_paid_row_check: % row(s) violate invariant (status = PAID requires payment_tx_hash and paid_at)', + violating_count; + END IF; +END $$; + +-- 4. Add PAID-row CHECK constraint to enforce that future PAID rows always have +-- both payment_tx_hash and paid_at set. +ALTER TABLE invoices DROP CONSTRAINT IF EXISTS invoices_paid_row_check; +ALTER TABLE invoices ADD CONSTRAINT invoices_paid_row_check + CHECK ( + (status = 'PAID' AND payment_tx_hash IS NOT NULL AND paid_at IS NOT NULL) + OR status <> 'PAID' + ); diff --git a/docs/POSTGRES_CUTOVER.md b/docs/POSTGRES_CUTOVER.md index 870773a1..e4e8c83a 100644 --- a/docs/POSTGRES_CUTOVER.md +++ b/docs/POSTGRES_CUTOVER.md @@ -54,18 +54,20 @@ DATABASE_URL=postgresql://user:password@localhost:5432/quittance INVOICE_STORAGE=postgres ``` -### 2. Apply the schema (idempotent) +### 2. Apply the schema (versioned, forward-only migrations) ```bash cd backend npm run db:migrate ``` -The migration script (`db/schema.sql`) creates the invoices, transactions, -payment_events, and payment_monitor_checkpoints tables; adds all required -indexes; and adds the `payment_tx_hash` uniqueness constraint. Re-running the -migration on an existing database is safe — every `CREATE TABLE`, `ALTER TABLE`, -and `CREATE INDEX` statement uses `IF NOT EXISTS`. +The migration runner (`backend/src/db/migration-runner.ts`) applies numbered, +checksummed SQL migrations from `db/migrations/` under a PostgreSQL advisory lock +(`pg_advisory_lock`), with each migration committed in its own transaction. +Applied versions and SHA-256 checksums are tracked in the `schema_migrations` table. +Re-running the migration on an existing database is safe and idempotent — already applied +migrations are verified against stored checksums and skipped. Existing databases +built from the legacy schema adopt baseline `0001` automatically upon detection. ### 3. Verify the schema applied cleanly