feat: support typeorm slave option

docs/docs/06_plugins/01_available-plugins/01-transactional/05-typeorm-adapter.md

@@ -75,6 +75,40 @@ new TransactionalAdapterTypeOrm({
 
 The `tx` property on the `TransactionHost<TransactionalAdapterTypeOrm>` is typed as [`EntityManager`](https://typeorm.io/working-with-entity-manager).
 
+## Running on a read replica
+
+When the `DataSource` is configured with [replication](https://typeorm.io/docs/data-source/multiple-data-sources/#replication), you can route a whole transaction to a read replica by passing `mode: 'slave'` in the transaction options.
+
+This is needed because TypeORM only routes reads to a replica when they run on a query runner that was not already bound to the master. Inside a transaction every read is bound to the transaction's query runner, which TypeORM opens on the master by default - so reads in a `@Transactional()` method would otherwise all hit the master. Passing `mode: 'slave'` opens the transaction's query runner on a replica instead, so **every** read in the callback goes to the replica.
+
+```ts
+@Injectable()
+class UserService {
+    constructor(private readonly userRepository: UserRepository) {}
+
+    // highlight-next-line
+    @Transactional<TransactionalAdapterTypeOrm>({ mode: 'slave' })
+    async getReportData() {
+        // every read here runs against a read replica
+        return await this.userRepository.getActiveUsers();
+    }
+}
+```
+
+The same option works with the function wrapper:
+
+```ts
+await this.txHost.withTransaction({ mode: 'slave' }, async () => {
+    // ...
+});
+```
+
+:::note
+
+`mode` defaults to `'master'`. Nested transactions are not supported in `'slave'` mode - propagating a nested transaction with `mode: 'slave'` throws an error.
+
+:::
+
 ## Example
 
 ```ts title="user.service.ts"

packages/transactional-adapters/transactional-adapter-typeorm/src/lib/transactional-adapter-typeorm.ts

@@ -1,6 +1,7 @@
 import { TransactionalAdapter } from '@nestjs-cls/transactional';
 import { DataSource, EntityManager } from 'typeorm';
 import type { IsolationLevel } from 'typeorm/driver/types/IsolationLevel.js';
+import type { ReplicationMode } from 'typeorm/driver/types/ReplicationMode.js';
 
 export interface TypeOrmTransactionalAdapterOptions {
     /**
@@ -17,6 +18,21 @@ export interface TypeOrmTransactionalAdapterOptions {
 
 export interface TypeOrmTransactionOptions {
     isolationLevel: IsolationLevel;
+
+    /**
+     * Which replica to run the transaction against, using TypeORM's
+     * {@link ReplicationMode}. Only has an effect when the `DataSource` is
+     * configured with replication. Defaults to `'master'`.
+     *
+     * - `'master'` runs on the primary/write connection.
+     * - `'slave'` runs on a read replica. Unlike TypeORM's default behaviour -
+     *   which only routes plain `find`-style reads to a replica - this forces
+     *   the whole transaction (including custom query-builder reads) onto the
+     *   replica.
+     *
+     * Note: nested transactions are not supported in `'slave'` mode.
+     */
+    mode?: ReplicationMode;
 }
 
 export class TransactionalAdapterTypeOrm implements TransactionalAdapter<
@@ -39,6 +55,14 @@ export class TransactionalAdapterTypeOrm implements TransactionalAdapter<
             fn: (...args: any[]) => Promise<any>,
             setClient: (client?: EntityManager) => void,
         ) => {
+            if (options?.mode === 'slave') {
+                return this.wrapWithReplicaTransaction(
+                    dataSource,
+                    options,
+                    fn,
+                    setClient,
+                );
+            }
             return dataSource.transaction(options?.isolationLevel, (trx) => {
                 setClient(trx);
                 return fn();
@@ -51,11 +75,43 @@ export class TransactionalAdapterTypeOrm implements TransactionalAdapter<
             setClient: (client?: EntityManager) => void,
             client: EntityManager,
         ) => {
+            if (options?.mode === 'slave') {
+                throw new Error(
+                    'Nested transactions are not supported in slave mode',
+                );
+            }
             return client.transaction(options?.isolationLevel, (trx) => {
                 setClient(trx);
                 return fn();
             });
         },
         getFallbackInstance: () => dataSource.manager,
     });
+
+    /**
+     * Runs a transaction on a replica (read) query runner so that every read in
+     * the callback - including custom query-builder reads, which TypeORM would
+     * otherwise send to the master - is routed to the replica.
+     */
+    private async wrapWithReplicaTransaction(
+        dataSource: DataSource,
+        options: TypeOrmTransactionOptions,
+        fn: (...args: any[]) => Promise<any>,
+        setClient: (client?: EntityManager) => void,
+    ) {
+        const queryRunner = dataSource.createQueryRunner('slave');
+        await queryRunner.connect();
+        await queryRunner.startTransaction(options?.isolationLevel);
+        setClient(queryRunner.manager);
+        try {
+            const result = await fn();
+            await queryRunner.commitTransaction();
+            return result;
+        } catch (error) {
+            await queryRunner.rollbackTransaction();
+            throw error;
+        } finally {
+            await queryRunner.release();
+        }
+    }
 }

packages/transactional-adapters/transactional-adapter-typeorm/test/transactional-adapter-typeorm.spec.ts

@@ -1,8 +1,9 @@
+import { jest } from '@jest/globals';
 import {
     ClsPluginTransactional,
     InjectTransaction,
     Propagation,
-    Transaction,
+    type Transaction,
     Transactional,
     TransactionHost,
 } from '@nestjs-cls/transactional';
@@ -25,13 +26,23 @@ class User {
     email?: string;
 }
 
-const dataSource = new DataSource({
-    type: 'postgres',
+const connectionOptions = {
     host: 'localhost',
     port: 5446,
     username: 'postgres',
     password: 'postgres',
     database: 'postgres',
+};
+
+const dataSource = new DataSource({
+    type: 'postgres',
+    // Configured with replication so the `slave` transaction mode has a replica
+    // to route to. Both master and replicas point at the same database (there
+    // is a single test instance), which is enough to exercise the routing.
+    replication: {
+        master: connectionOptions,
+        slaves: [connectionOptions],
+    },
     entities: [User],
     synchronize: true,
 });
@@ -112,6 +123,48 @@ class UserService {
         throw new Error('Rollback');
     }
 
+    @Transactional<TransactionalAdapterTypeOrm>({
+        mode: 'slave',
+        isolationLevel: 'SERIALIZABLE',
+    })
+    async readInSlaveModeWithIsolationLevel(id: number) {
+        return await this.userRepository.getUserById(id);
+    }
+
+    @Transactional<TransactionalAdapterTypeOrm>({ mode: 'slave' })
+    async readInSlaveMode(id: number) {
+        return await this.dataSource
+            .getRepository(User)
+            .createQueryBuilder('user')
+            .where('user.id = :id', { id })
+            .getOne();
+    }
+
+    async readInSlaveModeWithFunctionWrapper(id: number) {
+        return this.transactionHost.withTransaction(
+            { mode: 'slave' },
+            async () => this.userRepository.getUserById(id),
+        );
+    }
+
+    @Transactional<TransactionalAdapterTypeOrm>({ mode: 'slave' })
+    async slaveModeError() {
+        await this.userRepository.getUserById(1);
+        throw new Error('Rollback');
+    }
+
+    @Transactional<TransactionalAdapterTypeOrm>({ mode: 'slave' })
+    async slaveModeHasNested() {
+        await this.nestedSlaveTransaction();
+    }
+
+    @Transactional<TransactionalAdapterTypeOrm>(Propagation.Nested, {
+        mode: 'slave',
+    })
+    async nestedSlaveTransaction() {
+        await this.userRepository.getUserById(1);
+    }
+
     @Transactional()
     async transactionalHasNested(name?: string) {
         await this.nestedTransaction(name);
@@ -253,6 +306,86 @@ describe('Transactional', () => {
             );
         });
     });
+
+    describe('Slave mode', () => {
+        let createQueryRunnerSpy: ReturnType<typeof jest.spyOn>;
+        let startTransactionSpy: ReturnType<typeof jest.spyOn>;
+
+        beforeEach(() => {
+            const original = dataSource.createQueryRunner.bind(dataSource);
+            createQueryRunnerSpy = jest
+                .spyOn(dataSource, 'createQueryRunner')
+                .mockImplementation((mode) => {
+                    const queryRunner = original(mode);
+                    startTransactionSpy = jest.spyOn(
+                        queryRunner,
+                        'startTransaction',
+                    );
+                    return queryRunner;
+                });
+        });
+
+        afterEach(() => {
+            startTransactionSpy?.mockRestore();
+            createQueryRunnerSpy.mockRestore();
+        });
+
+        it('should route the transaction to a replica query runner', async () => {
+            const user = await callingService.withoutTransaction();
+
+            const found = await callingService.readInSlaveMode(user.r1.id);
+
+            expect(found).toEqual(user.r1);
+            expect(createQueryRunnerSpy).toHaveBeenCalledWith('slave');
+        });
+
+        it('should route to a replica with the function wrapper', async () => {
+            const user = await callingService.withoutTransaction();
+
+            const found =
+                await callingService.readInSlaveModeWithFunctionWrapper(
+                    user.r1.id,
+                );
+
+            expect(found).toEqual(user.r1);
+            expect(createQueryRunnerSpy).toHaveBeenCalledWith('slave');
+        });
+
+        it('should start the replica transaction with the given isolation level', async () => {
+            const user = await callingService.withoutTransaction();
+
+            const found =
+                await callingService.readInSlaveModeWithIsolationLevel(
+                    user.r1.id,
+                );
+
+            expect(found).toEqual(user.r1);
+            expect(createQueryRunnerSpy).toHaveBeenCalledWith('slave');
+
+            const queryRunner = createQueryRunnerSpy.mock.results[0]
+                .value as ReturnType<DataSource['createQueryRunner']>;
+            expect(startTransactionSpy).toHaveBeenCalledWith('SERIALIZABLE');
+            expect(queryRunner.isReleased).toBe(true);
+        });
+
+        it('should release the query runner on error', async () => {
+            await expect(callingService.slaveModeError()).rejects.toThrow(
+                new Error('Rollback'),
+            );
+
+            const queryRunner = createQueryRunnerSpy.mock.results[0]
+                .value as ReturnType<DataSource['createQueryRunner']>;
+            expect(queryRunner.isReleased).toBe(true);
+        });
+
+        it('should not support nested transactions', async () => {
+            await expect(callingService.slaveModeHasNested()).rejects.toThrow(
+                new Error(
+                    'Nested transactions are not supported in slave mode',
+                ),
+            );
+        });
+    });
 });
 
 describe('Default options', () => {