db-migrations: use pg for new migrations

Makefile

@@ -131,7 +131,8 @@ test-coverage: node_version
 
 .PHONY: lint
 lint: node_version
-	npx eslint --cache --max-warnings 0 .
+	# max-warnings set to take account of legacy database migrations
+	npx eslint --cache --max-warnings 15 .
 
 
 ################################################################################

lib/bin/check-migrations.js

@@ -7,11 +7,15 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { withDatabase, checkMigrations } = require('../model/migrate');
+const { withKnex, checkKnexMigrations, checkPostKnexMigrations } = require('../model/migrate');
+
+// REVIEW why is check-migrations required in the first place?
 
 (async () => {
   try {
-    await withDatabase(require('config').get('default.database'))(checkMigrations);
+    const config = require('config').get('default.database');
+    await withKnex(config)(checkKnexMigrations);
+    await checkPostKnexMigrations(config);
   } catch (err) {
     console.error('Error:', err.message);
     process.exit(1);

lib/bin/check-open-db-queries.js

@@ -7,11 +7,11 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { withDatabase } = require('../model/migrate');
+const { withKnex } = require('../model/migrate');
 
 (async () => {
   try {
-    const { rows } = await withDatabase(require('config').get('default.database'))((db) =>
+    const { rows } = await withKnex(require('config').get('default.database'))((db) =>
       db.raw('SELECT COUNT(*) FROM pg_stat_activity WHERE usename=CURRENT_USER'));
     const queryCount = rows[0].count - 1; // the count query will appear as one of the open queries
 

lib/bin/run-migrations.js

@@ -7,13 +7,20 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { withDatabase, migrate } = require('../model/migrate');
+const { // eslint-disable-line object-curly-newline
+  withKnex,
+  knexMigrations,
+
+  postKnexMigrations,
+} = require('../model/migrate'); // eslint-disable-line object-curly-newline
 
 (async () => {
   try {
-    await withDatabase(require('config').get('default.database'))(migrate);
+    const config = require('config').get('default.database');
+    await withKnex(config)(knexMigrations);
+    await postKnexMigrations(config);
   } catch (err) {
-    console.error('Error:', err.message);
+    console.error(err);
     process.exit(1);
   }
 })();

lib/model/knexfile.js

@@ -28,10 +28,10 @@ NODE_CONFIG_DIR=../../config DEBUG=knex:query,knex:bindings npx knex migrate:up
 */
 
 const config = require('config');
-const { connectionObject } = require('../util/db');
+const { knexConfig } = require('../util/db');
 
 module.exports = {
   client: 'pg',
-  connection: connectionObject(config.get('default.database'))
+  connection: knexConfig(config.get('default.database'))
 };
 

lib/model/migrate.js

@@ -10,28 +10,207 @@
 // This is a variety of functions helpful for connecting to and performing
 // top-level operations with a database, like migrations.
 
+const { lstatSync, readdirSync } = require('node:fs');
+
+const _ = require('lodash'); // eslint-disable-line import/no-extraneous-dependencies
 const knex = require('knex');
-const { connectionObject } = require('../util/db');
+const pg = require('pg');
+const { knexConfig } = require('../util/db');
 
 // Connects to the postgres database specified in configuration and returns it.
-const connect = (config) => knex({ client: 'pg', connection: connectionObject(config) });
+const initKnex = (config) => knex({ client: 'pg', connection: knexConfig(config) });
+
+const legacyPath = `${__dirname}/migrations/legacy`;
+const postKnexPath = `${__dirname}/migrations`;
 
 // Connects to a database, passes it to a function for operations, then ensures its closure.
-const withDatabase = (config) => (mutator) => {
-  const db = connect(config);
+const withKnex = (config) => (mutator) => {
+  const db = initKnex(config);
   return mutator(db).finally(() => db.destroy());
 };
 
 // Given a database, initiates migrations on it.
-const migrate = (db) => db.migrate.latest({ directory: `${__dirname}/migrations` });
+const knexMigrations = (db) => db.migrate.latest({ directory: legacyPath });
+
+const withPg = async (config, fn) => {
+  const log = (...args) => console.log('[withPg]', ...args); // eslint-disable-line no-console
+  log('ENTRY');
+
+  const { Client } = pg;
+  const client = new Client(config);
+
+  log('client created');
+
+  log('Connecting to client...');
+  await client.connect();
+  log('Client connected OK.');
+
+  try {
+    await fn(client);
+  } finally {
+    log('Ending client...');
+    await client.end();
+    log('Client ended.');
+  }
+};
+
+const getPostKnexMigrationsToRun = async client => {
+  const log = (...args) => console.log('[getPostKnexMigrationsToRun]', ...args); // eslint-disable-line no-console
+  log('ENTRY');
+
+  const migrationsDir = postKnexPath;
+  const allMigrations = readdirSync(migrationsDir)
+    .filter(f => f.endsWith('.js') && lstatSync(`${migrationsDir}/${f}`).isFile())
+    .sort();
+  log('allMigrations:', allMigrations);
+
+  const alreadyRun = (await client.query('SELECT name FROM post_knex_migrations')).rows.map(r => r.name);
+  log('alreadyRun:', alreadyRun);
+
+  const toRunNames = allMigrations.filter(m => !alreadyRun.includes(m));
+  log('toRunNames:', toRunNames);
+
+  const toRun = toRunNames.map(name => {
+    const path = `${migrationsDir}/${name}`;
+    const migration = require(path); // eslint-disable-line import/no-dynamic-require
+    return { name, path, migration };
+  });
+  log('toRun:', toRun);
+
+  return toRun;
+};
+
+const postKnexMigrations = async (config) => {
+  const log = (...args) => console.log('[postKnexMigrations]', ...args); // eslint-disable-line no-console
+  log('ENTRY');
+
+  // In the main, this migrator is written to behave similarly to knex's:
+  //
+  //   * expects transaction property async .up({ raw })
+  //   * provides implementation of db.raw()
+  //   * runs all new migrations in the same transaction
+  //
+  // Notable differences
+  //
+  //   * uses new post_knex_migrations table
+  //   * ONLY provides db.raw()-equivalent function to transactions - no knex query builder etc.
+  //   * ONLY implements up(); will throw if a transaction has other properties, except for `down()` which is currently ignored TODO implement this if it's useful to devs
+  //   * gets list of migrations to run _after_ acquiring db lock
+  //   * sets migration_time to be the start of the migration batch's transaction rather than some other intermediate time
+
+  await withPg(config, async client => {
+    try {
+      log('Starting transaction...');
+      await client.query('BEGIN'); // TODO do we need a specific transaction type?
+      log('Transaction started.');
+
+      log('Acquiring knex lock...');
+      // TODO do this... if it's useful.  Need to think of _some_ way to prevent new migrations and old migrations running simultaneously.
+      log('Knex lock acquired');
+
+      log('Creating new table if not exists...');
+      // N.B. this table is created to be similar to the legacy knex-created table.
+      // The key difference is that name, batch and migration_time columns are
+      // not nullable.
+      const maxLen = 255;
+      await client.query(`
+        CREATE TABLE IF NOT EXISTS post_knex_migrations (
+          id             SERIAL                      PRIMARY KEY,
+          name           VARCHAR(${maxLen})          NOT NULL,
+          batch          INTEGER                     NOT NULL,
+          migration_time TIMESTAMP(3) WITH TIME ZONE NOT NULL
+      )`);
+      log('Table now definitely exists.');
+
+      log('Acquiring lock on post_knex_migrations table...');
+      await client.query('LOCK TABLE post_knex_migrations IN EXCLUSIVE MODE NOWAIT');
+      log('Lock acquired.');
+
+      const toRun = await getPostKnexMigrationsToRun(client);
+
+      if (!toRun.length) {
+        log('No migrations to run - exiting.');
+        await client.query('ROLLBACK');
+        return;
+      }
+
+      log('Validating', toRun.length, 'migrations...');
+      for (const { migration, name } of toRun) {
+        log('Validing migration:', name, '...');
+
+        if (name.length > maxLen) throw new Error(`Migration name '${name}' is too long - max length is ${maxLen}, but got ${name.length}`);
+
+        // TODO check for illegal chars in name?
+
+        const keys = Object.keys(migration);
+        const unexpectedKeys = _.omit(keys, 'up', 'down');
+        if (unexpectedKeys.length) throw new Error(`Unexpected key(s) found in migration ${name}: ${unexpectedKeys}`);
+
+        if (!migration.up) throw new Error(`Required prop .up not found in migration ${name}`);
+        if (typeof migration.up !== 'function') {
+          throw new Error(`Required prop .up of migration ${name} has incorrect type - expected 'function', but got '${typeof migration.up}'`);
+        }
+
+        if (migration.down && typeof migration.down !== 'function') {
+          throw new Error(`Optional prop .down of migration ${name} has incorrect type - expected 'function' but got '${typeof migration.down}'`);
+        }
+
+        log('Migration', name, 'looks valid.');
+      }
+      log(toRun.length, 'migrations look valid.');
+
+      log('Running', toRun.length, 'migrations...');
+      for (const { migration, name } of toRun) {
+        log('Running migration:', name);
+        await migration.up(client); // eslint-disable-line no-await-in-loop
+        log('Migration complete:', name);
+      }
+      log(toRun.length, 'migrations ran OK.');
+
+      const { lastBatch } = (await client.query(`SELECT COALESCE(MAX(batch), 0) AS "lastBatch" FROM post_knex_migrations`)).rows[0];
+      log('lastBatch:', lastBatch);
+
+      // Note that migration_time is CLOCK_TIMESTAMP() to match knex implementation.
+      // TODO confirm in relevant version of knex source code that this is actually the case, and link here.
+      const namesJson = JSON.stringify(toRun.map(m => m.name));
+      // See: https://www.postgresql.org/docs/current/functions-json.html
+      await client.query(`
+        INSERT INTO post_knex_migrations(name, batch, migration_time)
+          SELECT value#>>'{}' AS name
+               , ${lastBatch + 1} AS batch
+               , CLOCK_TIMESTAMP() AS migration_time
+            FROM JSON_ARRAY_ELEMENTS($1)
+      `, [ namesJson ]);
+
+      log('Committing migrations...');
+      await client.query('COMMIT');
+      log('Migrations committed.');
+    } catch (err) {
+      log('Caught error; rolling back', err);
+      await client.query('ROLLBACK');
+      throw err;
+    }
+  });
+};
 
 // Checks for pending migrations and returns an exit code of 1 if any are
 // still pending/unapplied (e.g. automatically running migrations just failed).
-const checkMigrations = (db) => db.migrate.list({ directory: `${__dirname}/migrations` })
+const checkKnexMigrations = (db) => db.migrate.list({ directory: legacyPath })
   .then((res) => {
     if (res[1].length > 0)
       process.exitCode = 1;
   });
 
-module.exports = { checkMigrations, connect, withDatabase, migrate };
+// Checks for pending migrations and returns an exit code of 1 if any are
+// still pending/unapplied (e.g. automatically running migrations just failed).
+const checkPostKnexMigrations = async config => {
+  const log = (...args) => console.log('[checkPostKnexMigrations]', ...args); // eslint-disable-line no-console
+  log('ENTRY');
+
+  await withPg(config, async client => {
+    const toRun = await getPostKnexMigrationsToRun(client);
+    if (toRun.length) process.exitCode = 1;
+  });
+};
 
+module.exports = { checkKnexMigrations, checkPostKnexMigrations, initKnex, withKnex, withPg, knexMigrations, postKnexMigrations };

lib/model/migrations/.eslintrc.json

@@ -0,0 +1,9 @@
+{
+  "extends": "../../../.eslintrc.json",
+  "rules": {
+    "no-restricted-modules": [
+      "error",
+      { "patterns": [ "../*" ] }
+    ]
+  }
+}

lib/model/migrations/20250113-01-disable-nullable-blob-content-types.js

@@ -0,0 +1,23 @@
+// Copyright 2025 ODK Central Developers
+// See the NOTICE file at the top-level directory of this distribution and at
+// https://github.com/getodk/central-backend/blob/master/NOTICE.
+// This file is part of ODK Central. It is subject to the license terms in
+// the LICENSE file found in the top-level directory of this distribution and at
+// https://www.apache.org/licenses/LICENSE-2.0. No part of ODK Central,
+// including this file, may be copied, modified, propagated, or distributed
+// except according to the terms contained in the LICENSE file.
+
+const up = (db) => db.query(`
+  ALTER TABLE blobs
+    ALTER COLUMN "contentType" TYPE TEXT USING(COALESCE("contentType", 'application/octet-stream')),
+    ALTER COLUMN "contentType" SET DEFAULT 'application/octet-stream',
+    ALTER COLUMN "contentType" SET NOT NULL
+`);
+
+const down = (db) => db.query(`
+  ALTER TABLE blobs
+    ALTER COLUMN "contentType" DROP NOT NULL,
+    ALTER COLUMN "contentType" DROP DEFAULT
+`);
+
+module.exports = { up, down };

lib/model/migrations/legacy/.eslintrc.json

@@ -0,0 +1,9 @@
+{
+  "extends": "../../../../.eslintrc.json",
+  "rules": {
+    "no-restricted-modules": [
+      "warn",
+      { "patterns": [ "../*" ] }
+    ]
+  }
+}

lib/model/migrations/20180727-02-add-md5-to-blobs.js → lib/model/migrations/legacy/20180727-02-add-md5-to-blobs.js

@@ -8,7 +8,7 @@
 // except according to the terms contained in the LICENSE file.
 //
 
-const { md5sum } = require('../../util/crypto');
+const { md5sum } = require('../../../util/crypto');
 
 const up = (knex) =>
   knex.schema.table('blobs', (blobs) => { blobs.string('md5', 32); })

lib/model/migrations/20180727-03-add-form-attachments-table.js → lib/model/migrations/legacy/20180727-03-add-form-attachments-table.js

@@ -23,7 +23,7 @@
 
     fa.index([ 'formId' ]);
   }).then(() => {
-    const { expectedFormAttachments } = require('../../data/schema');
+    const { expectedFormAttachments } = require('../../../data/schema');
     const { uniq, pluck } = require('ramda');
 
     // now add all expected attachments on extant forms.

lib/model/migrations/20190520-01-add-form-versioning.js → lib/model/migrations/legacy/20190520-01-add-form-versioning.js

@@ -7,7 +7,7 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { shasum, sha256sum } = require('../../util/crypto');
+const { shasum, sha256sum } = require('../../../util/crypto');
 const assert = require('assert').strict;
 
 const check = (message, query) =>

lib/model/migrations/20191007-01-backfill-client-audits.js → lib/model/migrations/legacy/20191007-01-backfill-client-audits.js

@@ -7,9 +7,9 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { parseClientAudits } = require('../../data/client-audits');
-const { getFormFields } = require('../../data/schema');
-const { traverseXml, findOne, root, node, text } = require('../../util/xml');
+const { parseClientAudits } = require('../../../data/client-audits');
+const { getFormFields } = require('../../../data/schema');
+const { traverseXml, findOne, root, node, text } = require('../../../util/xml');
 
 const up = (db) => new Promise((resolve, reject) => {
   const work = [];

lib/model/migrations/20191231-02-add-schema-storage.js → lib/model/migrations/legacy/20191231-02-add-schema-storage.js

@@ -7,7 +7,7 @@
 // including this file, may be copied, modified, propagated, or distributed
 // except according to the terms contained in the LICENSE file.
 
-const { getFormFields } = require('../../data/schema');
+const { getFormFields } = require('../../../data/schema');
 
 const up = async (db) => {
   await db.schema.createTable('form_fields', (fields) => {
@@ -51,7 +51,7 @@
               // this config hardcoding would be dangerous with tests except that
               // tests will never invoke this path.
               const config = require('config').get('default.database');
-              const db2 = require('../migrate').connect(config);
+              const db2 = require('../../migrate').connect(config);
               return db2.select('projectId', 'xmlFormId').from('forms').where({ currentDefId: formDefId })
                 .then(([{ projectId, xmlFormId }]) => {
                   process.stderr.write(`\n!!!!\nThe database upgrade to v0.8 has failed because the Form '${xmlFormId}' in Project ${projectId} has an invalid schema. It tries to bind multiple instance nodes at the path ${path}.\n!!!!\n\n`);