feat(db-postgres): improve transaction support by passing req to migr…

…ations

packages/db-postgres/src/migrate.ts

@@ -1,8 +1,12 @@
 /* eslint-disable no-restricted-syntax, no-await-in-loop */
 import type { Payload } from 'payload'
 import type { Migration } from 'payload/database'
+import type { PayloadRequest } from 'payload/dist/express/types'
 
 import { readMigrationFiles } from 'payload/database'
+import { commitTransaction } from 'payload/dist/utilities/commitTransaction'
+import { initTransaction } from 'payload/dist/utilities/initTransaction'
+import { killTransaction } from 'payload/dist/utilities/killTransaction'
 import prompts from 'prompts'
 
 import type { PostgresAdapter } from './types'
@@ -42,11 +46,11 @@ export async function migrate(this: PostgresAdapter): Promise<void> {
     const { confirm: runMigrations } = await prompts(
       {
         name: 'confirm',
+        type: 'confirm',
         initial: false,
         message:
           "It looks like you've run Payload in dev mode, meaning you've dynamically pushed changes to your database.\n\n" +
           "If you'd like to run migrations, data loss will occur. Would you like to proceed?",
-        type: 'confirm',
       },
       {
         onCancel: () => {
@@ -79,14 +83,16 @@ async function runMigrationFile(payload: Payload, migration: Migration, batch: n
   const { generateDrizzleJson } = require('drizzle-kit/utils')
 
   const start = Date.now()
+  const req = {} as PayloadRequest
 
   payload.logger.info({ msg: `Migrating: ${migration.name}` })
 
   const pgAdapter = payload.db
   const drizzleJSON = generateDrizzleJson(pgAdapter.schema)
 
   try {
-    await migration.up({ payload })
+    await initTransaction(req)
+    await migration.up({ payload, req })
     payload.logger.info({ msg: `Migrated:  ${migration.name} (${Date.now() - start}ms)` })
     await payload.create({
       collection: 'payload-migrations',
@@ -95,8 +101,11 @@ async function runMigrationFile(payload: Payload, migration: Migration, batch: n
         batch,
         schema: drizzleJSON,
       },
+      req,
     })
+    await commitTransaction(req)
   } catch (err: unknown) {
+    await killTransaction(req)
     payload.logger.error({
       err,
       msg: parseError(err, `Error running migration ${migration.name}`),

packages/db-postgres/src/migrateDown.ts

@@ -2,6 +2,9 @@
 import type { PayloadRequest } from 'payload/types'
 
 import { getMigrations, readMigrationFiles } from 'payload/database'
+import { commitTransaction } from 'payload/dist/utilities/commitTransaction'
+import { initTransaction } from 'payload/dist/utilities/initTransaction'
+import { killTransaction } from 'payload/dist/utilities/killTransaction'
 
 import type { PostgresAdapter } from './types'
 
@@ -32,12 +35,12 @@ export async function migrateDown(this: PostgresAdapter): Promise<void> {
     }
 
     const start = Date.now()
-    let transactionID
+    const req = {} as PayloadRequest
 
     try {
       payload.logger.info({ msg: `Migrating down: ${migrationFile.name}` })
-      transactionID = await this.beginTransaction()
-      await migrationFile.down({ payload })
+      await initTransaction(req)
+      await migrationFile.down({ payload, req })
       payload.logger.info({
         msg: `Migrated down:  ${migrationFile.name} (${Date.now() - start}ms)`,
       })
@@ -47,15 +50,13 @@ export async function migrateDown(this: PostgresAdapter): Promise<void> {
         await payload.delete({
           id: migration.id,
           collection: 'payload-migrations',
-          req: {
-            transactionID,
-          } as PayloadRequest,
+          req,
         })
       }
 
-      await this.commitTransaction(transactionID)
+      await commitTransaction(req)
     } catch (err: unknown) {
-      await this.rollbackTransaction(transactionID)
+      await killTransaction(req)
 
       payload.logger.error({
         err,

packages/db-postgres/src/migrateFresh.ts

@@ -2,6 +2,9 @@ import type { PayloadRequest } from 'payload/types'
 
 import { sql } from 'drizzle-orm'
 import { readMigrationFiles } from 'payload/database'
+import { commitTransaction } from 'payload/dist/utilities/commitTransaction'
+import { initTransaction } from 'payload/dist/utilities/initTransaction'
+import { killTransaction } from 'payload/dist/utilities/killTransaction'
 import prompts from 'prompts'
 
 import type { PostgresAdapter } from './types'
@@ -17,9 +20,9 @@ export async function migrateFresh(this: PostgresAdapter): Promise<void> {
   const { confirm: acceptWarning } = await prompts(
     {
       name: 'confirm',
+      type: 'confirm',
       initial: false,
       message: `WARNING: This will drop your database and run all migrations. Are you sure you want to proceed?`,
-      type: 'confirm',
     },
     {
       onCancel: () => {
@@ -36,36 +39,35 @@ export async function migrateFresh(this: PostgresAdapter): Promise<void> {
     msg: `Dropping database.`,
   })
 
-  await this.drizzle.execute(sql`drop schema public cascade;\ncreate schema public;`)
+  await this.drizzle.execute(sql`drop schema public cascade;
+  create schema public;`)
 
   const migrationFiles = await readMigrationFiles({ payload })
   payload.logger.debug({
     msg: `Found ${migrationFiles.length} migration files.`,
   })
 
-  let transactionID
+  const req = {} as PayloadRequest
   // Run all migrate up
   for (const migration of migrationFiles) {
     payload.logger.info({ msg: `Migrating: ${migration.name}` })
     try {
       const start = Date.now()
-      transactionID = await this.beginTransaction()
-      await migration.up({ payload })
+      await initTransaction(req)
+      await migration.up({ payload, req })
       await payload.create({
         collection: 'payload-migrations',
         data: {
           name: migration.name,
           batch: 1,
         },
-        req: {
-          transactionID,
-        } as PayloadRequest,
+        req,
       })
-      await this.commitTransaction(transactionID)
+      await commitTransaction(req)
 
       payload.logger.info({ msg: `Migrated:  ${migration.name} (${Date.now() - start}ms)` })
     } catch (err: unknown) {
-      await this.rollbackTransaction(transactionID)
+      await killTransaction(req)
       payload.logger.error({
         err,
         msg: parseError(err, `Error running migration ${migration.name}. Rolling back`),

packages/db-postgres/src/migrateRefresh.ts

@@ -2,7 +2,9 @@
 import type { PayloadRequest } from 'payload/types'
 
 import { getMigrations, readMigrationFiles } from 'payload/database'
-import { DatabaseError } from 'pg'
+import { commitTransaction } from 'payload/dist/utilities/commitTransaction'
+import { initTransaction } from 'payload/dist/utilities/initTransaction'
+import { killTransaction } from 'payload/dist/utilities/killTransaction'
 
 import type { PostgresAdapter } from './types'
 
@@ -29,7 +31,7 @@ export async function migrateRefresh(this: PostgresAdapter) {
     msg: `Rolling back batch ${latestBatch} consisting of ${existingMigrations.length} migration(s).`,
   })
 
-  let transactionID
+  const req = {} as PayloadRequest
 
   // Reverse order of migrations to rollback
   existingMigrations.reverse()
@@ -43,8 +45,8 @@ export async function migrateRefresh(this: PostgresAdapter) {
 
       payload.logger.info({ msg: `Migrating down: ${migration.name}` })
       const start = Date.now()
-      transactionID = await this.beginTransaction()
-      await migrationFile.down({ payload })
+      await initTransaction(req)
+      await migrationFile.down({ payload, req })
       payload.logger.info({
         msg: `Migrated down:  ${migration.name} (${Date.now() - start}ms)`,
       })
@@ -53,18 +55,17 @@ export async function migrateRefresh(this: PostgresAdapter) {
       if (tableExists) {
         await payload.delete({
           collection: 'payload-migrations',
-          req: {
-            transactionID,
-          } as PayloadRequest,
+          req,
           where: {
             name: {
               equals: migration.name,
             },
           },
         })
       }
+      await commitTransaction(req)
     } catch (err: unknown) {
-      await this.rollbackTransaction(transactionID)
+      await killTransaction(req)
       payload.logger.error({
         err,
         msg: parseError(err, `Error running migration ${migration.name}. Rolling back.`),
@@ -78,23 +79,21 @@ export async function migrateRefresh(this: PostgresAdapter) {
     payload.logger.info({ msg: `Migrating: ${migration.name}` })
     try {
       const start = Date.now()
-      transactionID = await this.beginTransaction()
-      await migration.up({ payload })
+      await initTransaction(req)
+      await migration.up({ payload, req })
       await payload.create({
         collection: 'payload-migrations',
         data: {
           name: migration.name,
           executed: true,
         },
-        req: {
-          transactionID,
-        } as PayloadRequest,
+        req,
       })
-      await this.commitTransaction(transactionID)
+      await commitTransaction(req)
 
       payload.logger.info({ msg: `Migrated:  ${migration.name} (${Date.now() - start}ms)` })
     } catch (err: unknown) {
-      await this.rollbackTransaction(transactionID)
+      await killTransaction(req)
       payload.logger.error({
         err,
         msg: parseError(err, `Error running migration ${migration.name}. Rolling back.`),

packages/db-postgres/src/migrateReset.ts

@@ -2,6 +2,9 @@
 import type { PayloadRequest } from 'payload/types'
 
 import { getMigrations, readMigrationFiles } from 'payload/database'
+import { commitTransaction } from 'payload/dist/utilities/commitTransaction'
+import { initTransaction } from 'payload/dist/utilities/initTransaction'
+import { killTransaction } from 'payload/dist/utilities/killTransaction'
 
 import type { PostgresAdapter } from './types'
 
@@ -21,10 +24,10 @@ export async function migrateReset(this: PostgresAdapter): Promise<void> {
     return
   }
 
+  const req = {} as PayloadRequest
+
   // Rollback all migrations in order
   for (const migration of existingMigrations) {
-    let transactionID
-
     const migrationFile = migrationFiles.find((m) => m.name === migration.name)
     try {
       if (!migrationFile) {
@@ -33,8 +36,8 @@ export async function migrateReset(this: PostgresAdapter): Promise<void> {
 
       const start = Date.now()
       payload.logger.info({ msg: `Migrating down: ${migrationFile.name}` })
-      transactionID = await this.beginTransaction()
-      await migrationFile.down({ payload })
+      await initTransaction(req)
+      await migrationFile.down({ payload, req })
       payload.logger.info({
         msg: `Migrated down:  ${migrationFile.name} (${Date.now() - start}ms)`,
       })
@@ -44,19 +47,17 @@ export async function migrateReset(this: PostgresAdapter): Promise<void> {
         await payload.delete({
           id: migration.id,
           collection: 'payload-migrations',
-          req: {
-            transactionID,
-          } as PayloadRequest,
+          req,
         })
       }
 
-      await this.commitTransaction(transactionID)
+      await commitTransaction(req)
     } catch (err: unknown) {
       let msg = `Error running migration ${migrationFile.name}.`
 
       if (err instanceof Error) msg += ` ${err.message}`
 
-      await this.rollbackTransaction(transactionID)
+      await killTransaction(req)
       payload.logger.error({
         err,
         msg,