pf_app/lib/migrations.js

// Migration bookkeeping shared by the runner (scripts/migrate.js) and the
// startup check in server.js.
//
// Migrations are plain .sql files in setup_sql/migrations, applied in filename
// order and recorded in pf.schema_version. They are immutable once applied: the
// runner stores a checksum and refuses to proceed if a file has changed, since
// editing an applied migration means databases silently disagree about what the
// schema is.
//
// pf.dim_period is deliberately not a migration — it is a parameterised data
// load (fiscal year start month), so it stays in setup_sql/gen_dim_period.sql.

const fs   = require('fs');
const path = require('path');
const crypto = require('crypto');

const MIGRATIONS_DIR = path.join(__dirname, '..', 'setup_sql', 'migrations');

// Bookkeeping table has to exist before the first migration can be recorded, so
// the runner creates it rather than a migration doing so.
const BOOKKEEPING_DDL = `
    CREATE SCHEMA IF NOT EXISTS pf;
    CREATE TABLE IF NOT EXISTS pf.schema_version (
        filename   text PRIMARY KEY,
        checksum   text NOT NULL,
        applied_at timestamptz NOT NULL DEFAULT now(),
        applied_by text
    );
`;

function checksum(sql) {
    return crypto.createHash('sha256').update(sql).digest('hex').slice(0, 16);
}

// every .sql file on disk, in apply order
function readMigrations() {
    if (!fs.existsSync(MIGRATIONS_DIR)) return [];
    return fs.readdirSync(MIGRATIONS_DIR)
        .filter(f => f.endsWith('.sql'))
        .sort()
        .map(filename => {
            const sql = fs.readFileSync(path.join(MIGRATIONS_DIR, filename), 'utf8');
            return { filename, sql, checksum: checksum(sql) };
        });
}

async function ensureBookkeeping(client) {
    await client.query(BOOKKEEPING_DDL);
}

async function readApplied(client) {
    const { rows } = await client.query(
        `SELECT filename, checksum, applied_at FROM pf.schema_version ORDER BY filename`
    );
    return rows;
}

// Compare disk against the database. Returns pending migrations plus any
// integrity problems worth refusing to run on.
async function status(client) {
    await ensureBookkeeping(client);
    const onDisk  = readMigrations();
    const applied = await readApplied(client);
    const appliedBy = new Map(applied.map(r => [r.filename, r]));
    const diskBy    = new Map(onDisk.map(m => [m.filename, m]));

    const pending  = onDisk.filter(m => !appliedBy.has(m.filename));
    // an applied file whose contents changed — the schema is now undefined
    const modified = onDisk
        .filter(m => appliedBy.has(m.filename) && appliedBy.get(m.filename).checksum !== m.checksum)
        .map(m => m.filename);
    // recorded as applied but no longer on disk — someone deleted or renamed it
    const missing  = applied.filter(r => !diskBy.has(r.filename)).map(r => r.filename);
    // a pending migration ordered before an applied one would apply out of sequence
    const lastApplied = applied.length ? applied[applied.length - 1].filename : null;
    const outOfOrder  = lastApplied ? pending.filter(m => m.filename < lastApplied).map(m => m.filename) : [];

    return { onDisk, applied, pending, modified, missing, outOfOrder };
}

// Apply pending migrations, each in its own transaction so a failure leaves
// earlier ones committed and the failing one fully rolled back.
async function migrate(client, { user, log = console.log } = {}) {
    const st = await status(client);
    if (st.modified.length) {
        throw new Error(
            `Applied migrations have been modified: ${st.modified.join(', ')}. ` +
            `Migrations are immutable — revert the edit and add a new migration instead.`
        );
    }
    if (st.missing.length) {
        throw new Error(
            `Migrations recorded as applied are missing from disk: ${st.missing.join(', ')}.`
        );
    }
    if (st.outOfOrder.length) {
        throw new Error(
            `Migrations would apply out of order: ${st.outOfOrder.join(', ')} sort before ` +
            `already-applied migrations. Renumber them after the latest applied migration.`
        );
    }
    if (!st.pending.length) {
        log('Schema is up to date — no migrations to apply.');
        return [];
    }

    const done = [];
    for (const m of st.pending) {
        log(`applying ${m.filename} …`);
        try {
            await client.query('BEGIN');
            await client.query(m.sql);
            await client.query(
                `INSERT INTO pf.schema_version (filename, checksum, applied_by) VALUES ($1, $2, $3)`,
                [m.filename, m.checksum, user || null]
            );
            await client.query('COMMIT');
            done.push(m.filename);
        } catch (err) {
            await client.query('ROLLBACK').catch(() => {});
            throw new Error(`${m.filename} failed: ${err.message}`);
        }
    }
    log(`Applied ${done.length} migration(s).`);
    return done;
}

// Record migrations as applied without running them — for a database whose schema
// already matches, from before migrations existed.
async function baseline(client, { user, upTo, log = console.log } = {}) {
    const st = await status(client);
    const target = upTo
        ? st.pending.filter(m => m.filename <= upTo)
        : st.pending;
    if (!target.length) {
        log('Nothing to baseline — no pending migrations.');
        return [];
    }
    await client.query('BEGIN');
    try {
        for (const m of target) {
            await client.query(
                `INSERT INTO pf.schema_version (filename, checksum, applied_by)
                 VALUES ($1, $2, $3)
                 ON CONFLICT (filename) DO NOTHING`,
                [m.filename, m.checksum, user ? `${user} (baseline)` : 'baseline']
            );
        }
        await client.query('COMMIT');
    } catch (err) {
        await client.query('ROLLBACK').catch(() => {});
        throw err;
    }
    log(`Marked ${target.length} migration(s) as applied without running them.`);
    return target.map(m => m.filename);
}

module.exports = { MIGRATIONS_DIR, readMigrations, status, migrate, baseline, checksum };