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: @Post() create(@Body() body: CreateUserRequest) { return this.transactionExecutor.execute(async () => new RegisterUser(this.ids, this.users, this.settings).execute(body), ); } The approach is sound, and for most operations nothing more is needed. 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. That starting point is described in detail, along with the enlistment mechanism and its edge cases, in Transactions in NestJS and TypeORM without passing the EntityManager around. This article is about a kind of operation that approach cannot handle — one where the use case has to wait on an external service before it writes — and about the generalization it takes when one shows up: extracting that executor's contract into a ten-line interface that lives in the domain. What this buys is not only that the use case can decide its own scope without depending on the ORM. The interface is then put to the test with two implementations, one over PostgreSQL and one over MongoDB, running the same use cases and the same tests against both. That check is the only way to know whether the contract generalizes anything or is the same executor under a new name, and it is also what marks how far it reaches: there are differences between engines that no abstraction hides, and it is worth knowing which before trusting it. The problem: waiting on a third party inside the transaction An open transaction is not free. It holds a pooled connection and the locks on the rows already written, and it holds them for as long as the block it wraps takes to finish. That does not matter when the block only talks to the database. It starts to matter the moment the use case calls an external service: a payment gateway, a file store, an AI model. The transaction stays alive while something that has nothing to do with the database is awaited, and that wait is measured in seconds or minutes. @Post('reports') generateReport(@Body() body: GenerateReportRequest) { return this.transactionExecutor.execute(async () => new GenerateReport(this.ids, this.reports, this.content).execute(body), ); } // ...and inside that use case: async execute(props: GenerateReportProps): Promise { const generated = await this.content.generate(props.prompt); // minutes const report = new Report({ id: this.ids.create(), userId: props.userId, content: generated }); await this.reports.save(report); return report.id; } Neither file looks wrong on its own. The controller does what it does in every other endpoint, and the use case calls a service and saves the result. The problem exists only in the relationship between the two, and it is invisible from either. The result is a system that carries far less load than it should, with connections busy waiting on third parties and requests blocked on rows nobody will release until some external service answers. The fix is easy to state: only database operations belong inside the transaction. Applying it is what breaks the original approach, because the controller cannot carry it out. The controller knows the request is one unit of work, but not which part of the use case is the write and which is the wait. That knowledge lives one level down. Moving the scope couples the application layer If the use case decides the scope, it needs something to open the unit with. The immediate move is to inject the executor that already exists: export class GenerateReport { constructor(private readonly executor: TypeOrmTransactionExecutor) {} } With that signature, the application layer now depends on the class that implements transactions over TypeORM, and through it on the whole ORM. It is the same defect that shows up when a domain repository declares manager?: EntityManager on its methods, only one layer higher. A layer that should be reasonable, testable and instantiable without knowing how data is persisted no longer can be. And the use case is precisely where the logic worth keeping clean lives. The move solves one problem and creates another, and the second is solved by generalizing the executor rather than injecting it as it is. The solution: the UnitOfWork port The use case does not need that particular class, but any collaborator able to run a block as a single unit. That description is the description of an interface, and since it is a need of the application layer, declaring it falls to the domain: export interface UnitOfWorkProps { work: () => Promise; // 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; } export interface UnitOfWork { // Joins the unit already active in this async context, if there is one. run(props: UnitOfWorkProps): Promise; } That is the whole contract, and work taking no arguments is what keeps a persistence type from reaching the application layer. onError is not decoration. It is the seam for whatever must outlive the rollback, and it exists because of a trap worth naming: anything written in the catch on the failed unit's own connection is reverted along with the failure it was meant to record. The adapter invokes onError after rolling back and outside the unit's context, so repositories used there write on the normal connection. The concrete executor becomes an implementation of that contract. 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: export class TypeOrmUnitOfWork implements UnitOfWork { private static readonly storage = new AsyncLocalStorage(); constructor(private readonly dataSource: DataSource) {} getManagerIfActive(): EntityManager | null { return TypeOrmUnitOfWork.storage.getStore() ?? null; } async run({ work, onError }: UnitOfWorkProps): Promise { 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(); } } } 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: protected get repository(): Repository { const manager: EntityManager | null = this.transactionExecutor.getManagerIfActive(); return manager ? manager.getRepository(this.target) : this.dataSource.getRepository(this.target); } That is the mechanism by which a repository ends up inside the transaction without receiving anything. It is developed, with its edge cases, in Transactions in NestJS and TypeORM without passing the EntityManager around. With that, the use case depends on the port and knows nothing about the implementation: 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 { // 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; } } In the controller, that endpoint stops opening anything: @Post('reports') async generateReport(@Body() body: GenerateReportRequest) { const useCase = new GenerateReport( this.uow, this.ids, this.reports, this.failures, this.content, ); return { id: await useCase.execute(body) }; } Compared with the handler at the top of this article, the call that wrapped the body is gone. The boundary was not removed: it moved inward, to the only place that knows where to put it. Use cases that do not need to decide their scope do not change at all. They still know nothing about the port, and the boundary is opened by whoever calls them, exactly as before: export class RegisterUser { constructor( private readonly ids: IdGenerator, private readonly users: UserRepository, private readonly settings: UserSettingsRepository, ) {} async execute(props: { email: string; name: string }): Promise { const existing = await this.users.findByEmail(props.email); if (existing) throw new UserAlreadyExistsError(props.email); const user = new User({ id: this.ids.create(), email: props.email, name: props.name }); await this.users.save(user); await this.settings.save( new UserSettings({ id: this.ids.create(), userId: user.id, theme: 'light' }), ); return user.id; } } Nested units: joining instead of opening another Moving the scope into the use case has a consequence worth settling. From the moment a use case can open its own unit without knowing whether someone further up the stack already opened one, nesting stops being a rarity and becomes the expected case. That is why the first line of run() returns work() when a unit is already active. Without that guard a second transaction would open, on a different connection, committing on its own and capable of blocking on a row its own caller is holding. Checking that they do join admits a direct proof. An independent transaction cannot read another's uncommitted writes at any isolation level, so if the inner unit sees what the outer one wrote, there can only be one transaction: await uow.run({ work: async () => { await users.save(new User({ id: ids.create(), email, name: 'Outer' })); await uow.run({ work: async () => { // Uncommitted: visible only from inside the same transaction. sawOuterWriteFromInside = (await users.findByEmail(email)) !== null; }, }); }, }); expect(sawOuterWriteFromInside).toBe(true); Testing the generalization: the MongoDB adapter So far there is an interface and a single implementation, which is not a generalization but a rename. 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. MongoDB meets that requirement, and it also presents the problem this article started from in a more severe form. A transaction there does not only hold resources: it has a lifetime, and the server aborts it once that runs out. The clock starts at its first operation, and a read counts. With the server's limit lowered to one second, this sequence opens a transaction, performs a read, waits and then tries to write: const session = await connection.startSession(); session.startTransaction(); // First operation of the transaction: the clock starts here. await users.countDocuments({}).session(session); await sleep(5_000); // the server's limit is 1 second // Fails: the transaction has already been aborted. await users.create([{ _id: id, email: 'slow@example.com', name: 'Slow' }], { session }); The write fails. The same sequence without the wait commits without trouble, so the only thing that differs between them is how long is spent inside the transaction. Nothing had been written when the clock started running: the read was enough. What degrades performance in PostgreSQL prevents the operation from completing here. The fix is not to raise the limit, which exists to protect the cluster, but to leave the wait outside the unit — which is precisely what the port lets the use case decide. The adapter is then written against the same interface, and it is worth comparing it line by line with the TypeORM one: export class MongoUnitOfWork implements UnitOfWork { private static readonly storage = new AsyncLocalStorage(); constructor(private readonly connection: Connection) {} async run({ work, onError }: UnitOfWorkProps): Promise { 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(); if (onError) { try { await onError(error); } catch {} } throw error; } finally { await session.endSession(); } } } The session takes the place of the queryRunner and the rest is identical: the same join guard, the same storage.run publishing the unit, the same commit inside the callback and the same onError outside it. That both implementations take this shape was not the goal but the signal that the contract describes something both engines genuinely do. The real difference shows up on the repository side. Mongoose offers no manager shortcut: the session has to ride on every operation, reads included, and omitting it on a single call leaves that call outside the transaction with no error. An equivalent base class absorbs that difference, so concrete repositories for either engine end up written the same way. Absorbing those asymmetries so the port can stay the same is exactly the adapter's job. The same suite over PostgreSQL and MongoDB With two implementations available, the check can be run against both from the same material. The tests below are tests of the use cases, not of the adapters: they describe what must happen on commit and on rollback, without mentioning at any point which engine is underneath. describe.each(ALL_CONTEXTS)('UnitOfWork contract: %s', (_name, makeContext) => { it('commits both writes as a single unit', async () => { await ctx.uow.run({ work: () => registerUser().execute({ email: 'ana@example.com', name: 'Ana' }), }); expect(await ctx.users.countAll()).toBe(1); expect(await ctx.settings.countAll()).toBe(1); }); it('runs onError after the rollback, so the record survives it', async () => { await expect( ctx.uow.run({ work: async () => { await registerUser().execute({ email: 'ana@example.com', name: 'Ana' }); throw new Error('boom'); }, onError: async () => { await ctx.failures.save(new FailureLog({ /* ... */ })); }, }), ).rejects.toThrow('boom'); expect(await ctx.users.countAll()).toBe(0); // the work was reverted expect(await ctx.failures.countAll()).toBe(1); // the record of it was not }); }); The same tests run against PostgreSQL and against MongoDB, and pass on both. Had the interface come out shaped like the TypeORM executor, the MongoDB adapter could not have implemented it without distorting it, and this would not be green. The repository also includes a third implementation, in memory, which serves to run these same tests without infrastructure and for one additional check: both engines ship native transactions, so an adapter that satisfies the contract without having any rules out that the port is a transaction in disguise. With any test of this kind it is worth breaking on purpose what it claims to check and confirming it goes red. Removing the join guard from the adapters makes the nesting tests fail on all of them. A green test that cannot fail proves nothing. It is worth being precise about the scope of that check. The port unifies the API, not the guarantees: the two engines differ in isolation levels, in lock behaviour and in failure modes, and those differences survive the abstraction untouched. The transaction lifetime in MongoDB is one of them, and it is precisely the one that forced the scope down into the use case. Two things are guaranteed. That the decision about scope is taken in the layer that knows it, without that layer depending on the engine. And that a change of engine stays confined to infrastructure — the adapters and the repositories for the new engine — without touching a single use case or a single use-case test. When to generalize and when not to The generalization pays for itself once there is at least one use case that needs to decide its own scope, and those cases have a recognizable shape: slow work that is not database work, interleaved with writes that are. Charging and recording the charge. The use case calls the payment gateway and persists the result. The call takes seconds and cannot sit inside the transaction. Uploading a file and saving its metadata. The upload to object storage goes outside; the row referencing it, inside. Generating content with an external service and persisting it, which is the GenerateReport case in this article. In all three the controller has no way to get it right: wrapping everything keeps the transaction open across the wait, and wrapping nothing leaves the writes without atomicity. The boundary sits in the middle of the use case, and only it knows where. If no operation has that shape and all of them fit inside a transaction opened at the controller, the port adds nothing: injecting the concrete executor wherever it is needed is enough, and the interface would only add indirection nobody uses. And even with the port in place, outside those cases the boundary should still be opened at the controller. That the use case can take the scope does not mean it should, and in the controllers the transactional shape of the system can be read at a glance. Checklist The port lives in the domain and imports nothing from the engine. work takes no arguments: if it takes a manager or a session, it is not a port. run() joins the active unit instead of opening a second one. Whatever must outlive the rollback is written from onError, outside the unit. Only database operations go inside the unit. An interface with a single implementation does not generalize anything yet. Use-case tests run against every adapter, not against one. Every test is validated by deliberately breaking what it claims to check. The complete code is in a runnable repository — the adapters, the use cases, the NestJS wiring and the shared suite, with Docker Compose to bring up PostgreSQL and MongoDB: https://github.com/JoseCarlosGarcia/unit-of-work-demo