Generalizing Transactions in NestJS: A Domain Port over TypeORM and MongoDB

Generalizing Transactions in NestJS: A Domain Port over TypeORM and MongoDB

Transaction control usually starts out tied to whatever implements it. An executor - a concrete class that opens a transaction over the project's ORM - gets injected into the controller and wraps the body of the handler, so that everything a request writes either commits or is discarded as a block.

The Problem with External Services

An HTTP request corresponds exactly to one unit of work, the use case has no idea it is inside a transaction, and repositories enlist themselves in whichever one is active through AsyncLocalStorage. However, when the use case calls an external service before it writes, the transaction stays alive while the wait is measured in seconds or minutes.

The Original Approach

@Post()
create(@Body() body: CreateUserRequest) {
  return this.transactionExecutor.execute(async () => new RegisterUser(this.ids, this.users, this.settings).execute(body));
}

The Limitations of the Original Approach

The original approach is sound, but it cannot handle operations where the use case has to wait on an external service before it writes. This is because the transaction stays alive while the wait is measured in seconds or minutes.

The Solution: Generalizing the Executor

The solution is to generalize the executor rather than injecting it as it is. This involves creating an interface that defines the contract of the executor, and then implementing that interface for different databases.

The UnitOfWork Port

The use case does not need the particular class that implements the executor, but any collaborator able to run a block as a single unit. This description is the description of an interface, and since it is a need of the application layer, declaring it falls to the domain.

The UnitOfWork Interface

export interface UnitOfWorkProps<T> {
  work: () => Promise<T>;
  // Runs after the rollback, on a clean connection.
  // Writing this inside the // failed unit would revert it along with the failure it records.
  onError?: (error: unknown) => Promise<void> | void;
}

export interface UnitOfWork {
  // Joins the unit already active in this async context, if there is one.
  run<T>(props: UnitOfWorkProps<T>): Promise<T>;
}

The Concrete Executor

The concrete executor becomes an implementation of the UnitOfWork interface. It changes name, because it is no longer a standalone class but the adapter of a port, and execute becomes run to match the declared signature.

The TypeORM Implementation

export class TypeOrmUnitOfWork implements UnitOfWork {
  private static readonly storage = new AsyncLocalStorage<EntityManager>();

  constructor(private readonly dataSource: DataSource) {}

  getManagerIfActive(): EntityManager | null {
    return TypeOrmUnitOfWork.storage.getStore() ?? null;
  }

  async run<T>({ work, onError }: UnitOfWorkProps<T>): Promise<T> {
    if (TypeOrmUnitOfWork.storage.getStore()) return work();

    const queryRunner = this.dataSource.createQueryRunner();
    await queryRunner.connect();
    await queryRunner.startTransaction();

    try {
      return await TypeOrmUnitOfWork.storage.run(queryRunner.manager, async () => {
        const result = await work();
        await queryRunner.commitTransaction();
        return result;
      });
    } catch (error) {
      await queryRunner.rollbackTransaction();
      // Outside the run(): repositories used here write on the pooled
      // connection, so what they write survives the rollback.
      Best effort: a
      // failure while recording the failure must not replace it.
      if (onError) {
        try {
          await onError(error);
        } catch {}
      }
      throw error;
    } finally {
      await queryRunner.release();
    }
  }
}

The Repository Side

On the repository side, enlisting means reading the unit the executor has just published. A base class settles the decision for all of them, so no concrete repository ever asks the question again.

The Use Case

The use case does not need that particular class, but any collaborator able to run a block as a single unit. This description is the description of an interface, and since it is a need of the application layer, declaring it falls to the domain.

The GenerateReport Use Case

export class GenerateReport {
  constructor(private readonly uow: UnitOfWork, private readonly ids: IdGenerator, private readonly reports: ReportRepository, private readonly failures: FailureLogRepository, private readonly content: SlowContentService) {}

  async execute(props: { userId: string; prompt: string }): Promise<string> {
    // Outside the unit on purpose: a transaction held open across this wait
    // would pin a connection and its locks until it returned.
    const generated = await this.content.generate(props.prompt);
    const report = new Report({ id: this.ids.create(), userId: props.userId, content: generated });
    await this.uow.run({ work: async () => { await this.reports.save(report); }, onError: async (error) => { await this.failures.save(new FailureLog({ id: this.ids.create(), operation: 'generate-report', message: error instanceof Error ? error.message : String(error), })); }, });
    return report.id;
  }
}

The Controller

The controller no longer opens a transaction, but instead calls the use case, which decides its own scope.

Nested Units

The use case can now open its own unit without knowing whether someone further up the stack already opened one. This means that nesting stops being a rarity and becomes the expected case.

The MongoDB Adapter

The MongoDB adapter is written against the same interface as the TypeORM implementation. It is worth comparing it line by line with the TypeORM one.

The MongoUnitOfWork Implementation

export class MongoUnitOfWork implements UnitOfWork {
  private static readonly storage = new AsyncLocalStorage<ClientSession>();

  constructor(private readonly connection: Connection) {}

  async run<T>({ work, onError }: UnitOfWorkProps<T>): Promise<T> {
    if (MongoUnitOfWork.storage.getStore()) return work();

    const session = await this.connection.startSession();
    session.startTransaction();

    try {
      return await MongoUnitOfWork.storage.run(session, async () => {
        const result = await work();
        await session.commitTransaction();
        return result;
      });
    } catch (error) {
      // A transaction the server already aborted, by outliving its lifetime,
      // is no longer in progress, and aborting it again throws.
      if (session.inTransaction()) {
        await session.abortTransaction();
      }
      throw error;
    }
  }
}

Testing the Generalization

The only way to know whether the port generalizes anything is to write a second implementation over a technology that does not resemble the first.