feat(migrations): allow providing transaction context

Closes #851

docs/docs/migrations.md

@@ -123,6 +123,16 @@ Then run this script via `ts-node` (or compile it to plain JS and use `node`):
 $ ts-node migrate
 ```
 
+## Providing transaction context
+
+In some cases you might want to control the transaction context yourself:
+
+```ts
+await orm.em.transactional(async em => {
+  await migrator.up({ transaction: em.getTransactionContext() });
+});
+```
+
 ## Importing migrations statically
 
 If you do not want to dynamically import a folder (e.g. when bundling your code with webpack) you can import migrations

docs/versioned_docs/version-4.0/migrations.md

@@ -123,6 +123,16 @@ Then run this script via `ts-node` (or compile it to plain JS and use `node`):
 $ ts-node migrate
 ```
 
+## Providing transaction context
+
+In some cases you might want to control the transaction context yourself:
+
+```ts
+await orm.em.transactional(async em => {
+  await migrator.up({ transaction: em.getTransactionContext() });
+});
+```
+
 ## Importing migrations statically
 
 If you do not want to dynamically import a folder (e.g. when bundling your code with webpack) you can import migrations
@@ -143,7 +153,7 @@ await MikroORM.init({
 });
 ```
 
-With the help of (webpacks context module api)[https://webpack.js.org/guides/dependency-management/#context-module-api]
+With the help of [webpack's context module api](https://webpack.js.org/guides/dependency-management/#context-module-api)
 we can dynamically import the migrations making it possible to import all files in a folder.
 
 ```typescript

packages/migrations/src/Migrator.ts

@@ -1,5 +1,5 @@
 import umzug, { Umzug, migrationsList } from 'umzug';
-import { Utils, Constructor, Dictionary } from '@mikro-orm/core';
+import { Utils, Constructor, Dictionary, Transaction } from '@mikro-orm/core';
 import { SchemaGenerator, EntityManager } from '@mikro-orm/knex';
 import { Migration } from './Migration';
 import { MigrationRunner } from './MigrationRunner';
@@ -159,15 +159,21 @@ export class Migrator {
       return this.umzug[method](this.prefix(options as string[]));
     }
 
-    return this.driver.getConnection().transactional(async trx => {
-      this.runner.setMasterMigration(trx);
-      this.storage.setMasterMigration(trx);
-      const ret = await this.umzug[method](this.prefix(options as string[]));
-      this.runner.unsetMasterMigration();
-      this.storage.unsetMasterMigration();
+    if (Utils.isObject<MigrateOptions>(options) && options.transaction) {
+      return this.runInTransaction(options.transaction, method, options);
+    }
 
-      return ret;
-    });
+    return this.driver.getConnection().transactional(trx => this.runInTransaction(trx, method, options));
+  }
+
+  private async runInTransaction(trx: Transaction, method: 'up' | 'down', options: string | string[] | undefined | MigrateOptions) {
+    this.runner.setMasterMigration(trx);
+    this.storage.setMasterMigration(trx);
+    const ret = await this.umzug[method](this.prefix(options as string[]));
+    this.runner.unsetMasterMigration();
+    this.storage.unsetMasterMigration();
+
+    return ret;
   }
 
 }

packages/migrations/src/typings.ts

