feat: add support for global transactions

Author: thetutlage

adonis-typings/database.ts

@@ -688,6 +688,25 @@ declare module '@ioc:Adonis/Lucid/Database' {
       sql: string,
       bindings?: { [key: string]: StrictValuesWithoutRaw } | StrictValuesWithoutRaw[]
     ): RawBuilderContract
+
+    /**
+     * Begin a new global transaction. Multiple calls to this
+     * method is a noop
+     */
+    beginGlobalTransaction (
+      connectionName?: string,
+      options?: Exclude<DatabaseClientOptions, 'mode'>,
+    ): Promise<TransactionClientContract>
+
+    /**
+     * Commit an existing global transaction
+     */
+    commitGlobalTransaction (connectionName?: string): Promise<void>
+
+    /**
+     * Rollback an existing global transaction
+     */
+    rollbackGlobalTransaction (connectionName?: string): Promise<void>
   }
 
   const Database: DatabaseContract

src/Database/index.ts

@@ -17,6 +17,7 @@ import {
   DatabaseContract,
   DatabaseClientOptions,
   DatabaseConfigContract,
+  TransactionClientContract,
   ConnectionManagerContract,
 } from '@ioc:Adonis/Lucid/Database'
 
@@ -51,6 +52,11 @@ export class Database implements DatabaseContract {
   public InsertQueryBuilder = InsertQueryBuilder
   public ModelQueryBuilder = ModelQueryBuilder
 
+  /**
+   * A store of global transactions
+   */
+  public connectionGlobalTransactions: Map<string, TransactionClientContract> = new Map()
+
   constructor (
     private config: DatabaseConfigContract,
     private logger: LoggerContract,
@@ -106,6 +112,16 @@ export class Database implements DatabaseContract {
       throw new Exception(`Invalid mode ${options.mode}. Must be read or write`)
     }
 
+    /**
+     * Return the global transaction when it already exists.
+     */
+    if (this.connectionGlobalTransactions.has(connection)) {
+      this.logger.trace({ connection }, 'using pre-existing global transaction connection')
+      const globalTransactionClient = this.connectionGlobalTransactions.get(connection)!
+      globalTransactionClient.profiler = options.profiler
+      return globalTransactionClient
+    }
+
     /**
      * Fetching connection for the given name
      */
@@ -203,4 +219,72 @@ export class Database implements DatabaseContract {
   public raw (sql: string, bindings?: any) {
     return new RawBuilder(sql, bindings)
   }
+
+  /**
+   * Begin a new global transaction
+   */
+  public async beginGlobalTransaction (
+    connectionName?: string,
+    options?: Omit<DatabaseClientOptions, 'mode'>,
+  ) {
+    connectionName = connectionName || this.primaryConnectionName
+
+    /**
+     * Return global transaction as it is
+     */
+    const globalTrx = this.connectionGlobalTransactions.get(connectionName)
+    if (globalTrx) {
+      return globalTrx
+    }
+
+    /**
+     * Create a new transaction and store a reference to it
+     */
+    const trx = await this.connection(connectionName, options).transaction()
+    this.connectionGlobalTransactions.set(trx.connectionName, trx)
+
+    /**
+     * Listen for events to drop the reference when transaction
+     * is over
+     */
+    trx.on('commit', ($trx) => {
+      this.connectionGlobalTransactions.delete($trx.connectionName)
+    })
+
+    trx.on('rollback', ($trx) => {
+      this.connectionGlobalTransactions.delete($trx.connectionName)
+    })
+
+    return trx
+  }
+
+  /**
+   * Commit an existing global transaction
+   */
+  public async commitGlobalTransaction (connectionName?: string) {
+    connectionName = connectionName || this.primaryConnectionName
+    const trx = this.connectionGlobalTransactions.get(connectionName)
+
+    if (!trx) {
+      // eslint-disable-next-line max-len
+      throw new Exception('Cannot commit a non-existing global transaction. Make sure you are not calling "commitGlobalTransaction" twice')
+    }
+
+    await trx.commit()
+  }
+
+  /**
+   * Rollback an existing global transaction
+   */
+  public async rollbackGlobalTransaction (connectionName?: string) {
+    connectionName = connectionName || this.primaryConnectionName
+    const trx = this.connectionGlobalTransactions.get(connectionName)
+
+    if (!trx) {
+      // eslint-disable-next-line max-len
+      throw new Exception('Cannot rollback a non-existing global transaction. Make sure you are not calling "commitGlobalTransaction" twice')
+    }
+
+    await trx.rollback()
+  }
 }

test/database/database.spec.ts

@@ -280,3 +280,76 @@ test.group('Database | extend', (group) => {
     await db.manager.closeAll()
   })
 })
+
+test.group('Database | global transaction', (group) => {
+  group.before(async () => {
+    await setup()
+  })
+
+  group.after(async () => {
+    await cleanup()
+  })
+
+  test('perform queries inside a global transaction', async (assert) => {
+    const config = {
+      connection: 'primary',
+      connections: { primary: getConfig() },
+    }
+
+    const db = new Database(config, getLogger(), getProfiler())
+    await db.beginGlobalTransaction()
+
+    await db.table('users').insert({ username: 'virk' })
+    await db.rollbackGlobalTransaction()
+
+    const users = await db.from('users')
+    assert.lengthOf(users, 0)
+    assert.equal(db.connectionGlobalTransactions.size, 0)
+
+    await db.manager.closeAll()
+  })
+
+  test('create transactions inside a global transaction', async (assert) => {
+    const config = {
+      connection: 'primary',
+      connections: { primary: getConfig() },
+    }
+
+    const db = new Database(config, getLogger(), getProfiler())
+    await db.beginGlobalTransaction()
+    const trx = await db.transaction()
+
+    await trx.table('users').insert({ username: 'virk' })
+    await trx.commit()
+
+    await db.rollbackGlobalTransaction()
+
+    const users = await db.from('users')
+    assert.lengthOf(users, 0)
+    assert.equal(db.connectionGlobalTransactions.size, 0)
+
+    await db.manager.closeAll()
+  })
+
+  test('multiple calls to beginGlobalTransaction must be a noop', async (assert) => {
+    const config = {
+      connection: 'primary',
+      connections: { primary: getConfig() },
+    }
+
+    const db = new Database(config, getLogger(), getProfiler())
+    await db.beginGlobalTransaction()
+    await db.beginGlobalTransaction()
+    await db.beginGlobalTransaction()
+
+    await db.table('users').insert({ username: 'virk' })
+
+    await db.rollbackGlobalTransaction()
+
+    const users = await db.from('users')
+    assert.lengthOf(users, 0)
+    assert.equal(db.connectionGlobalTransactions.size, 0)
+
+    await db.manager.closeAll()
+  })
+})