Fix transaction support for migrations (#3084)

* Fix transaction support for migrations

* Clarify warning

* Clarify warning message

* Simplify logic, make enabling/disabling more consistent

* Fix tests

* Fix test

* Reduce duplication

src/migrate/Migrator.js

@@ -49,12 +49,16 @@ export default class Migrator {
   constructor(knex) {
     // Clone knex instance and remove post-processing that is unnecessary for internal queries from a cloned config
     if (isFunction(knex)) {
-      this.knex = knex.withUserParams({
-        ...knex.userParams,
-      });
-      this.knex.disableProcessing();
+      if (!knex.isTransaction) {
+        this.knex = knex.withUserParams({
+          ...knex.userParams,
+        });
+      } else {
+        this.knex = knex;
+      }
     } else {
       this.knex = Object.assign({}, knex);
+      this.knex.userParams = this.knex.userParams || {};
     }
 
     this.config = getMergedConfig(this.knex.client.config.migrations);
@@ -66,6 +70,7 @@ export default class Migrator {
 
   // Migrators to the latest configuration.
   latest(config) {
+    this._disableProcessing();
     this.config = getMergedConfig(config, this.config);
 
     return migrationListResolver
@@ -101,6 +106,7 @@ export default class Migrator {
 
   // Rollback the last "batch", or all, of migrations that were run.
   rollback(config, all = false) {
+    this._disableProcessing();
     return Promise.try(() => {
       this.config = getMergedConfig(config, this.config);
 
@@ -117,6 +123,7 @@ export default class Migrator {
   }
 
   status(config) {
+    this._disableProcessing();
     this.config = getMergedConfig(config, this.config);
 
     return Promise.all([
@@ -130,6 +137,7 @@ export default class Migrator {
   // Retrieves and returns the current migration version we're on, as a promise.
   // If no migrations have been run yet, return "none".
   currentVersion(config) {
+    this._disableProcessing();
     this.config = getMergedConfig(config, this.config);
 
     return migrationListResolver
@@ -155,6 +163,12 @@ export default class Migrator {
     return this.generator.make(name, this.config);
   }
 
+  _disableProcessing() {
+    if (this.knex.disableProcessing) {
+      this.knex.disableProcessing();
+    }
+  }
+
   _isLocked(trx) {
     const tableName = getLockTableName(this.config.tableName);
     return getTable(this.knex, tableName, this.config.schemaName)

src/transaction.js

@@ -193,7 +193,8 @@ function makeTransactor(trx, connection, trxClient) {
     );
   };
 
-  transactor.userParams = trx.userParams;
+  transactor.isTransaction = true;
+  transactor.userParams = trx.userParams || {};
 
   transactor.transaction = function(container, options) {
     return trxClient.transaction(container, options, trx);

test/unit/knex.js

@@ -99,10 +99,7 @@ describe('knex', () => {
     const knexWithParams = knex.withUserParams({ userParam: '451' });
 
     expect(knexWithParams.migrate.knex.userParams).to.deep.equal({
-      isProcessingDisabled: true,
-      postProcessResponse: undefined,
       userParam: '451',
-      wrapIdentifier: undefined,
     });
   });
 

test/unit/migrate/Migrator.js

@@ -26,7 +26,7 @@ describe('Migrator', () => {
 
     it('latest', (done) => {
       expect(() => {
-        knex.migrate
+        return knex.migrate
           .latest({
             directory: 'test/unit/migrate/migrations',
           })
@@ -37,6 +37,49 @@ describe('Migrator', () => {
     });
   });
 
+  describe('supports running migrations in transaction', (done) => {
+    let migrationSource;
+    let knex;
+    let wasProcessed = false;
+    let wasWrapped = false;
+    beforeEach(() => {
+      migrationSource = new FsMigrations('test/unit/migrate/migrations/');
+      knex = Knex({
+        ...sqliteConfig,
+        connection: ':memory:',
+        migrationSource,
+        postProcessResponse: (response) => {
+          wasProcessed = true;
+          return response;
+        },
+        wrapIdentifier: (value, wrap) => {
+          wasWrapped = true;
+          return wrap(value);
+        },
+      });
+    });
+
+    afterEach(() => {
+      knex.destroy();
+    });
+
+    it('latest', (done) => {
+      expect(() => {
+        return knex.transaction((txn) => {
+          txn.migrate
+            .latest({
+              directory: 'test/unit/migrate/migrations',
+            })
+            .then(() => {
+              expect(wasProcessed).to.equal(false);
+              expect(wasWrapped).to.equal(false);
+              done();
+            });
+        });
+      }).not.to.throw();
+    });
+  });
+
   describe('uses postProcessResponse for migrations', (done) => {
     let migrationSource;
     let knex;
@@ -56,8 +99,9 @@ describe('Migrator', () => {
         ...sqliteConfig,
         connection: ':memory:',
         migrationSource,
-        postProcessResponse: () => {
+        postProcessResponse: (response) => {
           wasPostProcessed = true;
+          return response;
         },
       });