fix(db): make PostgreSQL migrations atomic

package.json

@@ -110,6 +110,7 @@
     }
   },
   "devDependencies": {
+    "@electric-sql/pglite": "^0.3.15",
     "@nuxt/devtools": "^3.1.1",
     "@nuxt/eslint-config": "^1.15.1",
     "@nuxt/module-builder": "^1.0.2",

src/db/lib/migrations.ts

@@ -8,6 +8,12 @@ function getRelativePath(fullPath: string) {
   return relative(process.cwd(), fullPath)
 }
 
+function dollarQuote(value: string) {
+  let tag = '$nuxthub$'
+  while (value.includes(tag)) tag = `${tag.slice(0, -1)}_$`
+  return `${tag}${value}${tag}`
+}
+
 export async function applyDatabaseMigrations(hub: ResolvedHubConfig, db: any) {
   if (!hub.db) return
   // Create a logger for this function (at runtime so we can have the debug level when run by the CLI)
@@ -19,11 +25,19 @@ export async function applyDatabaseMigrations(hub: ResolvedHubConfig, db: any) {
   const getRows = (result: any) => (dialect === 'mysql' ? result[0] : result.results || result.rows || result) || []
 
   const createMigrationsTableQuery = getCreateMigrationsTableQuery({ dialect: hub.db.dialect })
+  const createMigrationsTableStatement = dialect === 'postgresql'
+    ? `DO ${dollarQuote(`
+BEGIN
+  PERFORM pg_advisory_xact_lock(hashtext('nuxthub'), hashtext('migrations'));
+  ${createMigrationsTableQuery}
+END
+`)};`
+    : createMigrationsTableQuery
   log.debug('Creating migrations table if not exists...')
   const drizzleOrmPkg = 'drizzle-orm'
   const sql = await import(drizzleOrmPkg).then(m => m.sql)
   try {
-    await db[execute](sql.raw(createMigrationsTableQuery))
+    await db[execute](sql.raw(createMigrationsTableStatement))
   } catch (error: any) {
     const message = error.cause?.message || error.message
     log.error(`Failed to create migrations table\n${message}`)
@@ -54,10 +68,19 @@ export async function applyDatabaseMigrations(hub: ResolvedHubConfig, db: any) {
   }
 
   for (const migration of pendingMigrations) {
-    let query = await migrationsStorage.getItem<string>(migration.filename)
+    const query = await migrationsStorage.getItem<string>(migration.filename)
     if (!query) continue
-    query += `\nINSERT INTO _hub_migrations (name) values ('${migration.name}');`
-    const queries = splitSqlQueries(query)
+    const queries = dialect === 'postgresql'
+      ? [`DO ${dollarQuote(`
+BEGIN
+  PERFORM pg_advisory_xact_lock(hashtext('nuxthub'), hashtext('migrations'));
+  IF NOT EXISTS (SELECT 1 FROM _hub_migrations WHERE name = ${dollarQuote(migration.name)}) THEN
+    EXECUTE ${dollarQuote(query)};
+    INSERT INTO _hub_migrations (name) VALUES (${dollarQuote(migration.name)});
+  END IF;
+END
+`)};`]
+      : splitSqlQueries(`${query}\nINSERT INTO _hub_migrations (name) values ('${migration.name}');`)
 
     try {
       log.debug(`Applying database migration \`${getRelativePath(join(hub.dir!, 'db/migrations', migration.filename))}\`...`)

test/database.migrations.test.ts

@@ -0,0 +1,89 @@
+import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'pathe'
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { PGlite } from '@electric-sql/pglite'
+import { drizzle } from 'drizzle-orm/pglite'
+import type { ResolvedHubConfig } from '../src/types'
+import { applyDatabaseMigrations } from '../src/db/lib/migrations'
+
+describe('PostgreSQL migrations', () => {
+  let rootDir: string
+  let client: PGlite
+  let db: ReturnType<typeof drizzle>
+  let executedQueries: string[]
+
+  beforeEach(async () => {
+    rootDir = await mkdtemp(join(tmpdir(), 'nuxthub-postgresql-migrations-'))
+    await mkdir(join(rootDir, 'db/migrations'), { recursive: true })
+    client = new PGlite()
+    executedQueries = []
+    db = drizzle(client, {
+      logger: {
+        logQuery(query) {
+          executedQueries.push(query)
+        }
+      }
+    })
+  })
+
+  afterEach(async () => {
+    await client.close()
+    await rm(rootDir, { recursive: true, force: true })
+  })
+
+  function hub() {
+    return {
+      dir: rootDir,
+      db: { dialect: 'postgresql' }
+    } as ResolvedHubConfig
+  }
+
+  it('rolls back failed migration statements and tracking', async () => {
+    await writeFile(join(rootDir, 'db/migrations/0001_failure.postgresql.sql'), `
+      CREATE TABLE partially_applied (id integer);
+      ALTER TABLE missing_table ADD COLUMN value integer;
+    `)
+
+    expect(await applyDatabaseMigrations(hub(), db)).toBe(false)
+
+    const result = await client.query(`
+      SELECT
+        to_regclass('partially_applied') IS NOT NULL AS ddl_applied,
+        (SELECT count(*) FROM _hub_migrations) AS tracker_rows
+    `)
+    expect(result.rows).toEqual([{ ddl_applied: false, tracker_rows: 0 }])
+  })
+
+  it('serializes overlapping migration attempts and records once', async () => {
+    await writeFile(join(rootDir, 'db/migrations/0001_once.postgresql.sql'), `
+      CREATE FUNCTION migration_value() RETURNS text AS $$
+      BEGIN
+        RETURN '$nuxthub$';
+      END;
+      $$ LANGUAGE plpgsql;
+      CREATE TABLE applied_once (value text DEFAULT '$nuxthub$');
+    `)
+
+    expect(await Promise.all([
+      applyDatabaseMigrations(hub(), db),
+      applyDatabaseMigrations(hub(), db)
+    ])).toEqual([true, true])
+
+    const atomicQueries = executedQueries.filter(query => query.startsWith('DO '))
+    expect(atomicQueries).toHaveLength(4)
+    expect(atomicQueries.every((query) => {
+      const lock = query.indexOf('pg_advisory_xact_lock')
+      const guardedOperation = Math.max(query.indexOf('CREATE TABLE IF NOT EXISTS'), query.indexOf('IF NOT EXISTS (SELECT'))
+      return lock !== -1 && lock < guardedOperation
+    })).toBe(true)
+
+    const result = await client.query(`
+      SELECT
+        to_regclass('applied_once') IS NOT NULL AS ddl_applied,
+        migration_value() AS function_value,
+        (SELECT count(*) FROM _hub_migrations) AS tracker_rows
+    `)
+    expect(result.rows).toEqual([{ ddl_applied: true, function_value: '$nuxthub$', tracker_rows: 1 }])
+  })
+})