@@ -1,3 +1,5 @@
+import { Transaction } from '@mikro-orm/core';
+
 declare module 'umzug' {
 
   interface MigrationDefinitionWithName extends UmzugMigration {
@@ -11,6 +13,6 @@ declare module 'umzug' {
 }
 
 export type UmzugMigration = { name?: string; path?: string; file: string };
-export type MigrateOptions = { from?: string | number; to?: string | number; migrations?: string[] };
+export type MigrateOptions = { from?: string | number; to?: string | number; migrations?: string[]; transaction?: Transaction };
 export type MigrationResult = { fileName: string; code: string; diff: string[] };
 export type MigrationRow = { name: string; executed_at: Date };

tests/Migrator.test.ts

@@ -77,7 +77,9 @@ describe('Migrator', () => {
   });
 
   test('generate initial migration', async () => {
+    await orm.em.getKnex().schema.dropTableIfExists(orm.config.get('migrations').tableName!);
     const getExecutedMigrationsMock = jest.spyOn<any, any>(Migrator.prototype, 'getExecutedMigrations');
+    const getPendingMigrationsMock = jest.spyOn<any, any>(Migrator.prototype, 'getPendingMigrations');
     getExecutedMigrationsMock.mockResolvedValueOnce(['test.ts']);
     const migrator = new Migrator(orm.em);
     const err = 'Initial migration cannot be created, as some migrations already exist';
@@ -89,6 +91,8 @@ describe('Migrator', () => {
     const dateMock = jest.spyOn(Date.prototype, 'toISOString');
     dateMock.mockReturnValue('2019-10-13T21:48:13.382Z');
 
+    getExecutedMigrationsMock.mockResolvedValueOnce([]);
+    getPendingMigrationsMock.mockResolvedValueOnce([]);
     const migration = await migrator.createMigration(undefined, false, true);
     expect(logMigrationMock).toBeCalledWith('Migration20191013214813.ts');
     expect(migration).toMatchSnapshot('initial-migration-dump');
@@ -191,7 +195,6 @@ describe('Migrator', () => {
     const path = process.cwd() + '/temp/migrations';
 
     const migration = await migrator.createMigration(path, true);
-    await writeFile(path + '/' + migration.fileName, migration.code.replace(`'mikro-orm'`, `'@mikro-orm/migrations'`));
     const migratorMock = jest.spyOn(Migration.prototype, 'down');
     migratorMock.mockImplementation(async () => void 0);
 
@@ -217,6 +220,34 @@ describe('Migrator', () => {
     expect(calls).toMatchSnapshot('all-or-nothing');
   });
 
+  test('up/down with explicit transaction', async () => {
+    await orm.em.getKnex().schema.dropTableIfExists(orm.config.get('migrations').tableName!);
+    const migrator = new Migrator(orm.em);
+    const path = process.cwd() + '/temp/migrations';
+
+    const migration = await migrator.createMigration(path, true);
+    const migratorMock = jest.spyOn(Migration.prototype, 'down');
+    migratorMock.mockImplementation(async () => void 0);
+
+    const mock = jest.fn();
+    const logger = new Logger(mock, ['query']);
+    Object.assign(orm.config, { logger });
+
+    await orm.em.transactional(async em => {
+      await migrator.up({ transaction: em.getTransactionContext() });
+      await migrator.down({ transaction: em.getTransactionContext() });
+    });
+
+    await remove(path + '/' + migration.fileName);
+    const calls = mock.mock.calls.map(call => {
+      return call[0]
+        .replace(/ \[took \d+ ms]/, '')
+        .replace(/\[query] /, '')
+        .replace(/ trx\d+/, 'trx_xx');
+    });
+    expect(calls).toMatchSnapshot('explicit-tx');
+  });
+
   test('up/down params [all or nothing disabled]', async () => {
     await orm.em.getKnex().schema.dropTableIfExists(orm.config.get('migrations').tableName!);
     const migrator = new Migrator(orm.em);

tests/__snapshots__/Migrator.test.ts.snap

@@ -545,3 +545,24 @@ Array [
   "commit (via write connection '127.0.0.1')",
 ]
 `;
+
+exports[`Migrator up/down with explicit transaction: explicit-tx 1`] = `
+Array [
+  "begin (via write connection '127.0.0.1')",
+  "select table_name as table_name from information_schema.tables where table_type = 'BASE TABLE' and table_schema = schema() (via write connection '127.0.0.1')",
+  "create table \`mikro_orm_migrations\` (\`id\` int unsigned not null auto_increment primary key, \`name\` varchar(255), \`executed_at\` datetime default current_timestamp) (via write connection '127.0.0.1')",
+  "select * from \`mikro_orm_migrations\` order by \`id\` asc (via write connection '127.0.0.1')",
+  "select * from \`mikro_orm_migrations\` order by \`id\` asc (via write connection '127.0.0.1')",
+  "savepointtrx_xx (via write connection '127.0.0.1')",
+  "select 1 (via write connection '127.0.0.1')",
+  "release savepointtrx_xx (via write connection '127.0.0.1')",
+  "insert into \`mikro_orm_migrations\` (\`name\`) values (?) (via write connection '127.0.0.1')",
+  "select table_name as table_name from information_schema.tables where table_type = 'BASE TABLE' and table_schema = schema() (via write connection '127.0.0.1')",
+  "select * from \`mikro_orm_migrations\` order by \`id\` asc (via write connection '127.0.0.1')",
+  "select * from \`mikro_orm_migrations\` order by \`id\` asc (via write connection '127.0.0.1')",
+  "savepointtrx_xx (via write connection '127.0.0.1')",
+  "release savepointtrx_xx (via write connection '127.0.0.1')",
+  "delete from \`mikro_orm_migrations\` where \`name\` = ? (via write connection '127.0.0.1')",
+  "commit (via write connection '127.0.0.1')",
+]
+`;