diff --git a/package.json b/package.json index 1d346730b..23d095019 100644 --- a/package.json +++ b/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", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c99399b20..a6fcaed9b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -87,6 +87,9 @@ importers: specifier: ^4.2.1 version: 4.3.6 devDependencies: + '@electric-sql/pglite': + specifier: ^0.3.15 + version: 0.3.15 '@nuxt/devtools': specifier: ^3.1.1 version: 3.2.4(@vitejs/devtools@0.1.15(@pnpm/logger@1001.0.1)(@upstash/redis@1.36.2)(@vercel/blob@2.2.0)(aws4fetch@1.0.20)(db0@0.3.4(@electric-sql/pglite@0.3.15)(@libsql/client@0.17.0)(better-sqlite3@12.6.2)(drizzle-orm@0.45.2(@cloudflare/workers-types@4.20260214.0)(@electric-sql/pglite@0.3.15)(@libsql/client@0.17.0)(@opentelemetry/api@1.9.0)(@upstash/redis@1.36.2)(better-sqlite3@12.6.2)(mysql2@3.17.1)(postgres@3.4.8))(mysql2@3.17.1))(ioredis@5.11.0)(typescript@5.9.3)(vite@7.3.1(@types/node@25.2.3)(jiti@2.6.1)(lightningcss@1.31.1)(terser@5.48.0)(yaml@2.8.2)))(vite@7.3.1(@types/node@25.2.3)(jiti@2.6.1)(lightningcss@1.31.1)(terser@5.48.0)(yaml@2.8.2))(vue@3.5.33(typescript@5.9.3)) diff --git a/src/db/lib/migrations.ts b/src/db/lib/migrations.ts index 1f1cf2d61..941f44b25 100644 --- a/src/db/lib/migrations.ts +++ b/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(migration.filename) + const query = await migrationsStorage.getItem(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))}\`...`) diff --git a/test/database.migrations.test.ts b/test/database.migrations.test.ts new file mode 100644 index 000000000..456aa2023 --- /dev/null +++ b/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 + 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 }]) + }) +})